diff --git a/docs/ANNOTATIONS_REFACTOR.md b/docs/ANNOTATIONS_REFACTOR.md new file mode 100644 index 0000000..b0e3876 --- /dev/null +++ b/docs/ANNOTATIONS_REFACTOR.md @@ -0,0 +1,703 @@ +# Annotations Table Refactoring + +## Overview + +Refactor manifest annotations from individual columns (`title`, `description`, `source_url`, etc.) to a normalized key-value table. This enables flexible annotation storage without schema changes for new OCI annotations. + +## Motivation + +**Current Problems:** +- Each new annotation (e.g., `org.opencontainers.image.version`) requires schema change +- Many NULL columns in manifests table +- Rigid schema doesn't match OCI's flexible annotation model + +**Benefits:** +- ✅ Add any annotation without code/schema changes +- ✅ Normalized database design +- ✅ Easy to query "all repos with annotation X" +- ✅ Simple queries (no joins needed for repository pages) + +## Database Schema Changes + +### 1. New Table: `repository_annotations` + +```sql +CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_repository_annotations_did_repo ON repository_annotations(did, repository); +CREATE INDEX IF NOT EXISTS idx_repository_annotations_key ON repository_annotations(key); +``` + +**Key Design Decisions:** +- Primary key: `(did, repository, key)` - one value per annotation per repository +- No `manifest_id` foreign key - annotations are repository-level, not manifest-level +- `updated_at` - track when annotation was last updated (from most recent manifest) +- Stored at repository level because that's where they're displayed + +### 2. Drop Columns from `manifests` Table + +Remove these columns (migration will preserve data by copying to annotations table): +- `title` +- `description` +- `source_url` +- `documentation_url` +- `licenses` +- `icon_url` +- `readme_url` +- `version` + +Keep only core manifest metadata: +- `id`, `did`, `repository`, `digest` +- `hold_endpoint`, `schema_version`, `media_type` +- `config_digest`, `config_size` +- `created_at` + +## Migration Strategy + +### Migration File: `0004_refactor_annotations_table.yaml` + +```yaml +description: Migrate manifest annotations to separate table +query: | + -- Step 1: Create new annotations table + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_repository_annotations_did_repo ON repository_annotations(did, repository); + CREATE INDEX IF NOT EXISTS idx_repository_annotations_key ON repository_annotations(key); + + -- Step 2: Migrate existing data from manifests to annotations + -- For each repository, use the most recent manifest with non-empty data + INSERT OR REPLACE INTO repository_annotations (did, repository, key, value, updated_at) + SELECT + m.did, + m.repository, + 'org.opencontainers.image.title' as key, + m.title as value, + m.created_at as updated_at + FROM manifests m + WHERE m.title IS NOT NULL AND m.title != '' + AND m.created_at = ( + SELECT MAX(created_at) FROM manifests m2 + WHERE m2.did = m.did AND m2.repository = m.repository + AND m2.title IS NOT NULL AND m2.title != '' + ); + + INSERT OR REPLACE INTO repository_annotations (did, repository, key, value, updated_at) + SELECT m.did, m.repository, 'org.opencontainers.image.description', m.description, m.created_at + FROM manifests m + WHERE m.description IS NOT NULL AND m.description != '' + AND m.created_at = ( + SELECT MAX(created_at) FROM manifests m2 + WHERE m2.did = m.did AND m2.repository = m.repository + AND m2.description IS NOT NULL AND m2.description != '' + ); + + INSERT OR REPLACE INTO repository_annotations (did, repository, key, value, updated_at) + SELECT m.did, m.repository, 'org.opencontainers.image.source', m.source_url, m.created_at + FROM manifests m + WHERE m.source_url IS NOT NULL AND m.source_url != '' + AND m.created_at = ( + SELECT MAX(created_at) FROM manifests m2 + WHERE m2.did = m.did AND m2.repository = m.repository + AND m2.source_url IS NOT NULL AND m2.source_url != '' + ); + + INSERT OR REPLACE INTO repository_annotations (did, repository, key, value, updated_at) + SELECT m.did, m.repository, 'org.opencontainers.image.documentation', m.documentation_url, m.created_at + FROM manifests m + WHERE m.documentation_url IS NOT NULL AND m.documentation_url != '' + AND m.created_at = ( + SELECT MAX(created_at) FROM manifests m2 + WHERE m2.did = m.did AND m2.repository = m.repository + AND m2.documentation_url IS NOT NULL AND m2.documentation_url != '' + ); + + INSERT OR REPLACE INTO repository_annotations (did, repository, key, value, updated_at) + SELECT m.did, m.repository, 'org.opencontainers.image.licenses', m.licenses, m.created_at + FROM manifests m + WHERE m.licenses IS NOT NULL AND m.licenses != '' + AND m.created_at = ( + SELECT MAX(created_at) FROM manifests m2 + WHERE m2.did = m.did AND m2.repository = m.repository + AND m2.licenses IS NOT NULL AND m2.licenses != '' + ); + + INSERT OR REPLACE INTO repository_annotations (did, repository, key, value, updated_at) + SELECT m.did, m.repository, 'io.atcr.icon', m.icon_url, m.created_at + FROM manifests m + WHERE m.icon_url IS NOT NULL AND m.icon_url != '' + AND m.created_at = ( + SELECT MAX(created_at) FROM manifests m2 + WHERE m2.did = m.did AND m2.repository = m.repository + AND m2.icon_url IS NOT NULL AND m2.icon_url != '' + ); + + INSERT OR REPLACE INTO repository_annotations (did, repository, key, value, updated_at) + SELECT m.did, m.repository, 'io.atcr.readme', m.readme_url, m.created_at + FROM manifests m + WHERE m.readme_url IS NOT NULL AND m.readme_url != '' + AND m.created_at = ( + SELECT MAX(created_at) FROM manifests m2 + WHERE m2.did = m.did AND m2.repository = m.repository + AND m2.readme_url IS NOT NULL AND m2.readme_url != '' + ); + + -- Step 3: Drop old columns from manifests table + -- SQLite requires recreating table to drop columns + CREATE TABLE manifests_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + did TEXT NOT NULL, + repository TEXT NOT NULL, + digest TEXT NOT NULL, + hold_endpoint TEXT NOT NULL, + schema_version INTEGER NOT NULL, + media_type TEXT NOT NULL, + config_digest TEXT, + config_size INTEGER, + created_at TIMESTAMP NOT NULL, + UNIQUE(did, repository, digest), + FOREIGN KEY(did) REFERENCES users(did) ON DELETE CASCADE + ); + + -- Copy data to new table + INSERT INTO manifests_new + SELECT id, did, repository, digest, hold_endpoint, schema_version, media_type, + config_digest, config_size, created_at + FROM manifests; + + -- Replace old table + DROP TABLE manifests; + ALTER TABLE manifests_new RENAME TO manifests; + + -- Recreate indexes + CREATE INDEX IF NOT EXISTS idx_manifests_did_repo ON manifests(did, repository); + CREATE INDEX IF NOT EXISTS idx_manifests_created_at ON manifests(created_at DESC); + CREATE INDEX IF NOT EXISTS idx_manifests_digest ON manifests(digest); +``` + +## Code Changes + +### 1. Database Helper Functions + +**New file: `pkg/appview/db/annotations.go`** + +```go +package db + +import ( + "database/sql" + "time" +) + +// GetRepositoryAnnotations retrieves all annotations for a repository +func GetRepositoryAnnotations(db *sql.DB, did, repository string) (map[string]string, error) { + rows, err := db.Query(` + SELECT key, value + FROM repository_annotations + WHERE did = ? AND repository = ? + `, did, repository) + if err != nil { + return nil, err + } + defer rows.Close() + + annotations := make(map[string]string) + for rows.Next() { + var key, value string + if err := rows.Scan(&key, &value); err != nil { + return nil, err + } + annotations[key] = value + } + + return annotations, rows.Err() +} + +// UpsertRepositoryAnnotations replaces all annotations for a repository +// Only called when manifest has at least one non-empty annotation +func UpsertRepositoryAnnotations(db *sql.DB, did, repository string, annotations map[string]string) error { + tx, err := db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + // Delete existing annotations + _, err = tx.Exec(` + DELETE FROM repository_annotations + WHERE did = ? AND repository = ? + `, did, repository) + if err != nil { + return err + } + + // Insert new annotations + stmt, err := tx.Prepare(` + INSERT INTO repository_annotations (did, repository, key, value, updated_at) + VALUES (?, ?, ?, ?, ?) + `) + if err != nil { + return err + } + defer stmt.Close() + + now := time.Now() + for key, value := range annotations { + _, err = stmt.Exec(did, repository, key, value, now) + if err != nil { + return err + } + } + + return tx.Commit() +} + +// DeleteRepositoryAnnotations removes all annotations for a repository +func DeleteRepositoryAnnotations(db *sql.DB, did, repository string) error { + _, err := db.Exec(` + DELETE FROM repository_annotations + WHERE did = ? AND repository = ? + `, did, repository) + return err +} +``` + +### 2. Update Backfill Worker + +**File: `pkg/appview/jetstream/backfill.go`** + +In `processManifestRecord()` function, after extracting annotations: + +```go +// Extract OCI annotations from manifest +var title, description, sourceURL, documentationURL, licenses, iconURL, readmeURL 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"] + readmeURL = manifestRecord.Annotations["io.atcr.readme"] +} + +// Prepare manifest for insertion (WITHOUT annotation fields) +manifest := &db.Manifest{ + DID: did, + Repository: manifestRecord.Repository, + Digest: manifestRecord.Digest, + MediaType: manifestRecord.MediaType, + SchemaVersion: manifestRecord.SchemaVersion, + HoldEndpoint: manifestRecord.HoldEndpoint, + CreatedAt: manifestRecord.CreatedAt, + // NO annotation fields +} + +// Set config fields only for image manifests (not manifest lists) +if !isManifestList && manifestRecord.Config != nil { + manifest.ConfigDigest = manifestRecord.Config.Digest + manifest.ConfigSize = manifestRecord.Config.Size +} + +// Insert manifest +manifestID, err := db.InsertManifest(b.db, manifest) +if err != nil { + return fmt.Errorf("failed to insert manifest: %w", err) +} + +// Update repository annotations ONLY if manifest has at least one non-empty annotation +if manifestRecord.Annotations != nil { + hasData := false + for _, value := range manifestRecord.Annotations { + if value != "" { + hasData = true + break + } + } + + if hasData { + // Replace all annotations for this repository + err = db.UpsertRepositoryAnnotations(b.db, did, manifestRecord.Repository, manifestRecord.Annotations) + if err != nil { + return fmt.Errorf("failed to upsert annotations: %w", err) + } + } +} +``` + +### 3. Update Jetstream Worker + +**File: `pkg/appview/jetstream/worker.go`** + +Same changes as backfill - in `processManifestCommit()` function: + +```go +// Extract OCI annotations from manifest +var title, description, sourceURL, documentationURL, licenses, iconURL, readmeURL 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"] + readmeURL = manifestRecord.Annotations["io.atcr.readme"] +} + +// Prepare manifest for insertion (WITHOUT annotation fields) +manifest := &db.Manifest{ + DID: commit.DID, + Repository: manifestRecord.Repository, + Digest: manifestRecord.Digest, + MediaType: manifestRecord.MediaType, + SchemaVersion: manifestRecord.SchemaVersion, + HoldEndpoint: manifestRecord.HoldEndpoint, + CreatedAt: manifestRecord.CreatedAt, + // NO annotation fields +} + +// Set config fields only for image manifests (not manifest lists) +if !isManifestList && manifestRecord.Config != nil { + manifest.ConfigDigest = manifestRecord.Config.Digest + manifest.ConfigSize = manifestRecord.Config.Size +} + +// Insert manifest +manifestID, err := db.InsertManifest(w.db, manifest) +if err != nil { + return fmt.Errorf("failed to insert manifest: %w", err) +} + +// Update repository annotations ONLY if manifest has at least one non-empty annotation +if manifestRecord.Annotations != nil { + hasData := false + for _, value := range manifestRecord.Annotations { + if value != "" { + hasData = true + break + } + } + + if hasData { + // Replace all annotations for this repository + err = db.UpsertRepositoryAnnotations(w.db, commit.DID, manifestRecord.Repository, manifestRecord.Annotations) + if err != nil { + return fmt.Errorf("failed to upsert annotations: %w", err) + } + } +} +``` + +### 4. Update Database Queries + +**File: `pkg/appview/db/queries.go`** + +Replace `GetRepositoryMetadata()` function: + +```go +// GetRepositoryMetadata retrieves metadata for a repository from annotations table +func GetRepositoryMetadata(db *sql.DB, did string, repository string) (title, description, sourceURL, documentationURL, licenses, iconURL, readmeURL, version string, err error) { + annotations, err := GetRepositoryAnnotations(db, did, repository) + if err != nil { + return "", "", "", "", "", "", "", "", err + } + + title = annotations["org.opencontainers.image.title"] + description = annotations["org.opencontainers.image.description"] + sourceURL = annotations["org.opencontainers.image.source"] + documentationURL = annotations["org.opencontainers.image.documentation"] + licenses = annotations["org.opencontainers.image.licenses"] + iconURL = annotations["io.atcr.icon"] + readmeURL = annotations["io.atcr.readme"] + version = annotations["org.opencontainers.image.version"] + + return title, description, sourceURL, documentationURL, licenses, iconURL, readmeURL, version, nil +} +``` + +Update `InsertManifest()` to remove annotation columns: + +```go +func InsertManifest(db *sql.DB, manifest *Manifest) (int64, error) { + _, err := db.Exec(` + INSERT INTO manifests + (did, repository, digest, hold_endpoint, schema_version, media_type, + config_digest, config_size, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(did, repository, digest) DO UPDATE SET + hold_endpoint = excluded.hold_endpoint, + schema_version = excluded.schema_version, + media_type = excluded.media_type, + config_digest = excluded.config_digest, + config_size = excluded.config_size + `, manifest.DID, manifest.Repository, manifest.Digest, manifest.HoldEndpoint, + manifest.SchemaVersion, manifest.MediaType, manifest.ConfigDigest, + manifest.ConfigSize, manifest.CreatedAt) + + if err != nil { + return 0, err + } + + // Query for the ID (works for both insert and update) + var id int64 + err = db.QueryRow(` + SELECT id FROM manifests + WHERE did = ? AND repository = ? AND digest = ? + `, manifest.DID, manifest.Repository, manifest.Digest).Scan(&id) + + if err != nil { + return 0, fmt.Errorf("failed to get manifest ID after upsert: %w", err) + } + + return id, nil +} +``` + +Similar updates needed for: +- `GetUserRepositories()` - fetch annotations separately and populate Repository struct +- `GetRecentPushes()` - join with annotations or fetch separately +- `SearchPushes()` - can now search annotations table directly + +### 5. Update Models + +**File: `pkg/appview/db/models.go`** + +Remove annotation fields from `Manifest` struct: + +```go +type Manifest struct { + ID int64 + DID string + Repository string + Digest string + HoldEndpoint string + SchemaVersion int + MediaType string + ConfigDigest string + ConfigSize int64 + CreatedAt time.Time + // Removed: Title, Description, SourceURL, DocumentationURL, Licenses, IconURL, ReadmeURL +} +``` + +Keep annotation fields on `Repository` struct (populated from annotations table): + +```go +type Repository struct { + Name string + TagCount int + ManifestCount int + LastPush time.Time + Tags []Tag + Manifests []Manifest + Title string + Description string + SourceURL string + DocumentationURL string + Licenses string + IconURL string + ReadmeURL string + Version string // NEW +} +``` + +### 6. Update Schema.sql + +**File: `pkg/appview/db/schema.sql`** + +Add new table: + +```sql +CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_repository_annotations_did_repo ON repository_annotations(did, repository); +CREATE INDEX IF NOT EXISTS idx_repository_annotations_key ON repository_annotations(key); +``` + +Update manifests table (remove annotation columns): + +```sql +CREATE TABLE IF NOT EXISTS manifests ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + did TEXT NOT NULL, + repository TEXT NOT NULL, + digest TEXT NOT NULL, + hold_endpoint TEXT NOT NULL, + schema_version INTEGER NOT NULL, + media_type TEXT NOT NULL, + config_digest TEXT, + config_size INTEGER, + created_at TIMESTAMP NOT NULL, + UNIQUE(did, repository, digest), + FOREIGN KEY(did) REFERENCES users(did) ON DELETE CASCADE +); +``` + +## Update Logic Summary + +**Key Decision: Only update annotations when manifest has data** + +``` +For each manifest processed (backfill or jetstream): + 1. Parse manifest.Annotations map + 2. Check if ANY annotation has non-empty value + 3. IF hasData: + DELETE all annotations for (did, repository) + INSERT all annotations from manifest (including empty ones) + ELSE: + SKIP (don't touch existing annotations) +``` + +**Why this works:** +- Manifest lists have no annotations or all empty → skip, preserve existing +- Platform manifests have real data → replace everything +- Removing annotation from Dockerfile → it's gone (not in new INSERT) +- Can't accidentally clear data (need at least one non-empty value) + +## UI/Template Changes + +### Handler Updates + +**File: `pkg/appview/handlers/repository.go`** + +Update the handler to include version: + +```go +// Fetch repository metadata from annotations +title, description, sourceURL, documentationURL, licenses, iconURL, readmeURL, version, err := db.GetRepositoryMetadata(h.DB, owner.DID, repository) +if err != nil { + log.Printf("Failed to fetch repository metadata: %v", err) + // Continue without metadata on error +} else { + repo.Title = title + repo.Description = description + repo.SourceURL = sourceURL + repo.DocumentationURL = documentationURL + repo.Licenses = licenses + repo.IconURL = iconURL + repo.ReadmeURL = readmeURL + repo.Version = version // NEW +} +``` + +### Template Updates + +**File: `pkg/appview/templates/pages/repository.html`** + +Update the metadata section condition to include version: + +```html + +{{ if or .Repository.Licenses .Repository.SourceURL .Repository.DocumentationURL .Repository.Version }} +
+ + {{ if .Repository.Version }} + + {{ .Repository.Version }} + + {{ end }} + + + {{ if .Repository.Licenses }} + {{ range parseLicenses .Repository.Licenses }} + {{ if .IsValid }} + + {{ .SPDXID }} + + {{ else }} + + {{ .Name }} + + {{ end }} + {{ end }} + {{ end }} + + + {{ if .Repository.SourceURL }} + + Source + + {{ end }} + + + {{ if .Repository.DocumentationURL }} + + Documentation + + {{ end }} +
+{{ end }} +``` + +### CSS Updates + +**File: `pkg/appview/static/css/style.css`** + +Add styling for version badge (different color from license badge): + +```css +.version-badge { + background: #0969da; /* GitHub blue */ + color: white; + padding: 0.25rem 0.5rem; + border-radius: 0.25rem; + font-size: 0.875rem; + font-weight: 500; + display: inline-block; +} +``` + +### Data Flow Summary + +**Before refactor:** +``` +DB columns → GetRepositoryMetadata() → Handler assigns to Repository struct → Template displays +``` + +**After refactor:** +``` +annotations table → GetRepositoryAnnotations() → GetRepositoryMetadata() extracts known fields → +Handler assigns to Repository struct → Template displays (same as before) +``` + +**Key point:** Templates still access `.Repository.Title`, `.Repository.Version`, etc. - the source just changed from DB columns to annotations table. The abstraction layer hides this complexity. + +## Benefits Recap + +1. **Flexible**: Support any OCI annotation without code changes +2. **Clean**: No NULL columns in manifests table +3. **Simple queries**: `SELECT * FROM repository_annotations WHERE did=? AND repo=?` +4. **Safe updates**: Only update when manifest has data +5. **Natural deletion**: Remove annotation from Dockerfile → it's deleted on next push +6. **Extensible**: Future features (annotation search, filtering) are trivial + +## Testing Checklist + +After migration: +- [ ] Verify existing repositories show annotations correctly +- [ ] Push new manifest with annotations → updates correctly +- [ ] Push manifest list → doesn't clear annotations +- [ ] Remove annotation from Dockerfile and push → annotation deleted +- [ ] Backfill re-run → annotations repopulated correctly +- [ ] Search still works (if implemented) diff --git a/pkg/appview/db/hold_store.go b/pkg/appview/db/hold_store.go index 9d0572a..8f6bb97 100644 --- a/pkg/appview/db/hold_store.go +++ b/pkg/appview/db/hold_store.go @@ -8,14 +8,14 @@ import ( // HoldCaptainRecord represents a cached captain record from a hold's PDS type HoldCaptainRecord struct { - HoldDID string - OwnerDID string - Public bool - AllowAllCrew bool - DeployedAt string - Region string - Provider string - UpdatedAt time.Time + HoldDID string `json:"-"` // Set manually, not from JSON + OwnerDID string `json:"owner"` + Public bool `json:"public"` + AllowAllCrew bool `json:"allowAllCrew"` + DeployedAt string `json:"deployedAt"` + Region string `json:"region"` + Provider string `json:"provider"` + UpdatedAt time.Time `json:"-"` // Set manually, not from JSON } // GetCaptainRecord retrieves a captain record from the cache diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index 0bb4321..f6eeade 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -8,9 +8,9 @@ import ( "strings" "time" - "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" + "atcr.io/pkg/appview" "atcr.io/pkg/appview/db" "atcr.io/pkg/atproto" ) @@ -19,9 +19,9 @@ import ( type BackfillWorker struct { db *sql.DB client *atproto.Client - directory identity.Directory - defaultHoldDID string // Default hold DID from AppView config (e.g., "did:web:hold01.atcr.io") - testMode bool // If true, suppress warnings for external holds + processor *Processor // Shared processor for DB operations + defaultHoldDID string // Default hold DID from AppView config (e.g., "did:web:hold01.atcr.io") + testMode bool // If true, suppress warnings for external holds } // BackfillState tracks backfill progress @@ -44,8 +44,8 @@ func NewBackfillWorker(database *sql.DB, relayEndpoint, defaultHoldDID string, t return &BackfillWorker{ db: database, - client: client, // This points to the relay - directory: identity.DefaultDirectory(), + client: client, // This points to the relay + processor: NewProcessor(database, false), // No cache for batch processing defaultHoldDID: defaultHoldDID, testMode: testMode, }, nil @@ -132,7 +132,7 @@ 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 and get their PDS endpoint - if err := b.ensureUser(ctx, did); err != nil { + if err := b.processor.EnsureUser(ctx, did); err != nil { return 0, fmt.Errorf("failed to ensure user: %w", err) } @@ -142,7 +142,7 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin return 0, fmt.Errorf("invalid DID %s: %w", did, err) } - ident, err := b.directory.LookupDID(ctx, didParsed) + ident, err := b.processor.directory.LookupDID(ctx, didParsed) if err != nil { return 0, fmt.Errorf("failed to resolve DID to PDS: %w", err) } @@ -173,12 +173,13 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin // Process each record for _, record := range records { // Track what we found for deletion reconciliation - if collection == atproto.ManifestCollection { + switch collection { + case 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 { + case atproto.TagCollection: var tagRecord atproto.TagRecord if err := json.Unmarshal(record.Value, &tagRecord); err == nil { foundTags = append(foundTags, struct{ Repository, Tag string }{ @@ -186,7 +187,7 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin Tag: tagRecord.Tag, }) } - } else if collection == atproto.StarCollection { + case 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) @@ -278,195 +279,21 @@ func (b *BackfillWorker) reconcileDeletions(did, collection string, foundManifes func (b *BackfillWorker) processRecord(ctx context.Context, did, collection string, record *atproto.Record) error { switch collection { case atproto.ManifestCollection: - return b.processManifestRecord(did, record) + _, err := b.processor.ProcessManifest(context.Background(), did, record.Value) + return err case atproto.TagCollection: - return b.processTagRecord(did, record) + return b.processor.ProcessTag(context.Background(), did, record.Value) case atproto.StarCollection: - return b.processStarRecord(did, record) + return b.processor.ProcessStar(context.Background(), did, record.Value) case atproto.SailorProfileCollection: - return b.processSailorProfileRecord(ctx, did, record) + return b.processor.ProcessSailorProfile(ctx, did, record.Value, b.queryCaptainRecordWrapper) 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, readmeURL 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"] - readmeURL = manifestRecord.Annotations["io.atcr.readme"] - } - - // Detect manifest type - isManifestList := len(manifestRecord.Manifests) > 0 - - // Prepare manifest for insertion - manifest := &db.Manifest{ - DID: did, - Repository: manifestRecord.Repository, - Digest: manifestRecord.Digest, - MediaType: manifestRecord.MediaType, - SchemaVersion: manifestRecord.SchemaVersion, - HoldEndpoint: manifestRecord.HoldEndpoint, - CreatedAt: manifestRecord.CreatedAt, - Title: title, - Description: description, - SourceURL: sourceURL, - DocumentationURL: documentationURL, - Licenses: licenses, - IconURL: iconURL, - ReadmeURL: readmeURL, - } - - // Set config fields only for image manifests (not manifest lists) - if !isManifestList && manifestRecord.Config != nil { - manifest.ConfigDigest = manifestRecord.Config.Digest - manifest.ConfigSize = manifestRecord.Config.Size - } - - // Platform info is only stored for multi-arch images in manifest_references table - // Single-arch images don't need platform display (it's obvious) - - // Insert manifest (or get existing ID if already exists) - manifestID, err := db.InsertManifest(b.db, manifest) - if err != nil { - // If manifest already exists, get its ID so we can still insert references/layers - if strings.Contains(err.Error(), "UNIQUE constraint failed") { - // Query for existing manifest ID - var existingID int64 - err := b.db.QueryRow(` - SELECT id FROM manifests - WHERE did = ? AND repository = ? AND digest = ? - `, manifest.DID, manifest.Repository, manifest.Digest).Scan(&existingID) - - if err != nil { - return fmt.Errorf("failed to get existing manifest ID: %w", err) - } - manifestID = existingID - } else { - return fmt.Errorf("failed to insert manifest: %w", err) - } - } - - if isManifestList { - // Insert manifest references (for manifest lists/indexes) - for i, ref := range manifestRecord.Manifests { - platformArch := "" - platformOS := "" - platformVariant := "" - platformOSVersion := "" - - if ref.Platform != nil { - platformArch = ref.Platform.Architecture - platformOS = ref.Platform.OS - platformVariant = ref.Platform.Variant - platformOSVersion = ref.Platform.OSVersion - } - - if err := db.InsertManifestReference(b.db, &db.ManifestReference{ - ManifestID: manifestID, - Digest: ref.Digest, - MediaType: ref.MediaType, - Size: ref.Size, - PlatformArchitecture: platformArch, - PlatformOS: platformOS, - PlatformVariant: platformVariant, - PlatformOSVersion: platformOSVersion, - ReferenceIndex: i, - }); err != nil { - // Continue on error - reference might already exist - continue - } - } - } else { - // Insert layers (for image manifests) - 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) - } - - // Extract digest from tag record (tries manifest field first, falls back to manifestDigest) - manifestDigest, err := tagRecord.GetManifestDigest() - if err != nil { - return fmt.Errorf("failed to get manifest digest from tag record: %w", err) - } - - // Insert or update tag - return db.UpsertTag(b.db, &db.Tag{ - DID: did, - Repository: tagRecord.Repository, - Tag: tagRecord.Tag, - Digest: 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) -} - -// processSailorProfileRecord processes a sailor profile record -// Extracts defaultHold and queries the hold's captain record to cache it -func (b *BackfillWorker) processSailorProfileRecord(ctx context.Context, did string, record *atproto.Record) error { - var profileRecord atproto.SailorProfileRecord - if err := json.Unmarshal(record.Value, &profileRecord); err != nil { - return fmt.Errorf("failed to unmarshal sailor profile: %w", err) - } - - // Skip if no default hold set - if profileRecord.DefaultHold == "" { - return nil - } - - // Convert hold URL/DID to canonical DID - holdDID := atproto.ResolveHoldDIDFromURL(profileRecord.DefaultHold) - if holdDID == "" { - fmt.Printf("WARNING [backfill]: Invalid hold reference in profile for %s: %s\n", did, profileRecord.DefaultHold) - return nil - } - - // Query and cache the captain record +// queryCaptainRecordWrapper wraps queryCaptainRecord with backfill-specific logic +func (b *BackfillWorker) queryCaptainRecordWrapper(ctx context.Context, holdDID string) error { if err := b.queryCaptainRecord(ctx, holdDID); err != nil { // In test mode, only warn about default hold (local hold) // External/production holds may not have captain records yet (dev ahead of prod) @@ -478,7 +305,6 @@ func (b *BackfillWorker) processSailorProfileRecord(ctx context.Context, did str // Don't fail the whole backfill - just skip this hold return nil } - return nil } @@ -494,11 +320,7 @@ func (b *BackfillWorker) queryCaptainRecord(ctx context.Context, holdDID string) } // Resolve hold DID to URL - // For did:web, we need to fetch .well-known/did.json - holdURL, err := resolveHoldDIDToURL(ctx, holdDID) - if err != nil { - return fmt.Errorf("failed to resolve hold DID to URL: %w", err) - } + holdURL := appview.ResolveHoldURL(holdDID) // Create client for hold's PDS holdClient := atproto.NewClient(holdURL, holdDID, "") @@ -522,150 +344,20 @@ func (b *BackfillWorker) queryCaptainRecord(ctx context.Context, holdDID string) return fmt.Errorf("failed to get captain record: %w", err) } - // Parse captain record from the record's Value field - var captainRecord struct { - Owner string `json:"owner"` - Public bool `json:"public"` - AllowAllCrew bool `json:"allowAllCrew"` - DeployedAt string `json:"deployedAt"` - Region string `json:"region"` - Provider string `json:"provider"` - } - + // Parse captain record directly into db struct + var captainRecord db.HoldCaptainRecord if err := json.Unmarshal(record.Value, &captainRecord); err != nil { return fmt.Errorf("failed to parse captain record: %w", err) } - // Cache in database - dbRecord := &db.HoldCaptainRecord{ - HoldDID: holdDID, - OwnerDID: captainRecord.Owner, - Public: captainRecord.Public, - AllowAllCrew: captainRecord.AllowAllCrew, - DeployedAt: captainRecord.DeployedAt, - Region: captainRecord.Region, - Provider: captainRecord.Provider, - UpdatedAt: time.Now(), - } + // Set fields not from JSON + captainRecord.HoldDID = holdDID + captainRecord.UpdatedAt = time.Now() - if err := db.UpsertCaptainRecord(b.db, dbRecord); err != nil { + if err := db.UpsertCaptainRecord(b.db, &captainRecord); err != nil { return fmt.Errorf("failed to cache captain record: %w", err) } - fmt.Printf("Backfill: Cached captain record for hold %s (owner: %s)\n", holdDID, captainRecord.Owner) + fmt.Printf("Backfill: Cached captain record for hold %s (owner: %s)\n", holdDID, captainRecord.OwnerDID) return nil } - -// resolveHoldDIDToURL resolves a hold DID to its service endpoint URL -// Fetches the DID document and returns both the canonical DID and service endpoint -func resolveHoldDIDToURL(ctx context.Context, inputDID string) (string, error) { - // For did:web, construct the .well-known URL - if !strings.HasPrefix(inputDID, "did:web:") { - return "", fmt.Errorf("only did:web is supported, got: %s", inputDID) - } - - // Extract hostname from did:web:hostname[:port] - hostname := strings.TrimPrefix(inputDID, "did:web:") - - // Try HTTP first (for local Docker), then HTTPS - var serviceEndpoint string - for _, scheme := range []string{"http", "https"} { - testURL := fmt.Sprintf("%s://%s/.well-known/did.json", scheme, hostname) - - // Fetch DID document (use NewClient to initialize httpClient) - client := atproto.NewClient("", "", "") - didDoc, err := client.FetchDIDDocument(ctx, testURL) - if err == nil && didDoc != nil { - // Extract service endpoint from DID document - for _, service := range didDoc.Service { - if service.Type == "AtprotoPersonalDataServer" || service.Type == "AtcrHoldService" { - serviceEndpoint = service.ServiceEndpoint - break - } - } - - if serviceEndpoint != "" { - fmt.Printf("DEBUG [backfill]: Resolved %s → canonical DID: %s, endpoint: %s\n", - inputDID, didDoc.ID, serviceEndpoint) - return serviceEndpoint, nil - } - } - } - - // Fallback: assume the hold service is at the root of the hostname - // Try HTTP first for local development - url := fmt.Sprintf("http://%s", hostname) - fmt.Printf("WARNING [backfill]: Failed to fetch DID document for %s, using fallback URL: %s\n", inputDID, url) - return url, nil -} - -// 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) -} diff --git a/pkg/appview/jetstream/processor.go b/pkg/appview/jetstream/processor.go new file mode 100644 index 0000000..aae5b9f --- /dev/null +++ b/pkg/appview/jetstream/processor.go @@ -0,0 +1,320 @@ +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" +) + +// Processor handles shared database operations for both Worker (live) and Backfill (sync) +// This eliminates code duplication between the two data ingestion paths +type Processor struct { + db *sql.DB + directory identity.Directory + userCache *UserCache // Optional - enabled for Worker, disabled for Backfill + useCache bool +} + +// NewProcessor creates a new shared processor +// useCache: true for Worker (live streaming), false for Backfill (batch processing) +func NewProcessor(database *sql.DB, useCache bool) *Processor { + p := &Processor{ + db: database, + directory: identity.DefaultDirectory(), + useCache: useCache, + } + + if useCache { + p.userCache = &UserCache{ + cache: make(map[string]*db.User), + } + } + + return p +} + +// EnsureUser resolves and upserts a user by DID +// Uses cache if enabled (Worker), queries DB if cache disabled (Backfill) +func (p *Processor) EnsureUser(ctx context.Context, did string) error { + // Check cache first (if enabled) + if p.useCache && p.userCache != nil { + if user, ok := p.userCache.cache[did]; ok { + // Update last seen + user.LastSeen = time.Now() + return db.UpsertUser(p.db, user) + } + } else if !p.useCache { + // No cache - check if user already exists in DB + existingUser, err := db.GetUserByDID(p.db, did) + if err == nil && existingUser != nil { + // Update last seen + existingUser.LastSeen = time.Now() + return db.UpsertUser(p.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(), + } + if p.useCache { + p.userCache.cache[did] = user + } + return db.UpsertUser(p.db, user) + } + + ident, err := p.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(), + } + if p.useCache { + p.userCache.cache[did] = user + } + return db.UpsertUser(p.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 [processor]: Failed to fetch profile for DID %s: %v\n", resolvedDID, err) + // Continue without avatar + } else { + avatar = profile.Avatar + } + + // Create user record + user := &db.User{ + DID: resolvedDID, + Handle: handle, + PDSEndpoint: pdsEndpoint, + Avatar: avatar, + LastSeen: time.Now(), + } + + // Cache if enabled + if p.useCache { + p.userCache.cache[did] = user + } + + // Upsert to database + return db.UpsertUser(p.db, user) +} + +// ProcessManifest processes a manifest record and stores it in the database +// Returns the manifest ID for further processing (layers/references) +func (p *Processor) ProcessManifest(ctx context.Context, did string, recordData []byte) (int64, error) { + // Unmarshal manifest record + var manifestRecord atproto.ManifestRecord + if err := json.Unmarshal(recordData, &manifestRecord); err != nil { + return 0, fmt.Errorf("failed to unmarshal manifest: %w", err) + } + // Extract OCI annotations from manifest + var title, description, sourceURL, documentationURL, licenses, iconURL, readmeURL 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"] + readmeURL = manifestRecord.Annotations["io.atcr.readme"] + } + + // Detect manifest type + isManifestList := len(manifestRecord.Manifests) > 0 + + // Prepare manifest for insertion + manifest := &db.Manifest{ + DID: did, + Repository: manifestRecord.Repository, + Digest: manifestRecord.Digest, + MediaType: manifestRecord.MediaType, + SchemaVersion: manifestRecord.SchemaVersion, + HoldEndpoint: manifestRecord.HoldEndpoint, + CreatedAt: manifestRecord.CreatedAt, + Title: title, + Description: description, + SourceURL: sourceURL, + DocumentationURL: documentationURL, + Licenses: licenses, + IconURL: iconURL, + ReadmeURL: readmeURL, + } + + // Set config fields only for image manifests (not manifest lists) + if !isManifestList && manifestRecord.Config != nil { + manifest.ConfigDigest = manifestRecord.Config.Digest + manifest.ConfigSize = manifestRecord.Config.Size + } + + // Insert manifest + manifestID, err := db.InsertManifest(p.db, manifest) + if err != nil { + // For backfill: if manifest already exists, get its ID + if strings.Contains(err.Error(), "UNIQUE constraint failed") { + var existingID int64 + err := p.db.QueryRow(` + SELECT id FROM manifests + WHERE did = ? AND repository = ? AND digest = ? + `, manifest.DID, manifest.Repository, manifest.Digest).Scan(&existingID) + + if err != nil { + return 0, fmt.Errorf("failed to get existing manifest ID: %w", err) + } + manifestID = existingID + } else { + return 0, fmt.Errorf("failed to insert manifest: %w", err) + } + } + + // Insert manifest references or layers + if isManifestList { + // Insert manifest references (for manifest lists/indexes) + for i, ref := range manifestRecord.Manifests { + platformArch := "" + platformOS := "" + platformVariant := "" + platformOSVersion := "" + + if ref.Platform != nil { + platformArch = ref.Platform.Architecture + platformOS = ref.Platform.OS + platformVariant = ref.Platform.Variant + platformOSVersion = ref.Platform.OSVersion + } + + if err := db.InsertManifestReference(p.db, &db.ManifestReference{ + ManifestID: manifestID, + Digest: ref.Digest, + MediaType: ref.MediaType, + Size: ref.Size, + PlatformArchitecture: platformArch, + PlatformOS: platformOS, + PlatformVariant: platformVariant, + PlatformOSVersion: platformOSVersion, + ReferenceIndex: i, + }); err != nil { + // Continue on error - reference might already exist + continue + } + } + } else { + // Insert layers (for image manifests) + for i, layer := range manifestRecord.Layers { + if err := db.InsertLayer(p.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 manifestID, nil +} + +// ProcessTag processes a tag record and stores it in the database +func (p *Processor) ProcessTag(ctx context.Context, did string, recordData []byte) error { + // Unmarshal tag record + var tagRecord atproto.TagRecord + if err := json.Unmarshal(recordData, &tagRecord); err != nil { + return fmt.Errorf("failed to unmarshal tag: %w", err) + } + // Extract digest from tag record (tries manifest field first, falls back to manifestDigest) + manifestDigest, err := tagRecord.GetManifestDigest() + if err != nil { + return fmt.Errorf("failed to get manifest digest from tag record: %w", err) + } + + // Insert or update tag + return db.UpsertTag(p.db, &db.Tag{ + DID: did, + Repository: tagRecord.Repository, + Tag: tagRecord.Tag, + Digest: manifestDigest, + CreatedAt: tagRecord.UpdatedAt, + }) +} + +// ProcessStar processes a star record and stores it in the database +func (p *Processor) ProcessStar(ctx context.Context, did string, recordData []byte) error { + // Unmarshal star record + var starRecord atproto.StarRecord + if err := json.Unmarshal(recordData, &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(p.db, did, starRecord.Subject.DID, starRecord.Subject.Repository, starRecord.CreatedAt) +} + +// ProcessSailorProfile processes a sailor profile record +// This is primarily used by backfill to cache captain records for holds +func (p *Processor) ProcessSailorProfile(ctx context.Context, did string, recordData []byte, queryCaptainFn func(context.Context, string) error) error { + // Unmarshal sailor profile record + var profileRecord atproto.SailorProfileRecord + if err := json.Unmarshal(recordData, &profileRecord); err != nil { + return fmt.Errorf("failed to unmarshal sailor profile: %w", err) + } + + // Skip if no default hold set + if profileRecord.DefaultHold == "" { + return nil + } + + // Convert hold URL/DID to canonical DID + holdDID := atproto.ResolveHoldDIDFromURL(profileRecord.DefaultHold) + if holdDID == "" { + fmt.Printf("WARNING [processor]: Invalid hold reference in profile for %s: %s\n", did, profileRecord.DefaultHold) + return nil + } + + // Query and cache the captain record using provided function + // This allows backfill-specific logic (retries, test mode handling) without duplicating it here + if queryCaptainFn != nil { + return queryCaptainFn(ctx, holdDID) + } + + return nil +} diff --git a/pkg/appview/jetstream/processor_test.go b/pkg/appview/jetstream/processor_test.go new file mode 100644 index 0000000..160091f --- /dev/null +++ b/pkg/appview/jetstream/processor_test.go @@ -0,0 +1,540 @@ +package jetstream + +import ( + "context" + "database/sql" + "encoding/json" + "testing" + "time" + + "atcr.io/pkg/atproto" + _ "github.com/mattn/go-sqlite3" +) + +// setupTestDB creates an in-memory SQLite database for testing +func setupTestDB(t *testing.T) *sql.DB { + database, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("Failed to open test database: %v", err) + } + + // Create schema + schema := ` + CREATE TABLE users ( + did TEXT PRIMARY KEY, + handle TEXT NOT NULL, + pds_endpoint TEXT NOT NULL, + avatar TEXT, + last_seen TIMESTAMP NOT NULL + ); + + CREATE TABLE manifests ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + did TEXT NOT NULL, + repository TEXT NOT NULL, + digest TEXT NOT NULL, + hold_endpoint TEXT NOT NULL, + schema_version INTEGER NOT NULL, + media_type TEXT NOT NULL, + 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 layers ( + manifest_id INTEGER NOT NULL, + digest TEXT NOT NULL, + size INTEGER NOT NULL, + media_type TEXT NOT NULL, + layer_index INTEGER NOT NULL, + PRIMARY KEY(manifest_id, layer_index) + ); + + CREATE TABLE manifest_references ( + manifest_id INTEGER NOT NULL, + digest TEXT NOT NULL, + media_type TEXT NOT NULL, + size INTEGER NOT NULL, + platform_architecture TEXT, + platform_os TEXT, + platform_variant TEXT, + platform_os_version TEXT, + reference_index INTEGER NOT NULL, + PRIMARY KEY(manifest_id, reference_index) + ); + + CREATE TABLE tags ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + did TEXT NOT NULL, + repository TEXT NOT NULL, + tag TEXT NOT NULL, + digest TEXT NOT NULL, + created_at TIMESTAMP NOT NULL, + UNIQUE(did, repository, tag) + ); + + CREATE TABLE stars ( + starrer_did TEXT NOT NULL, + owner_did TEXT NOT NULL, + repository TEXT NOT NULL, + created_at TIMESTAMP NOT NULL, + PRIMARY KEY(starrer_did, owner_did, repository) + ); + ` + + if _, err := database.Exec(schema); err != nil { + t.Fatalf("Failed to create schema: %v", err) + } + + return database +} + +func TestNewProcessor(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + tests := []struct { + name string + useCache bool + }{ + {"with cache", true}, + {"without cache", false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := NewProcessor(database, tt.useCache) + if p == nil { + t.Fatal("NewProcessor returned nil") + } + if p.db != database { + t.Error("Processor database not set correctly") + } + if p.useCache != tt.useCache { + t.Errorf("useCache = %v, want %v", p.useCache, tt.useCache) + } + if tt.useCache && p.userCache == nil { + t.Error("Cache enabled but userCache is nil") + } + if !tt.useCache && p.userCache != nil { + t.Error("Cache disabled but userCache is not nil") + } + }) + } +} + +func TestProcessManifest_ImageManifest(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + p := NewProcessor(database, false) + ctx := context.Background() + + // Create test manifest record + manifestRecord := &atproto.ManifestRecord{ + Repository: "test-app", + Digest: "sha256:abc123", + MediaType: "application/vnd.oci.image.manifest.v1+json", + SchemaVersion: 2, + HoldEndpoint: "did:web:hold01.atcr.io", + CreatedAt: time.Now(), + Config: &atproto.BlobReference{ + Digest: "sha256:config123", + Size: 1234, + }, + Layers: []atproto.BlobReference{ + {Digest: "sha256:layer1", Size: 5000, MediaType: "application/vnd.oci.image.layer.v1.tar+gzip"}, + {Digest: "sha256:layer2", Size: 3000, MediaType: "application/vnd.oci.image.layer.v1.tar+gzip"}, + }, + Annotations: map[string]string{ + "org.opencontainers.image.title": "Test App", + "org.opencontainers.image.description": "A test application", + "org.opencontainers.image.source": "https://github.com/test/app", + "org.opencontainers.image.licenses": "MIT", + "io.atcr.icon": "https://example.com/icon.png", + }, + } + + // Marshal to bytes for ProcessManifest + recordBytes, err := json.Marshal(manifestRecord) + if err != nil { + t.Fatalf("Failed to marshal manifest: %v", err) + } + + // Process manifest + manifestID, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes) + if err != nil { + t.Fatalf("ProcessManifest failed: %v", err) + } + if manifestID == 0 { + t.Error("Expected non-zero manifest ID") + } + + // Verify manifest was inserted + var count int + err = database.QueryRow("SELECT COUNT(*) FROM manifests WHERE did = ? AND repository = ? AND digest = ?", + "did:plc:test123", "test-app", "sha256:abc123").Scan(&count) + if err != nil { + t.Fatalf("Failed to query manifests: %v", err) + } + if count != 1 { + t.Errorf("Expected 1 manifest, got %d", count) + } + + // Verify annotations were stored + var title, source string + err = database.QueryRow("SELECT title, source_url FROM manifests WHERE id = ?", manifestID).Scan(&title, &source) + if err != nil { + t.Fatalf("Failed to query manifest fields: %v", err) + } + if title != "Test App" { + t.Errorf("title = %q, want %q", title, "Test App") + } + if source != "https://github.com/test/app" { + t.Errorf("source_url = %q, want %q", source, "https://github.com/test/app") + } + + // Verify layers were inserted + var layerCount int + err = database.QueryRow("SELECT COUNT(*) FROM layers WHERE manifest_id = ?", manifestID).Scan(&layerCount) + if err != nil { + t.Fatalf("Failed to query layers: %v", err) + } + if layerCount != 2 { + t.Errorf("Expected 2 layers, got %d", layerCount) + } + + // Verify no manifest references (this is an image, not a list) + var refCount int + err = database.QueryRow("SELECT COUNT(*) FROM manifest_references WHERE manifest_id = ?", manifestID).Scan(&refCount) + if err != nil { + t.Fatalf("Failed to query manifest_references: %v", err) + } + if refCount != 0 { + t.Errorf("Expected 0 manifest references, got %d", refCount) + } +} + +func TestProcessManifest_ManifestList(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + p := NewProcessor(database, false) + ctx := context.Background() + + // Create test manifest list record + manifestRecord := &atproto.ManifestRecord{ + Repository: "test-app", + Digest: "sha256:list123", + MediaType: "application/vnd.oci.image.index.v1+json", + SchemaVersion: 2, + HoldEndpoint: "did:web:hold01.atcr.io", + CreatedAt: time.Now(), + Manifests: []atproto.ManifestReference{ + { + Digest: "sha256:amd64manifest", + MediaType: "application/vnd.oci.image.manifest.v1+json", + Size: 1000, + Platform: &atproto.Platform{ + Architecture: "amd64", + OS: "linux", + }, + }, + { + Digest: "sha256:arm64manifest", + MediaType: "application/vnd.oci.image.manifest.v1+json", + Size: 1100, + Platform: &atproto.Platform{ + Architecture: "arm64", + OS: "linux", + Variant: "v8", + }, + }, + }, + } + + // Marshal to bytes for ProcessManifest + recordBytes, err := json.Marshal(manifestRecord) + if err != nil { + t.Fatalf("Failed to marshal manifest: %v", err) + } + + // Process manifest list + manifestID, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes) + if err != nil { + t.Fatalf("ProcessManifest failed: %v", err) + } + + // Verify manifest references were inserted + var refCount int + err = database.QueryRow("SELECT COUNT(*) FROM manifest_references WHERE manifest_id = ?", manifestID).Scan(&refCount) + if err != nil { + t.Fatalf("Failed to query manifest_references: %v", err) + } + if refCount != 2 { + t.Errorf("Expected 2 manifest references, got %d", refCount) + } + + // Verify platform info was stored + var arch, os string + err = database.QueryRow("SELECT platform_architecture, platform_os FROM manifest_references WHERE manifest_id = ? AND reference_index = 0", manifestID).Scan(&arch, &os) + if err != nil { + t.Fatalf("Failed to query platform info: %v", err) + } + if arch != "amd64" { + t.Errorf("platform_architecture = %q, want %q", arch, "amd64") + } + if os != "linux" { + t.Errorf("platform_os = %q, want %q", os, "linux") + } + + // Verify no layers (this is a list, not an image) + var layerCount int + err = database.QueryRow("SELECT COUNT(*) FROM layers WHERE manifest_id = ?", manifestID).Scan(&layerCount) + if err != nil { + t.Fatalf("Failed to query layers: %v", err) + } + if layerCount != 0 { + t.Errorf("Expected 0 layers, got %d", layerCount) + } +} + +func TestProcessTag(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + p := NewProcessor(database, false) + ctx := context.Background() + + // Create test tag record (using ManifestDigest field for simplicity) + tagRecord := &atproto.TagRecord{ + Repository: "test-app", + Tag: "latest", + ManifestDigest: "sha256:abc123", + UpdatedAt: time.Now(), + } + + // Marshal to bytes for ProcessTag + recordBytes, err := json.Marshal(tagRecord) + if err != nil { + t.Fatalf("Failed to marshal tag: %v", err) + } + + // Process tag + err = p.ProcessTag(ctx, "did:plc:test123", recordBytes) + if err != nil { + t.Fatalf("ProcessTag failed: %v", err) + } + + // Verify tag was inserted + var count int + err = database.QueryRow("SELECT COUNT(*) FROM tags WHERE did = ? AND repository = ? AND tag = ?", + "did:plc:test123", "test-app", "latest").Scan(&count) + if err != nil { + t.Fatalf("Failed to query tags: %v", err) + } + if count != 1 { + t.Errorf("Expected 1 tag, got %d", count) + } + + // Verify digest was stored + var digest string + err = database.QueryRow("SELECT digest FROM tags WHERE did = ? AND repository = ? AND tag = ?", + "did:plc:test123", "test-app", "latest").Scan(&digest) + if err != nil { + t.Fatalf("Failed to query tag digest: %v", err) + } + if digest != "sha256:abc123" { + t.Errorf("digest = %q, want %q", digest, "sha256:abc123") + } + + // Test upserting same tag with new digest + tagRecord.ManifestDigest = "sha256:newdigest" + recordBytes, err = json.Marshal(tagRecord) + if err != nil { + t.Fatalf("Failed to marshal tag: %v", err) + } + err = p.ProcessTag(ctx, "did:plc:test123", recordBytes) + if err != nil { + t.Fatalf("ProcessTag (upsert) failed: %v", err) + } + + // Verify tag was updated + err = database.QueryRow("SELECT digest FROM tags WHERE did = ? AND repository = ? AND tag = ?", + "did:plc:test123", "test-app", "latest").Scan(&digest) + if err != nil { + t.Fatalf("Failed to query updated tag: %v", err) + } + if digest != "sha256:newdigest" { + t.Errorf("digest = %q, want %q", digest, "sha256:newdigest") + } + + // Verify still only one tag (upsert, not insert) + err = database.QueryRow("SELECT COUNT(*) FROM tags WHERE did = ? AND repository = ? AND tag = ?", + "did:plc:test123", "test-app", "latest").Scan(&count) + if err != nil { + t.Fatalf("Failed to query tags after upsert: %v", err) + } + if count != 1 { + t.Errorf("Expected 1 tag after upsert, got %d", count) + } +} + +func TestProcessStar(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + p := NewProcessor(database, false) + ctx := context.Background() + + // Create test star record + starRecord := &atproto.StarRecord{ + Subject: atproto.StarSubject{ + DID: "did:plc:owner123", + Repository: "test-app", + }, + CreatedAt: time.Now(), + } + + // Marshal to bytes for ProcessStar + recordBytes, err := json.Marshal(starRecord) + if err != nil { + t.Fatalf("Failed to marshal star: %v", err) + } + + // Process star + err = p.ProcessStar(ctx, "did:plc:starrer123", recordBytes) + if err != nil { + t.Fatalf("ProcessStar failed: %v", err) + } + + // Verify star was inserted + var count int + err = database.QueryRow("SELECT COUNT(*) FROM stars WHERE starrer_did = ? AND owner_did = ? AND repository = ?", + "did:plc:starrer123", "did:plc:owner123", "test-app").Scan(&count) + if err != nil { + t.Fatalf("Failed to query stars: %v", err) + } + if count != 1 { + t.Errorf("Expected 1 star, got %d", count) + } + + // Test upserting same star (should be idempotent) + recordBytes, err = json.Marshal(starRecord) + if err != nil { + t.Fatalf("Failed to marshal star: %v", err) + } + err = p.ProcessStar(ctx, "did:plc:starrer123", recordBytes) + if err != nil { + t.Fatalf("ProcessStar (upsert) failed: %v", err) + } + + // Verify still only one star + err = database.QueryRow("SELECT COUNT(*) FROM stars WHERE starrer_did = ? AND owner_did = ? AND repository = ?", + "did:plc:starrer123", "did:plc:owner123", "test-app").Scan(&count) + if err != nil { + t.Fatalf("Failed to query stars after upsert: %v", err) + } + if count != 1 { + t.Errorf("Expected 1 star after upsert, got %d", count) + } +} + +func TestProcessManifest_Duplicate(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + p := NewProcessor(database, false) + ctx := context.Background() + + manifestRecord := &atproto.ManifestRecord{ + Repository: "test-app", + Digest: "sha256:abc123", + MediaType: "application/vnd.oci.image.manifest.v1+json", + SchemaVersion: 2, + HoldEndpoint: "did:web:hold01.atcr.io", + CreatedAt: time.Now(), + } + + // Marshal to bytes for ProcessManifest + recordBytes, err := json.Marshal(manifestRecord) + if err != nil { + t.Fatalf("Failed to marshal manifest: %v", err) + } + + // Insert first time + id1, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes) + if err != nil { + t.Fatalf("First ProcessManifest failed: %v", err) + } + + // Insert duplicate + id2, err := p.ProcessManifest(ctx, "did:plc:test123", recordBytes) + if err != nil { + t.Fatalf("Duplicate ProcessManifest failed: %v", err) + } + + // Should return existing ID + if id1 != id2 { + t.Errorf("Duplicate manifest got different ID: %d vs %d", id1, id2) + } + + // Verify only one manifest exists + var count int + err = database.QueryRow("SELECT COUNT(*) FROM manifests WHERE did = ? AND digest = ?", + "did:plc:test123", "sha256:abc123").Scan(&count) + if err != nil { + t.Fatalf("Failed to query manifests: %v", err) + } + if count != 1 { + t.Errorf("Expected 1 manifest, got %d", count) + } +} + +func TestProcessManifest_EmptyAnnotations(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + p := NewProcessor(database, false) + ctx := context.Background() + + // Manifest with nil annotations + manifestRecord := &atproto.ManifestRecord{ + Repository: "test-app", + Digest: "sha256:abc123", + MediaType: "application/vnd.oci.image.manifest.v1+json", + SchemaVersion: 2, + HoldEndpoint: "did:web:hold01.atcr.io", + CreatedAt: time.Now(), + Annotations: nil, + } + + // Marshal to bytes for ProcessManifest + recordBytes, err := json.Marshal(manifestRecord) + if err != nil { + t.Fatalf("Failed to marshal manifest: %v", err) + } + + manifestID, 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) + if err != nil { + t.Fatalf("Failed to query title: %v", err) + } + if title != "" { + t.Errorf("Expected empty title, got %q", title) + } +} diff --git a/pkg/appview/jetstream/worker.go b/pkg/appview/jetstream/worker.go index 6162e1b..6cfe778 100644 --- a/pkg/appview/jetstream/worker.go +++ b/pkg/appview/jetstream/worker.go @@ -9,9 +9,6 @@ import ( "sync" "time" - "github.com/bluesky-social/indigo/atproto/identity" - "github.com/bluesky-social/indigo/atproto/syntax" - "atcr.io/pkg/appview/db" "atcr.io/pkg/atproto" "github.com/gorilla/websocket" @@ -33,8 +30,7 @@ type Worker struct { startCursor int64 wantedCollections []string debugCollectionCount int - userCache *UserCache - directory identity.Directory + processor *Processor // Shared processor for DB operations eventCallback EventCallback connStartTime time.Time // Track when connection started for debugging @@ -65,10 +61,7 @@ func NewWorker(database *sql.DB, jetstreamURL string, startCursor int64) *Worker atproto.TagCollection, // io.atcr.tag atproto.StarCollection, // io.atcr.sailor.star }, - userCache: &UserCache{ - cache: make(map[string]*db.User), - }, - directory: identity.DefaultDirectory(), + processor: NewProcessor(database, true), // Use cache for live streaming } } @@ -333,86 +326,10 @@ func (w *Worker) processMessage(message []byte) error { } } -// ensureUser resolves and upserts a user by DID -func (w *Worker) ensureUser(ctx context.Context, did string) error { - // Check cache first - if user, ok := w.userCache.cache[did]; ok { - // Update last seen - user.LastSeen = time.Now() - return db.UpsertUser(w.db, user) - } - - // Resolve DID to get handle and PDS endpoint - didParsed, err := syntax.ParseDID(did) - if err != nil { - fmt.Printf("WARNING: Invalid DID %s: %v (using DID as handle)\n", did, err) - // Fallback: use DID as handle - user := &db.User{ - DID: did, - Handle: did, - PDSEndpoint: "https://bsky.social", // Default PDS endpoint as fallback - LastSeen: time.Now(), - } - w.userCache.cache[did] = user - return db.UpsertUser(w.db, user) - } - - ident, err := w.directory.LookupDID(ctx, didParsed) - if err != nil { - fmt.Printf("WARNING: Failed to resolve DID %s: %v (using DID as handle)\n", did, err) - // Fallback: use DID as handle - user := &db.User{ - DID: did, - Handle: did, - PDSEndpoint: "https://bsky.social", // Default PDS endpoint as fallback - LastSeen: time.Now(), - } - w.userCache.cache[did] = user - return db.UpsertUser(w.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 [worker]: Failed to fetch profile for DID %s: %v\n", resolvedDID, err) - // Continue without avatar - } else { - avatar = profile.Avatar - } - - // Cache the user - user := &db.User{ - DID: resolvedDID, - Handle: handle, - PDSEndpoint: pdsEndpoint, - Avatar: avatar, - LastSeen: time.Now(), - } - w.userCache.cache[did] = user - - // Upsert to database - return db.UpsertUser(w.db, user) -} - // processManifest processes a manifest commit event func (w *Worker) processManifest(commit *CommitEvent) error { // Resolve and upsert user with handle/PDS endpoint - if err := w.ensureUser(context.Background(), commit.DID); err != nil { + if err := w.processor.EnsureUser(context.Background(), commit.DID); err != nil { return fmt.Errorf("failed to ensure user: %w", err) } @@ -427,118 +344,25 @@ func (w *Worker) processManifest(commit *CommitEvent) error { } // Parse manifest record - var manifestRecord atproto.ManifestRecord - if commit.Record != nil { - recordBytes, err := json.Marshal(commit.Record) - if err != nil { - return fmt.Errorf("failed to marshal record: %w", err) - } - if err := json.Unmarshal(recordBytes, &manifestRecord); err != nil { - return fmt.Errorf("failed to unmarshal manifest: %w", err) - } - } else { - // No record data, can't process - return nil + if commit.Record == nil { + return nil // No record data, can't process } - // Extract OCI annotations from manifest - var title, description, sourceURL, documentationURL, licenses, iconURL, readmeURL 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"] - readmeURL = manifestRecord.Annotations["io.atcr.readme"] - } - - // Detect manifest type - isManifestList := len(manifestRecord.Manifests) > 0 - - // Prepare manifest for insertion - manifest := &db.Manifest{ - DID: commit.DID, - Repository: manifestRecord.Repository, - Digest: manifestRecord.Digest, - MediaType: manifestRecord.MediaType, - SchemaVersion: manifestRecord.SchemaVersion, - HoldEndpoint: manifestRecord.HoldEndpoint, - CreatedAt: manifestRecord.CreatedAt, - Title: title, - Description: description, - SourceURL: sourceURL, - DocumentationURL: documentationURL, - Licenses: licenses, - IconURL: iconURL, - ReadmeURL: readmeURL, - } - - // Set config fields only for image manifests (not manifest lists) - if !isManifestList && manifestRecord.Config != nil { - manifest.ConfigDigest = manifestRecord.Config.Digest - manifest.ConfigSize = manifestRecord.Config.Size - } - - // Insert manifest - manifestID, err := db.InsertManifest(w.db, manifest) + // Marshal map to bytes for processing + recordBytes, err := json.Marshal(commit.Record) if err != nil { - return fmt.Errorf("failed to insert manifest: %w", err) + return fmt.Errorf("failed to marshal record: %w", err) } - if isManifestList { - // Insert manifest references (for manifest lists/indexes) - for i, ref := range manifestRecord.Manifests { - platformArch := "" - platformOS := "" - platformVariant := "" - platformOSVersion := "" - - if ref.Platform != nil { - platformArch = ref.Platform.Architecture - platformOS = ref.Platform.OS - platformVariant = ref.Platform.Variant - platformOSVersion = ref.Platform.OSVersion - } - - if err := db.InsertManifestReference(w.db, &db.ManifestReference{ - ManifestID: manifestID, - Digest: ref.Digest, - MediaType: ref.MediaType, - Size: ref.Size, - PlatformArchitecture: platformArch, - PlatformOS: platformOS, - PlatformVariant: platformVariant, - PlatformOSVersion: platformOSVersion, - ReferenceIndex: i, - }); err != nil { - // Continue on error - reference might already exist - continue - } - } - } else { - // Insert layers (for image manifests) - for i, layer := range manifestRecord.Layers { - if err := db.InsertLayer(w.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 + // Use shared processor for DB operations + _, err = w.processor.ProcessManifest(context.Background(), commit.DID, recordBytes) + return err } // processTag processes a tag commit event func (w *Worker) processTag(commit *CommitEvent) error { // Resolve and upsert user with handle/PDS endpoint - if err := w.ensureUser(context.Background(), commit.DID); err != nil { + if err := w.processor.EnsureUser(context.Background(), commit.DID); err != nil { return fmt.Errorf("failed to ensure user: %w", err) } @@ -557,39 +381,24 @@ func (w *Worker) processTag(commit *CommitEvent) error { } // Parse tag record - var tagRecord atproto.TagRecord - if commit.Record != nil { - recordBytes, err := json.Marshal(commit.Record) - if err != nil { - return fmt.Errorf("failed to marshal record: %w", err) - } - if err := json.Unmarshal(recordBytes, &tagRecord); err != nil { - return fmt.Errorf("failed to unmarshal tag: %w", err) - } - } else { + if commit.Record == nil { return nil } - // Extract digest from tag record (tries manifest field first, falls back to manifestDigest) - manifestDigest, err := tagRecord.GetManifestDigest() + // Marshal map to bytes for processing + recordBytes, err := json.Marshal(commit.Record) if err != nil { - return fmt.Errorf("failed to get manifest digest from tag record: %w", err) + return fmt.Errorf("failed to marshal record: %w", err) } - // Insert or update tag - return db.UpsertTag(w.db, &db.Tag{ - DID: commit.DID, - Repository: tagRecord.Repository, - Tag: tagRecord.Tag, - Digest: manifestDigest, - CreatedAt: tagRecord.UpdatedAt, - }) + // Use shared processor for DB operations + return w.processor.ProcessTag(context.Background(), commit.DID, recordBytes) } // processStar processes a star commit event func (w *Worker) processStar(commit *CommitEvent) error { // Resolve and upsert the user who starred (starrer) - if err := w.ensureUser(context.Background(), commit.DID); err != nil { + if err := w.processor.EnsureUser(context.Background(), commit.DID); err != nil { return fmt.Errorf("failed to ensure user: %w", err) } @@ -606,21 +415,18 @@ func (w *Worker) processStar(commit *CommitEvent) error { } // Parse star record - var starRecord atproto.StarRecord - if commit.Record != nil { - recordBytes, err := json.Marshal(commit.Record) - if err != nil { - return fmt.Errorf("failed to marshal record: %w", err) - } - if err := json.Unmarshal(recordBytes, &starRecord); err != nil { - return fmt.Errorf("failed to unmarshal star: %w", err) - } - } else { + if commit.Record == nil { return nil } - // Upsert the star record (idempotent - star count will be calculated on demand) - return db.UpsertStar(w.db, commit.DID, starRecord.Subject.DID, starRecord.Subject.Repository, starRecord.CreatedAt) + // Marshal map to bytes for processing + recordBytes, err := json.Marshal(commit.Record) + if err != nil { + return fmt.Errorf("failed to marshal record: %w", err) + } + + // Use shared processor for DB operations + return w.processor.ProcessStar(context.Background(), commit.DID, recordBytes) } // JetstreamEvent represents a Jetstream event diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index 3e334d4..da8fd3a 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -40,7 +40,7 @@ type ProxyBlobStore struct { // NewProxyBlobStore creates a new proxy blob store func NewProxyBlobStore(ctx *RegistryContext) *ProxyBlobStore { // Resolve DID to URL once at construction time - holdURL := resolveHoldURL(ctx.HoldDID) + holdURL := appview.ResolveHoldURL(ctx.HoldDID) fmt.Printf("DEBUG [proxy_blob_store]: NewProxyBlobStore created with holdDID=%s, holdURL=%s, userDID=%s, repo=%s\n", ctx.HoldDID, holdURL, ctx.DID, ctx.Repository) @@ -108,12 +108,6 @@ func (p *ProxyBlobStore) checkWriteAccess(ctx context.Context) error { return nil } -// resolveHoldURL converts a hold identifier (DID or URL) to an HTTP URL -// Deprecated: Use appview.ResolveHoldURL instead -func resolveHoldURL(holdDID string) string { - return appview.ResolveHoldURL(holdDID) -} - // Stat returns the descriptor for a blob func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) { // Check read access diff --git a/pkg/appview/storage/proxy_blob_store_test.go b/pkg/appview/storage/proxy_blob_store_test.go index 9691621..6e33f2f 100644 --- a/pkg/appview/storage/proxy_blob_store_test.go +++ b/pkg/appview/storage/proxy_blob_store_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + "atcr.io/pkg/appview" "atcr.io/pkg/atproto" "atcr.io/pkg/auth/token" "github.com/opencontainers/go-digest" @@ -218,7 +219,7 @@ func TestResolveHoldURL(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - result := resolveHoldURL(tt.holdDID) + result := appview.ResolveHoldURL(tt.holdDID) if result != tt.expected { t.Errorf("Expected %s, got %s", tt.expected, result) } diff --git a/scripts/migrate-image.sh b/scripts/migrate-image.sh new file mode 100755 index 0000000..7bb992d --- /dev/null +++ b/scripts/migrate-image.sh @@ -0,0 +1,54 @@ +#!/bin/bash +set -e + +# Configuration +SOURCE_REGISTRY="ghcr.io/evanjarrett/hsm-secrets-operator" +TARGET_REGISTRY="atcr.io/evan.jarrett.net/hsm-secrets-operator" +TAG="latest" + +# Image digests +AMD64_DIGEST="sha256:274284a623810cf07c5b4735628832751926b7d192863681d5af1b4137f44254" +ARM64_DIGEST="sha256:b57929fd100033092766aad1c7e747deef9b1e3206756c11d0d7a7af74daedff" + +echo "=== Migrating multi-arch image from GHCR to ATCR ===" +echo "Source: ${SOURCE_REGISTRY}" +echo "Target: ${TARGET_REGISTRY}:${TAG}" +echo "" + +# Tag and push amd64 image +echo ">>> Tagging and pushing amd64 image..." +docker tag "${SOURCE_REGISTRY}@${AMD64_DIGEST}" "${TARGET_REGISTRY}:${TAG}-amd64" +docker push "${TARGET_REGISTRY}:${TAG}-amd64" +echo "" + +# Tag and push arm64 image +echo ">>> Tagging and pushing arm64 image..." +docker tag "${SOURCE_REGISTRY}@${ARM64_DIGEST}" "${TARGET_REGISTRY}:${TAG}-arm64" +docker push "${TARGET_REGISTRY}:${TAG}-arm64" +echo "" + +# Create multi-arch manifest using the pushed tags +echo ">>> Creating multi-arch manifest..." +docker manifest create "${TARGET_REGISTRY}:${TAG}" \ + --amend "${TARGET_REGISTRY}:${TAG}-amd64" \ + --amend "${TARGET_REGISTRY}:${TAG}-arm64" +echo "" + +# Annotate the manifest with platform information +echo ">>> Annotating manifest with platform information..." +docker manifest annotate "${TARGET_REGISTRY}:${TAG}" \ + "${TARGET_REGISTRY}:${TAG}-amd64" \ + --os linux --arch amd64 + +docker manifest annotate "${TARGET_REGISTRY}:${TAG}" \ + "${TARGET_REGISTRY}:${TAG}-arm64" \ + --os linux --arch arm64 +echo "" + +# Push the manifest list +echo ">>> Pushing multi-arch manifest..." +docker manifest push "${TARGET_REGISTRY}:${TAG}" +echo "" + +echo "=== Migration complete! ===" +echo "You can now pull: docker pull ${TARGET_REGISTRY}:${TAG}"