try and implement firehose cursor for subscribeRepo

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()

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()
 

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")

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