diff --git a/cmd/hold/main.go b/cmd/hold/main.go index 558f8de..04e7ce9 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -61,7 +61,19 @@ func main() { } // Create event broadcaster for subscribeRepos firehose - broadcaster = pds.NewEventBroadcaster(holdDID, 100) // Keep 100 events for backfill + // Database path: carstore creates db.sqlite3 inside cfg.Database.Path + var dbPath string + if cfg.Database.Path != ":memory:" { + dbPath = cfg.Database.Path + "/db.sqlite3" + } else { + dbPath = ":memory:" + } + broadcaster = pds.NewEventBroadcaster(holdDID, 100, dbPath) + + // Bootstrap events from existing repo records (one-time migration) + if err := broadcaster.BootstrapFromRepo(holdPDS); err != nil { + log.Printf("Warning: Failed to bootstrap events from repo: %v", err) + } // Wire up repo event handler to broadcaster holdPDS.RepomgrRef().SetEventHandler(broadcaster.SetRepoEventHandler(), true) @@ -175,6 +187,15 @@ func main() { } } + // Close broadcaster database connection + if broadcaster != nil { + if err := broadcaster.Close(); err != nil { + log.Printf("Warning: Failed to close broadcaster database: %v", err) + } else { + log.Printf("Broadcaster database closed") + } + } + // Graceful shutdown with 10 second timeout shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() diff --git a/pkg/hold/pds/events.go b/pkg/hold/pds/events.go index 8570efd..fd37d89 100644 --- a/pkg/hold/pds/events.go +++ b/pkg/hold/pds/events.go @@ -1,17 +1,25 @@ package pds import ( + "bytes" "context" + "database/sql" "encoding/json" + "fmt" "log" + "strings" "sync" "time" atproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/events" lexutil "github.com/bluesky-social/indigo/lex/util" + "github.com/bluesky-social/indigo/repo" "github.com/gorilla/websocket" "github.com/ipfs/go-cid" + "github.com/ipld/go-car" + carutil "github.com/ipld/go-car/util" + _ "github.com/mattn/go-sqlite3" ) // EventBroadcaster manages WebSocket connections and broadcasts repo events @@ -19,9 +27,11 @@ type EventBroadcaster struct { mu sync.RWMutex subscribers map[*Subscriber]bool eventSeq int64 - eventHistory []HistoricalEvent // Ring buffer for cursor backfill + eventHistory []HistoricalEvent // Ring buffer for cursor backfill (deprecated, kept for compatibility) maxHistory int - holdDID string // DID of the hold for setting repo field + holdDID string // DID of the hold for setting repo field + db *sql.DB // Database for persistent event storage + dbPath string // Path to database file } // Subscriber represents a WebSocket client subscribed to the firehose @@ -50,19 +60,229 @@ type RepoCommitEvent struct { Type string `json:"$type" cborgen:"$type"` // Always "#commit" } -// NewEventBroadcaster creates a new event broadcaster -func NewEventBroadcaster(holdDID string, maxHistory int) *EventBroadcaster { +// NewEventBroadcaster creates a new event broadcaster with persistent storage +// dbPath should point to the carstore database file (e.g., "/path/to/pds/db.sqlite3") +func NewEventBroadcaster(holdDID string, maxHistory int, dbPath string) *EventBroadcaster { if maxHistory <= 0 { maxHistory = 100 // Default to keeping 100 events } - return &EventBroadcaster{ + broadcaster := &EventBroadcaster{ subscribers: make(map[*Subscriber]bool), eventSeq: 0, eventHistory: make([]HistoricalEvent, 0, maxHistory), maxHistory: maxHistory, holdDID: holdDID, + dbPath: dbPath, } + + // Initialize database connection and schema + if dbPath != "" && dbPath != ":memory:" { + if err := broadcaster.initDatabase(); err != nil { + log.Printf("Warning: Failed to initialize event database: %v", err) + log.Printf("Events will not persist across restarts") + } + } + + return broadcaster +} + +// initDatabase opens database connection, creates table, and loads last sequence +func (b *EventBroadcaster) initDatabase() error { + // Open database connection + db, err := sql.Open("sqlite3", b.dbPath) + if err != nil { + return err + } + + // Test connection + if err := db.Ping(); err != nil { + db.Close() + return err + } + + b.db = db + + // Create events table if it doesn't exist + schema := ` + CREATE TABLE IF NOT EXISTS firehose_events ( + seq INTEGER PRIMARY KEY, + commit_cid TEXT NOT NULL, + rev TEXT NOT NULL, + since_rev TEXT, + repo_slice BLOB NOT NULL, + ops_json TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + CREATE INDEX IF NOT EXISTS idx_firehose_events_rev ON firehose_events(rev); + ` + + if _, err := db.Exec(schema); err != nil { + db.Close() + b.db = nil + return err + } + + // Load last sequence number from database + var lastSeq sql.NullInt64 + err = db.QueryRow("SELECT MAX(seq) FROM firehose_events").Scan(&lastSeq) + if err != nil { + log.Printf("Warning: Failed to load last event sequence: %v", err) + } else if lastSeq.Valid { + b.eventSeq = lastSeq.Int64 + log.Printf("Loaded event sequence from database: seq=%d", b.eventSeq) + } else { + // Database is empty but might have existing repo records + // This happens on first deployment after adding persistent events + log.Printf("No events in database - will bootstrap from repo if needed") + } + + return nil +} + +// BootstrapFromRepo generates synthetic events from all current records in the repo +// This is called once when deploying persistent events to an existing repo +func (b *EventBroadcaster) BootstrapFromRepo(pds *HoldPDS) error { + if b.db == nil { + return fmt.Errorf("database not initialized") + } + + // Check if we already have events + var count int64 + if err := b.db.QueryRow("SELECT COUNT(*) FROM firehose_events").Scan(&count); err != nil { + return fmt.Errorf("failed to check event count: %w", err) + } + + if count > 0 { + log.Printf("Database already has %d events, skipping bootstrap", count) + return nil + } + + ctx := context.Background() + + // Get current repo state + session, err := pds.carstore.ReadOnlySession(pds.uid) + if err != nil { + return fmt.Errorf("failed to create session: %w", err) + } + + head, err := pds.carstore.GetUserRepoHead(ctx, pds.uid) + if err != nil || !head.Defined() { + // Empty repo, nothing to bootstrap + log.Printf("Empty repo, no events to bootstrap") + return nil + } + + repoHandle, err := repo.OpenRepo(ctx, session, head) + if err != nil { + return fmt.Errorf("failed to open repo: %w", err) + } + + // Get current rev + rev, err := pds.repomgr.GetRepoRev(ctx, pds.uid) + if err != nil { + return fmt.Errorf("failed to get repo rev: %w", err) + } + + log.Printf("Bootstrapping firehose events from current repo state (head=%s, rev=%s)", head.String(), rev) + + var recordCount int64 + + // Walk all records in the repo and create synthetic events + err = repoHandle.ForEach(ctx, "", func(path string, recordCID cid.Cid) error { + // Get record value + _, recBytes, err := repoHandle.GetRecordBytes(ctx, path) + if err != nil { + log.Printf("Warning: failed to get record bytes for %s: %v", path, err) + return nil // Skip this record but continue + } + + recordValue, err := lexutil.CborDecodeValue(*recBytes) + if err != nil { + log.Printf("Warning: failed to decode record %s: %v", path, err) + return nil + } + + // Parse collection and rkey from path + parts := strings.Split(path, "/") + if len(parts) < 2 { + return nil // Invalid path + } + + collection := strings.Join(parts[:len(parts)-1], "/") + rkey := parts[len(parts)-1] + + // Create synthetic RepoOp + ops := []RepoOp{ + { + Kind: EvtKindCreateRecord, + Collection: collection, + Rkey: rkey, + RecCid: &recordCID, + Record: recordValue, + }, + } + + // Get CAR slice for this record (minimal - just the record block) + var carBuf bytes.Buffer + carHeader := &car.CarHeader{ + Roots: []cid.Cid{head}, + Version: 1, + } + if err := car.WriteHeader(carHeader, &carBuf); err != nil { + log.Printf("Warning: failed to write CAR header: %v", err) + return nil + } + + // Write the record block + if err := carutil.LdWrite(&carBuf, recordCID.Bytes(), *recBytes); err != nil { + log.Printf("Warning: failed to write record block: %v", err) + return nil + } + + // Create synthetic RepoEvent + repoEvent := &RepoEvent{ + NewRoot: head, + Rev: rev, + Since: nil, // No "since" for bootstrap events + RepoSlice: carBuf.Bytes(), + Ops: ops, + } + + // Convert to commit event and persist + b.mu.Lock() + b.eventSeq++ + seq := b.eventSeq + commitEvent := b.convertToCommitEvent(repoEvent, seq) + + // Persist to database + if err := b.persistEvent(commitEvent); err != nil { + b.mu.Unlock() + return fmt.Errorf("failed to persist bootstrap event seq=%d: %w", seq, err) + } + + // Also add to in-memory history for immediate use + b.addToHistory(seq, commitEvent) + b.mu.Unlock() + + recordCount++ + return nil + }) + + if err != nil { + return fmt.Errorf("failed to walk repo: %w", err) + } + + log.Printf("✅ Bootstrapped %d events from repo (seq now at %d)", recordCount, b.eventSeq) + return nil +} + +// Close closes the database connection +func (b *EventBroadcaster) Close() error { + if b.db != nil { + return b.db.Close() + } + return nil } // Subscribe adds a new WebSocket subscriber @@ -78,11 +298,23 @@ func (b *EventBroadcaster) Subscribe(conn *websocket.Conn, cursor int64) *Subscr currentSeq := b.eventSeq b.mu.Unlock() - // Send historical events if cursor is provided and < current seq - // cursor=0 means "replay all events from the beginning" - // cursor >= 0 triggers backfill, negative cursor means "no backfill" - if cursor >= 0 && cursor < currentSeq { - go b.backfillSubscriber(sub, cursor) + // Handle cursor-based backfill: + // - cursor < 0: No backfill, stream new events only + // - cursor >= 0: Backfill events from cursor onwards + // - cursor=0: Replay all events from beginning + // - cursor < currentSeq: Normal backfill + // - cursor >= currentSeq: Relay reconnecting after our restart, backfill from database + if cursor >= 0 { + if cursor < currentSeq { + // Normal case: relay is behind, backfill missing events + go b.backfillSubscriber(sub, cursor) + } else if cursor > currentSeq { + // Relay has cursor ahead of us - server was restarted + // Database should have the events if we had them before + log.Printf("Relay cursor %d > currentSeq %d (server restarted), attempting database backfill", cursor, currentSeq) + go b.backfillSubscriber(sub, cursor) + } + // else cursor == currentSeq: relay is caught up, just stream new events } // Start goroutine to handle sending events to this subscriber @@ -114,7 +346,14 @@ func (b *EventBroadcaster) Broadcast(ctx context.Context, event *RepoEvent) { // Convert RepoEvent to RepoCommitEvent commitEvent := b.convertToCommitEvent(event, seq) - // Store in history for backfill + // Persist event to database + if b.db != nil { + if err := b.persistEvent(commitEvent); err != nil { + log.Printf("Warning: Failed to persist event seq=%d to database: %v", seq, err) + } + } + + // Store in history for backfill (deprecated, but kept for compatibility) b.addToHistory(seq, commitEvent) // Broadcast to all subscribers @@ -129,6 +368,29 @@ func (b *EventBroadcaster) Broadcast(ctx context.Context, event *RepoEvent) { } } +// persistEvent stores an event in the database +func (b *EventBroadcaster) persistEvent(event *RepoCommitEvent) error { + // Serialize ops to JSON + opsJSON, err := json.Marshal(event.Ops) + if err != nil { + return err + } + + // Get since_rev value (may be nil) + var sinceRev sql.NullString + if event.Since != nil { + sinceRev = sql.NullString{String: *event.Since, Valid: true} + } + + // Insert event + query := ` + INSERT INTO firehose_events (seq, commit_cid, rev, since_rev, repo_slice, ops_json) + VALUES (?, ?, ?, ?, ?, ?) + ` + _, err = b.db.Exec(query, event.Seq, event.Commit, event.Rev, sinceRev, event.Blocks, opsJSON) + return err +} + // convertToCommitEvent converts a RepoEvent to a RepoCommitEvent func (b *EventBroadcaster) convertToCommitEvent(event *RepoEvent, seq int64) *RepoCommitEvent { // Convert RepoOps to atproto.SyncSubscribeRepos_RepoOp @@ -183,7 +445,93 @@ func (b *EventBroadcaster) addToHistory(seq int64, event *RepoCommitEvent) { } // backfillSubscriber sends historical events to a subscriber +// Query events from database where seq > cursor func (b *EventBroadcaster) backfillSubscriber(sub *Subscriber, cursor int64) { + // If database is available, use it for backfill + if b.db != nil { + if err := b.backfillFromDatabase(sub, cursor); err != nil { + log.Printf("Database backfill failed, falling back to in-memory: %v", err) + b.backfillFromMemory(sub, cursor) + } + return + } + + // Fall back to in-memory backfill + b.backfillFromMemory(sub, cursor) +} + +// backfillFromDatabase queries events from database and sends to subscriber +func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) error { + // Query events where seq > cursor, ordered by seq + query := ` + SELECT seq, commit_cid, rev, since_rev, repo_slice, ops_json + FROM firehose_events + WHERE seq > ? + ORDER BY seq ASC + ` + + rows, err := b.db.Query(query, cursor) + if err != nil { + return err + } + defer rows.Close() + + for rows.Next() { + var ( + seq int64 + commitCID string + rev string + sinceRev sql.NullString + repoSlice []byte + opsJSON []byte + ) + + if err := rows.Scan(&seq, &commitCID, &rev, &sinceRev, &repoSlice, &opsJSON); err != nil { + log.Printf("Error scanning event row: %v", err) + continue + } + + // Deserialize ops from JSON + var ops []*atproto.SyncSubscribeRepos_RepoOp + if err := json.Unmarshal(opsJSON, &ops); err != nil { + log.Printf("Error unmarshaling ops for seq=%d: %v", seq, err) + continue + } + + // Reconstruct event + var since *string + if sinceRev.Valid { + since = &sinceRev.String + } + + event := &RepoCommitEvent{ + Seq: seq, + Repo: b.holdDID, + Commit: commitCID, + Rev: rev, + Since: since, + Blocks: repoSlice, + Ops: ops, + Time: time.Now().Format(time.RFC3339), + Type: "#commit", + } + + // Send to subscriber + select { + case sub.send <- event: + // Sent successfully + case <-time.After(5 * time.Second): + // Timeout, subscriber too slow + log.Printf("Backfill timeout for subscriber at seq=%d", seq) + return nil + } + } + + return rows.Err() +} + +// backfillFromMemory sends events from in-memory ring buffer (fallback) +func (b *EventBroadcaster) backfillFromMemory(sub *Subscriber, cursor int64) { b.mu.RLock() defer b.mu.RUnlock() diff --git a/pkg/hold/pds/events_test.go b/pkg/hold/pds/events_test.go index 8cd9f5b..9634d5e 100644 --- a/pkg/hold/pds/events_test.go +++ b/pkg/hold/pds/events_test.go @@ -13,7 +13,7 @@ import ( // TestNewEventBroadcaster tests event broadcaster creation func TestNewEventBroadcaster(t *testing.T) { holdDID := "did:web:hold.example.com" - broadcaster := NewEventBroadcaster(holdDID, 100) + broadcaster := NewEventBroadcaster(holdDID, 100, "") if broadcaster.holdDID != holdDID { t.Errorf("Expected holdDID=%s, got %s", holdDID, broadcaster.holdDID) @@ -35,12 +35,12 @@ func TestNewEventBroadcaster(t *testing.T) { // TestNewEventBroadcaster_DefaultHistory tests default history size func TestNewEventBroadcaster_DefaultHistory(t *testing.T) { // Zero or negative maxHistory should default to 100 - broadcaster := NewEventBroadcaster("did:web:test", 0) + broadcaster := NewEventBroadcaster("did:web:test", 0, "") if broadcaster.maxHistory != 100 { t.Errorf("Expected default maxHistory=100 for input 0, got %d", broadcaster.maxHistory) } - broadcaster2 := NewEventBroadcaster("did:web:test", -5) + broadcaster2 := NewEventBroadcaster("did:web:test", -5, "") if broadcaster2.maxHistory != 100 { t.Errorf("Expected default maxHistory=100 for negative input, got %d", broadcaster2.maxHistory) } @@ -48,7 +48,7 @@ func TestNewEventBroadcaster_DefaultHistory(t *testing.T) { // TestGetCurrentSeq tests sequence number tracking func TestGetCurrentSeq(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") // Initial seq should be 0 seq := broadcaster.GetCurrentSeq() @@ -91,7 +91,7 @@ func TestGetCurrentSeq(t *testing.T) { // TestBroadcast tests event broadcasting func TestBroadcast(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") ctx := context.Background() testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") @@ -144,7 +144,7 @@ func TestBroadcast(t *testing.T) { // TestAddToHistory_RingBuffer tests ring buffer behavior func TestAddToHistory_RingBuffer(t *testing.T) { // Create broadcaster with small history - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 3) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 3, "") ctx := context.Background() testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") @@ -181,7 +181,7 @@ func TestAddToHistory_RingBuffer(t *testing.T) { // TestConvertToCommitEvent tests event conversion func TestConvertToCommitEvent(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") since := "prev-rev" @@ -296,7 +296,7 @@ func TestConvertToCommitEvent(t *testing.T) { // TestConvertToCommitEvent_NoSince tests event without since field func TestConvertToCommitEvent_NoSince(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") @@ -317,7 +317,7 @@ func TestConvertToCommitEvent_NoSince(t *testing.T) { // TestSetRepoEventHandler tests handler registration func TestSetRepoEventHandler(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") handler := broadcaster.SetRepoEventHandler() if handler == nil { @@ -385,7 +385,7 @@ func TestEncodeCBOR(t *testing.T) { // TestSubscribe_CursorZeroBackfill tests that cursor=0 replays all events func TestSubscribe_CursorZeroBackfill(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "") ctx := context.Background() testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") @@ -456,7 +456,7 @@ func TestSubscribe_CursorZeroBackfill(t *testing.T) { // TestSubscribe_MidCursorBackfill tests that cursor=N only gets events after N func TestSubscribe_MidCursorBackfill(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "") ctx := context.Background() testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") @@ -506,7 +506,7 @@ func TestSubscribe_MidCursorBackfill(t *testing.T) { // TestSubscribe_NegativeCursorNoBackfill tests that negative cursor means no backfill func TestSubscribe_NegativeCursorNoBackfill(t *testing.T) { - broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100) + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "") ctx := context.Background() testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") diff --git a/pkg/hold/pds/xrpc_test.go b/pkg/hold/pds/xrpc_test.go index 84e423d..a51eed1 100644 --- a/pkg/hold/pds/xrpc_test.go +++ b/pkg/hold/pds/xrpc_test.go @@ -2059,7 +2059,7 @@ func TestHandleSubscribeRepos(t *testing.T) { handler, ctx := setupTestXRPCHandler(t) // Create EventBroadcaster - broadcaster := NewEventBroadcaster(handler.pds.DID(), 100) + broadcaster := NewEventBroadcaster(handler.pds.DID(), 100, "") handler.broadcaster = broadcaster // Set up test HTTP server