Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
15306dc0c2 | ||
|
|
7c3cedc549 | ||
|
|
1c73f1b1ef |
@@ -20,7 +20,7 @@ jobs:
|
|||||||
|
|
||||||
- uses: actions/setup-node@v4
|
- uses: actions/setup-node@v4
|
||||||
with:
|
with:
|
||||||
node-version: '20'
|
node-version: '24'
|
||||||
|
|
||||||
- name: Install Wails CLI
|
- name: Install Wails CLI
|
||||||
run: go install github.com/wailsapp/wails/v2/cmd/wails@latest
|
run: go install github.com/wailsapp/wails/v2/cmd/wails@latest
|
||||||
@@ -61,7 +61,7 @@ jobs:
|
|||||||
- name: Build frontend
|
- name: Build frontend
|
||||||
run: |
|
run: |
|
||||||
cd web
|
cd web
|
||||||
npm ci
|
npm install
|
||||||
npm run build
|
npm run build
|
||||||
cp -r dist ../cmd/app/frontend/dist
|
cp -r dist ../cmd/app/frontend/dist
|
||||||
|
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
154
PROPOSAL-history-gossip.md
Normal file
154
PROPOSAL-history-gossip.md
Normal file
@@ -0,0 +1,154 @@
|
|||||||
|
# Proposal: P2P Message History Gossip
|
||||||
|
|
||||||
|
## Goals
|
||||||
|
|
||||||
|
- New peers joining a network can retrieve recent message history from existing peers
|
||||||
|
- No central storage — history lives only in peer daemons (SQLite)
|
||||||
|
- No new trust requirements — history is shared only over already-established encrypted DataChannels
|
||||||
|
- No duplicates in the local store or UI
|
||||||
|
- No conflicts — history gossip is append-only and idempotent
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Privacy Model
|
||||||
|
|
||||||
|
History is shared **peer-to-peer over the encrypted mesh**, never via the anchor. The anchor remains dumb — it sees only signaling blobs. A peer only receives history from peers they have successfully completed a YAW/2 handshake with, so the same trust boundary as live messages applies.
|
||||||
|
|
||||||
|
A peer can choose not to share history by ignoring `history_request` messages — the protocol is advisory, not mandatory.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Wire Protocol
|
||||||
|
|
||||||
|
Two new YAW/2 extension messages (added to EXTENSIONS.md):
|
||||||
|
|
||||||
|
### `history_request`
|
||||||
|
|
||||||
|
Sent by a newly-connected peer to one or more existing peers shortly after the handshake completes.
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"type": "history_request",
|
||||||
|
"room": "general",
|
||||||
|
"since": "2026-01-01T00:00:00Z",
|
||||||
|
"limit": 200
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
| Field | Type | Description |
|
||||||
|
|---------|-----------------|----------------------------------------------------------|
|
||||||
|
| `room` | string | Room name to request history for. One request per room. |
|
||||||
|
| `since` | ISO 8601 or "" | Only return messages newer than this timestamp. Empty = return up to `limit` most recent. |
|
||||||
|
| `limit` | int (max 500) | Maximum number of messages to return. Responder may return fewer. |
|
||||||
|
|
||||||
|
### `history_chunk`
|
||||||
|
|
||||||
|
Response from an existing peer. May be sent in multiple chunks if `limit` is large.
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"type": "history_chunk",
|
||||||
|
"room": "general",
|
||||||
|
"messages": [
|
||||||
|
{
|
||||||
|
"id": "sha256:<hex>",
|
||||||
|
"from": "<peer-alias>",
|
||||||
|
"from_id": "<hex-pubkey>",
|
||||||
|
"body": "hello",
|
||||||
|
"ts": "2026-06-01T12:00:00Z",
|
||||||
|
"room": "general"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"done": true
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
| Field | Type | Description |
|
||||||
|
|------------|---------|----------------------------------------------------------|
|
||||||
|
| `messages` | array | Ordered oldest-first. |
|
||||||
|
| `done` | bool | `true` on the final chunk. Receiver may display after this. |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Message Identity and Deduplication
|
||||||
|
|
||||||
|
Each message has a **content-addressed ID**:
|
||||||
|
|
||||||
|
```
|
||||||
|
id = "sha256:" + hex(SHA-256(from_id || room || ts || body))
|
||||||
|
```
|
||||||
|
|
||||||
|
- Computed by the original sender and included in every live message going forward
|
||||||
|
- The SQLite `messages` table gains an `id TEXT UNIQUE` column
|
||||||
|
- On insert, use `INSERT OR IGNORE` — receiving the same message twice (live or via gossip) is a no-op
|
||||||
|
- The UI sorts by `ts`, so late-arriving history slots in correctly without reordering visible messages
|
||||||
|
|
||||||
|
Legacy messages (before this feature) have no `id`. They are assigned a local-only ID on migration and are never gossipped (they have no canonical ID the receiver could deduplicate against).
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Daemon-Side Implementation
|
||||||
|
|
||||||
|
### SQLite schema change
|
||||||
|
|
||||||
|
```sql
|
||||||
|
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;
|
||||||
|
```
|
||||||
|
|
||||||
|
### Handling `history_request`
|
||||||
|
|
||||||
|
```
|
||||||
|
peer sends history_request{room, since, limit}
|
||||||
|
→ query SQLite: SELECT * FROM messages WHERE room=? AND ts>? ORDER BY ts ASC LIMIT ?
|
||||||
|
→ send history_chunk{room, messages, done:true}
|
||||||
|
```
|
||||||
|
|
||||||
|
Responder enforces:
|
||||||
|
- `limit` capped at 500
|
||||||
|
- Only messages the responder itself received or sent (no re-gossipping of gossipped history to avoid amplification)
|
||||||
|
- Rate limit: one `history_request` per peer per room per 60 seconds
|
||||||
|
|
||||||
|
### Handling `history_chunk`
|
||||||
|
|
||||||
|
```
|
||||||
|
for each message in chunk:
|
||||||
|
INSERT OR IGNORE INTO messages (msg_id, room, from_alias, from_id, body, ts) VALUES (...)
|
||||||
|
emit IPC event: history_loaded{room, count}
|
||||||
|
```
|
||||||
|
|
||||||
|
### When to request
|
||||||
|
|
||||||
|
- After handshake completes with the **first** peer in a network (only ask one peer — avoids fan-out)
|
||||||
|
- Request rooms the local peer knows about (from its own SQLite `rooms` table)
|
||||||
|
- If no rooms known yet: request `"general"` only; discover others from incoming live messages
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## IPC / UI Integration
|
||||||
|
|
||||||
|
New IPC event emitted after history is loaded:
|
||||||
|
|
||||||
|
```json
|
||||||
|
{ "type": "history_loaded", "network_id": "...", "room": "general", "count": 47 }
|
||||||
|
```
|
||||||
|
|
||||||
|
The web UI and TUI insert a visual separator above the first gossipped message:
|
||||||
|
|
||||||
|
```
|
||||||
|
── 47 earlier messages ─────────────────────────────
|
||||||
|
[12:03] alice: hey
|
||||||
|
[12:04] bob: yo
|
||||||
|
── live ─────────────────────────────────────────────
|
||||||
|
[14:22] you joined
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## What This Does Not Do
|
||||||
|
|
||||||
|
- **No conflict resolution** — messages are immutable append-only records; there is nothing to conflict
|
||||||
|
- **No ordering guarantee beyond timestamp** — if two peers sent messages at the same millisecond, both are stored; the UI sorts by `ts` then `msg_id` for a stable tiebreak
|
||||||
|
- **No full sync** — gossip is bounded by `limit` and `since`; it is not a replication protocol
|
||||||
|
- **No anchor involvement** — the anchor never sees history
|
||||||
|
- **No history from peers who were offline** — if no peer with history is online when you join, you get nothing (acceptable given the trust model)
|
||||||
@@ -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{
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user