amadan

eleven / messenger Public

server: emoji reactions and encrypted file uploads

- reactions table + add/remove/count/forMessages; WS react/unreact broadcast updated counts; /api/messages includes [{emoji,count,mine}].

- blobs table + POST /api/upload and GET /api/blob/{id} (token-gated, 10 MB cap); stores opaque encrypted bytes only.

- tests for both; newTestServer now returns the *Server so tests can seed the DB.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Paul Campbell ffa0d3ff52bd618cad9bc465254a0a3e29d110d2 parent cc96f4d
5 files changed, +411 −22
  • db.go +136 −3
  • db_test.go +42 −0
  • handlers_test.go +104 −7
  • hub.go +53 −12
  • main.go +76 −0
diff --git a/db.go b/db.go
index ea9c1a3..d76fbeb 100644
--- a/db.go
+++ b/db.go
@@ -83,6 +83,19 @@ CREATE TABLE IF NOT EXISTS push_subscriptions (
p256dh TEXT NOT NULL,
auth TEXT NOT NULL,
created_at INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS reactions (
+ message_id INTEGER NOT NULL,
+ token TEXT NOT NULL,
+ emoji TEXT NOT NULL,
+ created_at INTEGER NOT NULL,
+ PRIMARY KEY (message_id, token, emoji)
+);
+CREATE TABLE IF NOT EXISTS blobs (
+ id TEXT PRIMARY KEY,
+ data BLOB NOT NULL,
+ size INTEGER NOT NULL,
+ created_at INTEGER NOT NULL
);`)
return err
}
@@ -249,9 +262,10 @@ func (db *DB) revokeInvite(token string) error {
// ---- messages -----------------------------------------------------------
type Message struct {
- ID int64 `json:"id"`
- CreatedAt int64 `json:"created_at"`
- Blob string `json:"blob"`
+ ID int64 `json:"id"`
+ CreatedAt int64 `json:"created_at"`
+ Blob string `json:"blob"`
+ Reactions []ReactionView `json:"reactions,omitempty"`
}
func (db *DB) insertMessage(blob string) (Message, error) {
@@ -297,3 +311,122 @@ func (db *DB) messagesAfter(after int64, limit int) ([]Message, error) {
}
return out, rows.Err()
}
+
+// ---- blobs (encrypted attachments) --------------------------------------
+//
+// An uploaded file is encrypted in the browser before it ever reaches here, so
+// the server stores only opaque bytes — no filename, type, or content. The
+// referencing message (also encrypted) carries the metadata.
+
+func (db *DB) putBlob(id string, data []byte) error {
+ _, err := db.sql.Exec(
+ "INSERT INTO blobs(id, data, size, created_at) VALUES(?, ?, ?, ?)",
+ id, data, len(data), time.Now().Unix(),
+ )
+ return err
+}
+
+// getBlob returns the stored bytes, or (nil, nil) if the id is unknown.
+func (db *DB) getBlob(id string) ([]byte, error) {
+ var data []byte
+ err := db.sql.QueryRow("SELECT data FROM blobs WHERE id = ?", id).Scan(&data)
+ if err == sql.ErrNoRows {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return data, nil
+}
+
+// ---- reactions ----------------------------------------------------------
+//
+// A reaction is (message_id, token, emoji). The emoji is stored in the clear so
+// the server can count and de-duplicate them; message content stays encrypted.
+// The token identifies the reactor for toggling, never their screen name.
+
+// ReactionCount is an emoji and how many people used it (for live broadcast).
+type ReactionCount struct {
+ Emoji string `json:"emoji"`
+ Count int `json:"count"`
+}
+
+// ReactionView adds whether the requesting token is one of the reactors.
+type ReactionView struct {
+ Emoji string `json:"emoji"`
+ Count int `json:"count"`
+ Mine bool `json:"mine"`
+}
+
+func (db *DB) addReaction(messageID int64, token, emoji string) error {
+ _, err := db.sql.Exec(
+ "INSERT OR IGNORE INTO reactions(message_id, token, emoji, created_at) VALUES(?, ?, ?, ?)",
+ messageID, token, emoji, time.Now().Unix(),
+ )
+ return err
+}
+
+func (db *DB) removeReaction(messageID int64, token, emoji string) error {
+ _, err := db.sql.Exec(
+ "DELETE FROM reactions WHERE message_id = ? AND token = ? AND emoji = ?",
+ messageID, token, emoji,
+ )
+ return err
+}
+
+// reactionCounts returns the per-emoji totals for one message, oldest emoji first.
+func (db *DB) reactionCounts(messageID int64) ([]ReactionCount, error) {
+ rows, err := db.sql.Query(
+ "SELECT emoji, COUNT(*) FROM reactions WHERE message_id = ? GROUP BY emoji ORDER BY MIN(created_at)",
+ messageID,
+ )
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ out := []ReactionCount{}
+ for rows.Next() {
+ var rc ReactionCount
+ if err := rows.Scan(&rc.Emoji, &rc.Count); err != nil {
+ return nil, err
+ }
+ out = append(out, rc)
+ }
+ return out, rows.Err()
+}
+
+// reactionsForMessages returns per-message reaction views for the given message
+// ids, flagging which emojis the supplied token reacted with (for the toggle UI).
+func (db *DB) reactionsForMessages(ids []int64, token string) (map[int64][]ReactionView, error) {
+ out := map[int64][]ReactionView{}
+ if len(ids) == 0 {
+ return out, nil
+ }
+ ph := make([]string, len(ids))
+ args := make([]any, 0, len(ids)+1)
+ args = append(args, token) // first ? in the CASE expression
+ for i, id := range ids {
+ ph[i] = "?"
+ args = append(args, id)
+ }
+ q := "SELECT message_id, emoji, COUNT(*), " +
+ "SUM(CASE WHEN token = ? THEN 1 ELSE 0 END) " +
+ "FROM reactions WHERE message_id IN (" + strings.Join(ph, ",") + ") " +
+ "GROUP BY message_id, emoji ORDER BY MIN(created_at)"
+ rows, err := db.sql.Query(q, args...)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ for rows.Next() {
+ var mid int64
+ var rv ReactionView
+ var mine int
+ if err := rows.Scan(&mid, &rv.Emoji, &rv.Count, &mine); err != nil {
+ return nil, err
+ }
+ rv.Mine = mine > 0
+ out[mid] = append(out[mid], rv)
+ }
+ return out, rows.Err()
+}
diff --git a/db_test.go b/db_test.go
index abdc56a..baf5087 100644
--- a/db_test.go
+++ b/db_test.go
@@ -101,6 +101,48 @@ func TestPushSubscriptions(t *testing.T) {
}
}
+func TestReactions(t *testing.T) {
+ db := newTestDB(t)
+ m, _ := db.insertMessage("blob")
+
+ // Two people add 👍; one adds ❤️.
+ db.addReaction(m.ID, "tokA", "👍")
+ db.addReaction(m.ID, "tokB", "👍")
+ db.addReaction(m.ID, "tokA", "❤️")
+
+ // Duplicate add is a no-op (composite PK).
+ db.addReaction(m.ID, "tokA", "👍")
+
+ counts, _ := db.reactionCounts(m.ID)
+ got := map[string]int{}
+ for _, c := range counts {
+ got[c.Emoji] = c.Count
+ }
+ if got["👍"] != 2 || got["❤️"] != 1 {
+ t.Fatalf("counts = %+v", got)
+ }
+
+ // "mine" is computed per-token.
+ views, _ := db.reactionsForMessages([]int64{m.ID}, "tokB")
+ for _, v := range views[m.ID] {
+ if v.Emoji == "👍" && !v.Mine {
+ t.Fatal("tokB should own 👍")
+ }
+ if v.Emoji == "❤️" && v.Mine {
+ t.Fatal("tokB should NOT own ❤️")
+ }
+ }
+
+ // Removing toggles it off.
+ db.removeReaction(m.ID, "tokB", "👍")
+ counts, _ = db.reactionCounts(m.ID)
+ for _, c := range counts {
+ if c.Emoji == "👍" && c.Count != 1 {
+ t.Fatalf("after remove, 👍 count = %d", c.Count)
+ }
+ }
+}
+
func TestGroupName(t *testing.T) {
db := newTestDB(t)
if got := db.getGroupName(); got != defaultGroupName {
diff --git a/handlers_test.go b/handlers_test.go
index 81fadda..0b2cdb0 100644
--- a/handlers_test.go
+++ b/handlers_test.go
@@ -9,7 +9,7 @@ import (
"testing"
)
-func newTestServer(t *testing.T) *httptest.Server {
+func newTestServer(t *testing.T) (*httptest.Server, *Server) {
t.Helper()
db, err := openDB(t.TempDir() + "/test.db")
if err != nil {
@@ -26,7 +26,7 @@ func newTestServer(t *testing.T) *httptest.Server {
s := &Server{db: db, hub: hub, pages: pages, vapidPub: pub, vapidPriv: priv}
ts := httptest.NewServer(s.handler())
t.Cleanup(ts.Close)
- return ts
+ return ts, s
}
// adminPost issues a POST with the admin bearer token.
@@ -43,7 +43,7 @@ func adminPost(t *testing.T, url, body string) *http.Response {
}
func TestAdminAuthGating(t *testing.T) {
- ts := newTestServer(t)
+ ts, _ := newTestServer(t)
resp, _ := http.Post(ts.URL+"/api/admin/invites", "application/json", strings.NewReader(`{"label":"x"}`))
if resp.StatusCode != http.StatusUnauthorized {
@@ -59,7 +59,7 @@ func TestAdminAuthGating(t *testing.T) {
}
func TestInviteAndMessagesFlow(t *testing.T) {
- ts := newTestServer(t)
+ ts, _ := newTestServer(t)
resp := adminPost(t, ts.URL+"/api/admin/invites", `{"label":"Sophie"}`)
if resp.StatusCode != http.StatusOK {
@@ -95,8 +95,105 @@ func TestInviteAndMessagesFlow(t *testing.T) {
}
}
+func TestMessagesIncludeReactions(t *testing.T) {
+ ts, srv := newTestServer(t)
+
+ resp := adminPost(t, ts.URL+"/api/admin/invites", `{"label":"x"}`)
+ var inv struct {
+ Token string `json:"token"`
+ }
+ json.NewDecoder(resp.Body).Decode(&inv)
+
+ // Seed a message + reactions directly (they normally arrive over the WS).
+ m, _ := srv.db.insertMessage("ciphertext-blob")
+ srv.db.addReaction(m.ID, "someoneElse", "👍")
+ srv.db.addReaction(m.ID, inv.Token, "❤️")
+
+ resp, _ = http.Get(ts.URL + "/api/messages?token=" + inv.Token)
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("messages: %d", resp.StatusCode)
+ }
+ var out struct {
+ Messages []struct {
+ ID int64 `json:"id"`
+ Reactions []struct {
+ Emoji string `json:"emoji"`
+ Count int `json:"count"`
+ Mine bool `json:"mine"`
+ } `json:"reactions"`
+ } `json:"messages"`
+ }
+ if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+ t.Fatalf("decode: %v", err)
+ }
+ if len(out.Messages) != 1 {
+ t.Fatalf("want 1 message, got %d", len(out.Messages))
+ }
+ seen := map[string]struct{ count int; mine bool }{}
+ for _, r := range out.Messages[0].Reactions {
+ seen[r.Emoji] = struct{ count int; mine bool }{r.Count, r.Mine}
+ }
+ if seen["👍"].count != 1 || seen["👍"].mine {
+ t.Fatalf("👍 = %+v (want count 1, mine false)", seen["👍"])
+ }
+ if seen["❤️"].count != 1 || !seen["❤️"].mine {
+ t.Fatalf("❤️ = %+v (want count 1, mine true)", seen["❤️"])
+ }
+}
+
+func TestBlobUploadDownload(t *testing.T) {
+ ts, _ := newTestServer(t)
+ resp := adminPost(t, ts.URL+"/api/admin/invites", `{"label":"x"}`)
+ var inv struct {
+ Token string `json:"token"`
+ }
+ json.NewDecoder(resp.Body).Decode(&inv)
+
+ payload := []byte("pretend-this-is-encrypted-bytes\x00\x01\x02")
+
+ // Upload requires a valid token.
+ resp, _ = http.Post(ts.URL+"/api/upload?token=bogus", "application/octet-stream", strings.NewReader("x"))
+ if resp.StatusCode != http.StatusForbidden {
+ t.Fatalf("upload bad token: want 403, got %d", resp.StatusCode)
+ }
+
+ resp, err := http.Post(ts.URL+"/api/upload?token="+inv.Token, "application/octet-stream", strings.NewReader(string(payload)))
+ if err != nil || resp.StatusCode != http.StatusOK {
+ t.Fatalf("upload: err=%v status=%d", err, resp.StatusCode)
+ }
+ var up struct {
+ ID string `json:"id"`
+ }
+ json.NewDecoder(resp.Body).Decode(&up)
+ if up.ID == "" {
+ t.Fatal("no blob id returned")
+ }
+
+ // Download requires a valid token.
+ resp, _ = http.Get(ts.URL + "/api/blob/" + up.ID + "?token=bogus")
+ if resp.StatusCode != http.StatusForbidden {
+ t.Fatalf("download bad token: want 403, got %d", resp.StatusCode)
+ }
+
+ // Round-trips exactly.
+ resp, _ = http.Get(ts.URL + "/api/blob/" + up.ID + "?token=" + inv.Token)
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("download: %d", resp.StatusCode)
+ }
+ got, _ := io.ReadAll(resp.Body)
+ if string(got) != string(payload) {
+ t.Fatalf("blob mismatch: got %q", got)
+ }
+
+ // Unknown id is a 404.
+ resp, _ = http.Get(ts.URL + "/api/blob/nope?token=" + inv.Token)
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("missing blob: want 404, got %d", resp.StatusCode)
+ }
+}
+
func TestPushSubscribeGating(t *testing.T) {
- ts := newTestServer(t)
+ ts, _ := newTestServer(t)
body := `{"token":"bad","subscription":{"endpoint":"https://e","keys":{"p256dh":"p","auth":"a"}}}`
resp, _ := http.Post(ts.URL+"/api/push/subscribe", "application/json", strings.NewReader(body))
if resp.StatusCode != http.StatusForbidden {
@@ -105,7 +202,7 @@ func TestPushSubscribeGating(t *testing.T) {
}
func TestGroupNameEndpoints(t *testing.T) {
- ts := newTestServer(t)
+ ts, _ := newTestServer(t)
resp, _ := http.Get(ts.URL + "/api/group")
var g struct{ Name string }
@@ -139,7 +236,7 @@ func TestGroupNameEndpoints(t *testing.T) {
}
func TestGroupNameEscapedInHTML(t *testing.T) {
- ts := newTestServer(t)
+ ts, _ := newTestServer(t)
adminPost(t, ts.URL+"/api/admin/group", `{"name":"<script>x</script>"}`)
resp, _ := http.Get(ts.URL + "/")
body, _ := io.ReadAll(resp.Body)
diff --git a/hub.go b/hub.go
index 360b2c4..f9f90ba 100644
--- a/hub.go
+++ b/hub.go
@@ -63,8 +63,10 @@ func (h *Hub) run() {
}
type wsIn struct {
- Type string `json:"type"`
- Blob string `json:"blob"`
+ Type string `json:"type"`
+ Blob string `json:"blob"`
+ MessageID int64 `json:"message_id"`
+ Emoji string `json:"emoji"`
}
type wsOut struct {
@@ -74,6 +76,18 @@ type wsOut struct {
Blob string `json:"blob"`
}
+type wsReactionOut struct {
+ Type string `json:"type"`
+ MessageID int64 `json:"message_id"`
+ Reactions []ReactionCount `json:"reactions"`
+}
+
+// allowedReactions is the fixed emoji set the UI offers. Restricting it keeps
+// the (plaintext) emoji column to known values — no arbitrary strings.
+var allowedReactions = map[string]bool{
+ "👍": true, "❤️": true, "😂": true, "😮": true, "😢": true, "🎉": true,
+}
+
func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
token := r.URL.Query().Get("token")
exists, revoked, err := s.db.inviteStatus(token)
@@ -120,21 +134,48 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
if json.Unmarshal(data, &in) != nil {
continue
}
- if in.Type != "msg" || in.Blob == "" || len(in.Blob) > maxBlobBytes {
+ if in.Type != "msg" && in.Type != "react" && in.Type != "unreact" {
continue
}
- // Re-check revocation so a revoked client can't keep posting.
+ // Re-check revocation so a revoked client can't keep posting or reacting.
if _, rev, err := s.db.inviteStatus(token); err != nil || rev {
return
}
- m, err := s.db.insertMessage(in.Blob)
- if err != nil {
- log.Println("insert message:", err)
- continue
+
+ switch in.Type {
+ case "msg":
+ if in.Blob == "" || len(in.Blob) > maxBlobBytes {
+ continue
+ }
+ m, err := s.db.insertMessage(in.Blob)
+ if err != nil {
+ log.Println("insert message:", err)
+ continue
+ }
+ out, _ := json.Marshal(wsOut{Type: "msg", ID: m.ID, CreatedAt: m.CreatedAt, Blob: m.Blob})
+ s.hub.broadcast <- out
+ // Notify subscribers who don't have the app open (best-effort, async).
+ go s.pushNewMessage(m.Blob)
+
+ case "react", "unreact":
+ if in.MessageID <= 0 || !allowedReactions[in.Emoji] {
+ continue
+ }
+ if in.Type == "react" {
+ err = s.db.addReaction(in.MessageID, token, in.Emoji)
+ } else {
+ err = s.db.removeReaction(in.MessageID, token, in.Emoji)
+ }
+ if err != nil {
+ log.Println("reaction:", err)
+ continue
+ }
+ counts, err := s.db.reactionCounts(in.MessageID)
+ if err != nil {
+ continue
+ }
+ out, _ := json.Marshal(wsReactionOut{Type: "reaction", MessageID: in.MessageID, Reactions: counts})
+ s.hub.broadcast <- out
}
- out, _ := json.Marshal(wsOut{Type: "msg", ID: m.ID, CreatedAt: m.CreatedAt, Blob: m.Blob})
- s.hub.broadcast <- out
- // Notify subscribers who don't have the app open (best-effort, async).
- go s.pushNewMessage(m.Blob)
}
}
diff --git a/main.go b/main.go
index 22f9ef7..8d3869e 100644
--- a/main.go
+++ b/main.go
@@ -5,6 +5,7 @@ import (
"encoding/json"
"flag"
"html"
+ "io"
"io/fs"
"log"
"net/http"
@@ -13,6 +14,9 @@ import (
"strings"
)
+// maxUploadBytes caps an encrypted attachment upload (10 MB of ciphertext).
+const maxUploadBytes = 10 << 20
+
//go:embed web/pages/*.html web/static/*
var webFS embed.FS
@@ -92,6 +96,11 @@ func (s *Server) handler() http.Handler {
mux.HandleFunc("GET /api/messages", s.getMessages)
mux.HandleFunc("GET /ws", s.handleWS)
+ // Encrypted attachments: bytes are E2E-encrypted in the browser; the server
+ // stores/serves opaque blobs only. Both ends are token-gated.
+ mux.HandleFunc("POST /api/upload", s.uploadBlob)
+ mux.HandleFunc("GET /api/blob/{id}", s.getBlobHandler)
+
// Invite lifecycle (used by the join page)
mux.HandleFunc("GET /api/invite/{token}", s.getInvite)
mux.HandleFunc("POST /api/invite/{token}/claim", s.claimInvite)
@@ -161,9 +170,76 @@ func (s *Server) getMessages(w http.ResponseWriter, r *http.Request) {
httpError(w, http.StatusInternalServerError, "db error")
return
}
+ // Attach reactions (with a "mine" flag for this caller's token).
+ ids := make([]int64, len(msgs))
+ for i, m := range msgs {
+ ids[i] = m.ID
+ }
+ if reMap, err := s.db.reactionsForMessages(ids, token); err == nil {
+ for i := range msgs {
+ if rs := reMap[msgs[i].ID]; len(rs) > 0 {
+ msgs[i].Reactions = rs
+ }
+ }
+ }
writeJSON(w, http.StatusOK, map[string]any{"messages": msgs})
}
+// tokenOK reports whether a token is a known, non-revoked invite.
+func (s *Server) tokenOK(token string) bool {
+ exists, revoked, err := s.db.inviteStatus(token)
+ return err == nil && exists && !revoked
+}
+
+// ---- attachments --------------------------------------------------------
+
+func (s *Server) uploadBlob(w http.ResponseWriter, r *http.Request) {
+ if !s.tokenOK(r.URL.Query().Get("token")) {
+ httpError(w, http.StatusForbidden, "invalid invite")
+ return
+ }
+ r.Body = http.MaxBytesReader(w, r.Body, maxUploadBytes)
+ data, err := io.ReadAll(r.Body)
+ if err != nil {
+ httpError(w, http.StatusRequestEntityTooLarge, "file too large (max 10 MB)")
+ return
+ }
+ if len(data) == 0 {
+ httpError(w, http.StatusBadRequest, "empty upload")
+ return
+ }
+ id, err := randToken()
+ if err != nil {
+ httpError(w, http.StatusInternalServerError, "id error")
+ return
+ }
+ if err := s.db.putBlob(id, data); err != nil {
+ httpError(w, http.StatusInternalServerError, "db error")
+ return
+ }
+ writeJSON(w, http.StatusOK, map[string]any{"id": id})
+}
+
+func (s *Server) getBlobHandler(w http.ResponseWriter, r *http.Request) {
+ if !s.tokenOK(r.URL.Query().Get("token")) {
+ httpError(w, http.StatusForbidden, "invalid invite")
+ return
+ }
+ data, err := s.db.getBlob(r.PathValue("id"))
+ if err != nil {
+ httpError(w, http.StatusInternalServerError, "db error")
+ return
+ }
+ if data == nil {
+ http.NotFound(w, r)
+ return
+ }
+ w.Header().Set("Content-Type", "application/octet-stream")
+ // Blob ids are unique and content is immutable, so it's safe to cache hard.
+ w.Header().Set("Cache-Control", "private, max-age=31536000, immutable")
+ w.Write(data)
+}
+
// ---- invite (join flow) -------------------------------------------------
func (s *Server) getInvite(w http.ResponseWriter, r *http.Request) {