at-container-registry/pkg/appview/jetstream/backfill.go

package jetstream

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"

	"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
	directory identity.Directory
}

// 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, relayEndpoint string) (*BackfillWorker, error) {
	// Create client for relay - used only for listReposByCollection
	client := atproto.NewClient(relayEndpoint, "", "")

	return &BackfillWorker{
		db:        database,
		client:    client, // This points to the relay
		directory: identity.DefaultDirectory(),
	}, 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
		atproto.StarCollection,     // io.atcr.sailor.star
	}

	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 _, 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", repo.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 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
	didParsed, err := syntax.ParseDID(did)
	if err != nil {
		return 0, fmt.Errorf("invalid DID %s: %w", did, err)
	}

	ident, err := b.directory.LookupDID(ctx, didParsed)
	if err != nil {
		return 0, fmt.Errorf("failed to resolve DID to PDS: %w", err)
	}

	pdsEndpoint := ident.PDSEndpoint()
	if pdsEndpoint == "" {
		return 0, fmt.Errorf("no PDS endpoint found for DID %s", did)
	}

	// Create a client for this user's PDS
	pdsClient := atproto.NewClient(pdsEndpoint, "", "")

	var recordCursor string
	recordCount := 0

	// Track which records exist on the PDS for reconciliation
	var foundManifestDigests []string
	var foundTags []struct{ Repository, Tag string }
	foundStars := make(map[string]time.Time) // key: "ownerDID/repository", value: createdAt

	// Paginate through all records for this repo
	for {
		records, cursor, err := pdsClient.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,
					})
				}
			} else if collection == atproto.StarCollection {
				var starRecord atproto.StarRecord
				if err := json.Unmarshal(record.Value, &starRecord); err == nil {
					key := fmt.Sprintf("%s/%s", starRecord.Subject.DID, starRecord.Subject.Repository)
					foundStars[key] = starRecord.CreatedAt
				}
			}

			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, foundStars); err != nil {
		fmt.Printf("WARNING: Failed to reconcile deletions for %s: %v\n", did, err)
	}

	// After processing manifests, clean up orphaned tags (tags pointing to non-existent manifests)
	if collection == atproto.ManifestCollection {
		if err := db.CleanupOrphanedTags(b.db, did); err != nil {
			fmt.Printf("WARNING: Failed to cleanup orphaned tags 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 }, foundStars map[string]time.Time) 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)
		}

	case atproto.StarCollection:
		// Reconcile stars - delete stars that no longer exist on PDS
		// Star counts will be calculated on demand from the stars table
		if err := db.DeleteStarsNotInList(b.db, did, foundStars); err != nil {
			return fmt.Errorf("failed to delete orphaned stars: %w", err)
		}
	}

	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)
	case atproto.StarCollection:
		return b.processStarRecord(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)
	}

	// Extract OCI annotations from manifest
	var title, description, sourceURL, documentationURL, licenses, iconURL string
	if manifestRecord.Annotations != nil {
		title = manifestRecord.Annotations["org.opencontainers.image.title"]
		description = manifestRecord.Annotations["org.opencontainers.image.description"]
		sourceURL = manifestRecord.Annotations["org.opencontainers.image.source"]
		documentationURL = manifestRecord.Annotations["org.opencontainers.image.documentation"]
		licenses = manifestRecord.Annotations["org.opencontainers.image.licenses"]
		iconURL = manifestRecord.Annotations["io.atcr.icon"]
	}

	// 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,
		HoldEndpoint:     manifestRecord.HoldEndpoint,
		CreatedAt:        manifestRecord.CreatedAt,
		Title:            title,
		Description:      description,
		SourceURL:        sourceURL,
		DocumentationURL: documentationURL,
		Licenses:         licenses,
		IconURL:          iconURL,
	})
	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,
	})
}

// processStarRecord processes a star record
func (b *BackfillWorker) processStarRecord(did string, record *atproto.Record) error {
	var starRecord atproto.StarRecord
	if err := json.Unmarshal(record.Value, &starRecord); err != nil {
		return fmt.Errorf("failed to unmarshal star: %w", err)
	}

	// Upsert the star record (idempotent - won't duplicate)
	// The DID here is the starrer (user who starred)
	// The subject contains the owner DID and repository
	// Star count will be calculated on demand from the stars table
	return db.UpsertStar(b.db, did, starRecord.Subject.DID, starRecord.Subject.Repository, starRecord.CreatedAt)
}

// 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
	didParsed, err := syntax.ParseDID(did)
	if err != nil {
		// Fallback: use DID as handle
		user := &db.User{
			DID:         did,
			Handle:      did,
			PDSEndpoint: "https://bsky.social",
			LastSeen:    time.Now(),
		}
		return db.UpsertUser(b.db, user)
	}

	ident, err := b.directory.LookupDID(ctx, didParsed)
	if err != nil {
		// Fallback: use DID as handle
		user := &db.User{
			DID:         did,
			Handle:      did,
			PDSEndpoint: "https://bsky.social",
			LastSeen:    time.Now(),
		}
		return db.UpsertUser(b.db, user)
	}

	resolvedDID := ident.DID.String()
	handle := ident.Handle.String()
	pdsEndpoint := ident.PDSEndpoint()

	// If handle is invalid or PDS is missing, use defaults
	if handle == "handle.invalid" || handle == "" {
		handle = resolvedDID
	}
	if pdsEndpoint == "" {
		pdsEndpoint = "https://bsky.social"
	}

	// Fetch user's Bluesky profile (including avatar)
	// Use public Bluesky AppView API (doesn't require auth for public profiles)
	avatar := ""
	publicClient := atproto.NewClient("https://public.api.bsky.app", "", "")
	profile, err := publicClient.GetActorProfile(ctx, resolvedDID)
	if err != nil {
		fmt.Printf("WARNING [backfill]: Failed to fetch profile for DID %s: %v\n", resolvedDID, err)
		// Continue without avatar
	} else {
		avatar = profile.Avatar
	}

	// Upsert to database
	user := &db.User{
		DID:         resolvedDID,
		Handle:      handle,
		PDSEndpoint: pdsEndpoint,
		Avatar:      avatar,
		LastSeen:    time.Now(),
	}

	return db.UpsertUser(b.db, user)
}