backfill via com.atproto.sync.listReposByCollection

cmd/registry/serve.go

@@ -460,37 +460,40 @@ func initializeUI(config *configuration.Configuration, refresher *oauth.Refreshe
 		jetstreamURL = "wss://jetstream2.us-west.bsky.network/subscribe"
 	}
 
-	// Parse cursor for backfilling historical data
-	// Set to Unix microseconds timestamp to replay from that point
-	// Examples:
-	//   - 2 weeks ago: use `date -d '2 weeks ago' +%s` * 1000000
-	//   - Leave unset (or 0) to start from now
-	var startCursor int64
-	if cursorStr := os.Getenv("JETSTREAM_START_CURSOR"); cursorStr != "" {
-		if cursor, err := time.Parse(time.RFC3339, cursorStr); err == nil {
-			// Support RFC3339 format: "2025-09-23T00:00:00Z"
-			startCursor = cursor.UnixMicro()
-			fmt.Printf("Jetstream: Starting from %s (%d microseconds)\n", cursorStr, startCursor)
-		} else if cursor, err := time.ParseDuration(cursorStr); err == nil {
-			// Support duration format: "-336h" (2 weeks ago)
-			startCursor = time.Now().Add(cursor).UnixMicro()
-			fmt.Printf("Jetstream: Starting from %s ago (%d microseconds)\n", cursorStr, startCursor)
-		} else {
-			fmt.Printf("Warning: Invalid JETSTREAM_START_CURSOR format: %s\n", cursorStr)
-		}
-	}
-
-	worker := jetstream.NewWorker(database, jetstreamURL, startCursor)
+	// Start real-time Jetstream worker (no cursor = start from now)
+	worker := jetstream.NewWorker(database, jetstreamURL, 0)
 	go func() {
 		for {
 			if err := worker.Start(context.Background()); err != nil {
-				fmt.Printf("Jetstream worker error: %v, reconnecting in 10s...\n", err)
+				fmt.Printf("Jetstream: Real-time worker error: %v, reconnecting in 10s...\n", err)
 				time.Sleep(10 * time.Second)
 			}
 		}
 	}()
+	fmt.Println("Jetstream: Real-time worker started")
 
-	fmt.Println("Jetstream worker started")
+	// Start backfill worker if enabled
+	if backfillEnabled := os.Getenv("ATCR_BACKFILL_ENABLED"); backfillEnabled == "true" {
+		// Get BGS endpoint for sync API (defaults to Bluesky's BGS)
+		bgsEndpoint := os.Getenv("ATCR_BGS_ENDPOINT")
+		if bgsEndpoint == "" {
+			bgsEndpoint = "https://bsky.network"
+		}
+
+		backfillWorker, err := jetstream.NewBackfillWorker(database, bgsEndpoint)
+		if err != nil {
+			fmt.Printf("Warning: Failed to create backfill worker: %v\n", err)
+		} else {
+			go func() {
+				fmt.Printf("Backfill: Starting sync-based backfill from %s...\n", bgsEndpoint)
+				if err := backfillWorker.Start(context.Background()); err != nil {
+					fmt.Printf("Backfill: Finished with error: %v\n", err)
+				} else {
+					fmt.Println("Backfill: Completed successfully!")
+				}
+			}()
+		}
+	}
 
 	return database, sessionStore, templates, router
 }

docker-compose.yml

@@ -10,8 +10,9 @@ services:
     environment:
       - ATCR_TOKEN_STORAGE_PATH=/var/lib/atcr/tokens/oauth-tokens.json
       - ATCR_UI_ENABLED=true
-      # Jetstream backfill: Replay 5 days of historical events
-      # - JETSTREAM_START_CURSOR=-120h
+      # Jetstream backfill: Replay historical events (runs once until caught up)
+      # Examples: -120h (5 days ago), -336h (2 weeks ago), 2025-10-01T00:00:00Z
+      - JETSTREAM_BACKFILL_START=-121h
     volumes:
       # Auth keys (JWT signing keys)
       - atcr-auth:/var/lib/atcr/auth

go.mod

@@ -9,6 +9,10 @@ require (
 	github.com/distribution/reference v0.6.0
 	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/google/uuid v1.6.0
+	github.com/gorilla/mux v1.8.1
+	github.com/gorilla/websocket v1.5.3
+	github.com/klauspost/compress v1.18.0
+	github.com/mattn/go-sqlite3 v1.14.32
 	github.com/opencontainers/go-digest v1.0.0
 	github.com/spf13/cobra v1.8.0
 )
@@ -29,15 +33,11 @@ require (
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gorilla/handlers v1.5.2 // indirect
-	github.com/gorilla/mux v1.8.1 // indirect
-	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.23.0 // indirect
 	github.com/hashicorp/golang-lru/arc/v2 v2.0.6 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
-	github.com/klauspost/compress v1.18.0 // indirect
-	github.com/mattn/go-sqlite3 v1.14.32 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opencontainers/image-spec v1.1.0 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect

go.sum

@@ -94,8 +94,6 @@ github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfC
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
-github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
-github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=

pkg/appview/db/queries.go

@@ -2,6 +2,8 @@ package db
 
 import (
 	"database/sql"
+	"fmt"
+	"strings"
 	"time"
 )
 
@@ -91,11 +93,11 @@ func GetUserRepositories(db *sql.DB, did string) ([]Repository, error) {
 		if lastPushStr != "" {
 			// Try multiple timestamp formats
 			formats := []string{
-				time.RFC3339Nano,                        // 2006-01-02T15:04:05.999999999Z07:00
-				"2006-01-02 15:04:05.999999999-07:00",   // SQLite with microseconds and timezone
-				"2006-01-02 15:04:05.999999999",         // SQLite with microseconds
-				time.RFC3339,                             // 2006-01-02T15:04:05Z07:00
-				"2006-01-02 15:04:05",                    // SQLite default
+				time.RFC3339Nano,                      // 2006-01-02T15:04:05.999999999Z07:00
+				"2006-01-02 15:04:05.999999999-07:00", // SQLite with microseconds and timezone
+				"2006-01-02 15:04:05.999999999",       // SQLite with microseconds
+				time.RFC3339,                          // 2006-01-02T15:04:05Z07:00
+				"2006-01-02 15:04:05",                 // SQLite default
 			}
 
 			for _, format := range formats {
@@ -162,6 +164,25 @@ func GetUserRepositories(db *sql.DB, did string) ([]Repository, error) {
 	return repos, nil
 }
 
+// GetUserByDID retrieves a user by DID
+func GetUserByDID(db *sql.DB, did string) (*User, error) {
+	var user User
+	err := db.QueryRow(`
+        SELECT did, handle, pds_endpoint, last_seen
+        FROM users
+        WHERE did = ?
+    `, did).Scan(&user.DID, &user.Handle, &user.PDSEndpoint, &user.LastSeen)
+
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return &user, nil
+}
+
 // UpsertUser inserts or updates a user record
 func UpsertUser(db *sql.DB, user *User) error {
 	_, err := db.Exec(`
@@ -175,6 +196,131 @@ func UpsertUser(db *sql.DB, user *User) error {
 	return err
 }
 
+// GetManifestDigestsForDID returns all manifest digests for a DID
+func GetManifestDigestsForDID(db *sql.DB, did string) ([]string, error) {
+	rows, err := db.Query(`
+        SELECT digest FROM manifests WHERE did = ?
+    `, did)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var digests []string
+	for rows.Next() {
+		var digest string
+		if err := rows.Scan(&digest); err != nil {
+			return nil, err
+		}
+		digests = append(digests, digest)
+	}
+
+	return digests, rows.Err()
+}
+
+// DeleteManifestsNotInList deletes all manifests for a DID that are not in the provided list
+func DeleteManifestsNotInList(db *sql.DB, did string, keepDigests []string) error {
+	if len(keepDigests) == 0 {
+		// No manifests to keep - delete all for this DID
+		_, err := db.Exec(`DELETE FROM manifests WHERE did = ?`, did)
+		return err
+	}
+
+	// Build placeholders for IN clause
+	placeholders := make([]string, len(keepDigests))
+	args := []interface{}{did}
+	for i, digest := range keepDigests {
+		placeholders[i] = "?"
+		args = append(args, digest)
+	}
+
+	query := fmt.Sprintf(`
+        DELETE FROM manifests
+        WHERE did = ? AND digest NOT IN (%s)
+    `, strings.Join(placeholders, ","))
+
+	_, err := db.Exec(query, args...)
+	return err
+}
+
+// GetTagsForDID returns all (repository, tag) pairs for a DID
+func GetTagsForDID(db *sql.DB, did string) ([]struct{ Repository, Tag string }, error) {
+	rows, err := db.Query(`
+        SELECT repository, tag FROM tags WHERE did = ?
+    `, did)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var tags []struct{ Repository, Tag string }
+	for rows.Next() {
+		var t struct{ Repository, Tag string }
+		if err := rows.Scan(&t.Repository, &t.Tag); err != nil {
+			return nil, err
+		}
+		tags = append(tags, t)
+	}
+
+	return tags, rows.Err()
+}
+
+// DeleteTagsNotInList deletes all tags for a DID that are not in the provided list
+func DeleteTagsNotInList(db *sql.DB, did string, keepTags []struct{ Repository, Tag string }) error {
+	if len(keepTags) == 0 {
+		// No tags to keep - delete all for this DID
+		_, err := db.Exec(`DELETE FROM tags WHERE did = ?`, did)
+		return err
+	}
+
+	// For tags, we need to check (repository, tag) pairs
+	// Build a DELETE query that excludes the pairs we want to keep
+	tx, err := db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	// First, get all current tags
+	rows, err := tx.Query(`SELECT id, repository, tag FROM tags WHERE did = ?`, did)
+	if err != nil {
+		return err
+	}
+
+	var toDelete []int64
+	for rows.Next() {
+		var id int64
+		var repo, tag string
+		if err := rows.Scan(&id, &repo, &tag); err != nil {
+			rows.Close()
+			return err
+		}
+
+		// Check if this tag should be kept
+		found := false
+		for _, keep := range keepTags {
+			if keep.Repository == repo && keep.Tag == tag {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			toDelete = append(toDelete, id)
+		}
+	}
+	rows.Close()
+
+	// Delete tags not in keep list
+	for _, id := range toDelete {
+		if _, err := tx.Exec(`DELETE FROM tags WHERE id = ?`, id); err != nil {
+			return err
+		}
+	}
+
+	return tx.Commit()
+}
+
 // InsertManifest inserts a new manifest record
 func InsertManifest(db *sql.DB, manifest *Manifest) (int64, error) {
 	result, err := db.Exec(`
@@ -311,3 +457,83 @@ func IsManifestTagged(db *sql.DB, did, repository, digest string) (bool, error)
 
 	return count > 0, nil
 }
+
+// BackfillState represents the backfill progress
+type BackfillState struct {
+	StartCursor   int64
+	CurrentCursor int64
+	Completed     bool
+	UpdatedAt     time.Time
+}
+
+// GetBackfillState retrieves the backfill state
+func GetBackfillState(db *sql.DB) (*BackfillState, error) {
+	var state BackfillState
+	var updatedAtStr string
+
+	err := db.QueryRow(`
+		SELECT start_cursor, current_cursor, completed, updated_at
+		FROM backfill_state
+		WHERE id = 1
+	`).Scan(&state.StartCursor, &state.CurrentCursor, &state.Completed, &updatedAtStr)
+
+	if err == sql.ErrNoRows {
+		return nil, nil // No backfill state exists
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	// Parse timestamp
+	if updatedAtStr != "" {
+		formats := []string{
+			time.RFC3339Nano,
+			"2006-01-02 15:04:05.999999999-07:00",
+			"2006-01-02 15:04:05.999999999",
+			time.RFC3339,
+			"2006-01-02 15:04:05",
+		}
+		for _, format := range formats {
+			if t, err := time.Parse(format, updatedAtStr); err == nil {
+				state.UpdatedAt = t
+				break
+			}
+		}
+	}
+
+	return &state, nil
+}
+
+// UpsertBackfillState updates or creates backfill state
+func UpsertBackfillState(db *sql.DB, state *BackfillState) error {
+	_, err := db.Exec(`
+        INSERT INTO backfill_state (id, start_cursor, current_cursor, completed, updated_at)
+        VALUES (1, ?, ?, ?, datetime('now'))
+        ON CONFLICT(id) DO UPDATE SET
+            start_cursor = excluded.start_cursor,
+            current_cursor = excluded.current_cursor,
+            completed = excluded.completed,
+            updated_at = excluded.updated_at
+    `, state.StartCursor, state.CurrentCursor, state.Completed)
+	return err
+}
+
+// UpdateBackfillCursor updates just the current cursor position
+func UpdateBackfillCursor(db *sql.DB, cursor int64) error {
+	_, err := db.Exec(`
+        UPDATE backfill_state
+        SET current_cursor = ?, updated_at = datetime('now')
+        WHERE id = 1
+    `, cursor)
+	return err
+}
+
+// MarkBackfillCompleted marks the backfill as completed
+func MarkBackfillCompleted(db *sql.DB) error {
+	_, err := db.Exec(`
+        UPDATE backfill_state
+        SET completed = 1, updated_at = datetime('now')
+        WHERE id = 1
+    `)
+	return err
+}

pkg/appview/db/schema.go

@@ -63,6 +63,14 @@ CREATE TABLE IF NOT EXISTS firehose_cursor (
     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
+);
 `
 
 // InitDB initializes the SQLite database with the schema

pkg/appview/handlers/settings.go

@@ -48,10 +48,10 @@ func (h *SettingsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	data := struct {
 		User    *db.User
 		Profile struct {
-			Handle       string
-			DID          string
-			PDSEndpoint  string
-			DefaultHold  string
+			Handle      string
+			DID         string
+			PDSEndpoint string
+			DefaultHold string
 		}
 		SessionExpiry time.Time
 		Query         string

pkg/appview/jetstream/backfill.go

@@ -0,0 +1,330 @@
+package jetstream
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"atcr.io/pkg/appview/db"
+	"atcr.io/pkg/atproto"
+)
+
+// BackfillWorker uses com.atproto.sync.listReposByCollection to backfill historical data
+type BackfillWorker struct {
+	db       *sql.DB
+	client   *atproto.Client
+	resolver *atproto.Resolver
+}
+
+// BackfillState tracks backfill progress
+type BackfillState struct {
+	Collection       string
+	RepoCursor       string // Cursor for listReposByCollection
+	CurrentDID       string // Current DID being processed
+	RecordCursor     string // Cursor for listRecords within current DID
+	ProcessedRepos   int
+	ProcessedRecords int
+	Completed        bool
+}
+
+// NewBackfillWorker creates a backfill worker using sync API
+func NewBackfillWorker(database *sql.DB, pdsEndpoint string) (*BackfillWorker, error) {
+	// Create client without auth - sync endpoints are public
+	client := atproto.NewClient(pdsEndpoint, "", "")
+
+	return &BackfillWorker{
+		db:       database,
+		client:   client,
+		resolver: atproto.NewResolver(),
+	}, nil
+}
+
+// Start runs the backfill for all ATCR collections
+func (b *BackfillWorker) Start(ctx context.Context) error {
+	fmt.Println("Backfill: Starting sync-based backfill...")
+
+	collections := []string{
+		atproto.ManifestCollection, // io.atcr.manifest
+		atproto.TagCollection,      // io.atcr.tag
+	}
+
+	for _, collection := range collections {
+		fmt.Printf("Backfill: Processing collection: %s\n", collection)
+
+		if err := b.backfillCollection(ctx, collection); err != nil {
+			return fmt.Errorf("failed to backfill collection %s: %w", collection, err)
+		}
+
+		fmt.Printf("Backfill: Completed collection: %s\n", collection)
+	}
+
+	fmt.Println("Backfill: All collections completed!")
+	return nil
+}
+
+// backfillCollection backfills a single collection
+func (b *BackfillWorker) backfillCollection(ctx context.Context, collection string) error {
+	var repoCursor string
+	processedRepos := 0
+	processedRecords := 0
+
+	// Paginate through all repos with this collection
+	for {
+		// List repos that have records in this collection
+		result, err := b.client.ListReposByCollection(ctx, collection, 1000, repoCursor)
+		if err != nil {
+			return fmt.Errorf("failed to list repos: %w", err)
+		}
+
+		fmt.Printf("Backfill: Found %d repos with %s (cursor: %s)\n", len(result.Repos), collection, repoCursor)
+
+		// Process each repo (DID)
+		for _, did := range result.Repos {
+			recordCount, err := b.backfillRepo(ctx, did, collection)
+			if err != nil {
+				fmt.Printf("WARNING: Failed to backfill repo %s: %v\n", did, err)
+				continue
+			}
+
+			processedRepos++
+			processedRecords += recordCount
+
+			if processedRepos%10 == 0 {
+				fmt.Printf("Backfill: Progress - %d repos, %d records\n", processedRepos, processedRecords)
+			}
+		}
+
+		// Check if there are more pages
+		if result.Cursor == "" {
+			break
+		}
+
+		repoCursor = result.Cursor
+	}
+
+	fmt.Printf("Backfill: Collection %s complete - %d repos, %d records\n", collection, processedRepos, processedRecords)
+	return nil
+}
+
+// backfillRepo backfills all records for a single repo/DID
+func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection string) (int, error) {
+	// Ensure user exists in database
+	if err := b.ensureUser(ctx, did); err != nil {
+		return 0, fmt.Errorf("failed to ensure user: %w", err)
+	}
+
+	var recordCursor string
+	recordCount := 0
+
+	// Track which records exist on the PDS for reconciliation
+	var foundManifestDigests []string
+	var foundTags []struct{ Repository, Tag string }
+
+	// Paginate through all records for this repo
+	for {
+		records, cursor, err := b.client.ListRecordsForRepo(ctx, did, collection, 100, recordCursor)
+		if err != nil {
+			return recordCount, fmt.Errorf("failed to list records: %w", err)
+		}
+
+		// Process each record
+		for _, record := range records {
+			// Track what we found for deletion reconciliation
+			if collection == atproto.ManifestCollection {
+				var manifestRecord atproto.ManifestRecord
+				if err := json.Unmarshal(record.Value, &manifestRecord); err == nil {
+					foundManifestDigests = append(foundManifestDigests, manifestRecord.Digest)
+				}
+			} else if collection == atproto.TagCollection {
+				var tagRecord atproto.TagRecord
+				if err := json.Unmarshal(record.Value, &tagRecord); err == nil {
+					foundTags = append(foundTags, struct{ Repository, Tag string }{
+						Repository: tagRecord.Repository,
+						Tag:        tagRecord.Tag,
+					})
+				}
+			}
+
+			if err := b.processRecord(ctx, did, collection, &record); err != nil {
+				fmt.Printf("WARNING: Failed to process record %s: %v\n", record.URI, err)
+				continue
+			}
+			recordCount++
+		}
+
+		// Check if there are more pages
+		if cursor == "" {
+			break
+		}
+
+		recordCursor = cursor
+	}
+
+	// Reconcile deletions - remove records from DB that no longer exist on PDS
+	if err := b.reconcileDeletions(did, collection, foundManifestDigests, foundTags); err != nil {
+		fmt.Printf("WARNING: Failed to reconcile deletions for %s: %v\n", did, err)
+	}
+
+	return recordCount, nil
+}
+
+// reconcileDeletions removes records from the database that no longer exist on the PDS
+func (b *BackfillWorker) reconcileDeletions(did, collection string, foundManifestDigests []string, foundTags []struct{ Repository, Tag string }) error {
+	switch collection {
+	case atproto.ManifestCollection:
+		// Get current manifests in DB
+		dbDigests, err := db.GetManifestDigestsForDID(b.db, did)
+		if err != nil {
+			return fmt.Errorf("failed to get DB manifests: %w", err)
+		}
+
+		// Delete manifests not found on PDS
+		if err := db.DeleteManifestsNotInList(b.db, did, foundManifestDigests); err != nil {
+			return fmt.Errorf("failed to delete orphaned manifests: %w", err)
+		}
+
+		// Log deletions
+		deleted := len(dbDigests) - len(foundManifestDigests)
+		if deleted > 0 {
+			fmt.Printf("Backfill: Deleted %d orphaned manifests for %s\n", deleted, did)
+		}
+
+	case atproto.TagCollection:
+		// Get current tags in DB
+		dbTags, err := db.GetTagsForDID(b.db, did)
+		if err != nil {
+			return fmt.Errorf("failed to get DB tags: %w", err)
+		}
+
+		// Delete tags not found on PDS
+		if err := db.DeleteTagsNotInList(b.db, did, foundTags); err != nil {
+			return fmt.Errorf("failed to delete orphaned tags: %w", err)
+		}
+
+		// Log deletions
+		deleted := len(dbTags) - len(foundTags)
+		if deleted > 0 {
+			fmt.Printf("Backfill: Deleted %d orphaned tags for %s\n", deleted, did)
+		}
+	}
+
+	return nil
+}
+
+// processRecord processes a single record and stores it in the database
+func (b *BackfillWorker) processRecord(ctx context.Context, did, collection string, record *atproto.Record) error {
+	switch collection {
+	case atproto.ManifestCollection:
+		return b.processManifestRecord(did, record)
+	case atproto.TagCollection:
+		return b.processTagRecord(did, record)
+	default:
+		return fmt.Errorf("unsupported collection: %s", collection)
+	}
+}
+
+// processManifestRecord processes a manifest record
+func (b *BackfillWorker) processManifestRecord(did string, record *atproto.Record) error {
+	var manifestRecord atproto.ManifestRecord
+	if err := json.Unmarshal(record.Value, &manifestRecord); err != nil {
+		return fmt.Errorf("failed to unmarshal manifest: %w", err)
+	}
+
+	// Serialize full manifest as JSON for storage
+	manifestJSON, err := json.Marshal(manifestRecord)
+	if err != nil {
+		return fmt.Errorf("failed to marshal manifest: %w", err)
+	}
+
+	// Insert manifest
+	manifestID, err := db.InsertManifest(b.db, &db.Manifest{
+		DID:           did,
+		Repository:    manifestRecord.Repository,
+		Digest:        manifestRecord.Digest,
+		MediaType:     manifestRecord.MediaType,
+		SchemaVersion: manifestRecord.SchemaVersion,
+		ConfigDigest:  manifestRecord.Config.Digest,
+		ConfigSize:    manifestRecord.Config.Size,
+		RawManifest:   string(manifestJSON),
+		HoldEndpoint:  manifestRecord.HoldEndpoint,
+		CreatedAt:     manifestRecord.CreatedAt,
+	})
+	if err != nil {
+		// Skip if already exists
+		if strings.Contains(err.Error(), "UNIQUE constraint failed") {
+			return nil
+		}
+		return fmt.Errorf("failed to insert manifest: %w", err)
+	}
+
+	// Insert layers
+	for i, layer := range manifestRecord.Layers {
+		if err := db.InsertLayer(b.db, &db.Layer{
+			ManifestID: manifestID,
+			Digest:     layer.Digest,
+			MediaType:  layer.MediaType,
+			Size:       layer.Size,
+			LayerIndex: i,
+		}); err != nil {
+			// Continue on error - layer might already exist
+			continue
+		}
+	}
+
+	return nil
+}
+
+// processTagRecord processes a tag record
+func (b *BackfillWorker) processTagRecord(did string, record *atproto.Record) error {
+	var tagRecord atproto.TagRecord
+	if err := json.Unmarshal(record.Value, &tagRecord); err != nil {
+		return fmt.Errorf("failed to unmarshal tag: %w", err)
+	}
+
+	// Insert or update tag
+	return db.UpsertTag(b.db, &db.Tag{
+		DID:        did,
+		Repository: tagRecord.Repository,
+		Tag:        tagRecord.Tag,
+		Digest:     tagRecord.ManifestDigest,
+		CreatedAt:  tagRecord.UpdatedAt,
+	})
+}
+
+// ensureUser resolves and upserts a user by DID
+func (b *BackfillWorker) ensureUser(ctx context.Context, did string) error {
+	// Check if user already exists
+	existingUser, err := db.GetUserByDID(b.db, did)
+	if err == nil && existingUser != nil {
+		// Update last seen
+		existingUser.LastSeen = time.Now()
+		return db.UpsertUser(b.db, existingUser)
+	}
+
+	// Resolve DID to get handle and PDS endpoint
+	resolvedDID, pdsEndpoint, err := b.resolver.ResolveIdentity(ctx, did)
+	if err != nil {
+		// Fallback: use DID as handle
+		resolvedDID = did
+		pdsEndpoint = "https://bsky.social"
+	}
+
+	// Get handle from DID document
+	handle, err := b.resolver.ResolveHandleFromDID(ctx, resolvedDID)
+	if err != nil {
+		handle = resolvedDID // Fallback to DID
+	}
+
+	// Upsert to database
+	user := &db.User{
+		DID:         resolvedDID,
+		Handle:      handle,
+		PDSEndpoint: pdsEndpoint,
+		LastSeen:    time.Now(),
+	}
+
+	return db.UpsertUser(b.db, user)
+}

pkg/appview/jetstream/worker.go

@@ -20,6 +20,9 @@ type UserCache struct {
 	cache map[string]*db.User
 }
 
+// EventCallback is called for each processed event
+type EventCallback func(timeUS int64)
+
 // Worker consumes Jetstream events and populates the UI database
 type Worker struct {
 	db                   *sql.DB
@@ -29,6 +32,7 @@ type Worker struct {
 	debugCollectionCount int
 	userCache            *UserCache
 	resolver             *atproto.Resolver
+	eventCallback        EventCallback
 }
 
 // NewWorker creates a new Jetstream worker
@@ -44,7 +48,7 @@ func NewWorker(database *sql.DB, jetstreamURL string, startCursor int64) *Worker
 		startCursor:  startCursor,
 		wantedCollections: []string{
 			atproto.ManifestCollection, // io.atcr.manifest
-			atproto.TagCollection,       // io.atcr.tag
+			atproto.TagCollection,      // io.atcr.tag
 		},
 		userCache: &UserCache{
 			cache: make(map[string]*db.User),
@@ -132,6 +136,11 @@ func (w *Worker) Start(ctx context.Context) error {
 	}
 }
 
+// SetEventCallback sets a callback to be called for each event
+func (w *Worker) SetEventCallback(cb EventCallback) {
+	w.eventCallback = cb
+}
+
 // processMessage processes a single Jetstream event
 func (w *Worker) processMessage(message []byte) error {
 	var event JetstreamEvent
@@ -139,6 +148,11 @@ func (w *Worker) processMessage(message []byte) error {
 		return fmt.Errorf("failed to unmarshal event: %w", err)
 	}
 
+	// Call callback if set
+	if w.eventCallback != nil {
+		w.eventCallback(event.TimeUS)
+	}
+
 	// Only process commit events
 	if event.Kind != "commit" {
 		return nil
@@ -325,12 +339,12 @@ func (w *Worker) processTag(commit *CommitEvent) error {
 
 // JetstreamEvent represents a Jetstream event
 type JetstreamEvent struct {
-	DID       string        `json:"did"`
-	TimeUS    int64         `json:"time_us"`
-	Kind      string        `json:"kind"` // "commit", "identity", "account"
-	Commit    *CommitEvent  `json:"commit,omitempty"`
-	Identity  *IdentityInfo `json:"identity,omitempty"`
-	Account   *AccountInfo  `json:"account,omitempty"`
+	DID      string        `json:"did"`
+	TimeUS   int64         `json:"time_us"`
+	Kind     string        `json:"kind"` // "commit", "identity", "account"
+	Commit   *CommitEvent  `json:"commit,omitempty"`
+	Identity *IdentityInfo `json:"identity,omitempty"`
+	Account  *AccountInfo  `json:"account,omitempty"`
 }
 
 // CommitEvent represents a commit event (create/update/delete)

pkg/atproto/client.go

@@ -299,3 +299,97 @@ func (c *Client) GetBlob(ctx context.Context, cid string) ([]byte, error) {
 
 	return data, nil
 }
+
+// ListReposByCollectionResult represents the response from com.atproto.sync.listReposByCollection
+type ListReposByCollectionResult struct {
+	Repos  []string `json:"repos"` // Array of DIDs
+	Cursor string   `json:"cursor,omitempty"`
+}
+
+// ListReposByCollection lists all repos (DIDs) that have records in a collection
+// This is a network-wide query, not limited to a single PDS
+func (c *Client) ListReposByCollection(ctx context.Context, collection string, limit int, cursor string) (*ListReposByCollectionResult, error) {
+	// Build URL with query parameters
+	url := fmt.Sprintf("%s/xrpc/com.atproto.sync.listReposByCollection?collection=%s", c.pdsEndpoint, collection)
+
+	if limit > 0 {
+		url += fmt.Sprintf("&limit=%d", limit)
+	}
+	if cursor != "" {
+		url += fmt.Sprintf("&cursor=%s", cursor)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	// This endpoint typically doesn't require auth for public data
+	// but we include it if available
+	if c.accessToken != "" {
+		req.Header.Set("Authorization", c.authHeader())
+	}
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list repos by collection: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("list repos by collection failed with status %d: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	var result ListReposByCollectionResult
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return nil, fmt.Errorf("failed to decode response: %w", err)
+	}
+
+	return &result, nil
+}
+
+// ListRecordsForRepo lists records in a collection for a specific repo (DID)
+// This differs from ListRecords which uses the client's DID
+func (c *Client) ListRecordsForRepo(ctx context.Context, repoDID, collection string, limit int, cursor string) ([]Record, string, error) {
+	url := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s",
+		c.pdsEndpoint, repoDID, collection)
+
+	if limit > 0 {
+		url += fmt.Sprintf("&limit=%d", limit)
+	}
+	if cursor != "" {
+		url += fmt.Sprintf("&cursor=%s", cursor)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
+	if err != nil {
+		return nil, "", err
+	}
+
+	// This endpoint typically doesn't require auth for public records
+	if c.accessToken != "" {
+		req.Header.Set("Authorization", c.authHeader())
+	}
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to list records: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return nil, "", fmt.Errorf("list records failed with status %d: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	var result struct {
+		Records []Record `json:"records"`
+		Cursor  string   `json:"cursor,omitempty"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return nil, "", fmt.Errorf("failed to decode response: %w", err)
+	}
+
+	return result.Records, result.Cursor, nil
+}

pkg/auth/oauth/client.go

@@ -45,7 +45,7 @@ func NewClientWithKey(baseURL string, dpopKey *ecdsa.PrivateKey) *Client {
 		dpopKey:       dpopKey,
 		dpopTransport: NewDPoPTransport(http.DefaultTransport, dpopKey),
 		resolver:      atproto.NewResolver(),
-		baseUrl:      baseURL,
+		baseUrl:       baseURL,
 	}
 }
 
@@ -201,7 +201,7 @@ func (c *Client) RefreshToken(ctx context.Context, refreshToken string) (*oauth2
 	return newToken, nil
 }
 
-func (c *Client) ClientID() (string) {
+func (c *Client) ClientID() string {
 	return c.ClientIDWithScopes(c.GetDefaultScopes())
 }
 

pkg/auth/oauth/server.go

@@ -21,14 +21,14 @@ type UISessionStore interface {
 
 // Server handles OAuth authorization for the AppView
 type Server struct {
-	storage         *RefreshTokenStorage
-	sessionManager  *session.Manager
-	resolver        *atproto.Resolver
-	refresher       *Refresher
-	uiSessionStore  UISessionStore
-	baseURL         string
-	states          map[string]*OAuthState
-	statesMu        sync.RWMutex
+	storage        *RefreshTokenStorage
+	sessionManager *session.Manager
+	resolver       *atproto.Resolver
+	refresher      *Refresher
+	uiSessionStore UISessionStore
+	baseURL        string
+	states         map[string]*OAuthState
+	statesMu       sync.RWMutex
 }
 
 // OAuthState tracks an in-progress OAuth flow