diff --git a/cmd/appview/serve.go b/cmd/appview/serve.go index 5668239..dda4465 100644 --- a/cmd/appview/serve.go +++ b/cmd/appview/serve.go @@ -601,11 +601,14 @@ func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.S jetstreamURL = "wss://jetstream2.us-west.bsky.network/subscribe" } - // Start real-time Jetstream worker (no cursor = start from now) - worker := jetstream.NewWorker(database, jetstreamURL, 0) + // Start real-time Jetstream worker with cursor tracking for reconnects go func() { + var lastCursor int64 = 0 // Start from now on first connect for { + worker := jetstream.NewWorker(database, jetstreamURL, lastCursor) if err := worker.Start(context.Background()); err != nil { + // Save cursor from this connection for next reconnect + lastCursor = worker.GetLastCursor() fmt.Printf("Jetstream: Real-time worker error: %v, reconnecting in 10s...\n", err) time.Sleep(10 * time.Second) } @@ -613,8 +616,8 @@ func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.S }() fmt.Println("Jetstream: Real-time worker started") - // Start backfill worker if enabled - if backfillEnabled := os.Getenv("ATCR_BACKFILL_ENABLED"); backfillEnabled == "true" { + // Start backfill worker (enabled by default, set ATCR_BACKFILL_ENABLED=false to disable) + if backfillEnabled := os.Getenv("ATCR_BACKFILL_ENABLED"); backfillEnabled != "false" { // Get relay endpoint for sync API (defaults to Bluesky's relay) relayEndpoint := os.Getenv("ATCR_RELAY_ENDPOINT") if relayEndpoint == "" { diff --git a/docker-compose.yml b/docker-compose.yml index 75f34ad..520c55b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,7 +21,7 @@ services: ATCR_LOG_LEVEL: info volumes: # Auth keys (JWT signing keys) - - atcr-auth:/var/lib/atcr/auth + # - atcr-auth:/var/lib/atcr/auth # UI database (includes OAuth sessions, devices, and Jetstream cache) - atcr-ui:/var/lib/atcr restart: unless-stopped diff --git a/pkg/appview/db/schema.go b/pkg/appview/db/schema.go index 601f884..172b32c 100644 --- a/pkg/appview/db/schema.go +++ b/pkg/appview/db/schema.go @@ -80,20 +80,6 @@ CREATE TABLE IF NOT EXISTS tags ( ); CREATE INDEX IF NOT EXISTS idx_tags_did_repo ON tags(did, repository); -CREATE TABLE IF NOT EXISTS firehose_cursor ( - id INTEGER PRIMARY KEY CHECK (id = 1), - cursor INTEGER NOT NULL, - updated_at TIMESTAMP NOT NULL -); - -CREATE TABLE IF NOT EXISTS backfill_state ( - id INTEGER PRIMARY KEY CHECK (id = 1), - start_cursor INTEGER NOT NULL, - current_cursor INTEGER NOT NULL, - completed BOOLEAN NOT NULL DEFAULT 0, - updated_at TIMESTAMP NOT NULL -); - CREATE TABLE IF NOT EXISTS oauth_sessions ( session_key TEXT PRIMARY KEY, account_did TEXT NOT NULL, diff --git a/pkg/appview/jetstream/worker.go b/pkg/appview/jetstream/worker.go index 2e5818a..54c95ab 100644 --- a/pkg/appview/jetstream/worker.go +++ b/pkg/appview/jetstream/worker.go @@ -44,6 +44,10 @@ type Worker struct { pongsReceived int64 lastPongTime time.Time pongMutex sync.Mutex + + // In-memory cursor tracking for reconnects + lastCursor int64 + cursorMutex sync.RWMutex } // NewWorker creates a new Jetstream worker @@ -83,10 +87,14 @@ func (w *Worker) Start(ctx context.Context) error { q.Add("wantedCollections", collection) } - // Add cursor if specified (for backfilling historical data) + // Add cursor if specified (for backfilling historical data or reconnects) if w.startCursor > 0 { q.Set("cursor", fmt.Sprintf("%d", w.startCursor)) - fmt.Printf("Starting from cursor: %d (replaying historical events)\n", w.startCursor) + + // Calculate lag (cursor is in microseconds) + now := time.Now().UnixMicro() + lagSeconds := float64(now-w.startCursor) / 1_000_000.0 + fmt.Printf("Jetstream: Starting from cursor %d (%.1f seconds behind live)\n", w.startCursor, lagSeconds) } // Disable compression for now to debug @@ -263,6 +271,13 @@ func (w *Worker) SetEventCallback(cb EventCallback) { w.eventCallback = cb } +// GetLastCursor returns the last processed cursor (time_us) for reconnects +func (w *Worker) GetLastCursor() int64 { + w.cursorMutex.RLock() + defer w.cursorMutex.RUnlock() + return w.lastCursor +} + // processMessage processes a single Jetstream event func (w *Worker) processMessage(message []byte) error { var event JetstreamEvent @@ -270,6 +285,11 @@ func (w *Worker) processMessage(message []byte) error { return fmt.Errorf("failed to unmarshal event: %w", err) } + // Update cursor for reconnects (do this first, even if processing fails) + w.cursorMutex.Lock() + w.lastCursor = event.TimeUS + w.cursorMutex.Unlock() + // Call callback if set if w.eventCallback != nil { w.eventCallback(event.TimeUS)