diff --git a/cmd/registry/serve.go b/cmd/registry/serve.go index 7f259b7..da1a41d 100644 --- a/cmd/registry/serve.go +++ b/cmd/registry/serve.go @@ -474,18 +474,18 @@ func initializeUI(config *configuration.Configuration, refresher *oauth.Refreshe // 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" + // Get relay endpoint for sync API (defaults to Bluesky's relay) + relayEndpoint := os.Getenv("ATCR_RELAY_ENDPOINT") + if relayEndpoint == "" { + relayEndpoint = "https://relay1.us-east.bsky.network" } - backfillWorker, err := jetstream.NewBackfillWorker(database, bgsEndpoint) + backfillWorker, err := jetstream.NewBackfillWorker(database, relayEndpoint) 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) + fmt.Printf("Backfill: Starting sync-based backfill from %s...\n", relayEndpoint) if err := backfillWorker.Start(context.Background()); err != nil { fmt.Printf("Backfill: Finished with error: %v\n", err) } else { diff --git a/docker-compose.yml b/docker-compose.yml index ff7eba6..de510fb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -10,9 +10,7 @@ services: environment: - ATCR_TOKEN_STORAGE_PATH=/var/lib/atcr/tokens/oauth-tokens.json - ATCR_UI_ENABLED=true - # 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 + - ATCR_BACKFILL_ENABLED=true volumes: # Auth keys (JWT signing keys) - atcr-auth:/var/lib/atcr/auth diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index 656a2b0..d73c1b5 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -31,13 +31,13 @@ type BackfillState struct { } // 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, "", "") +func NewBackfillWorker(database *sql.DB, relayEndpoint string) (*BackfillWorker, error) { + // Create client for relay - used only for listReposByCollection + client := atproto.NewClient(relayEndpoint, "", "") return &BackfillWorker{ db: database, - client: client, + client: client, // This points to the relay resolver: atproto.NewResolver(), }, nil } @@ -82,10 +82,10 @@ func (b *BackfillWorker) backfillCollection(ctx context.Context, collection stri 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) + for _, repo := range result.Repos { + recordCount, err := b.backfillRepo(ctx, repo.DID, collection) if err != nil { - fmt.Printf("WARNING: Failed to backfill repo %s: %v\n", did, err) + fmt.Printf("WARNING: Failed to backfill repo %s: %v\n", repo.DID, err) continue } @@ -111,11 +111,20 @@ func (b *BackfillWorker) backfillCollection(ctx context.Context, collection stri // 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 + // Ensure user exists in database and get their PDS endpoint if err := b.ensureUser(ctx, did); err != nil { return 0, fmt.Errorf("failed to ensure user: %w", err) } + // Resolve DID to get user's PDS endpoint + _, pdsEndpoint, err := b.resolver.ResolveIdentity(ctx, did) + if err != nil { + return 0, fmt.Errorf("failed to resolve DID to PDS: %w", err) + } + + // Create a client for this user's PDS + pdsClient := atproto.NewClient(pdsEndpoint, "", "") + var recordCursor string recordCount := 0 @@ -125,7 +134,7 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin // Paginate through all records for this repo for { - records, cursor, err := b.client.ListRecordsForRepo(ctx, did, collection, 100, recordCursor) + records, cursor, err := pdsClient.ListRecordsForRepo(ctx, did, collection, 100, recordCursor) if err != nil { return recordCount, fmt.Errorf("failed to list records: %w", err) } diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index 9e49d41..15fc972 100644 --- a/pkg/atproto/client.go +++ b/pkg/atproto/client.go @@ -302,8 +302,13 @@ func (c *Client) GetBlob(ctx context.Context, cid string) ([]byte, error) { // ListReposByCollectionResult represents the response from com.atproto.sync.listReposByCollection type ListReposByCollectionResult struct { - Repos []string `json:"repos"` // Array of DIDs - Cursor string `json:"cursor,omitempty"` + Repos []RepoRef `json:"repos"` // Array of repo references + Cursor string `json:"cursor,omitempty"` +} + +// RepoRef represents a repository reference in listReposByCollection response +type RepoRef struct { + DID string `json:"did"` } // ListReposByCollection lists all repos (DIDs) that have records in a collection