diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index ebd4dae..26c99d0 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -652,6 +652,52 @@ func GetManifest(db *sql.DB, digest string) (*Manifest, error) { return &m, nil } +// GetNewestManifestForRepo returns the newest manifest for a specific repository +// Used by backfill to ensure annotations come from the most recent manifest +func GetNewestManifestForRepo(db *sql.DB, did, repository string) (*Manifest, error) { + var m Manifest + err := db.QueryRow(` + SELECT id, did, repository, digest, hold_endpoint, schema_version, media_type, + config_digest, config_size, created_at + FROM manifests + WHERE did = ? AND repository = ? + ORDER BY created_at DESC + LIMIT 1 + `, did, repository).Scan( + &m.ID, &m.DID, &m.Repository, &m.Digest, + &m.HoldEndpoint, &m.SchemaVersion, &m.MediaType, + &m.ConfigDigest, &m.ConfigSize, &m.CreatedAt, + ) + if err != nil { + return nil, err + } + return &m, nil +} + +// GetRepositoriesForDID returns all unique repository names for a DID +// Used by backfill to reconcile annotations for all repositories +func GetRepositoriesForDID(db *sql.DB, did string) ([]string, error) { + rows, err := db.Query(` + SELECT DISTINCT repository + FROM manifests + WHERE did = ? + `, did) + if err != nil { + return nil, err + } + defer rows.Close() + + var repositories []string + for rows.Next() { + var repo string + if err := rows.Scan(&repo); err != nil { + return nil, err + } + repositories = append(repositories, repo) + } + return repositories, rows.Err() +} + // GetLayersForManifest fetches all layers for a manifest func GetLayersForManifest(db *sql.DB, manifestID int64) ([]Layer, error) { rows, err := db.Query(` diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index f6eeade..42238a8 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -152,8 +152,9 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin return 0, fmt.Errorf("no PDS endpoint found for DID %s", did) } - // Create a client for this user's PDS - pdsClient := atproto.NewClient(pdsEndpoint, "", "") + // Create a client for this user's PDS with the user's DID + // This allows GetRecord to work properly with the repo parameter + pdsClient := atproto.NewClient(pdsEndpoint, did, "") var recordCursor string recordCount := 0 @@ -220,6 +221,12 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin if err := db.CleanupOrphanedTags(b.db, did); err != nil { fmt.Printf("WARNING: Failed to cleanup orphaned tags for %s: %v\n", did, err) } + + // Reconcile annotations - ensure they come from newest manifest per repository + // This fixes out-of-order backfill where older manifests can overwrite newer annotations + if err := b.reconcileAnnotations(ctx, did, pdsClient); err != nil { + fmt.Printf("WARNING: Failed to reconcile annotations for %s: %v\n", did, err) + } } return recordCount, nil @@ -361,3 +368,60 @@ func (b *BackfillWorker) queryCaptainRecord(ctx context.Context, holdDID string) fmt.Printf("Backfill: Cached captain record for hold %s (owner: %s)\n", holdDID, captainRecord.OwnerDID) return nil } + +// reconcileAnnotations ensures annotations come from the newest manifest in each repository +// This fixes the out-of-order backfill issue where older manifests can overwrite newer annotations +func (b *BackfillWorker) reconcileAnnotations(ctx context.Context, did string, pdsClient *atproto.Client) error { + // Get all repositories for this DID + repositories, err := db.GetRepositoriesForDID(b.db, did) + if err != nil { + return fmt.Errorf("failed to get repositories: %w", err) + } + + for _, repo := range repositories { + // Find newest manifest for this repository + newestManifest, err := db.GetNewestManifestForRepo(b.db, did, repo) + if err != nil { + fmt.Printf("WARNING [backfill]: Failed to get newest manifest for %s/%s: %v\n", did, repo, err) + continue // Skip on error + } + + // Fetch the full manifest record from PDS using the digest as rkey + rkey := strings.TrimPrefix(newestManifest.Digest, "sha256:") + record, err := pdsClient.GetRecord(ctx, atproto.ManifestCollection, rkey) + if err != nil { + fmt.Printf("WARNING [backfill]: Failed to fetch manifest record for %s/%s: %v\n", did, repo, err) + continue // Skip on error + } + + // Parse manifest record + var manifestRecord atproto.ManifestRecord + if err := json.Unmarshal(record.Value, &manifestRecord); err != nil { + fmt.Printf("WARNING [backfill]: Failed to parse manifest record for %s/%s: %v\n", did, repo, err) + continue + } + + // Update annotations from newest manifest only + if manifestRecord.Annotations != nil && len(manifestRecord.Annotations) > 0 { + // Filter out empty annotations + hasData := false + for _, value := range manifestRecord.Annotations { + if value != "" { + hasData = true + break + } + } + + if hasData { + err = db.UpsertRepositoryAnnotations(b.db, did, repo, manifestRecord.Annotations) + if err != nil { + fmt.Printf("WARNING [backfill]: Failed to reconcile annotations for %s/%s: %v\n", did, repo, err) + } else { + fmt.Printf("Backfill: Reconciled annotations for %s/%s from newest manifest %s\n", did, repo, newestManifest.Digest) + } + } + } + } + + return nil +} diff --git a/pkg/appview/jetstream/processor_test.go b/pkg/appview/jetstream/processor_test.go index 160091f..51e211b 100644 --- a/pkg/appview/jetstream/processor_test.go +++ b/pkg/appview/jetstream/processor_test.go @@ -39,16 +39,19 @@ func setupTestDB(t *testing.T) *sql.DB { config_digest TEXT, config_size INTEGER, created_at TIMESTAMP NOT NULL, - title TEXT, - description TEXT, - source_url TEXT, - documentation_url TEXT, - licenses TEXT, - icon_url TEXT, - readme_url TEXT, UNIQUE(did, repository, digest) ); + CREATE TABLE repository_annotations ( + did TEXT NOT NULL, + repository TEXT NOT NULL, + key TEXT NOT NULL, + value TEXT NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(did, repository, key), + FOREIGN KEY(did) REFERENCES users(did) ON DELETE CASCADE + ); + CREATE TABLE layers ( manifest_id INTEGER NOT NULL, digest TEXT NOT NULL, @@ -189,17 +192,24 @@ func TestProcessManifest_ImageManifest(t *testing.T) { t.Errorf("Expected 1 manifest, got %d", count) } - // Verify annotations were stored + // Verify annotations were stored in repository_annotations table var title, source string - err = database.QueryRow("SELECT title, source_url FROM manifests WHERE id = ?", manifestID).Scan(&title, &source) + err = database.QueryRow("SELECT value FROM repository_annotations WHERE did = ? AND repository = ? AND key = ?", + "did:plc:test123", "test-app", "org.opencontainers.image.title").Scan(&title) if err != nil { - t.Fatalf("Failed to query manifest fields: %v", err) + t.Fatalf("Failed to query title annotation: %v", err) } if title != "Test App" { t.Errorf("title = %q, want %q", title, "Test App") } + + err = database.QueryRow("SELECT value FROM repository_annotations WHERE did = ? AND repository = ? AND key = ?", + "did:plc:test123", "test-app", "org.opencontainers.image.source").Scan(&source) + if err != nil { + t.Fatalf("Failed to query source annotation: %v", err) + } if source != "https://github.com/test/app" { - t.Errorf("source_url = %q, want %q", source, "https://github.com/test/app") + t.Errorf("source = %q, want %q", source, "https://github.com/test/app") } // Verify layers were inserted @@ -523,18 +533,19 @@ func TestProcessManifest_EmptyAnnotations(t *testing.T) { t.Fatalf("Failed to marshal manifest: %v", err) } - manifestID, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes) + _, err = p.ProcessManifest(ctx, "did:plc:test123", recordBytes) if err != nil { t.Fatalf("ProcessManifest failed: %v", err) } - // Verify annotation fields are empty strings (not NULL) - var title string - err = database.QueryRow("SELECT title FROM manifests WHERE id = ?", manifestID).Scan(&title) + // Verify no annotations were stored (nil annotations should not create entries) + var annotationCount int + err = database.QueryRow("SELECT COUNT(*) FROM repository_annotations WHERE did = ? AND repository = ?", + "did:plc:test123", "test-app").Scan(&annotationCount) if err != nil { - t.Fatalf("Failed to query title: %v", err) + t.Fatalf("Failed to query annotations: %v", err) } - if title != "" { - t.Errorf("Expected empty title, got %q", title) + if annotationCount != 0 { + t.Errorf("Expected 0 annotations for nil annotations, got %d", annotationCount) } }