mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-09-21 01:34:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

change to transactions for database

Browse Source
This commit is contained in:
Evan Jarrett
2026-02-10 20:58:24 -06:00
parent 4ac2b97c33
commit b251c8857f
11 changed files with 148 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config-hold.example.yaml
+6
View File
@@ -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.
deploy/upcloud/configs/appview.yaml.tmpl
+9 -2
View File
@@ -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"
deploy/upcloud/configs/hold.yaml.tmpl
+4
View File
@@ -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:
pkg/appview/db/annotations.go
+9 -17
View File
@@ -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 = ?
pkg/appview/db/dbtx.go
+13
View File
@@ -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)
}
pkg/appview/db/delete.go
+2 -3
View File
@@ -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 {
pkg/appview/db/export.go
+5 -5
View File
@@ -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{},
pkg/appview/db/hold_store.go
+9 -9
View File
@@ -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
pkg/appview/db/queries.go
+60 -67
View File
@@ -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
pkg/appview/jetstream/backfill.go
+29 -5
View File
@@ -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
pkg/appview/jetstream/processor.go
+2 -3
View File
@@ -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()