From b251c8857fc4195dd410e2990b841071a44f0643 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Tue, 10 Feb 2026 20:58:24 -0600 Subject: [PATCH] change to transactions for database --- config-hold.example.yaml | 6 ++ deploy/upcloud/configs/appview.yaml.tmpl | 11 +- deploy/upcloud/configs/hold.yaml.tmpl | 4 + pkg/appview/db/annotations.go | 26 ++--- pkg/appview/db/dbtx.go | 13 +++ pkg/appview/db/delete.go | 5 +- pkg/appview/db/export.go | 10 +- pkg/appview/db/hold_store.go | 18 ++-- pkg/appview/db/queries.go | 127 +++++++++++------------ pkg/appview/jetstream/backfill.go | 34 +++++- pkg/appview/jetstream/processor.go | 5 +- 11 files changed, 148 insertions(+), 111 deletions(-) create mode 100644 pkg/appview/db/dbtx.go diff --git a/config-hold.example.yaml b/config-hold.example.yaml index 5165223..7d25194 100644 --- a/config-hold.example.yaml +++ b/config-hold.example.yaml @@ -65,6 +65,12 @@ database: path: /var/lib/atcr-hold # PDS signing key path. Defaults to {database.path}/signing.key. key_path: "" + # libSQL sync URL (libsql://...). Works with Turso cloud, Bunny DB, or self-hosted libsql-server. Leave empty for local-only SQLite. + libsql_sync_url: "" + # Auth token for libSQL sync. Required if libsql_sync_url is set. + libsql_auth_token: "" + # How often to sync with remote libSQL server. Default: 60s. + libsql_sync_interval: 1m0s # Admin panel settings. admin: # Enable the web-based admin panel for crew and storage management. diff --git a/deploy/upcloud/configs/appview.yaml.tmpl b/deploy/upcloud/configs/appview.yaml.tmpl index 825d628..3c9a5a5 100644 --- a/deploy/upcloud/configs/appview.yaml.tmpl +++ b/deploy/upcloud/configs/appview.yaml.tmpl @@ -13,6 +13,7 @@ server: default_hold_did: "{{.HoldDid}}" oauth_key_path: "{{.BasePath}}/oauth/client.key" client_name: Seamark + test_mode: false client_short_name: Seamark registry_domains: - "buoy.cr" @@ -27,9 +28,15 @@ health: cache_ttl: 15m0s check_interval: 15m0s jetstream: - url: wss://jetstream2.us-west.bsky.network/subscribe + urls: + - wss://jetstream2.us-west.bsky.network/subscribe + - wss://jetstream1.us-west.bsky.network/subscribe + - wss://jetstream2.us-east.bsky.network/subscribe + - wss://jetstream1.us-east.bsky.network/subscribe backfill_enabled: true - relay_endpoint: https://relay1.us-east.bsky.network + relay_endpoints: + - https://relay1.us-east.bsky.network + - https://relay1.us-west.bsky.network auth: key_path: "{{.BasePath}}/auth/private-key.pem" cert_path: "{{.BasePath}}/auth/private-key.crt" diff --git a/deploy/upcloud/configs/hold.yaml.tmpl b/deploy/upcloud/configs/hold.yaml.tmpl index 952a811..950cd0b 100644 --- a/deploy/upcloud/configs/hold.yaml.tmpl +++ b/deploy/upcloud/configs/hold.yaml.tmpl @@ -17,6 +17,7 @@ server: addr: :8080 public_url: "https://{{.HoldDomain}}" public: false + test_mode: false relay_endpoint: "" read_timeout: 5m0s write_timeout: 5m0s @@ -29,6 +30,9 @@ registration: database: path: "{{.BasePath}}" key_path: "" + libsql_sync_url: "" + libsql_auth_token: "" + libsql_sync_interval: 1m0s admin: enabled: true quota: diff --git a/pkg/appview/db/annotations.go b/pkg/appview/db/annotations.go index 0d43688..d2554e7 100644 --- a/pkg/appview/db/annotations.go +++ b/pkg/appview/db/annotations.go @@ -1,12 +1,9 @@ package db -import ( - "database/sql" - "time" -) +import "time" // GetRepositoryAnnotations retrieves all annotations for a repository -func GetRepositoryAnnotations(db *sql.DB, did, repository string) (map[string]string, error) { +func GetRepositoryAnnotations(db DBTX, did, repository string) (map[string]string, error) { rows, err := db.Query(` SELECT key, value FROM repository_annotations @@ -30,16 +27,11 @@ func GetRepositoryAnnotations(db *sql.DB, did, repository string) (map[string]st } // 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() - +// Only called when manifest has at least one non-empty annotation. +// Atomicity is provided by the caller's transaction when used during backfill. +func UpsertRepositoryAnnotations(db DBTX, did, repository string, annotations map[string]string) error { // Delete existing annotations - _, err = tx.Exec(` + _, err := db.Exec(` DELETE FROM repository_annotations WHERE did = ? AND repository = ? `, did, repository) @@ -48,7 +40,7 @@ func UpsertRepositoryAnnotations(db *sql.DB, did, repository string, annotations } // Insert new annotations - stmt, err := tx.Prepare(` + stmt, err := db.Prepare(` INSERT INTO repository_annotations (did, repository, key, value, updated_at) VALUES (?, ?, ?, ?, ?) `) @@ -65,11 +57,11 @@ func UpsertRepositoryAnnotations(db *sql.DB, did, repository string, annotations } } - return tx.Commit() + return nil } // DeleteRepositoryAnnotations removes all annotations for a repository -func DeleteRepositoryAnnotations(db *sql.DB, did, repository string) error { +func DeleteRepositoryAnnotations(db DBTX, did, repository string) error { _, err := db.Exec(` DELETE FROM repository_annotations WHERE did = ? AND repository = ? diff --git a/pkg/appview/db/dbtx.go b/pkg/appview/db/dbtx.go new file mode 100644 index 0000000..86e3d08 --- /dev/null +++ b/pkg/appview/db/dbtx.go @@ -0,0 +1,13 @@ +package db + +import "database/sql" + +// DBTX is an interface satisfied by both *sql.DB and *sql.Tx. +// All query functions in this package accept DBTX to allow callers +// to choose whether operations run in a transaction or standalone. +type DBTX interface { + Exec(query string, args ...any) (sql.Result, error) + Query(query string, args ...any) (*sql.Rows, error) + QueryRow(query string, args ...any) *sql.Row + Prepare(query string) (*sql.Stmt, error) +} diff --git a/pkg/appview/db/delete.go b/pkg/appview/db/delete.go index b86fd58..47eebe6 100644 --- a/pkg/appview/db/delete.go +++ b/pkg/appview/db/delete.go @@ -2,7 +2,6 @@ package db import ( "context" - "database/sql" "fmt" "log/slog" ) @@ -17,7 +16,7 @@ import ( // // This should be called AFTER remote cleanup (hold services, PDS records) // since we need the OAuth tokens to authenticate those requests. -func DeleteUserDataFull(db *sql.DB, oauthStore *OAuthStore, did string) error { +func DeleteUserDataFull(db DBTX, oauthStore *OAuthStore, did string) error { slog.Info("Starting full user data deletion", "did", did) // 1. Delete non-cascading hold membership tables @@ -48,7 +47,7 @@ func DeleteUserDataFull(db *sql.DB, oauthStore *OAuthStore, did string) error { // deleteHoldMembershipData deletes non-cascading hold membership tables. // These tables don't have foreign keys to the users table. -func deleteHoldMembershipData(db *sql.DB, did string) error { +func deleteHoldMembershipData(db DBTX, did string) error { // Delete from hold_crew_approvals (where user is the approved member) result, err := db.Exec(`DELETE FROM hold_crew_approvals WHERE user_did = ?`, did) if err != nil { diff --git a/pkg/appview/db/export.go b/pkg/appview/db/export.go index 6b69df3..137b86b 100644 --- a/pkg/appview/db/export.go +++ b/pkg/appview/db/export.go @@ -75,7 +75,7 @@ type CachedDataNote struct { // ExportUserData gathers all user data for GDPR export // Only includes data we originate, not cached PDS data -func ExportUserData(db *sql.DB, did string) (*UserDataExport, error) { +func ExportUserData(db DBTX, did string) (*UserDataExport, error) { export := &UserDataExport{ ExportedAt: time.Now().UTC(), ExportVersion: "1.0", @@ -128,7 +128,7 @@ func ExportUserData(db *sql.DB, did string) (*UserDataExport, error) { } // getDevicesForExport retrieves sanitized device records -func getDevicesForExport(db *sql.DB, did string) ([]DeviceExport, error) { +func getDevicesForExport(db DBTX, did string) ([]DeviceExport, error) { rows, err := db.Query(` SELECT id, name, ip_address, location, user_agent, created_at, last_used FROM devices @@ -169,7 +169,7 @@ func getDevicesForExport(db *sql.DB, did string) ([]DeviceExport, error) { } // getOAuthSessionsForExport retrieves sanitized OAuth session records -func getOAuthSessionsForExport(db *sql.DB, did string) ([]OAuthSessionExport, error) { +func getOAuthSessionsForExport(db DBTX, did string) ([]OAuthSessionExport, error) { rows, err := db.Query(` SELECT session_id, created_at, updated_at FROM oauth_sessions @@ -199,7 +199,7 @@ func getOAuthSessionsForExport(db *sql.DB, did string) ([]OAuthSessionExport, er } // getUISessionsForExport retrieves sanitized UI session records -func getUISessionsForExport(db *sql.DB, did string) ([]UISessionExport, error) { +func getUISessionsForExport(db DBTX, did string) ([]UISessionExport, error) { rows, err := db.Query(` SELECT id, expires_at, created_at FROM ui_sessions @@ -229,7 +229,7 @@ func getUISessionsForExport(db *sql.DB, did string) ([]UISessionExport, error) { } // getHoldMembershipsForExport retrieves hold approval and denial records -func getHoldMembershipsForExport(db *sql.DB, did string) (HoldMembershipsExport, error) { +func getHoldMembershipsForExport(db DBTX, did string) (HoldMembershipsExport, error) { memberships := HoldMembershipsExport{ Approvals: []HoldApprovalExport{}, Denials: []HoldDenialExport{}, diff --git a/pkg/appview/db/hold_store.go b/pkg/appview/db/hold_store.go index c4b9607..9fc3816 100644 --- a/pkg/appview/db/hold_store.go +++ b/pkg/appview/db/hold_store.go @@ -30,7 +30,7 @@ type HoldCaptainRecord struct { // GetCaptainRecord retrieves a captain record from the cache // Returns nil if not found (cache miss) -func GetCaptainRecord(db *sql.DB, holdDID string) (*HoldCaptainRecord, error) { +func GetCaptainRecord(db DBTX, holdDID string) (*HoldCaptainRecord, error) { query := ` SELECT hold_did, owner_did, public, allow_all_crew, deployed_at, region, updated_at @@ -71,7 +71,7 @@ func GetCaptainRecord(db *sql.DB, holdDID string) (*HoldCaptainRecord, error) { } // UpsertCaptainRecord inserts or updates a captain record in the cache -func UpsertCaptainRecord(db *sql.DB, record *HoldCaptainRecord) error { +func UpsertCaptainRecord(db DBTX, record *HoldCaptainRecord) error { query := ` INSERT INTO hold_captain_records ( hold_did, owner_did, public, allow_all_crew, @@ -104,7 +104,7 @@ func UpsertCaptainRecord(db *sql.DB, record *HoldCaptainRecord) error { } // ListHoldDIDs returns all known hold DIDs from the cache -func ListHoldDIDs(db *sql.DB) ([]string, error) { +func ListHoldDIDs(db DBTX) ([]string, error) { query := ` SELECT hold_did FROM hold_captain_records @@ -143,7 +143,7 @@ func nullString(s string) sql.NullString { // GetCaptainRecordsForOwner retrieves all captain records where the user is the owner // Used for GDPR export to find all holds owned by a user -func GetCaptainRecordsForOwner(db *sql.DB, ownerDID string) ([]*HoldCaptainRecord, error) { +func GetCaptainRecordsForOwner(db DBTX, ownerDID string) ([]*HoldCaptainRecord, error) { query := ` SELECT hold_did, owner_did, public, allow_all_crew, deployed_at, region, updated_at @@ -198,7 +198,7 @@ func GetCaptainRecordsForOwner(db *sql.DB, ownerDID string) ([]*HoldCaptainRecor } // DeleteCaptainRecord removes a captain record from the cache -func DeleteCaptainRecord(db *sql.DB, holdDID string) error { +func DeleteCaptainRecord(db DBTX, holdDID string) error { // Note: hold_crew_members doesn't have CASCADE, so delete crew first _, err := db.Exec(`DELETE FROM hold_crew_members WHERE hold_did = ?`, holdDID) if err != nil { @@ -226,7 +226,7 @@ type CrewMember struct { } // UpsertCrewMember inserts or updates a crew member record -func UpsertCrewMember(db *sql.DB, member *CrewMember) error { +func UpsertCrewMember(db DBTX, member *CrewMember) error { query := ` INSERT INTO hold_crew_members ( hold_did, member_did, rkey, role, permissions, tier, added_at, updated_at @@ -257,7 +257,7 @@ func UpsertCrewMember(db *sql.DB, member *CrewMember) error { } // DeleteCrewMemberByRkey removes a crew member by rkey (for delete events from Jetstream) -func DeleteCrewMemberByRkey(db *sql.DB, holdDID, rkey string) error { +func DeleteCrewMemberByRkey(db DBTX, holdDID, rkey string) error { _, err := db.Exec(`DELETE FROM hold_crew_members WHERE hold_did = ? AND rkey = ?`, holdDID, rkey) if err != nil { return fmt.Errorf("failed to delete crew member by rkey: %w", err) @@ -278,7 +278,7 @@ type AvailableHold struct { // GetAvailableHolds returns all holds available to a user, grouped by membership type // Results are ordered: owner first, then crew, then eligible, then public -func GetAvailableHolds(db *sql.DB, userDID string) ([]AvailableHold, error) { +func GetAvailableHolds(db DBTX, userDID string) ([]AvailableHold, error) { query := ` SELECT h.hold_did, @@ -352,7 +352,7 @@ func GetAvailableHolds(db *sql.DB, userDID string) ([]AvailableHold, error) { } // GetCrewMemberships returns all holds where a user is a crew member -func GetCrewMemberships(db *sql.DB, memberDID string) ([]CrewMember, error) { +func GetCrewMemberships(db DBTX, memberDID string) ([]CrewMember, error) { query := ` SELECT hold_did, member_did, rkey, role, permissions, tier, added_at, created_at, updated_at FROM hold_crew_members diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index cc7eb44..53aa812 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -55,7 +55,7 @@ func escapeLikePattern(s string) string { // SearchRepositories searches for repositories matching the query across handles, DIDs, repositories, and annotations // Returns RepoCardData (one per repository) instead of individual pushes/tags -func SearchRepositories(db *sql.DB, query string, limit, offset int, currentUserDID string) ([]RepoCardData, int, error) { +func SearchRepositories(db DBTX, query string, limit, offset int, currentUserDID string) ([]RepoCardData, int, error) { // Escape LIKE wildcards so they're treated literally query = escapeLikePattern(query) @@ -181,7 +181,7 @@ func SearchRepositories(db *sql.DB, query string, limit, offset int, currentUser } // GetUserRepositories fetches all repositories for a user -func GetUserRepositories(db *sql.DB, did string) ([]Repository, error) { +func GetUserRepositories(db DBTX, did string) ([]Repository, error) { // Get repository summary rows, err := db.Query(` SELECT @@ -310,12 +310,12 @@ func GetUserRepositories(db *sql.DB, did string) ([]Repository, error) { // GetRepositoryMetadata retrieves metadata for a repository from annotations table // Returns a map of annotation key -> value for easy access in templates and handlers -func GetRepositoryMetadata(db *sql.DB, did string, repository string) (map[string]string, error) { +func GetRepositoryMetadata(db DBTX, did string, repository string) (map[string]string, error) { return GetRepositoryAnnotations(db, did, repository) } // GetUserByDID retrieves a user by DID -func GetUserByDID(db *sql.DB, did string) (*User, error) { +func GetUserByDID(db DBTX, did string) (*User, error) { var user User var avatar sql.NullString err := db.QueryRow(` @@ -340,7 +340,7 @@ func GetUserByDID(db *sql.DB, did string) (*User, error) { } // GetUserByHandle retrieves a user by handle -func GetUserByHandle(db *sql.DB, handle string) (*User, error) { +func GetUserByHandle(db DBTX, handle string) (*User, error) { var user User var avatar sql.NullString err := db.QueryRow(` @@ -365,7 +365,7 @@ func GetUserByHandle(db *sql.DB, handle string) (*User, error) { } // UpsertUser inserts or updates a user record -func UpsertUser(db *sql.DB, user *User) error { +func UpsertUser(db DBTX, user *User) error { _, err := db.Exec(` INSERT INTO users (did, handle, pds_endpoint, avatar, last_seen) VALUES (?, ?, ?, ?, ?) @@ -380,7 +380,7 @@ func UpsertUser(db *sql.DB, user *User) error { // UpsertUserIgnoreAvatar inserts or updates a user record, but preserves existing avatar on update // This is useful when avatar fetch fails, and we don't want to overwrite an existing avatar with empty string -func UpsertUserIgnoreAvatar(db *sql.DB, user *User) error { +func UpsertUserIgnoreAvatar(db DBTX, user *User) error { _, err := db.Exec(` INSERT INTO users (did, handle, pds_endpoint, avatar, last_seen) VALUES (?, ?, ?, ?, ?) @@ -394,7 +394,7 @@ func UpsertUserIgnoreAvatar(db *sql.DB, user *User) error { // UpdateUserLastSeen updates only the last_seen timestamp for a user // This is more efficient than UpsertUser when only updating activity timestamp -func UpdateUserLastSeen(db *sql.DB, did string) error { +func UpdateUserLastSeen(db DBTX, did string) error { _, err := db.Exec(` UPDATE users SET last_seen = ? WHERE did = ? `, time.Now(), did) @@ -403,7 +403,7 @@ func UpdateUserLastSeen(db *sql.DB, did string) error { // UpdateUserHandle updates a user's handle when an identity change event is received // This is called when Jetstream receives an identity event indicating a handle change -func UpdateUserHandle(db *sql.DB, did string, newHandle string) error { +func UpdateUserHandle(db DBTX, did string, newHandle string) error { _, err := db.Exec(` UPDATE users SET handle = ?, last_seen = ? WHERE did = ? `, newHandle, time.Now(), did) @@ -412,7 +412,7 @@ func UpdateUserHandle(db *sql.DB, did string, newHandle string) error { // UpdateUserAvatar updates a user's avatar URL when a profile change is detected // This is called when Jetstream receives an app.bsky.actor.profile update -func UpdateUserAvatar(db *sql.DB, did string, avatarURL string) error { +func UpdateUserAvatar(db DBTX, did string, avatarURL string) error { _, err := db.Exec(` UPDATE users SET avatar = ?, last_seen = ? WHERE did = ? `, avatarURL, time.Now(), did) @@ -420,7 +420,7 @@ func UpdateUserAvatar(db *sql.DB, did string, avatarURL string) error { } // GetManifestDigestsForDID returns all manifest digests for a DID -func GetManifestDigestsForDID(db *sql.DB, did string) ([]string, error) { +func GetManifestDigestsForDID(db DBTX, did string) ([]string, error) { rows, err := db.Query(` SELECT digest FROM manifests WHERE did = ? `, did) @@ -442,7 +442,7 @@ func GetManifestDigestsForDID(db *sql.DB, did string) ([]string, error) { } // DeleteManifestsNotInList deletes all manifests for a DID that are not in the provided list -func DeleteManifestsNotInList(db *sql.DB, did string, keepDigests []string) error { +func DeleteManifestsNotInList(db DBTX, did string, keepDigests []string) error { if len(keepDigests) == 0 { // No manifests to keep - delete all for this DID _, err := db.Exec(`DELETE FROM manifests WHERE did = ?`, did) @@ -467,7 +467,7 @@ func DeleteManifestsNotInList(db *sql.DB, did string, keepDigests []string) erro } // GetTagsForDID returns all (repository, tag) pairs for a DID -func GetTagsForDID(db *sql.DB, did string) ([]struct{ Repository, Tag string }, error) { +func GetTagsForDID(db DBTX, did string) ([]struct{ Repository, Tag string }, error) { rows, err := db.Query(` SELECT repository, tag FROM tags WHERE did = ? `, did) @@ -488,24 +488,17 @@ func GetTagsForDID(db *sql.DB, did string) ([]struct{ Repository, Tag string }, return tags, rows.Err() } -// DeleteTagsNotInList deletes all tags for a DID that are not in the provided list -func DeleteTagsNotInList(db *sql.DB, did string, keepTags []struct{ Repository, Tag string }) error { +// DeleteTagsNotInList deletes all tags for a DID that are not in the provided list. +// Atomicity is provided by the caller's transaction when used during backfill. +func DeleteTagsNotInList(db DBTX, did string, keepTags []struct{ Repository, Tag string }) error { if len(keepTags) == 0 { // No tags to keep - delete all for this DID _, err := db.Exec(`DELETE FROM tags WHERE did = ?`, did) return err } - // For tags, we need to check (repository, tag) pairs - // Build a DELETE query that excludes the pairs we want to keep - tx, err := db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - // First, get all current tags - rows, err := tx.Query(`SELECT id, repository, tag FROM tags WHERE did = ?`, did) + rows, err := db.Query(`SELECT id, repository, tag FROM tags WHERE did = ?`, did) if err != nil { return err } @@ -536,19 +529,19 @@ func DeleteTagsNotInList(db *sql.DB, did string, keepTags []struct{ Repository, // Delete tags not in keep list for _, id := range toDelete { - if _, err := tx.Exec(`DELETE FROM tags WHERE id = ?`, id); err != nil { + if _, err := db.Exec(`DELETE FROM tags WHERE id = ?`, id); err != nil { return err } } - return tx.Commit() + return nil } // InsertManifest inserts or updates a manifest record // Uses UPSERT to update core metadata if manifest already exists // Returns the manifest ID (works correctly for both insert and update) // Note: Annotations are stored separately in repository_annotations table -func InsertManifest(db *sql.DB, manifest *Manifest) (int64, error) { +func InsertManifest(db DBTX, manifest *Manifest) (int64, error) { _, err := db.Exec(` INSERT INTO manifests (did, repository, digest, hold_endpoint, schema_version, media_type, @@ -584,7 +577,7 @@ func InsertManifest(db *sql.DB, manifest *Manifest) (int64, error) { } // InsertLayer inserts a new layer record -func InsertLayer(db *sql.DB, layer *Layer) error { +func InsertLayer(db DBTX, layer *Layer) error { _, err := db.Exec(` INSERT INTO layers (manifest_id, digest, size, media_type, layer_index) VALUES (?, ?, ?, ?, ?) @@ -593,7 +586,7 @@ func InsertLayer(db *sql.DB, layer *Layer) error { } // UpsertTag inserts or updates a tag record -func UpsertTag(db *sql.DB, tag *Tag) error { +func UpsertTag(db DBTX, tag *Tag) error { _, err := db.Exec(` INSERT INTO tags (did, repository, tag, digest, created_at) VALUES (?, ?, ?, ?, ?) @@ -605,7 +598,7 @@ func UpsertTag(db *sql.DB, tag *Tag) error { } // DeleteTag deletes a tag record -func DeleteTag(db *sql.DB, did, repository, tag string) error { +func DeleteTag(db DBTX, did, repository, tag string) error { _, err := db.Exec(` DELETE FROM tags WHERE did = ? AND repository = ? AND tag = ? `, did, repository, tag) @@ -616,7 +609,7 @@ func DeleteTag(db *sql.DB, did, repository, tag string) error { // Only multi-arch tags (manifest lists) have platform info in manifest_references // Single-arch tags will have empty Platforms slice (platform is obvious for single-arch) // Attestation references (unknown/unknown platforms) are filtered out but tracked via HasAttestations -func GetTagsWithPlatforms(db *sql.DB, did, repository string) ([]TagWithPlatforms, error) { +func GetTagsWithPlatforms(db DBTX, did, repository string) ([]TagWithPlatforms, error) { rows, err := db.Query(` SELECT t.id, @@ -700,7 +693,7 @@ func GetTagsWithPlatforms(db *sql.DB, did, repository string) ([]TagWithPlatform // DeleteManifest deletes a manifest and its associated layers // If repository is empty, deletes all manifests matching did and digest -func DeleteManifest(db *sql.DB, did, repository, digest string) error { +func DeleteManifest(db DBTX, did, repository, digest string) error { var err error if repository == "" { // Delete by DID + digest only (used when repository is unknown, e.g., Jetstream DELETE events) @@ -718,7 +711,7 @@ func DeleteManifest(db *sql.DB, did, repository, digest string) error { // // Due to ON DELETE CASCADE in the schema, deleting from users will automatically // cascade to: manifests, tags, layers, references, annotations, stars, repo_pages, etc. -func DeleteUserData(db *sql.DB, did string) error { +func DeleteUserData(db DBTX, did string) error { result, err := db.Exec(`DELETE FROM users WHERE did = ?`, did) if err != nil { return fmt.Errorf("failed to delete user: %w", err) @@ -735,7 +728,7 @@ func DeleteUserData(db *sql.DB, did string) error { // GetManifest fetches a single manifest by digest // Note: Annotations are stored separately in repository_annotations table -func GetManifest(db *sql.DB, digest string) (*Manifest, error) { +func GetManifest(db DBTX, digest string) (*Manifest, error) { var m Manifest err := db.QueryRow(` @@ -756,7 +749,7 @@ func GetManifest(db *sql.DB, digest string) (*Manifest, error) { // GetNewestManifestForRepo returns the newest manifest for a specific repository // Used by backfill to ensure annotations come from the most recent manifest -func GetNewestManifestForRepo(db *sql.DB, did, repository string) (*Manifest, error) { +func GetNewestManifestForRepo(db DBTX, did, repository string) (*Manifest, error) { var m Manifest err := db.QueryRow(` SELECT id, did, repository, digest, hold_endpoint, schema_version, media_type, @@ -779,7 +772,7 @@ func GetNewestManifestForRepo(db *sql.DB, did, repository string) (*Manifest, er // GetLatestHoldDIDForRepo returns the hold DID from the most recent manifest for a repository // Returns empty string if no manifests exist (e.g., first push) // This is used instead of the in-memory cache to determine which hold to use for blob operations -func GetLatestHoldDIDForRepo(db *sql.DB, did, repository string) (string, error) { +func GetLatestHoldDIDForRepo(db DBTX, did, repository string) (string, error) { var holdDID string err := db.QueryRow(` SELECT hold_endpoint @@ -802,7 +795,7 @@ func GetLatestHoldDIDForRepo(db *sql.DB, did, repository string) (string, error) // GetRepositoriesForDID returns all unique repository names for a DID // Used by backfill to reconcile annotations for all repositories -func GetRepositoriesForDID(db *sql.DB, did string) ([]string, error) { +func GetRepositoriesForDID(db DBTX, did string) ([]string, error) { rows, err := db.Query(` SELECT DISTINCT repository FROM manifests @@ -825,7 +818,7 @@ func GetRepositoriesForDID(db *sql.DB, did string) ([]string, error) { } // GetLayersForManifest fetches all layers for a manifest -func GetLayersForManifest(db *sql.DB, manifestID int64) ([]Layer, error) { +func GetLayersForManifest(db DBTX, manifestID int64) ([]Layer, error) { rows, err := db.Query(` SELECT manifest_id, digest, size, media_type, layer_index FROM layers @@ -851,7 +844,7 @@ func GetLayersForManifest(db *sql.DB, manifestID int64) ([]Layer, error) { } // InsertManifestReference inserts a new manifest reference record (for manifest lists/indexes) -func InsertManifestReference(db *sql.DB, ref *ManifestReference) error { +func InsertManifestReference(db DBTX, ref *ManifestReference) error { _, err := db.Exec(` INSERT INTO manifest_references (manifest_id, digest, size, media_type, platform_architecture, platform_os, @@ -866,7 +859,7 @@ func InsertManifestReference(db *sql.DB, ref *ManifestReference) error { } // GetManifestReferencesForManifest fetches all manifest references for a manifest list/index -func GetManifestReferencesForManifest(db *sql.DB, manifestID int64) ([]ManifestReference, error) { +func GetManifestReferencesForManifest(db DBTX, manifestID int64) ([]ManifestReference, error) { rows, err := db.Query(` SELECT manifest_id, digest, size, media_type, platform_architecture, platform_os, platform_variant, platform_os_version, @@ -914,7 +907,7 @@ func GetManifestReferencesForManifest(db *sql.DB, manifestID int64) ([]ManifestR // GetTopLevelManifests returns only manifest lists and orphaned single-arch manifests // Filters out platform-specific manifests that are referenced by manifest lists // Note: Annotations are stored separately in repository_annotations table - use GetRepositoryMetadata to fetch them -func GetTopLevelManifests(db *sql.DB, did, repository string, limit, offset int) ([]ManifestWithMetadata, error) { +func GetTopLevelManifests(db DBTX, did, repository string, limit, offset int) ([]ManifestWithMetadata, error) { rows, err := db.Query(` WITH manifest_list_children AS ( -- Get all digests that are children of manifest lists @@ -1047,7 +1040,7 @@ func GetTopLevelManifests(db *sql.DB, did, repository string, limit, offset int) // GetManifestDetail returns a manifest with full platform details and tags // Note: Annotations are stored separately in repository_annotations table - use GetRepositoryMetadata to fetch them -func GetManifestDetail(db *sql.DB, did, repository, digest string) (*ManifestWithMetadata, error) { +func GetManifestDetail(db DBTX, did, repository, digest string) (*ManifestWithMetadata, error) { // First, get the manifest and its tags var m ManifestWithMetadata var tags, configDigest sql.NullString @@ -1152,7 +1145,7 @@ func GetManifestDetail(db *sql.DB, did, repository, digest string) (*ManifestWit } // GetFirehoseCursor retrieves the current firehose cursor -func GetFirehoseCursor(db *sql.DB) (int64, error) { +func GetFirehoseCursor(db DBTX) (int64, error) { var cursor int64 err := db.QueryRow("SELECT cursor FROM firehose_cursor WHERE id = 1").Scan(&cursor) if err == sql.ErrNoRows { @@ -1162,7 +1155,7 @@ func GetFirehoseCursor(db *sql.DB) (int64, error) { } // UpdateFirehoseCursor updates the firehose cursor -func UpdateFirehoseCursor(db *sql.DB, cursor int64) error { +func UpdateFirehoseCursor(db DBTX, cursor int64) error { _, err := db.Exec(` INSERT INTO firehose_cursor (id, cursor, updated_at) VALUES (1, ?, datetime('now')) @@ -1174,7 +1167,7 @@ func UpdateFirehoseCursor(db *sql.DB, cursor int64) error { } // IsManifestTagged checks if a manifest has any tags -func IsManifestTagged(db *sql.DB, did, repository, digest string) (bool, error) { +func IsManifestTagged(db DBTX, did, repository, digest string) (bool, error) { var count int err := db.QueryRow(` SELECT COUNT(*) FROM tags @@ -1189,7 +1182,7 @@ func IsManifestTagged(db *sql.DB, did, repository, digest string) (bool, error) } // GetManifestTags retrieves all tags for a manifest -func GetManifestTags(db *sql.DB, did, repository, digest string) ([]string, error) { +func GetManifestTags(db DBTX, did, repository, digest string) ([]string, error) { rows, err := db.Query(` SELECT tag FROM tags WHERE did = ? AND repository = ? AND digest = ? @@ -1225,7 +1218,7 @@ type BackfillState struct { } // GetBackfillState retrieves the backfill state -func GetBackfillState(db *sql.DB) (*BackfillState, error) { +func GetBackfillState(db DBTX) (*BackfillState, error) { var state BackfillState var updatedAtStr string @@ -1263,7 +1256,7 @@ func GetBackfillState(db *sql.DB) (*BackfillState, error) { } // UpsertBackfillState updates or creates backfill state -func UpsertBackfillState(db *sql.DB, state *BackfillState) error { +func UpsertBackfillState(db DBTX, state *BackfillState) error { _, err := db.Exec(` INSERT INTO backfill_state (id, start_cursor, current_cursor, completed, updated_at) VALUES (1, ?, ?, ?, datetime('now')) @@ -1277,7 +1270,7 @@ func UpsertBackfillState(db *sql.DB, state *BackfillState) error { } // UpdateBackfillCursor updates just the current cursor position -func UpdateBackfillCursor(db *sql.DB, cursor int64) error { +func UpdateBackfillCursor(db DBTX, cursor int64) error { _, err := db.Exec(` UPDATE backfill_state SET current_cursor = ?, updated_at = datetime('now') @@ -1287,7 +1280,7 @@ func UpdateBackfillCursor(db *sql.DB, cursor int64) error { } // MarkBackfillCompleted marks the backfill as completed -func MarkBackfillCompleted(db *sql.DB) error { +func MarkBackfillCompleted(db DBTX) error { _, err := db.Exec(` UPDATE backfill_state SET completed = 1, updated_at = datetime('now') @@ -1297,7 +1290,7 @@ func MarkBackfillCompleted(db *sql.DB) error { } // GetRepository fetches a specific repository for a user -func GetRepository(db *sql.DB, did, repository string) (*Repository, error) { +func GetRepository(db DBTX, did, repository string) (*Repository, error) { // Get repository summary var r Repository r.Name = repository @@ -1412,7 +1405,7 @@ func GetRepository(db *sql.DB, did, repository string) (*Repository, error) { } // GetRepositoryStats fetches stats for a repository -func GetRepositoryStats(db *sql.DB, did, repository string) (*RepositoryStats, error) { +func GetRepositoryStats(db DBTX, did, repository string) (*RepositoryStats, error) { var stats RepositoryStats var lastPullStr, lastPushStr sql.NullString @@ -1463,7 +1456,7 @@ func GetRepositoryStats(db *sql.DB, did, repository string) (*RepositoryStats, e // UpsertRepositoryStats inserts or updates repository stats // Note: star_count is calculated dynamically from the stars table, not stored here -func UpsertRepositoryStats(db *sql.DB, stats *RepositoryStats) error { +func UpsertRepositoryStats(db DBTX, stats *RepositoryStats) error { _, err := db.Exec(` INSERT INTO repository_stats (did, repository, pull_count, last_pull, push_count, last_push) VALUES (?, ?, ?, ?, ?, ?) @@ -1477,7 +1470,7 @@ func UpsertRepositoryStats(db *sql.DB, stats *RepositoryStats) error { } // UpsertStar inserts or updates a star record (idempotent) -func UpsertStar(db *sql.DB, starrerDID, ownerDID, repository string, createdAt time.Time) error { +func UpsertStar(db DBTX, starrerDID, ownerDID, repository string, createdAt time.Time) error { _, err := db.Exec(` INSERT INTO stars (starrer_did, owner_did, repository, created_at) VALUES (?, ?, ?, ?) @@ -1488,7 +1481,7 @@ func UpsertStar(db *sql.DB, starrerDID, ownerDID, repository string, createdAt t } // DeleteStar deletes a star record -func DeleteStar(db *sql.DB, starrerDID, ownerDID, repository string) error { +func DeleteStar(db DBTX, starrerDID, ownerDID, repository string) error { _, err := db.Exec(` DELETE FROM stars WHERE starrer_did = ? AND owner_did = ? AND repository = ? @@ -1497,7 +1490,7 @@ func DeleteStar(db *sql.DB, starrerDID, ownerDID, repository string) error { } // RebuildStarCount rebuilds the star count for a specific repository from the stars table -func RebuildStarCount(db *sql.DB, ownerDID, repository string) error { +func RebuildStarCount(db DBTX, ownerDID, repository string) error { _, err := db.Exec(` INSERT INTO repository_stats (did, repository, star_count) VALUES (?, ?, ( @@ -1515,7 +1508,7 @@ func RebuildStarCount(db *sql.DB, ownerDID, repository string) error { // GetStarsForDID returns all stars created by a specific DID (for backfill reconciliation) // Returns a map of (ownerDID, repository) -> createdAt -func GetStarsForDID(db *sql.DB, starrerDID string) (map[string]time.Time, error) { +func GetStarsForDID(db DBTX, starrerDID string) (map[string]time.Time, error) { rows, err := db.Query(` SELECT owner_did, repository, created_at FROM stars @@ -1542,7 +1535,7 @@ func GetStarsForDID(db *sql.DB, starrerDID string) (map[string]time.Time, error) // CleanupOrphanedTags removes tags whose manifest digest no longer exists // This handles cases where manifests were deleted but tags pointing to them remain -func CleanupOrphanedTags(db *sql.DB, did string) error { +func CleanupOrphanedTags(db DBTX, did string) error { _, err := db.Exec(` DELETE FROM tags WHERE did = ? @@ -1557,7 +1550,7 @@ func CleanupOrphanedTags(db *sql.DB, did string) error { // DeleteStarsNotInList deletes stars from the database that are not in the provided list // This is used during backfill reconciliation to remove stars that no longer exist on PDS -func DeleteStarsNotInList(db *sql.DB, starrerDID string, foundStars map[string]time.Time) error { +func DeleteStarsNotInList(db DBTX, starrerDID string, foundStars map[string]time.Time) error { // Get current stars in DB currentStars, err := GetStarsForDID(db, starrerDID) if err != nil { @@ -1608,11 +1601,11 @@ func parseTimestamp(s string) (time.Time, error) { // HoldDIDDB wraps a sql.DB and implements the HoldDIDLookup interface for middleware // This is a minimal wrapper that only provides hold DID lookups for blob routing type HoldDIDDB struct { - db *sql.DB + db DBTX } // NewHoldDIDDB creates a new hold DID database wrapper -func NewHoldDIDDB(db *sql.DB) *HoldDIDDB { +func NewHoldDIDDB(db DBTX) *HoldDIDDB { return &HoldDIDDB{db: db} } @@ -1632,7 +1625,7 @@ const ( ) // GetRepoCards fetches repository cards with full data including Tag, Digest, and LastUpdated -func GetRepoCards(db *sql.DB, limit int, currentUserDID string, sortOrder RepoCardSortOrder) ([]RepoCardData, error) { +func GetRepoCards(db DBTX, limit int, currentUserDID string, sortOrder RepoCardSortOrder) ([]RepoCardData, error) { // Build ORDER BY clause based on sort order var orderBy string switch sortOrder { @@ -1713,7 +1706,7 @@ func GetRepoCards(db *sql.DB, limit int, currentUserDID string, sortOrder RepoCa } // GetUserRepoCards fetches repository cards for a specific user with full data -func GetUserRepoCards(db *sql.DB, userDID string, currentUserDID string) ([]RepoCardData, error) { +func GetUserRepoCards(db DBTX, userDID string, currentUserDID string) ([]RepoCardData, error) { query := ` WITH latest_manifests AS ( SELECT did, repository, MAX(id) as latest_id @@ -1795,7 +1788,7 @@ type RepoPage struct { } // UpsertRepoPage inserts or updates a repo page record -func UpsertRepoPage(db *sql.DB, did, repository, description, avatarCID string, createdAt, updatedAt time.Time) error { +func UpsertRepoPage(db DBTX, did, repository, description, avatarCID string, createdAt, updatedAt time.Time) error { _, err := db.Exec(` INSERT INTO repo_pages (did, repository, description, avatar_cid, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) @@ -1808,7 +1801,7 @@ func UpsertRepoPage(db *sql.DB, did, repository, description, avatarCID string, } // GetRepoPage retrieves a repo page record -func GetRepoPage(db *sql.DB, did, repository string) (*RepoPage, error) { +func GetRepoPage(db DBTX, did, repository string) (*RepoPage, error) { var rp RepoPage err := db.QueryRow(` SELECT did, repository, description, avatar_cid, created_at, updated_at @@ -1822,7 +1815,7 @@ func GetRepoPage(db *sql.DB, did, repository string) (*RepoPage, error) { } // DeleteRepoPage deletes a repo page record -func DeleteRepoPage(db *sql.DB, did, repository string) error { +func DeleteRepoPage(db DBTX, did, repository string) error { _, err := db.Exec(` DELETE FROM repo_pages WHERE did = ? AND repository = ? `, did, repository) @@ -1830,7 +1823,7 @@ func DeleteRepoPage(db *sql.DB, did, repository string) error { } // GetRepoPagesByDID returns all repo pages for a DID -func GetRepoPagesByDID(db *sql.DB, did string) ([]RepoPage, error) { +func GetRepoPagesByDID(db DBTX, did string) ([]RepoPage, error) { rows, err := db.Query(` SELECT did, repository, description, avatar_cid, created_at, updated_at FROM repo_pages diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index 498a899..b789db9 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -181,7 +181,9 @@ func (b *BackfillWorker) backfillCollection(ctx context.Context, collection stri return nil } -// backfillRepo backfills all records for a single repo/DID +// backfillRepo backfills all records for a single repo/DID. +// Per-record processing is wrapped in a single SQL transaction to batch writes +// (one commit per repo instead of per-statement). func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection string) (int, error) { // Resolve DID to get user's PDS endpoint pdsEndpoint, err := atproto.ResolveDIDToPDS(ctx, did) @@ -193,6 +195,16 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin // This allows GetRecord to work properly with the repo parameter pdsClient := atproto.NewClient(pdsEndpoint, did, "") + // Begin transaction for per-record processing (batches all writes into one commit) + tx, err := b.db.Begin() + if err != nil { + return 0, fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback() + + // Create a transactional processor — all DB writes go through this tx + txProcessor := NewProcessor(tx, false, b.processor.statsCache) + var recordCursor string recordCount := 0 @@ -235,7 +247,7 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin } } - if err := b.processRecord(ctx, did, collection, &record); err != nil { + if err := b.processRecordWith(ctx, txProcessor, did, collection, &record); err != nil { slog.Warn("Backfill failed to process record", "uri", record.URI, "error", err) continue } @@ -250,6 +262,13 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin recordCursor = cursor } + // Commit all per-record writes in one batch + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("failed to commit transaction: %w", err) + } + + // Reconciliation runs outside the transaction (involves network I/O and fewer writes) + // Reconcile deletions - remove records from DB that no longer exist on PDS if err := b.reconcileDeletions(did, collection, foundManifestDigests, foundTags, foundStars); err != nil { slog.Warn("Backfill failed to reconcile deletions", "did", did, "error", err) @@ -334,9 +353,14 @@ func (b *BackfillWorker) reconcileDeletions(did, collection string, foundManifes return nil } -// processRecord processes a single record using the unified ProcessRecord method. -// This ensures consistent handling (validation, user creation) between Worker and Backfill. +// processRecord processes a single record using the default processor. func (b *BackfillWorker) processRecord(ctx context.Context, did, collection string, record *atproto.Record) error { + return b.processRecordWith(ctx, b.processor, did, collection, record) +} + +// processRecordWith processes a single record using the given processor. +// This allows backfillRepo to use a transactional processor while other callers use the default. +func (b *BackfillWorker) processRecordWith(ctx context.Context, proc *Processor, did, collection string, record *atproto.Record) error { rkey := extractRkeyFromURI(record.URI) // For sailor profile collection, we need to pass the queryCaptainFn @@ -346,7 +370,7 @@ func (b *BackfillWorker) processRecord(ctx context.Context, did, collection stri queryCaptainFn = b.queryCaptainRecordWrapper } - return b.processor.ProcessRecord(ctx, did, collection, rkey, record.Value, false, queryCaptainFn) + return proc.ProcessRecord(ctx, did, collection, rkey, record.Value, false, queryCaptainFn) } // queryCaptainRecordWrapper wraps queryCaptainRecord with backfill-specific logic diff --git a/pkg/appview/jetstream/processor.go b/pkg/appview/jetstream/processor.go index 1371457..6bc4ce7 100644 --- a/pkg/appview/jetstream/processor.go +++ b/pkg/appview/jetstream/processor.go @@ -2,7 +2,6 @@ package jetstream import ( "context" - "database/sql" "encoding/json" "fmt" "log/slog" @@ -18,7 +17,7 @@ import ( // 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 + db db.DBTX userCache *UserCache // Optional - enabled for Worker, disabled for Backfill statsCache *StatsCache // In-memory cache for per-hold stats aggregation useCache bool @@ -28,7 +27,7 @@ type Processor struct { // NewProcessor creates a new shared processor // useCache: true for Worker (live streaming), false for Backfill (batch processing) // statsCache: shared stats cache for aggregating across holds (nil to skip stats processing) -func NewProcessor(database *sql.DB, useCache bool, statsCache *StatsCache) *Processor { +func NewProcessor(database db.DBTX, useCache bool, statsCache *StatsCache) *Processor { // Create lexicon catalog for debug validation logging dir := identity.DefaultDirectory() catalog := lexicon.NewResolvingCatalog()