Something went wrong. Try again.
A container registry that uses the AT Protocol for manifest storage and S3 for blob storage.
Something went wrong. Try again.
13 kB · 397 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398package jetstream
import ( "context" "database/sql" "encoding/json" "fmt" "log/slog" "strings" "time"
"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 pathstype Processor struct { db *sql.DB 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, 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 _, ok := p.userCache.cache[did]; ok { // User in cache - just update last seen timestamp return db.UpdateUserLastSeen(p.db, did) } } 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 { // User exists - just update last seen timestamp return db.UpdateUserLastSeen(p.db, did) } }
// Resolve DID to get handle and PDS endpoint resolvedDID, handle, pdsEndpoint, err := atproto.ResolveIdentity(ctx, did) if err != nil { return err }
// Fetch user's Bluesky profile record from their PDS (including avatar) avatarURL := "" client := atproto.NewClient(pdsEndpoint, "", "") profileRecord, err := client.GetProfileRecord(ctx, resolvedDID) if err != nil { slog.Warn("Failed to fetch profile record", "component", "processor", "did", resolvedDID, "error", err) // Continue without avatar } else if profileRecord.Avatar != nil && profileRecord.Avatar.Ref.Link != "" { avatarURL = atproto.BlobCDNURL(resolvedDID, profileRecord.Avatar.Ref.Link) }
// Create user record user := &db.User{ DID: resolvedDID, Handle: handle, PDSEndpoint: pdsEndpoint, Avatar: avatarURL, LastSeen: time.Now(), }
// Cache if enabled if p.useCache { p.userCache.cache[did] = user }
// Upsert to database // Use UpsertUser if we successfully fetched an avatar (to update existing users) // Use UpsertUserIgnoreAvatar if fetch failed (to preserve existing avatars) if avatarURL != "" { return db.UpsertUser(p.db, user) } return db.UpsertUserIgnoreAvatar(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) } // Detect manifest type isManifestList := len(manifestRecord.Manifests) > 0
// Extract hold DID from manifest (with fallback for legacy manifests) // New manifests use holdDid field (DID format) // Old manifests use holdEndpoint field (URL format) - convert to DID holdDID := manifestRecord.HoldDID if holdDID == "" && manifestRecord.HoldEndpoint != "" { // Legacy manifest - convert URL to DID holdDID = atproto.ResolveHoldDIDFromURL(manifestRecord.HoldEndpoint) }
// 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: holdDID, CreatedAt: manifestRecord.CreatedAt, // Annotations removed - stored separately in repository_annotations table }
// 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) } }
// 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(p.db, did, manifestRecord.Repository, manifestRecord.Annotations) if err != nil { return 0, fmt.Errorf("failed to upsert annotations: %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 }
// Detect attestation manifests from annotations isAttestation := false if ref.Annotations != nil { if refType, ok := ref.Annotations["vnd.docker.reference.type"]; ok { isAttestation = refType == "attestation-manifest" } }
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, IsAttestation: isAttestation, 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 databasefunc (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 databasefunc (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 holdsfunc (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 == "" { slog.Warn("Invalid hold reference in profile", "component", "processor", "did", did, "default_hold", 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}
// ProcessIdentity handles identity change events (handle updates)// This is called when Jetstream receives an identity event indicating a handle change.// The identity cache is invalidated to ensure the next lookup uses the new handle,// and the database is updated to reflect the change in the UI.//// Only processes events for users who already exist in our database (have ATCR activity).func (p *Processor) ProcessIdentity(ctx context.Context, did string, newHandle string) error { // Check if user exists in our database - only update if they're an ATCR user user, err := db.GetUserByDID(p.db, did) if err != nil { return fmt.Errorf("failed to check user existence: %w", err) }
// Skip if user doesn't exist - they don't have any ATCR activity (manifests, profiles, etc.) if user == nil { return nil }
// Update handle in database if err := db.UpdateUserHandle(p.db, did, newHandle); err != nil { slog.Warn("Failed to update user handle in database", "component", "processor", "did", did, "handle", newHandle, "error", err) // Continue to invalidate cache even if DB update fails }
// Invalidate cached identity data to force re-resolution on next lookup if err := atproto.InvalidateIdentity(ctx, did); err != nil { slog.Warn("Failed to invalidate identity cache", "component", "processor", "did", did, "error", err) return err }
slog.Info("Processed identity change event", "component", "processor", "did", did, "old_handle", user.Handle, "new_handle", newHandle)
return nil}
// ProcessAccount handles account status events (deactivation/reactivation)// This is called when Jetstream receives an account event indicating status changes.//// IMPORTANT: Deactivation events are ambiguous - they could indicate:// 1. Permanent account deactivation (user deleted account)// 2. PDS migration (account deactivated at old PDS, reactivated at new PDS)//// We DO NOT delete user data on deactivation events. Instead, we invalidate the// identity cache. On the next resolution attempt:// - If migrated: Resolution finds the new PDS and updates the database automatically// - If truly deactivated: Resolution fails and user won't appear in new queries//// This approach prevents data loss from PDS migrations while still handling deactivations.//// Only processes events for users who already exist in our database (have ATCR activity).func (p *Processor) ProcessAccount(ctx context.Context, did string, active bool, status string) error { // Only process deactivation events if active || status != "deactivated" { return nil }
// Check if user exists in our database - only update if they're an ATCR user user, err := db.GetUserByDID(p.db, did) if err != nil { return fmt.Errorf("failed to check user existence: %w", err) }
// Skip if user doesn't exist - they don't have any ATCR activity if user == nil { return nil }
// Invalidate cached identity data to force re-resolution on next lookup // This will discover if the account was migrated (new PDS) or truly deactivated (resolution fails) if err := atproto.InvalidateIdentity(ctx, did); err != nil { slog.Warn("Failed to invalidate identity cache for deactivated account", "component", "processor", "did", did, "error", err) return err }
slog.Info("Processed account deactivation event - cache invalidated", "component", "processor", "did", did, "handle", user.Handle, "status", status)
return nil}