fix backfill

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 {

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

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

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