refactor jetstream code to unify shared functionality between that and backfill. add tests

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

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

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

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

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

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

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