Add SQLite message and peer persistence (internal/store)
Each daemon writes to <data-dir>/messages.db on startup. Messages received or sent are stored immediately; duplicate mids (INSERT OR IGNORE) are safe to call multiple times. Peer aliases are upserted on peer_connected and again after hello verification when the real nick is known. Schema - messages(mid UNIQUE, room, from_peer, body, sent_at) — mid is the YAW/2 dedup key added in the proto migration; index on (room, sent_at) for efficient per-room queries. - peers(peer_id PK, alias, last_seen) — cache of every peer ever seen, used to resolve hex ids to names when peers are offline. Wiring - store.Open called in cmd/daemon/main.go, passed to mesh.New. - mesh.Mesh holds *store.Store (nil-safe; persistence is optional). - mesh.SaveMessage called in dispatchPeerMessage (incoming) and ipc CmdSendMessage (outgoing) so the local node's own messages are stored. - mesh.UpdatePeerAlias called after hello verification updates the alias with the verified nick rather than the placeholder short-id. Messages only accumulate from join time forward — no history replay to late-joining peers; each node's view starts from when it connected. test-network.sh: added SQLite verification block that queries each node's DB after the test and prints message + peer counts. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -7,12 +7,14 @@ import (
|
||||
"flag"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/waste-go/internal/anchor"
|
||||
"github.com/waste-go/internal/crypto"
|
||||
"github.com/waste-go/internal/ipc"
|
||||
"github.com/waste-go/internal/mesh"
|
||||
"github.com/waste-go/internal/proto"
|
||||
"github.com/waste-go/internal/store"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -22,13 +24,20 @@ func main() {
|
||||
anchorURL := flag.String("anchor", "", "anchor WebSocket URL, e.g. ws://your-vps:17339/ws")
|
||||
flag.Parse()
|
||||
|
||||
id, err := crypto.LoadOrCreate(expandHome(*dataDir), *alias)
|
||||
dir := expandHome(*dataDir)
|
||||
id, err := crypto.LoadOrCreate(dir, *alias)
|
||||
if err != nil {
|
||||
log.Fatalf("identity: %v", err)
|
||||
}
|
||||
log.Printf("daemon: local peer id: %s alias: %s", id.PeerID().Short(), id.Alias)
|
||||
|
||||
m := mesh.New(id)
|
||||
st, err := store.Open(filepath.Join(dir, "messages.db"))
|
||||
if err != nil {
|
||||
log.Fatalf("store: %v", err)
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
m := mesh.New(id, st)
|
||||
|
||||
// joinFn is passed to the IPC layer; it's called when the UI sends join_network.
|
||||
joinFn := func(ctx context.Context, networkName string) {
|
||||
|
||||
14
go.mod
14
go.mod
@@ -1,16 +1,20 @@
|
||||
module github.com/waste-go
|
||||
|
||||
go 1.24.0
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
filippo.io/edwards25519 v1.2.0
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/pion/webrtc/v3 v3.3.6
|
||||
golang.org/x/crypto v0.24.0
|
||||
nhooyr.io/websocket v1.8.17
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||
github.com/mattn/go-isatty v0.0.20 // indirect
|
||||
github.com/ncruces/go-strftime v1.0.0 // indirect
|
||||
github.com/pion/datachannel v1.5.8 // indirect
|
||||
github.com/pion/dtls/v2 v2.2.12 // indirect
|
||||
github.com/pion/ice/v2 v2.3.38 // indirect
|
||||
@@ -26,11 +30,15 @@ require (
|
||||
github.com/pion/stun v0.6.1 // indirect
|
||||
github.com/pion/transport/v2 v2.2.10 // indirect
|
||||
github.com/pion/turn/v2 v2.1.6 // indirect
|
||||
github.com/pion/webrtc/v3 v3.3.6 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
|
||||
github.com/stretchr/testify v1.9.0 // indirect
|
||||
github.com/wlynxg/anet v0.0.3 // indirect
|
||||
golang.org/x/net v0.22.0 // indirect
|
||||
golang.org/x/sys v0.21.0 // indirect
|
||||
golang.org/x/sys v0.44.0 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
modernc.org/libc v1.73.4 // indirect
|
||||
modernc.org/mathutil v1.7.1 // indirect
|
||||
modernc.org/memory v1.11.0 // indirect
|
||||
modernc.org/sqlite v1.53.0 // indirect
|
||||
)
|
||||
|
||||
27
go.sum
27
go.sum
@@ -3,12 +3,21 @@ filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
|
||||
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
|
||||
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
||||
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
|
||||
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
|
||||
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
|
||||
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
|
||||
github.com/pion/datachannel v1.5.8 h1:ph1P1NsGkazkjrvyMfhRBUAWMxugJjq2HfQifaOoSNo=
|
||||
github.com/pion/datachannel v1.5.8/go.mod h1:PgmdpoaNBLX9HNzNClmdki4DYW5JtI7Yibu8QzbL3tI=
|
||||
github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
|
||||
@@ -44,6 +53,8 @@ github.com/pion/transport/v2 v2.2.4/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLh
|
||||
github.com/pion/transport/v2 v2.2.10 h1:ucLBLE8nuxiHfvkFKnkDQRYWYfp8ejf4YBOPfaQpw6Q=
|
||||
github.com/pion/transport/v2 v2.2.10/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E=
|
||||
github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9SzK5f5xE0=
|
||||
github.com/pion/transport/v3 v3.0.2 h1:r+40RJR25S9w3jbA6/5uEPTzcdn7ncyU44RWCbHkLg4=
|
||||
github.com/pion/transport/v3 v3.0.2/go.mod h1:nIToODoOlb5If2jF9y2Igfx3PFYWfuXi37m0IlWa/D0=
|
||||
github.com/pion/turn/v2 v2.1.3/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
|
||||
github.com/pion/turn/v2 v2.1.6 h1:Xr2niVsiPTB0FPtt+yAWKFUkU1eotQbGgpTIld4x1Gc=
|
||||
github.com/pion/turn/v2 v2.1.6/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
|
||||
@@ -51,6 +62,8 @@ github.com/pion/webrtc/v3 v3.3.6 h1:7XAh4RPtlY1Vul6/GmZrv7z+NnxKA6If0KStXBI2ZLE=
|
||||
github.com/pion/webrtc/v3 v3.3.6/go.mod h1:zyN7th4mZpV27eXybfR/cnUf3J2DRy8zw/mdjD9JTNM=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
|
||||
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
|
||||
@@ -92,13 +105,14 @@ golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBc
|
||||
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
|
||||
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ=
|
||||
golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
|
||||
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
|
||||
@@ -119,9 +133,18 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc
|
||||
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
|
||||
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
modernc.org/libc v1.73.4 h1:+ra4Ui8ngyt8HDcO1FTDPWlkAh6yOdaO2yAoh8MddQA=
|
||||
modernc.org/libc v1.73.4/go.mod h1:DXZ3eO8qMCNn2SnmTNCiC71nJ9Rcq3PsnpU6Vc4rWK8=
|
||||
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
|
||||
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
|
||||
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
|
||||
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
|
||||
modernc.org/sqlite v1.53.0 h1:20WG8N9q4ji/dEqGk4uiI0c6OPjSeLTNYGFCc3+7c1M=
|
||||
modernc.org/sqlite v1.53.0/go.mod h1:xoEpOIpGrgT48H5iiyt/YXPCZPEzlfmfFwtk8Lklw8s=
|
||||
nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y=
|
||||
nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=
|
||||
|
||||
@@ -161,6 +161,7 @@ func handleClient(conn net.Conn, m *mesh.Mesh, doJoin func(string), doLeave func
|
||||
continue
|
||||
}
|
||||
m.Broadcast(payload)
|
||||
m.SaveMessage(msg)
|
||||
m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: msg})
|
||||
|
||||
case proto.CmdJoinNetwork:
|
||||
|
||||
@@ -2,10 +2,12 @@
|
||||
package mesh
|
||||
|
||||
import (
|
||||
"log"
|
||||
"sync"
|
||||
|
||||
"github.com/waste-go/internal/crypto"
|
||||
"github.com/waste-go/internal/proto"
|
||||
"github.com/waste-go/internal/store"
|
||||
)
|
||||
|
||||
// PeerConn is a live connection to one peer.
|
||||
@@ -19,6 +21,7 @@ type PeerConn struct {
|
||||
// All methods are safe to call from multiple goroutines.
|
||||
type Mesh struct {
|
||||
Identity *crypto.Identity
|
||||
Store *store.Store // may be nil if persistence is disabled
|
||||
|
||||
mu sync.RWMutex
|
||||
peers map[proto.PeerID]*PeerConn
|
||||
@@ -29,9 +32,11 @@ type Mesh struct {
|
||||
}
|
||||
|
||||
// New creates an empty mesh with the given identity.
|
||||
func New(id *crypto.Identity) *Mesh {
|
||||
// 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),
|
||||
}
|
||||
}
|
||||
@@ -44,12 +49,38 @@ func (m *Mesh) AddPeer(conn *PeerConn) {
|
||||
m.peers[conn.Info.ID] = conn
|
||||
m.mu.Unlock()
|
||||
|
||||
if m.Store != nil && conn.Info.Alias != "" {
|
||||
if err := m.Store.SavePeer(conn.Info.ID, conn.Info.Alias); err != nil {
|
||||
log.Printf("mesh: store peer %s: %v", conn.Info.ID.Short(), err)
|
||||
}
|
||||
}
|
||||
|
||||
m.emit(proto.IpcMessage{
|
||||
Type: proto.EvtPeerConnected,
|
||||
Peer: &conn.Info,
|
||||
})
|
||||
}
|
||||
|
||||
// SaveMessage persists a chat message if a store is configured.
|
||||
// Duplicate mids are silently dropped.
|
||||
func (m *Mesh) SaveMessage(msg *proto.ChatMessage) {
|
||||
if m.Store == nil || msg == nil {
|
||||
return
|
||||
}
|
||||
if err := m.Store.SaveMessage(msg); err != nil {
|
||||
log.Printf("mesh: store message %s: %v", msg.Mid, err)
|
||||
}
|
||||
}
|
||||
|
||||
// UpdatePeerAlias updates the cached alias for a peer after hello verification.
|
||||
func (m *Mesh) UpdatePeerAlias(id proto.PeerID, alias string) {
|
||||
if m.Store != nil && alias != "" {
|
||||
if err := m.Store.SavePeer(id, alias); err != nil {
|
||||
log.Printf("mesh: update peer alias %s: %v", id.Short(), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RemovePeer unregisters a peer and notifies subscribers.
|
||||
func (m *Mesh) RemovePeer(id proto.PeerID) {
|
||||
m.mu.Lock()
|
||||
|
||||
@@ -128,6 +128,7 @@ func handleDCMessage(data []byte, from proto.PeerID, localID *crypto.Identity, m
|
||||
conn.Info.PublicKey = hello.ID
|
||||
}
|
||||
m.mu.Unlock()
|
||||
m.UpdatePeerAlias(from, hello.Nick)
|
||||
m.Emit(proto.IpcMessage{
|
||||
Type: proto.EvtSessionReady,
|
||||
PeerID: peerIDPtr(from),
|
||||
@@ -148,6 +149,7 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
|
||||
switch msg.Type {
|
||||
case proto.MsgChat:
|
||||
if msg.Chat != nil {
|
||||
m.SaveMessage(msg.Chat)
|
||||
m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: msg.Chat})
|
||||
}
|
||||
case proto.MsgPeerGossip:
|
||||
|
||||
130
internal/store/store.go
Normal file
130
internal/store/store.go
Normal file
@@ -0,0 +1,130 @@
|
||||
// Package store persists messages and peer info to a local SQLite database.
|
||||
// Each daemon has its own database; nothing is shared across the network.
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
|
||||
"github.com/waste-go/internal/proto"
|
||||
)
|
||||
|
||||
const schema = `
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
mid TEXT NOT NULL UNIQUE,
|
||||
room TEXT NOT NULL,
|
||||
from_peer TEXT NOT NULL,
|
||||
body TEXT NOT NULL,
|
||||
sent_at DATETIME NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_room ON messages (room, sent_at);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS peers (
|
||||
peer_id TEXT PRIMARY KEY,
|
||||
alias TEXT NOT NULL,
|
||||
last_seen DATETIME NOT NULL
|
||||
);
|
||||
`
|
||||
|
||||
// Store is a local SQLite-backed message and peer store.
|
||||
type Store struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
// Open opens (or creates) the SQLite database at path and runs migrations.
|
||||
func Open(path string) (*Store, error) {
|
||||
db, err := sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open db: %w", err)
|
||||
}
|
||||
db.SetMaxOpenConns(1) // SQLite is single-writer
|
||||
if _, err := db.Exec(schema); err != nil {
|
||||
db.Close()
|
||||
return nil, fmt.Errorf("migrate db: %w", err)
|
||||
}
|
||||
return &Store{db: db}, nil
|
||||
}
|
||||
|
||||
// Close releases the database connection.
|
||||
func (s *Store) Close() error {
|
||||
return s.db.Close()
|
||||
}
|
||||
|
||||
// SaveMessage persists a chat message. Duplicate mids are silently ignored
|
||||
// (INSERT OR IGNORE), so calling this more than once is safe.
|
||||
func (s *Store) SaveMessage(msg *proto.ChatMessage) error {
|
||||
_, 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.Body, msg.SentAt.UTC(),
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// SavePeer upserts a peer's alias and last-seen timestamp.
|
||||
func (s *Store) SavePeer(peerID proto.PeerID, alias string) error {
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO peers (peer_id, alias, last_seen)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(peer_id) DO UPDATE SET alias=excluded.alias, last_seen=excluded.last_seen`,
|
||||
string(peerID), alias, time.Now().UTC(),
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// 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
|
||||
WHERE room = ?
|
||||
ORDER BY sent_at DESC
|
||||
LIMIT ?`,
|
||||
room, limit,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var msgs []proto.ChatMessage
|
||||
for rows.Next() {
|
||||
var m proto.ChatMessage
|
||||
var from string
|
||||
var sentAt time.Time
|
||||
if err := rows.Scan(&m.Mid, &from, &m.Body, &sentAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m.From = proto.PeerID(from)
|
||||
m.Room = room
|
||||
m.SentAt = sentAt
|
||||
msgs = append(msgs, m)
|
||||
}
|
||||
// Reverse so oldest-first.
|
||||
for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 {
|
||||
msgs[i], msgs[j] = msgs[j], msgs[i]
|
||||
}
|
||||
return msgs, rows.Err()
|
||||
}
|
||||
|
||||
// KnownPeers returns all peers seen since this daemon started storing data.
|
||||
func (s *Store) KnownPeers() (map[proto.PeerID]string, error) {
|
||||
rows, err := s.db.Query(`SELECT peer_id, alias FROM peers`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make(map[proto.PeerID]string)
|
||||
for rows.Next() {
|
||||
var id, alias string
|
||||
if err := rows.Scan(&id, &alias); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[proto.PeerID(id)] = alias
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
103
internal/store/store_test.go
Normal file
103
internal/store/store_test.go
Normal file
@@ -0,0 +1,103 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/waste-go/internal/proto"
|
||||
)
|
||||
|
||||
func TestRoundTrip(t *testing.T) {
|
||||
st, err := Open(filepath.Join(t.TempDir(), "test.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
msg := &proto.ChatMessage{
|
||||
Mid: "aabbccdd00112233",
|
||||
From: proto.PeerID("deadbeef"),
|
||||
Room: "general",
|
||||
Body: "hello world",
|
||||
SentAt: time.Now().UTC().Truncate(time.Second),
|
||||
}
|
||||
|
||||
if err := st.SaveMessage(msg); err != nil {
|
||||
t.Fatalf("SaveMessage: %v", err)
|
||||
}
|
||||
|
||||
// Duplicate mid must be silently ignored.
|
||||
if err := st.SaveMessage(msg); err != nil {
|
||||
t.Fatalf("duplicate SaveMessage: %v", err)
|
||||
}
|
||||
|
||||
msgs, err := st.RecentMessages("general", 10)
|
||||
if err != nil {
|
||||
t.Fatalf("RecentMessages: %v", err)
|
||||
}
|
||||
if len(msgs) != 1 {
|
||||
t.Fatalf("got %d messages, want 1", len(msgs))
|
||||
}
|
||||
got := msgs[0]
|
||||
if got.Mid != msg.Mid || got.Body != msg.Body || got.Room != msg.Room {
|
||||
t.Fatalf("message mismatch: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSavePeerAndKnownPeers(t *testing.T) {
|
||||
st, err := Open(filepath.Join(t.TempDir(), "test.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
if err := st.SavePeer("aabbccdd", "alice"); err != nil {
|
||||
t.Fatalf("SavePeer: %v", err)
|
||||
}
|
||||
// Upsert with updated alias.
|
||||
if err := st.SavePeer("aabbccdd", "alice-updated"); err != nil {
|
||||
t.Fatalf("SavePeer upsert: %v", err)
|
||||
}
|
||||
|
||||
peers, err := st.KnownPeers()
|
||||
if err != nil {
|
||||
t.Fatalf("KnownPeers: %v", err)
|
||||
}
|
||||
if peers["aabbccdd"] != "alice-updated" {
|
||||
t.Fatalf("alias = %q, want %q", peers["aabbccdd"], "alice-updated")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecentMessagesOrdering(t *testing.T) {
|
||||
st, err := Open(filepath.Join(t.TempDir(), "test.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
base := time.Now().UTC().Truncate(time.Second)
|
||||
for i := range 5 {
|
||||
st.SaveMessage(&proto.ChatMessage{
|
||||
Mid: string(rune('a'+i)) + "000000000000000",
|
||||
From: "deadbeef",
|
||||
Room: "general",
|
||||
Body: string(rune('a' + i)),
|
||||
SentAt: base.Add(time.Duration(i) * time.Second),
|
||||
})
|
||||
}
|
||||
|
||||
msgs, err := st.RecentMessages("general", 10)
|
||||
if err != nil {
|
||||
t.Fatalf("RecentMessages: %v", err)
|
||||
}
|
||||
if len(msgs) != 5 {
|
||||
t.Fatalf("got %d messages, want 5", len(msgs))
|
||||
}
|
||||
// Must be oldest-first.
|
||||
for i := 1; i < len(msgs); i++ {
|
||||
if msgs[i].SentAt.Before(msgs[i-1].SentAt) {
|
||||
t.Fatalf("messages not in ascending order at index %d", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -244,6 +244,21 @@ echo -e "${DIM}─────────────────────
|
||||
ipc "$CHARLIE_IPC" '{"type":"leave_network"}'
|
||||
sleep 1
|
||||
|
||||
echo ""
|
||||
echo -e "${DIM}────────────────────────────────────────────────────────${RESET}"
|
||||
echo -e "verifying persistence (SQLite)"
|
||||
echo -e "${DIM}────────────────────────────────────────────────────────${RESET}"
|
||||
for peer in alice bob charlie; do
|
||||
db="$DATA_ROOT/$peer/messages.db"
|
||||
if [ -f "$db" ]; then
|
||||
count=$(sqlite3 "$db" "SELECT COUNT(*) FROM messages;" 2>/dev/null || echo "?")
|
||||
peers=$(sqlite3 "$db" "SELECT COUNT(*) FROM peers;" 2>/dev/null || echo "?")
|
||||
echo -e " ${BOLD}${peer}${RESET}: ${count} messages, ${peers} known peers in messages.db"
|
||||
else
|
||||
echo -e " ${RED}${peer}: no messages.db found${RESET}"
|
||||
fi
|
||||
done
|
||||
|
||||
echo ""
|
||||
echo -e "${DIM}────────────────────────────────────────────────────────${RESET}"
|
||||
echo -e "${GREEN}${BOLD}test complete${RESET}"
|
||||
|
||||
Reference in New Issue
Block a user