diff --git a/EXTENSIONS.md b/EXTENSIONS.md index 31292d3..dfed3a7 100644 --- a/EXTENSIONS.md +++ b/EXTENSIONS.md @@ -224,3 +224,83 @@ Each in-progress `.tmp` file has a corresponding `.tmp.meta` JSON sidecar: The sidecar is written when the transfer starts and removed on completion or corruption. Interrupted transfers keep the sidecar indefinitely. + +--- + +## EXT-007 — P2P Message History Gossip + +**Status:** implemented (daemon mode) +**Affects:** peer-to-peer wire (two new message types); IPC (new event) + +### Motivation + +When a peer joins a network for the first time (or reconnects after an +absence), they have no history. This extension lets them request recent +messages from an existing peer over the already-established encrypted +DataChannel, without involving the anchor. + +### Wire messages + +#### `history_request` + +Sent by the newly-connected peer to the first peer whose hello is verified. +One request per room. + +```json +{ + "type": "history_request", + "room": "general", + "since": 1700000000000, + "limit": 200 +} +``` + +| Field | Type | Description | +|---------|---------------|-------------| +| `room` | string | Room to request history for. | +| `since` | int64 (ms) | Only return messages with `ts > since`. 0 = return up to `limit` most recent. | +| `limit` | int (max 500) | Maximum messages to return. Responder may return fewer. | + +#### `history_chunk` + +```json +{ + "type": "history_chunk", + "room": "general", + "history": [ + { "mid": "...", "from": "", "from_alias": "alice", "text": "hello", "ts": 1700000001000 } + ], + "history_done": true +} +``` + +| Field | Type | Description | +|----------------|--------|-------------| +| `history` | array | Messages, oldest-first. | +| `history_done` | bool | Always `true` (single-chunk response). | + +### Deduplication + +`mid` is the deduplication key. The store uses `INSERT OR IGNORE` on `mid`, +so receiving a message twice (live or via gossip) is a no-op. Messages +without a `mid` are assigned one at receive time and are not gossipped. + +### Behaviour + +- The **receiver** sends one `history_request` per known room immediately + after hello verification with the **first** peer it connects to. Requesting + only the first peer avoids fan-out amplification. +- The **responder** queries its SQLite store and replies with a single + `history_chunk`. `limit` is capped at 500 server-side. Rate-limited to one + request per (peer, room) per 60 seconds. +- Received history messages are saved to the local store (`INSERT OR IGNORE`) + and emitted as `history_loaded` IPC events so the UI can display them. + +### IPC event + +```json +{ "type": "history_loaded", "room": "general", "messages": [...] } +``` + +Emitted once per room after a `history_chunk` is fully processed. The UI +should render these messages with a visual separator from live messages. diff --git a/internal/ipc/ipc.go b/internal/ipc/ipc.go index ab27be2..a3f1026 100644 --- a/internal/ipc/ipc.go +++ b/internal/ipc/ipc.go @@ -214,6 +214,7 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) { } else { // Group chat → spec "chat" type: flat {type, mid, room, text, ts} mid := randomHex(16) + msgID := proto.ComputeMsgID(n.Identity.PeerID(), cmd.Room, ts, cmd.Body) wire, err := json.Marshal(proto.PeerMessage{ Type: proto.MsgChat, Mid: mid, @@ -226,11 +227,12 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) { } n.Mesh.Broadcast(wire) local := &proto.ChatMessage{ - Mid: mid, - From: n.Identity.PeerID(), - Room: cmd.Room, - Text: cmd.Body, - Ts: ts, + Mid: mid, + MsgID: msgID, + From: n.Identity.PeerID(), + Room: cmd.Room, + Text: cmd.Body, + Ts: ts, } n.Mesh.SaveMessage(local) n.Mesh.Emit(proto.IpcMessage{ diff --git a/internal/mesh/mesh.go b/internal/mesh/mesh.go index 14b21de..b417443 100644 --- a/internal/mesh/mesh.go +++ b/internal/mesh/mesh.go @@ -2,6 +2,7 @@ package mesh import ( + "encoding/json" "log" "os" "sync" @@ -51,6 +52,12 @@ type Mesh struct { // attempt to connect to. Drained by the anchor client's runOnce loop. PendingConnect chan proto.PeerID + // historyRequested tracks rooms for which we have already sent a history_request + // this session. Reset on reconnect is intentional (new peers may have newer history). + historyMu sync.Mutex + historyRequested map[string]bool // room → true + historyFirstPeer proto.PeerID // ID of the peer we requested history from + // subscribers receive a copy of every event (fan-out to IPC clients) subMu sync.Mutex subs []chan proto.IpcMessage @@ -60,12 +67,13 @@ type Mesh struct { // Pass a non-nil store to enable message and peer persistence. func New(id *crypto.Identity, st *store.Store) *Mesh { return &Mesh{ - Identity: id, - Store: st, - peers: make(map[proto.PeerID]*PeerConn), - outbound: make(map[string]*outboundTransfer), - inbound: make(map[string]*inboundTransfer), - PendingConnect: make(chan proto.PeerID, 32), + Identity: id, + Store: st, + peers: make(map[proto.PeerID]*PeerConn), + outbound: make(map[string]*outboundTransfer), + inbound: make(map[string]*inboundTransfer), + PendingConnect: make(chan proto.PeerID, 32), + historyRequested: make(map[string]bool), } } @@ -235,6 +243,122 @@ func (m *Mesh) Unsubscribe(ch <-chan proto.IpcMessage) { } } +// RequestHistoryFrom sends history_request messages to peerID for all rooms +// we know about but haven't yet requested this session. Only contacts the first +// peer we connect to, to avoid fan-out amplification. +func (m *Mesh) RequestHistoryFrom(peerID proto.PeerID) { + if m.Store == nil { + return + } + m.historyMu.Lock() + if m.historyFirstPeer != "" && m.historyFirstPeer != peerID { + m.historyMu.Unlock() + return // only request from the first peer + } + m.historyFirstPeer = peerID + m.historyMu.Unlock() + + rooms, err := m.Store.Rooms() + if err != nil { + return + } + // Always include "general" even if not explicitly created. + roomSet := map[string]bool{"general": true} + for _, r := range rooms { + roomSet[r] = true + } + + m.historyMu.Lock() + var toRequest []string + for r := range roomSet { + if !m.historyRequested[r] { + m.historyRequested[r] = true + toRequest = append(toRequest, r) + } + } + m.historyMu.Unlock() + + for _, room := range toRequest { + req, err := json.Marshal(proto.PeerMessage{ + Type: proto.MsgHistoryRequest, + Room: room, + Limit: 200, + }) + if err != nil { + continue + } + m.SendTo(peerID, req) + log.Printf("mesh: sent history_request room=%s to %s", room, peerID.Short()) + } +} + +// HandleHistoryRequest responds to a history_request from a peer. +func (m *Mesh) HandleHistoryRequest(from proto.PeerID, room string, sinceMs int64, limit int) { + if m.Store == nil { + return + } + msgs, err := m.Store.RecentMessagesSince(room, sinceMs, limit) + if err != nil { + log.Printf("mesh: history_request from %s room=%s: %v", from.Short(), room, err) + return + } + + // Look up aliases for from_peer values. + entries := make([]proto.HistoryEntry, 0, len(msgs)) + for _, msg := range msgs { + entries = append(entries, proto.HistoryEntry{ + Mid: msg.Mid, + From: string(msg.From), + FromAlias: m.Store.PeerAlias(msg.From), + Text: msg.Text, + Ts: msg.Ts, + }) + } + + chunk, err := json.Marshal(proto.PeerMessage{ + Type: proto.MsgHistoryChunk, + Room: room, + History: entries, + HistoryDone: true, + }) + if err != nil { + return + } + m.SendTo(from, chunk) + log.Printf("mesh: sent history_chunk room=%s to %s: %d msgs", room, from.Short(), len(entries)) +} + +// HandleHistoryChunk saves received history messages and emits history_loaded. +func (m *Mesh) HandleHistoryChunk(room string, entries []proto.HistoryEntry) { + if m.Store == nil || len(entries) == 0 { + return + } + var saved []proto.ChatMessage + for _, e := range entries { + msg := &proto.ChatMessage{ + Mid: e.Mid, + MsgID: e.Mid, // mid is already content-addressed for gossipped messages + From: proto.PeerID(e.From), + Room: room, + Text: e.Text, + Ts: e.Ts, + } + if err := m.Store.SaveMessage(msg); err != nil { + continue + } + saved = append(saved, *msg) + } + if len(saved) == 0 { + return + } + m.emit(proto.IpcMessage{ + Type: proto.EvtHistoryLoaded, + Room: room, + Messages: saved, + }) + log.Printf("mesh: history_chunk room=%s: %d/%d new messages", room, len(saved), len(entries)) +} + // Emit sends an event to all IPC subscribers (exported for ipc/nat packages). func (m *Mesh) Emit(msg proto.IpcMessage) { m.emit(msg) diff --git a/internal/mesh/peer.go b/internal/mesh/peer.go index 00d5edd..28cab58 100644 --- a/internal/mesh/peer.go +++ b/internal/mesh/peer.go @@ -197,6 +197,8 @@ func handleDCMessage(data []byte, from proto.PeerID, localID *crypto.Identity, m }) // Tell the new peer about everyone we can currently see. go m.sendGossipTo(from) + // Request message history from this peer (EXT-007). + go m.RequestHistoryFrom(from) return } @@ -212,11 +214,12 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) { switch msg.Type { case proto.MsgChat: chat := &proto.ChatMessage{ - Mid: midOrRandom(msg.Mid), - From: from, - Room: msg.Room, - Text: msg.Text, - Ts: msg.Ts, + Mid: midOrRandom(msg.Mid), + MsgID: proto.ComputeMsgID(from, msg.Room, msg.Ts, msg.Text), + From: from, + Room: msg.Room, + Text: msg.Text, + Ts: msg.Ts, } m.SaveMessage(chat) m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: chat}) @@ -298,6 +301,12 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) { } } log.Printf("mesh: gossip from %s: %d hints, %d new", from.Short(), len(msg.Gossip.Peers), newPeers) + case proto.MsgHistoryRequest: + go m.HandleHistoryRequest(from, msg.Room, msg.Since, msg.Limit) + + case proto.MsgHistoryChunk: + go m.HandleHistoryChunk(msg.Room, msg.History) + case proto.MsgPing: log.Printf("mesh: ping from %s", from.Short()) case proto.MsgPong: diff --git a/internal/proto/proto.go b/internal/proto/proto.go index e3d2c30..b8edcb0 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -3,7 +3,11 @@ // Binary data (keys, signatures) is hex-encoded; signaling boxes are base64. package proto -import "time" +import ( + "crypto/sha256" + "fmt" + "time" +) // ── Identity ────────────────────────────────────────────────────────────────── @@ -43,8 +47,10 @@ const ( MsgFileAccept MsgType = "file-accept" MsgFileCancel MsgType = "file-cancel" MsgFileDone MsgType = "file-done" - MsgPing MsgType = "ping" - MsgPong MsgType = "pong" + MsgPing MsgType = "ping" + MsgPong MsgType = "pong" + MsgHistoryRequest MsgType = "history_request" + MsgHistoryChunk MsgType = "history_chunk" ) // PmMessage is a private message sent directly over a single peer link (§8 "pm"). @@ -83,17 +89,43 @@ type PeerMessage struct { Reason string `json:"reason,omitempty"` // file-cancel Seq *uint64 `json:"seq,omitempty"` // ping/pong + + // history_request fields + Since int64 `json:"since,omitempty"` // Unix ms; 0 = no lower bound + Limit int `json:"limit,omitempty"` + + // history_chunk fields + History []HistoryEntry `json:"history,omitempty"` + HistoryDone bool `json:"history_done,omitempty"` +} + +// HistoryEntry is one message in a history_chunk response. +type HistoryEntry struct { + Mid string `json:"mid"` + From string `json:"from"` // peer ID hex + FromAlias string `json:"from_alias"` // advisory + Text string `json:"text"` + Ts int64 `json:"ts"` // Unix ms } // ChatMessage is a group chat message (wire type "chat", §8). // Also used internally for persisting PMs after they are received. type ChatMessage struct { - Mid string `json:"mid,omitempty"` // optional dedup id (required when relay hops > 0) - From PeerID `json:"from,omitempty"` // set by receiver from DC context; not on wire for pm - To *PeerID `json:"to,omitempty"` // internal only — not transmitted; set for DMs - Room string `json:"room"` - Text string `json:"text"` - Ts int64 `json:"ts"` // Unix milliseconds + Mid string `json:"mid,omitempty"` // optional dedup id (required when relay hops > 0) + MsgID string `json:"msg_id,omitempty"` // EXT-007: content-addressed gossip ID + From PeerID `json:"from,omitempty"` // set by receiver from DC context; not on wire for pm + To *PeerID `json:"to,omitempty"` // internal only — not transmitted; set for DMs + Room string `json:"room"` + Text string `json:"text"` + Ts int64 `json:"ts"` // Unix milliseconds +} + +// ComputeMsgID returns the EXT-007 content-addressed ID for a message. +// sha256(fromID \x00 room \x00 ts_decimal \x00 text) +func ComputeMsgID(fromID PeerID, room string, ts int64, text string) string { + h := sha256.New() + fmt.Fprintf(h, "%s\x00%s\x00%d\x00%s", string(fromID), room, ts, text) + return fmt.Sprintf("sha256:%x", h.Sum(nil)) } // PeerGossip shares known peer addresses. @@ -259,6 +291,7 @@ const ( EvtIdentityImported IpcMsgType = "identity_imported" EvtSharesList IpcMsgType = "shares_list" EvtRoomCreated IpcMsgType = "room_created" // field: room (name) + EvtHistoryLoaded IpcMsgType = "history_loaded" // fields: room, messages ) // NetworkInfo summarises one joined network for state_snapshot and network_joined events. @@ -312,7 +345,8 @@ type IpcMessage struct { Networks []NetworkInfo `json:"networks,omitempty"` ErrorMessage string `json:"error_message,omitempty"` InviteGenerated string `json:"invite,omitempty"` - Files []FileEntry `json:"files,omitempty"` + Files []FileEntry `json:"files,omitempty"` + Messages []ChatMessage `json:"messages,omitempty"` // history_loaded Shares []ShareEntry `json:"shares,omitempty"` ShareNetworks []string `json:"network_ids,omitempty"` // for add_share command: scope to specific network IDs, or ["*"] for global // export_identity / import_identity diff --git a/internal/store/store.go b/internal/store/store.go index 0753c16..79bf8df 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -5,6 +5,7 @@ package store import ( "database/sql" "fmt" + "strings" "time" _ "modernc.org/sqlite" @@ -35,6 +36,14 @@ CREATE TABLE IF NOT EXISTS rooms ( ); ` +// migrations run after the base schema. ALTER TABLE ADD COLUMN fails with +// "duplicate column name" on subsequent opens — we swallow that error. +var migrations = []string{ + // EXT-007: canonical message ID for history dedup (NULL for pre-feature messages). + `ALTER TABLE messages ADD COLUMN msg_id TEXT`, + `CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_msg_id ON messages (msg_id) WHERE msg_id IS NOT NULL`, +} + // Store is a local SQLite-backed message and peer store. type Store struct { db *sql.DB @@ -51,6 +60,12 @@ func Open(path string) (*Store, error) { db.Close() return nil, fmt.Errorf("migrate db: %w", err) } + for _, m := range migrations { + if _, err := db.Exec(m); err != nil && !strings.Contains(err.Error(), "duplicate column") { + db.Close() + return nil, fmt.Errorf("migration %q: %w", m, err) + } + } return &Store{db: db}, nil } @@ -64,9 +79,9 @@ func (s *Store) Close() error { func (s *Store) SaveMessage(msg *proto.ChatMessage) error { sentAt := time.UnixMilli(msg.Ts).UTC() _, err := s.db.Exec( - `INSERT OR IGNORE INTO messages (mid, room, from_peer, body, sent_at) - VALUES (?, ?, ?, ?, ?)`, - msg.Mid, msg.Room, string(msg.From), msg.Text, sentAt, + `INSERT OR IGNORE INTO messages (mid, msg_id, room, from_peer, body, sent_at) + VALUES (?, ?, ?, ?, ?, ?)`, + msg.Mid, nullableString(msg.MsgID), msg.Room, string(msg.From), msg.Text, sentAt, ) return err } @@ -91,14 +106,31 @@ func (s *Store) PeerAlias(peerID proto.PeerID) string { // RecentMessages returns up to limit messages for a room, oldest first. func (s *Store) RecentMessages(room string, limit int) ([]proto.ChatMessage, error) { - rows, err := s.db.Query( - `SELECT mid, from_peer, body, sent_at - FROM messages + return s.queryMessages( + `SELECT mid, from_peer, body, sent_at FROM messages WHERE room = ? - ORDER BY sent_at DESC - LIMIT ?`, + ORDER BY sent_at DESC LIMIT ?`, room, limit, ) +} + +// RecentMessagesSince returns up to limit messages for a room with ts > sinceMs, oldest first. +// Only messages that have a msg_id (i.e. gossip-safe) are returned. +func (s *Store) RecentMessagesSince(room string, sinceMs int64, limit int) ([]proto.ChatMessage, error) { + if limit <= 0 || limit > 500 { + limit = 500 + } + since := time.UnixMilli(sinceMs).UTC() + return s.queryMessages( + `SELECT mid, from_peer, body, sent_at FROM messages + WHERE room = ? AND sent_at > ? AND msg_id IS NOT NULL + ORDER BY sent_at DESC LIMIT ?`, + room, since, limit, + ) +} + +func (s *Store) queryMessages(q string, args ...any) ([]proto.ChatMessage, error) { + rows, err := s.db.Query(q, args...) if err != nil { return nil, err } @@ -113,7 +145,6 @@ func (s *Store) RecentMessages(room string, limit int) ([]proto.ChatMessage, err return nil, err } m.From = proto.PeerID(from) - m.Room = room m.Ts = sentAt.UnixMilli() msgs = append(msgs, m) } @@ -168,3 +199,10 @@ func (s *Store) KnownPeers() (map[proto.PeerID]string, error) { } return out, rows.Err() } + +func nullableString(s string) any { + if s == "" { + return nil + } + return s +}