From 678a11d1b7d786010b993a7621072e540834fec0 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Mon, 6 Oct 2025 21:24:46 -0500 Subject: [PATCH] backfill via com.atproto.sync.listReposByCollection --- cmd/registry/serve.go | 49 ++--- docker-compose.yml | 5 +- go.mod | 8 +- go.sum | 2 - pkg/appview/db/queries.go | 236 ++++++++++++++++++++- pkg/appview/db/schema.go | 8 + pkg/appview/handlers/settings.go | 8 +- pkg/appview/jetstream/backfill.go | 330 ++++++++++++++++++++++++++++++ pkg/appview/jetstream/worker.go | 28 ++- pkg/atproto/client.go | 94 +++++++++ pkg/auth/oauth/client.go | 4 +- pkg/auth/oauth/server.go | 16 +- 12 files changed, 731 insertions(+), 57 deletions(-) create mode 100644 pkg/appview/jetstream/backfill.go diff --git a/cmd/registry/serve.go b/cmd/registry/serve.go index c36b104..7f259b7 100644 --- a/cmd/registry/serve.go +++ b/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 } diff --git a/docker-compose.yml b/docker-compose.yml index 51eb54c..ff7eba6 100644 --- a/docker-compose.yml +++ b/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 diff --git a/go.mod b/go.mod index e03e8b6..a1c4bd8 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 3116e22..3bf4d52 100644 --- a/go.sum +++ b/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= diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 5f067ed..00c30ea 100644 --- a/pkg/appview/db/queries.go +++ b/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 +} diff --git a/pkg/appview/db/schema.go b/pkg/appview/db/schema.go index c5a6900..fcd963c 100644 --- a/pkg/appview/db/schema.go +++ b/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 diff --git a/pkg/appview/handlers/settings.go b/pkg/appview/handlers/settings.go index bedbf85..1bce92f 100644 --- a/pkg/appview/handlers/settings.go +++ b/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 diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go new file mode 100644 index 0000000..656a2b0 --- /dev/null +++ b/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) +} diff --git a/pkg/appview/jetstream/worker.go b/pkg/appview/jetstream/worker.go index 6f9cea0..ba6f8f5 100644 --- a/pkg/appview/jetstream/worker.go +++ b/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) diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index 22d0c0c..9e49d41 100644 --- a/pkg/atproto/client.go +++ b/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 +} diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index bc36390..d651d94 100644 --- a/pkg/auth/oauth/client.go +++ b/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()) } diff --git a/pkg/auth/oauth/server.go b/pkg/auth/oauth/server.go index 90462fb..4c9f942 100644 --- a/pkg/auth/oauth/server.go +++ b/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