feat: EXT-007 P2P message history gossip

After hello verification, the connecting peer sends history_request to
the first peer it meets (one per room, no fan-out). The responder queries
SQLite and replies with a history_chunk. Received history is stored via
INSERT OR IGNORE (mid dedup) and emitted as history_loaded IPC events.

- proto: MsgHistoryRequest/Chunk types, HistoryEntry, EvtHistoryLoaded,
  ComputeMsgID (sha256 content-addressed ID), MsgID field on ChatMessage
- store: ALTER TABLE ADD COLUMN msg_id + unique index migration (idempotent);
  RecentMessagesSince query (msg_id IS NOT NULL filter); msg_id persisted on save
- mesh: RequestHistoryFrom, HandleHistoryRequest, HandleHistoryChunk methods;
  historyRequested/historyFirstPeer state to ensure single-peer requests
- peer: dispatch history_request/history_chunk; RequestHistoryFrom after hello;
  stamp MsgID on incoming chat messages
- ipc: stamp MsgID on outgoing group chat messages
- EXTENSIONS.md: EXT-007 documented

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Fredrik Johansson
2026-06-28 23:08:17 +02:00
parent 1c73f1b1ef
commit 7c3cedc549
6 changed files with 322 additions and 35 deletions

View File

@@ -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 The sidecar is written when the transfer starts and removed on completion or
corruption. Interrupted transfers keep the sidecar indefinitely. 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": "<peer-id>", "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.

View File

@@ -214,6 +214,7 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) {
} else { } else {
// Group chat → spec "chat" type: flat {type, mid, room, text, ts} // Group chat → spec "chat" type: flat {type, mid, room, text, ts}
mid := randomHex(16) mid := randomHex(16)
msgID := proto.ComputeMsgID(n.Identity.PeerID(), cmd.Room, ts, cmd.Body)
wire, err := json.Marshal(proto.PeerMessage{ wire, err := json.Marshal(proto.PeerMessage{
Type: proto.MsgChat, Type: proto.MsgChat,
Mid: mid, Mid: mid,
@@ -226,11 +227,12 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) {
} }
n.Mesh.Broadcast(wire) n.Mesh.Broadcast(wire)
local := &proto.ChatMessage{ local := &proto.ChatMessage{
Mid: mid, Mid: mid,
From: n.Identity.PeerID(), MsgID: msgID,
Room: cmd.Room, From: n.Identity.PeerID(),
Text: cmd.Body, Room: cmd.Room,
Ts: ts, Text: cmd.Body,
Ts: ts,
} }
n.Mesh.SaveMessage(local) n.Mesh.SaveMessage(local)
n.Mesh.Emit(proto.IpcMessage{ n.Mesh.Emit(proto.IpcMessage{

View File

@@ -2,6 +2,7 @@
package mesh package mesh
import ( import (
"encoding/json"
"log" "log"
"os" "os"
"sync" "sync"
@@ -51,6 +52,12 @@ type Mesh struct {
// attempt to connect to. Drained by the anchor client's runOnce loop. // attempt to connect to. Drained by the anchor client's runOnce loop.
PendingConnect chan proto.PeerID 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) // subscribers receive a copy of every event (fan-out to IPC clients)
subMu sync.Mutex subMu sync.Mutex
subs []chan proto.IpcMessage subs []chan proto.IpcMessage
@@ -60,12 +67,13 @@ type Mesh struct {
// Pass a non-nil store to enable message and peer persistence. // Pass a non-nil store to enable message and peer persistence.
func New(id *crypto.Identity, st *store.Store) *Mesh { func New(id *crypto.Identity, st *store.Store) *Mesh {
return &Mesh{ return &Mesh{
Identity: id, Identity: id,
Store: st, Store: st,
peers: make(map[proto.PeerID]*PeerConn), peers: make(map[proto.PeerID]*PeerConn),
outbound: make(map[string]*outboundTransfer), outbound: make(map[string]*outboundTransfer),
inbound: make(map[string]*inboundTransfer), inbound: make(map[string]*inboundTransfer),
PendingConnect: make(chan proto.PeerID, 32), 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). // Emit sends an event to all IPC subscribers (exported for ipc/nat packages).
func (m *Mesh) Emit(msg proto.IpcMessage) { func (m *Mesh) Emit(msg proto.IpcMessage) {
m.emit(msg) m.emit(msg)

View File

@@ -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. // Tell the new peer about everyone we can currently see.
go m.sendGossipTo(from) go m.sendGossipTo(from)
// Request message history from this peer (EXT-007).
go m.RequestHistoryFrom(from)
return return
} }
@@ -212,11 +214,12 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
switch msg.Type { switch msg.Type {
case proto.MsgChat: case proto.MsgChat:
chat := &proto.ChatMessage{ chat := &proto.ChatMessage{
Mid: midOrRandom(msg.Mid), Mid: midOrRandom(msg.Mid),
From: from, MsgID: proto.ComputeMsgID(from, msg.Room, msg.Ts, msg.Text),
Room: msg.Room, From: from,
Text: msg.Text, Room: msg.Room,
Ts: msg.Ts, Text: msg.Text,
Ts: msg.Ts,
} }
m.SaveMessage(chat) m.SaveMessage(chat)
m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: 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) 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: case proto.MsgPing:
log.Printf("mesh: ping from %s", from.Short()) log.Printf("mesh: ping from %s", from.Short())
case proto.MsgPong: case proto.MsgPong:

View File

@@ -3,7 +3,11 @@
// Binary data (keys, signatures) is hex-encoded; signaling boxes are base64. // Binary data (keys, signatures) is hex-encoded; signaling boxes are base64.
package proto package proto
import "time" import (
"crypto/sha256"
"fmt"
"time"
)
// ── Identity ────────────────────────────────────────────────────────────────── // ── Identity ──────────────────────────────────────────────────────────────────
@@ -43,8 +47,10 @@ const (
MsgFileAccept MsgType = "file-accept" MsgFileAccept MsgType = "file-accept"
MsgFileCancel MsgType = "file-cancel" MsgFileCancel MsgType = "file-cancel"
MsgFileDone MsgType = "file-done" MsgFileDone MsgType = "file-done"
MsgPing MsgType = "ping" MsgPing MsgType = "ping"
MsgPong MsgType = "pong" 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"). // 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 Reason string `json:"reason,omitempty"` // file-cancel
Seq *uint64 `json:"seq,omitempty"` // ping/pong 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). // ChatMessage is a group chat message (wire type "chat", §8).
// Also used internally for persisting PMs after they are received. // Also used internally for persisting PMs after they are received.
type ChatMessage struct { type ChatMessage struct {
Mid string `json:"mid,omitempty"` // optional dedup id (required when relay hops > 0) 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 MsgID string `json:"msg_id,omitempty"` // EXT-007: content-addressed gossip ID
To *PeerID `json:"to,omitempty"` // internal only — not transmitted; set for DMs From PeerID `json:"from,omitempty"` // set by receiver from DC context; not on wire for pm
Room string `json:"room"` To *PeerID `json:"to,omitempty"` // internal only — not transmitted; set for DMs
Text string `json:"text"` Room string `json:"room"`
Ts int64 `json:"ts"` // Unix milliseconds 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. // PeerGossip shares known peer addresses.
@@ -259,6 +291,7 @@ const (
EvtIdentityImported IpcMsgType = "identity_imported" EvtIdentityImported IpcMsgType = "identity_imported"
EvtSharesList IpcMsgType = "shares_list" EvtSharesList IpcMsgType = "shares_list"
EvtRoomCreated IpcMsgType = "room_created" // field: room (name) 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. // NetworkInfo summarises one joined network for state_snapshot and network_joined events.
@@ -312,7 +345,8 @@ type IpcMessage struct {
Networks []NetworkInfo `json:"networks,omitempty"` Networks []NetworkInfo `json:"networks,omitempty"`
ErrorMessage string `json:"error_message,omitempty"` ErrorMessage string `json:"error_message,omitempty"`
InviteGenerated string `json:"invite,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"` Shares []ShareEntry `json:"shares,omitempty"`
ShareNetworks []string `json:"network_ids,omitempty"` // for add_share command: scope to specific network IDs, or ["*"] for global ShareNetworks []string `json:"network_ids,omitempty"` // for add_share command: scope to specific network IDs, or ["*"] for global
// export_identity / import_identity // export_identity / import_identity

View File

@@ -5,6 +5,7 @@ package store
import ( import (
"database/sql" "database/sql"
"fmt" "fmt"
"strings"
"time" "time"
_ "modernc.org/sqlite" _ "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. // Store is a local SQLite-backed message and peer store.
type Store struct { type Store struct {
db *sql.DB db *sql.DB
@@ -51,6 +60,12 @@ func Open(path string) (*Store, error) {
db.Close() db.Close()
return nil, fmt.Errorf("migrate db: %w", err) 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 return &Store{db: db}, nil
} }
@@ -64,9 +79,9 @@ func (s *Store) Close() error {
func (s *Store) SaveMessage(msg *proto.ChatMessage) error { func (s *Store) SaveMessage(msg *proto.ChatMessage) error {
sentAt := time.UnixMilli(msg.Ts).UTC() sentAt := time.UnixMilli(msg.Ts).UTC()
_, err := s.db.Exec( _, err := s.db.Exec(
`INSERT OR IGNORE INTO messages (mid, room, from_peer, body, sent_at) `INSERT OR IGNORE INTO messages (mid, msg_id, room, from_peer, body, sent_at)
VALUES (?, ?, ?, ?, ?)`, VALUES (?, ?, ?, ?, ?, ?)`,
msg.Mid, msg.Room, string(msg.From), msg.Text, sentAt, msg.Mid, nullableString(msg.MsgID), msg.Room, string(msg.From), msg.Text, sentAt,
) )
return err 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. // RecentMessages returns up to limit messages for a room, oldest first.
func (s *Store) RecentMessages(room string, limit int) ([]proto.ChatMessage, error) { func (s *Store) RecentMessages(room string, limit int) ([]proto.ChatMessage, error) {
rows, err := s.db.Query( return s.queryMessages(
`SELECT mid, from_peer, body, sent_at `SELECT mid, from_peer, body, sent_at FROM messages
FROM messages
WHERE room = ? WHERE room = ?
ORDER BY sent_at DESC ORDER BY sent_at DESC LIMIT ?`,
LIMIT ?`,
room, 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 { if err != nil {
return nil, err return nil, err
} }
@@ -113,7 +145,6 @@ func (s *Store) RecentMessages(room string, limit int) ([]proto.ChatMessage, err
return nil, err return nil, err
} }
m.From = proto.PeerID(from) m.From = proto.PeerID(from)
m.Room = room
m.Ts = sentAt.UnixMilli() m.Ts = sentAt.UnixMilli()
msgs = append(msgs, m) msgs = append(msgs, m)
} }
@@ -168,3 +199,10 @@ func (s *Store) KnownPeers() (map[proto.PeerID]string, error) {
} }
return out, rows.Err() return out, rows.Err()
} }
func nullableString(s string) any {
if s == "" {
return nil
}
return s
}