// 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" "strings" "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 ); CREATE TABLE IF NOT EXISTS rooms ( name TEXT PRIMARY KEY, created_at DATETIME NOT NULL ); ` // migrations run after the base schema. ALTER TABLE ADD COLUMN fails with // "duplicate column name" on subsequent opens — we swallow that error. // CREATE TABLE IF NOT EXISTS and CREATE INDEX IF NOT EXISTS are idempotent. 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`, // Reactions: (mid, emoji, from_peer) triple is the unique dedup key. `CREATE TABLE IF NOT EXISTS reactions ( mid TEXT NOT NULL, emoji TEXT NOT NULL, from_peer TEXT NOT NULL, reacted_at DATETIME NOT NULL, PRIMARY KEY (mid, emoji, from_peer) )`, } // 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) } 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 } // 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 { sentAt := time.UnixMilli(msg.Ts).UTC() _, err := s.db.Exec( `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 } // 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 } // PeerAlias returns the cached alias for a peer, or "" if unknown. func (s *Store) PeerAlias(peerID proto.PeerID) string { var alias string s.db.QueryRow(`SELECT alias FROM peers WHERE peer_id = ?`, string(peerID)).Scan(&alias) //nolint:errcheck return alias } // RecentMessages returns up to limit messages for a room, oldest first. func (s *Store) RecentMessages(room string, limit int) ([]proto.ChatMessage, error) { return s.queryMessages( `SELECT mid, from_peer, room, body, sent_at FROM messages WHERE room = ? ORDER BY sent_at DESC LIMIT ?`, room, limit, ) } // RecentMessagesSince returns up to limit messages for a room with ts > sinceMs, oldest first. // sinceMs == 0 returns the most recent messages regardless of timestamp. func (s *Store) RecentMessagesSince(room string, sinceMs int64, limit int) ([]proto.ChatMessage, error) { if limit <= 0 || limit > 500 { limit = 500 } if sinceMs == 0 { return s.queryMessages( `SELECT mid, from_peer, room, body, sent_at FROM messages WHERE room = ? ORDER BY sent_at DESC LIMIT ?`, room, limit, ) } since := time.UnixMilli(sinceMs).UTC() return s.queryMessages( `SELECT mid, from_peer, room, body, sent_at FROM messages WHERE room = ? AND sent_at > ? 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 } 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.Room, &m.Text, &sentAt); err != nil { return nil, err } m.From = proto.PeerID(from) m.Ts = sentAt.UnixMilli() 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() } // SaveRoom persists a room name. Duplicate names are silently ignored. func (s *Store) SaveRoom(name string) error { _, err := s.db.Exec( `INSERT OR IGNORE INTO rooms (name, created_at) VALUES (?, ?)`, name, time.Now().UTC(), ) return err } // Rooms returns all persisted room names, ordered by creation time. func (s *Store) Rooms() ([]string, error) { rows, err := s.db.Query(`SELECT name FROM rooms ORDER BY created_at`) if err != nil { return nil, err } defer rows.Close() var out []string for rows.Next() { var name string if err := rows.Scan(&name); err != nil { return nil, err } out = append(out, name) } return out, 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() } // SaveReaction persists a reaction. Duplicate (mid, emoji, from_peer) triples are silently ignored. func (s *Store) SaveReaction(mid, emoji, fromPeer string) error { _, err := s.db.Exec( `INSERT OR IGNORE INTO reactions (mid, emoji, from_peer, reacted_at) VALUES (?, ?, ?, ?)`, mid, emoji, fromPeer, time.Now().UTC(), ) return err } // ReactionsForRoom returns all reactions for messages in a given room. // Result: mid → emoji → []fromPeer (ordered by reaction time). func (s *Store) ReactionsForRoom(room string) (map[string]map[string][]string, error) { rows, err := s.db.Query(` SELECT r.mid, r.emoji, r.from_peer FROM reactions r JOIN messages m ON m.mid = r.mid WHERE m.room = ? ORDER BY r.reacted_at `, room) if err != nil { return nil, err } defer rows.Close() out := make(map[string]map[string][]string) for rows.Next() { var mid, emoji, from string if err := rows.Scan(&mid, &emoji, &from); err != nil { return nil, err } if out[mid] == nil { out[mid] = make(map[string][]string) } out[mid][emoji] = append(out[mid][emoji], from) } return out, rows.Err() } func nullableString(s string) any { if s == "" { return nil } return s }