From 7a6775d4fa10e3c23841619304417eb816efd8a5 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sun, 19 Apr 2026 18:04:57 -0500 Subject: [PATCH] missed new files for jetstream improvements --- pkg/appview/db/batch.go | 580 ++++++++++++++++++ pkg/appview/db/batch_test.go | 383 ++++++++++++ pkg/appview/db/conn.go | 69 +++ pkg/appview/db/conn_test.go | 28 + pkg/appview/db/jetstream_cursor.go | 33 + pkg/appview/db/jetstream_cursor_test.go | 50 ++ .../0022_create_jetstream_cursor.yaml | 7 + pkg/appview/jetstream/backfill_batch.go | 518 ++++++++++++++++ pkg/appview/src/css/main.css | 4 +- .../templates/components/repo-card.html | 2 +- 10 files changed, 1671 insertions(+), 3 deletions(-) create mode 100644 pkg/appview/db/batch.go create mode 100644 pkg/appview/db/batch_test.go create mode 100644 pkg/appview/db/conn.go create mode 100644 pkg/appview/db/conn_test.go create mode 100644 pkg/appview/db/jetstream_cursor.go create mode 100644 pkg/appview/db/jetstream_cursor_test.go create mode 100644 pkg/appview/db/migrations/0022_create_jetstream_cursor.yaml create mode 100644 pkg/appview/jetstream/backfill_batch.go diff --git a/pkg/appview/db/batch.go b/pkg/appview/db/batch.go new file mode 100644 index 0000000..c1cef82 --- /dev/null +++ b/pkg/appview/db/batch.go @@ -0,0 +1,580 @@ +package db + +import ( + "encoding/json" + "fmt" + "strings" + "time" +) + +// BatchSize is the maximum number of rows included in a single multi-row INSERT. +// Kept well under SQLite's default SQLITE_MAX_VARIABLE_NUMBER (32766) and any +// remote libsql parameter ceiling — at 11 columns this is 1100 placeholders. +const BatchSize = 100 + +// buildPlaceholders returns a comma-separated list of `rows` groups of the form +// `(?,?,?)`, each group containing `cols` placeholders. Used to construct the +// VALUES clause of multi-row INSERT statements. +func buildPlaceholders(rows, cols int) string { + if rows <= 0 || cols <= 0 { + return "" + } + group := "(" + strings.Repeat("?,", cols-1) + "?)" + var sb strings.Builder + sb.Grow((len(group) + 1) * rows) + for i := 0; i < rows; i++ { + if i > 0 { + sb.WriteByte(',') + } + sb.WriteString(group) + } + return sb.String() +} + +// chunk returns the half-open range [start, end) for the i-th chunk of size +// BatchSize within a slice of length n. +func chunk(n, i int) (start, end int) { + start = i * BatchSize + end = start + BatchSize + if end > n { + end = n + } + return start, end +} + +// BatchInsertManifests upserts a batch of manifests and returns a map of +// digest → manifest id for the inserted rows (both new and existing). Rows +// are keyed by (did, repository, digest); callers that need the id must +// group their input so that digest is unique per (did, repository) in one +// batch call. +// +// Implementation: one multi-row INSERT per sub-batch, followed by one SELECT +// to fetch ids back (libsql's RETURNING support across replica modes is +// uneven; a second SELECT is reliable and still a single round-trip per +// sub-batch). +func BatchInsertManifests(db DBTX, manifests []Manifest) (map[string]int64, error) { + out := make(map[string]int64, len(manifests)) + if len(manifests) == 0 { + return out, nil + } + + for i := 0; i*BatchSize < len(manifests); i++ { + start, end := chunk(len(manifests), i) + batch := manifests[start:end] + + const cols = 11 + args := make([]any, 0, len(batch)*cols) + for _, m := range batch { + args = append(args, + m.DID, m.Repository, m.Digest, m.HoldEndpoint, + m.SchemaVersion, m.MediaType, m.ConfigDigest, + m.ConfigSize, m.ArtifactType, + nullString(m.SubjectDigest), + m.CreatedAt, + ) + } + + query := ` + INSERT INTO manifests + (did, repository, digest, hold_endpoint, schema_version, media_type, + config_digest, config_size, artifact_type, subject_digest, created_at) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(did, repository, digest) DO UPDATE SET + hold_endpoint = excluded.hold_endpoint, + schema_version = excluded.schema_version, + media_type = excluded.media_type, + config_digest = excluded.config_digest, + config_size = excluded.config_size, + artifact_type = excluded.artifact_type, + subject_digest = excluded.subject_digest + WHERE excluded.hold_endpoint != manifests.hold_endpoint + OR excluded.schema_version != manifests.schema_version + OR excluded.media_type != manifests.media_type + OR excluded.config_digest IS NOT manifests.config_digest + OR excluded.config_size IS NOT manifests.config_size + OR excluded.artifact_type != manifests.artifact_type + OR excluded.subject_digest IS NOT manifests.subject_digest + ` + if _, err := db.Exec(query, args...); err != nil { + return nil, fmt.Errorf("batch insert manifests: %w", err) + } + + // Fetch ids for this sub-batch by (did, digest) — digests are unique enough + // that matching on (did, digest) avoids needing a three-column IN list. + // repository is included in the row to disambiguate if a user genuinely has + // the same digest across repos. + selectArgs := make([]any, 0, 1+2*len(batch)) + // Group by did (caller usually supplies one did per call, but be safe). + didSet := make(map[string]struct{}) + for _, m := range batch { + didSet[m.DID] = struct{}{} + } + // Build a per-did IN (?) query; usually exactly one iteration. + for did := range didSet { + digests := make([]string, 0, len(batch)) + for _, m := range batch { + if m.DID == did { + digests = append(digests, m.Digest) + } + } + selectArgs = append(selectArgs[:0], did) + for _, d := range digests { + selectArgs = append(selectArgs, d) + } + selectQuery := ` + SELECT repository, digest, id FROM manifests + WHERE did = ? AND digest IN (` + + strings.TrimSuffix(strings.Repeat("?,", len(digests)), ",") + `) + ` + rows, err := db.Query(selectQuery, selectArgs...) + if err != nil { + return nil, fmt.Errorf("batch select manifest ids: %w", err) + } + for rows.Next() { + var repo, digest string + var id int64 + if err := rows.Scan(&repo, &digest, &id); err != nil { + rows.Close() + return nil, fmt.Errorf("scan manifest id: %w", err) + } + // Key format matches what callers use: "did|repo|digest". + out[manifestKey(did, repo, digest)] = id + } + rows.Close() + } + } + return out, nil +} + +// ManifestKey builds the lookup key used by BatchInsertManifests' result map. +// Callers construct the same key from their in-memory Manifest structs to +// find the assigned id. +func ManifestKey(did, repository, digest string) string { + return manifestKey(did, repository, digest) +} + +func manifestKey(did, repository, digest string) string { + return did + "|" + repository + "|" + digest +} + +// BatchInsertLayers inserts a batch of layers, skipping any that already exist. +// Layers are immutable, so ON CONFLICT DO NOTHING matches the single-row +// InsertLayer semantics. +func BatchInsertLayers(db DBTX, layers []Layer) error { + if len(layers) == 0 { + return nil + } + for i := 0; i*BatchSize < len(layers); i++ { + start, end := chunk(len(layers), i) + batch := layers[start:end] + + const cols = 6 + args := make([]any, 0, len(batch)*cols) + for _, l := range batch { + var annotationsJSON any + if len(l.Annotations) > 0 { + b, err := json.Marshal(l.Annotations) + if err != nil { + return fmt.Errorf("marshal layer annotations: %w", err) + } + s := string(b) + annotationsJSON = &s + } + args = append(args, l.ManifestID, l.Digest, l.Size, l.MediaType, l.LayerIndex, annotationsJSON) + } + + query := ` + INSERT INTO layers (manifest_id, digest, size, media_type, layer_index, annotations) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(manifest_id, layer_index) DO NOTHING + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch insert layers: %w", err) + } + } + return nil +} + +// BatchInsertManifestReferences inserts a batch of manifest references. +// The table has PRIMARY KEY(manifest_id, reference_index); duplicates skip. +func BatchInsertManifestReferences(db DBTX, refs []ManifestReference) error { + if len(refs) == 0 { + return nil + } + for i := 0; i*BatchSize < len(refs); i++ { + start, end := chunk(len(refs), i) + batch := refs[start:end] + + const cols = 10 + args := make([]any, 0, len(batch)*cols) + for _, r := range batch { + args = append(args, + r.ManifestID, r.Digest, r.Size, r.MediaType, + r.PlatformArchitecture, r.PlatformOS, + r.PlatformVariant, r.PlatformOSVersion, + r.IsAttestation, r.ReferenceIndex, + ) + } + + query := ` + INSERT INTO manifest_references (manifest_id, digest, size, media_type, + platform_architecture, platform_os, + platform_variant, platform_os_version, + is_attestation, reference_index) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(manifest_id, reference_index) DO NOTHING + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch insert manifest references: %w", err) + } + } + return nil +} + +// BatchUpsertTags upserts a batch of tag records, matching UpsertTag semantics. +func BatchUpsertTags(db DBTX, tags []Tag) error { + if len(tags) == 0 { + return nil + } + for i := 0; i*BatchSize < len(tags); i++ { + start, end := chunk(len(tags), i) + batch := tags[start:end] + + const cols = 5 + args := make([]any, 0, len(batch)*cols) + for _, t := range batch { + args = append(args, t.DID, t.Repository, t.Tag, t.Digest, t.CreatedAt) + } + + query := ` + INSERT INTO tags (did, repository, tag, digest, created_at) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(did, repository, tag) DO UPDATE SET + digest = excluded.digest, + created_at = excluded.created_at + WHERE excluded.digest != tags.digest + OR excluded.created_at != tags.created_at + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert tags: %w", err) + } + } + return nil +} + +// StarInput is a struct projection of the UpsertStar argument list for use with BatchUpsertStars. +type StarInput struct { + StarrerDID string + OwnerDID string + Repository string + CreatedAt time.Time +} + +// BatchUpsertStars upserts a batch of stars. Stars are immutable. +func BatchUpsertStars(db DBTX, stars []StarInput) error { + if len(stars) == 0 { + return nil + } + for i := 0; i*BatchSize < len(stars); i++ { + start, end := chunk(len(stars), i) + batch := stars[start:end] + + const cols = 4 + args := make([]any, 0, len(batch)*cols) + for _, s := range batch { + args = append(args, s.StarrerDID, s.OwnerDID, s.Repository, s.CreatedAt) + } + + query := ` + INSERT INTO stars (starrer_did, owner_did, repository, created_at) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(starrer_did, owner_did, repository) DO NOTHING + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert stars: %w", err) + } + } + return nil +} + +// BatchUpsertRepoPages upserts a batch of repo page records. +func BatchUpsertRepoPages(db DBTX, pages []RepoPage) error { + if len(pages) == 0 { + return nil + } + for i := 0; i*BatchSize < len(pages); i++ { + start, end := chunk(len(pages), i) + batch := pages[start:end] + + const cols = 7 + args := make([]any, 0, len(batch)*cols) + for _, p := range batch { + args = append(args, + p.DID, p.Repository, p.Description, p.AvatarCID, + p.UserEdited, p.CreatedAt, p.UpdatedAt, + ) + } + + query := ` + INSERT INTO repo_pages (did, repository, description, avatar_cid, user_edited, created_at, updated_at) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(did, repository) DO UPDATE SET + description = excluded.description, + avatar_cid = excluded.avatar_cid, + user_edited = excluded.user_edited, + updated_at = excluded.updated_at + WHERE excluded.description IS NOT repo_pages.description + OR excluded.avatar_cid IS NOT repo_pages.avatar_cid + OR excluded.user_edited IS NOT repo_pages.user_edited + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert repo pages: %w", err) + } + } + return nil +} + +// BatchUpsertDailyStats upserts a batch of daily stats rows. +func BatchUpsertDailyStats(db DBTX, stats []DailyStats) error { + if len(stats) == 0 { + return nil + } + for i := 0; i*BatchSize < len(stats); i++ { + start, end := chunk(len(stats), i) + batch := stats[start:end] + + const cols = 5 + args := make([]any, 0, len(batch)*cols) + for _, s := range batch { + args = append(args, s.DID, s.Repository, s.Date, s.PullCount, s.PushCount) + } + + query := ` + INSERT INTO repository_stats_daily (did, repository, date, pull_count, push_count) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(did, repository, date) DO UPDATE SET + pull_count = excluded.pull_count, + push_count = excluded.push_count + WHERE excluded.pull_count != repository_stats_daily.pull_count + OR excluded.push_count != repository_stats_daily.push_count + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert daily stats: %w", err) + } + } + return nil +} + +// BatchUpsertRepositoryStats upserts aggregated repository stats. +func BatchUpsertRepositoryStats(db DBTX, stats []RepositoryStats) error { + if len(stats) == 0 { + return nil + } + for i := 0; i*BatchSize < len(stats); i++ { + start, end := chunk(len(stats), i) + batch := stats[start:end] + + const cols = 6 + args := make([]any, 0, len(batch)*cols) + for _, s := range batch { + args = append(args, + s.DID, s.Repository, s.PullCount, s.LastPull, s.PushCount, s.LastPush, + ) + } + + query := ` + INSERT INTO repository_stats (did, repository, pull_count, last_pull, push_count, last_push) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(did, repository) DO UPDATE SET + pull_count = excluded.pull_count, + last_pull = excluded.last_pull, + push_count = excluded.push_count, + last_push = excluded.last_push + WHERE excluded.pull_count != repository_stats.pull_count + OR excluded.last_pull IS NOT repository_stats.last_pull + OR excluded.push_count != repository_stats.push_count + OR excluded.last_push IS NOT repository_stats.last_push + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert repository stats: %w", err) + } + } + return nil +} + +// BatchUpsertCaptainRecords upserts a batch of captain records. +func BatchUpsertCaptainRecords(db DBTX, records []HoldCaptainRecord) error { + if len(records) == 0 { + return nil + } + for i := 0; i*BatchSize < len(records); i++ { + start, end := chunk(len(records), i) + batch := records[start:end] + + const cols = 8 + args := make([]any, 0, len(batch)*cols) + for _, r := range batch { + args = append(args, + r.HoldDID, r.OwnerDID, r.Public, r.AllowAllCrew, + nullString(r.DeployedAt), + nullString(r.Region), + nullString(r.Successor), + r.UpdatedAt, + ) + } + + query := ` + INSERT INTO hold_captain_records ( + hold_did, owner_did, public, allow_all_crew, + deployed_at, region, successor, updated_at + ) VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(hold_did) DO UPDATE SET + owner_did = excluded.owner_did, + public = excluded.public, + allow_all_crew = excluded.allow_all_crew, + deployed_at = excluded.deployed_at, + region = excluded.region, + successor = excluded.successor, + updated_at = excluded.updated_at + WHERE excluded.owner_did != hold_captain_records.owner_did + OR excluded.public != hold_captain_records.public + OR excluded.allow_all_crew != hold_captain_records.allow_all_crew + OR excluded.deployed_at IS NOT hold_captain_records.deployed_at + OR excluded.region IS NOT hold_captain_records.region + OR excluded.successor IS NOT hold_captain_records.successor + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert captain records: %w", err) + } + } + return nil +} + +// BatchUpsertCrewMembers upserts a batch of crew members. +func BatchUpsertCrewMembers(db DBTX, members []CrewMember) error { + if len(members) == 0 { + return nil + } + for i := 0; i*BatchSize < len(members); i++ { + start, end := chunk(len(members), i) + batch := members[start:end] + + // updated_at uses CURRENT_TIMESTAMP literal, so it's not a placeholder. + const cols = 7 + args := make([]any, 0, len(batch)*cols) + for _, m := range batch { + args = append(args, + m.HoldDID, m.MemberDID, m.Rkey, + nullString(m.Role), + nullString(m.Permissions), + nullString(m.Tier), + nullString(m.AddedAt), + ) + } + + // Replace each group with `(?,?,?,?,?,?,?,CURRENT_TIMESTAMP)` — we build it + // manually because buildPlaceholders only handles uniform placeholders. + group := "(" + strings.Repeat("?,", cols) + "CURRENT_TIMESTAMP)" + var sb strings.Builder + sb.Grow((len(group) + 1) * len(batch)) + for i := 0; i < len(batch); i++ { + if i > 0 { + sb.WriteByte(',') + } + sb.WriteString(group) + } + + query := ` + INSERT INTO hold_crew_members ( + hold_did, member_did, rkey, role, permissions, tier, added_at, updated_at + ) VALUES ` + sb.String() + ` + ON CONFLICT(hold_did, member_did) DO UPDATE SET + rkey = excluded.rkey, + role = excluded.role, + permissions = excluded.permissions, + tier = excluded.tier, + added_at = excluded.added_at, + updated_at = CURRENT_TIMESTAMP + WHERE excluded.rkey != hold_crew_members.rkey + OR excluded.role IS NOT hold_crew_members.role + OR excluded.permissions IS NOT hold_crew_members.permissions + OR excluded.tier IS NOT hold_crew_members.tier + OR excluded.added_at IS NOT hold_crew_members.added_at + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert crew members: %w", err) + } + } + return nil +} + +// AnnotationRow represents a single key/value annotation for a repository, +// used by BatchUpsertRepositoryAnnotations. +type AnnotationRow struct { + DID string + Repository string + Key string + Value string +} + +// BatchUpsertRepositoryAnnotations upserts annotation rows and deletes any +// stale keys for each (did, repository) represented in the input. The caller +// is responsible for pre-filtering: rows should represent only repositories +// whose newest manifest has at least one non-empty annotation, matching the +// single-row UpsertRepositoryAnnotations semantics. +func BatchUpsertRepositoryAnnotations(db DBTX, rows []AnnotationRow) error { + if len(rows) == 0 { + return nil + } + + // Group rows by (did, repository) so we can delete stale keys per repo. + type repoKey struct{ did, repo string } + keysByRepo := make(map[repoKey][]string) + for _, r := range rows { + k := repoKey{r.DID, r.Repository} + keysByRepo[k] = append(keysByRepo[k], r.Key) + } + + // Delete stale keys per repository in one statement each. We could batch + // further with OR chains, but DELETE is cheap and each repo has few keys. + for k, keys := range keysByRepo { + placeholders := strings.TrimSuffix(strings.Repeat("?,", len(keys)), ",") + args := make([]any, 0, 2+len(keys)) + args = append(args, k.did, k.repo) + for _, key := range keys { + args = append(args, key) + } + if _, err := db.Exec(` + DELETE FROM repository_annotations + WHERE did = ? AND repository = ? AND key NOT IN (`+placeholders+`) + `, args...); err != nil { + return fmt.Errorf("batch delete stale annotations: %w", err) + } + } + + // Upsert all annotation rows in sub-batches. + now := time.Now() + for i := 0; i*BatchSize < len(rows); i++ { + start, end := chunk(len(rows), i) + batch := rows[start:end] + + const cols = 5 + args := make([]any, 0, len(batch)*cols) + for _, r := range batch { + args = append(args, r.DID, r.Repository, r.Key, r.Value, now) + } + + query := ` + INSERT INTO repository_annotations (did, repository, key, value, updated_at) + VALUES ` + buildPlaceholders(len(batch), cols) + ` + ON CONFLICT(did, repository, key) DO UPDATE SET + value = excluded.value, + updated_at = excluded.updated_at + WHERE excluded.value != repository_annotations.value + ` + if _, err := db.Exec(query, args...); err != nil { + return fmt.Errorf("batch upsert annotations: %w", err) + } + } + return nil +} diff --git a/pkg/appview/db/batch_test.go b/pkg/appview/db/batch_test.go new file mode 100644 index 0000000..18f9229 --- /dev/null +++ b/pkg/appview/db/batch_test.go @@ -0,0 +1,383 @@ +package db + +import ( + "database/sql" + "fmt" + "strings" + "testing" + "time" +) + +// setupBatchTestDB spins up a fresh in-memory libsql database with the full +// schema applied, so every batch test can write realistic data without +// stubbing individual tables. +func setupBatchTestDB(t *testing.T) *sql.DB { + t.Helper() + safeName := strings.ReplaceAll(t.Name(), "/", "_") + d, err := InitDB(fmt.Sprintf("file:%s?mode=memory&cache=shared", safeName), LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + // Single conn to avoid cross-test contention in the shared in-memory cache. + d.SetMaxOpenConns(1) + t.Cleanup(func() { d.Close() }) + return d +} + +func createBatchTestUser(t *testing.T, d *sql.DB, did string) { + t.Helper() + _, err := d.Exec(` + INSERT OR IGNORE INTO users (did, handle, pds_endpoint, last_seen) + VALUES (?, ?, ?, datetime('now')) + `, did, did+".bsky.social", "https://pds.example.com") + if err != nil { + t.Fatalf("seed user: %v", err) + } +} + +func countRows(t *testing.T, d *sql.DB, query string, args ...any) int { + t.Helper() + var n int + if err := d.QueryRow(query, args...).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n +} + +func TestBuildPlaceholders(t *testing.T) { + cases := []struct { + rows, cols int + want string + }{ + {1, 1, "(?)"}, + {2, 1, "(?),(?)"}, + {1, 3, "(?,?,?)"}, + {3, 2, "(?,?),(?,?),(?,?)"}, + {0, 5, ""}, + {5, 0, ""}, + } + for _, c := range cases { + got := buildPlaceholders(c.rows, c.cols) + if got != c.want { + t.Errorf("buildPlaceholders(%d,%d) = %q, want %q", c.rows, c.cols, got, c.want) + } + } +} + +func TestBatchInsertManifests_InsertsAndReturnsIDs(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + now := time.Now() + manifests := []Manifest{ + {DID: "did:plc:alice", Repository: "app1", Digest: "sha256:aaa", HoldEndpoint: "did:web:hold", SchemaVersion: 2, MediaType: "application/vnd.oci.image.manifest.v1+json", ArtifactType: "container-image", CreatedAt: now}, + {DID: "did:plc:alice", Repository: "app2", Digest: "sha256:bbb", HoldEndpoint: "did:web:hold", SchemaVersion: 2, MediaType: "application/vnd.oci.image.manifest.v1+json", ArtifactType: "container-image", CreatedAt: now}, + } + + ids, err := BatchInsertManifests(d, manifests) + if err != nil { + t.Fatalf("batch insert: %v", err) + } + if len(ids) != 2 { + t.Fatalf("expected 2 ids, got %d", len(ids)) + } + if ids[ManifestKey("did:plc:alice", "app1", "sha256:aaa")] == 0 { + t.Errorf("missing id for app1") + } + if ids[ManifestKey("did:plc:alice", "app2", "sha256:bbb")] == 0 { + t.Errorf("missing id for app2") + } + if got := countRows(t, d, `SELECT COUNT(*) FROM manifests`); got != 2 { + t.Errorf("row count = %d, want 2", got) + } +} + +func TestBatchInsertManifests_Idempotent(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + now := time.Now() + m := []Manifest{{ + DID: "did:plc:alice", Repository: "app", Digest: "sha256:aaa", + HoldEndpoint: "did:web:hold", SchemaVersion: 2, + MediaType: "application/vnd.oci.image.manifest.v1+json", + ArtifactType: "container-image", CreatedAt: now, + }} + if _, err := BatchInsertManifests(d, m); err != nil { + t.Fatalf("first insert: %v", err) + } + if _, err := BatchInsertManifests(d, m); err != nil { + t.Fatalf("second insert: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM manifests`); got != 1 { + t.Errorf("expected idempotent; row count = %d", got) + } +} + +func TestBatchInsertManifests_Chunking(t *testing.T) { + // Exceed one sub-batch to exercise the chunk loop. + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + const n = BatchSize + 17 + now := time.Now() + manifests := make([]Manifest, n) + for i := 0; i < n; i++ { + manifests[i] = Manifest{ + DID: "did:plc:alice", Repository: "app", Digest: fmt.Sprintf("sha256:%04d", i), + HoldEndpoint: "did:web:hold", SchemaVersion: 2, + MediaType: "application/vnd.oci.image.manifest.v1+json", + ArtifactType: "container-image", CreatedAt: now, + } + } + ids, err := BatchInsertManifests(d, manifests) + if err != nil { + t.Fatalf("batch insert: %v", err) + } + if len(ids) != n { + t.Errorf("ids len = %d, want %d", len(ids), n) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM manifests`); got != n { + t.Errorf("row count = %d, want %d", got, n) + } +} + +func TestBatchInsertLayers_RespectsFK(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + now := time.Now() + ids, err := BatchInsertManifests(d, []Manifest{{ + DID: "did:plc:alice", Repository: "app", Digest: "sha256:aaa", + HoldEndpoint: "did:web:hold", SchemaVersion: 2, + MediaType: "application/vnd.oci.image.manifest.v1+json", + ArtifactType: "container-image", CreatedAt: now, + }}) + if err != nil { + t.Fatalf("insert manifest: %v", err) + } + mid := ids[ManifestKey("did:plc:alice", "app", "sha256:aaa")] + + layers := []Layer{ + {ManifestID: mid, Digest: "sha256:L0", Size: 100, MediaType: "application/vnd.oci.image.layer.v1.tar+gzip", LayerIndex: 0}, + {ManifestID: mid, Digest: "sha256:L1", Size: 200, MediaType: "application/vnd.oci.image.layer.v1.tar+gzip", LayerIndex: 1}, + } + if err := BatchInsertLayers(d, layers); err != nil { + t.Fatalf("batch insert layers: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM layers`); got != 2 { + t.Errorf("layers count = %d, want 2", got) + } + // Re-run to confirm ON CONFLICT DO NOTHING doesn't error. + if err := BatchInsertLayers(d, layers); err != nil { + t.Fatalf("idempotent layers: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM layers`); got != 2 { + t.Errorf("layers after re-insert = %d, want 2", got) + } +} + +func TestBatchUpsertTags_Idempotent(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + now := time.Now() + tags := []Tag{ + {DID: "did:plc:alice", Repository: "app", Tag: "v1", Digest: "sha256:aaa", CreatedAt: now}, + {DID: "did:plc:alice", Repository: "app", Tag: "v2", Digest: "sha256:bbb", CreatedAt: now}, + } + if err := BatchUpsertTags(d, tags); err != nil { + t.Fatalf("batch upsert: %v", err) + } + if err := BatchUpsertTags(d, tags); err != nil { + t.Fatalf("rerun: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM tags`); got != 2 { + t.Errorf("tags count = %d, want 2", got) + } +} + +func TestBatchUpsertStars(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + createBatchTestUser(t, d, "did:plc:bob") + + now := time.Now() + stars := []StarInput{ + {StarrerDID: "did:plc:bob", OwnerDID: "did:plc:alice", Repository: "app", CreatedAt: now}, + } + if err := BatchUpsertStars(d, stars); err != nil { + t.Fatalf("batch upsert stars: %v", err) + } + // Re-insert to confirm ON CONFLICT DO NOTHING. + if err := BatchUpsertStars(d, stars); err != nil { + t.Fatalf("rerun: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM stars`); got != 1 { + t.Errorf("stars count = %d, want 1", got) + } +} + +func TestBatchUpsertRepoPages(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + now := time.Now() + pages := []RepoPage{ + {DID: "did:plc:alice", Repository: "app", Description: "desc", CreatedAt: now, UpdatedAt: now}, + } + if err := BatchUpsertRepoPages(d, pages); err != nil { + t.Fatalf("batch upsert: %v", err) + } + // Update with new description. + pages[0].Description = "new desc" + if err := BatchUpsertRepoPages(d, pages); err != nil { + t.Fatalf("update: %v", err) + } + var desc string + if err := d.QueryRow(`SELECT description FROM repo_pages WHERE did=? AND repository=?`, + "did:plc:alice", "app").Scan(&desc); err != nil { + t.Fatalf("select: %v", err) + } + if desc != "new desc" { + t.Errorf("description = %q, want %q", desc, "new desc") + } +} + +func TestBatchUpsertDailyStats(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + stats := []DailyStats{ + {DID: "did:plc:alice", Repository: "app", Date: "2026-04-19", PullCount: 5, PushCount: 2}, + } + if err := BatchUpsertDailyStats(d, stats); err != nil { + t.Fatalf("upsert: %v", err) + } + stats[0].PullCount = 10 + if err := BatchUpsertDailyStats(d, stats); err != nil { + t.Fatalf("update: %v", err) + } + var pull int + if err := d.QueryRow(`SELECT pull_count FROM repository_stats_daily WHERE did=? AND repository=? AND date=?`, + "did:plc:alice", "app", "2026-04-19").Scan(&pull); err != nil { + t.Fatalf("select: %v", err) + } + if pull != 10 { + t.Errorf("pull = %d, want 10", pull) + } +} + +func TestBatchUpsertRepositoryAnnotations_DropsStaleKeys(t *testing.T) { + d := setupBatchTestDB(t) + createBatchTestUser(t, d, "did:plc:alice") + + rows := []AnnotationRow{ + {DID: "did:plc:alice", Repository: "app", Key: "a", Value: "1"}, + {DID: "did:plc:alice", Repository: "app", Key: "b", Value: "2"}, + } + if err := BatchUpsertRepositoryAnnotations(d, rows); err != nil { + t.Fatalf("initial: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM repository_annotations WHERE did=? AND repository=?`, + "did:plc:alice", "app"); got != 2 { + t.Errorf("initial count = %d, want 2", got) + } + + // Second call drops stale key "b". + rows = []AnnotationRow{ + {DID: "did:plc:alice", Repository: "app", Key: "a", Value: "1-updated"}, + } + if err := BatchUpsertRepositoryAnnotations(d, rows); err != nil { + t.Fatalf("update: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM repository_annotations WHERE did=? AND repository=?`, + "did:plc:alice", "app"); got != 1 { + t.Errorf("after update = %d, want 1", got) + } + var val string + if err := d.QueryRow(`SELECT value FROM repository_annotations WHERE key=? AND did=? AND repository=?`, + "a", "did:plc:alice", "app").Scan(&val); err != nil { + t.Fatalf("select: %v", err) + } + if val != "1-updated" { + t.Errorf("value = %q, want 1-updated", val) + } +} + +func TestBatchUpsertCaptainRecords(t *testing.T) { + d := setupBatchTestDB(t) + + now := time.Now() + records := []HoldCaptainRecord{ + {HoldDID: "did:web:hold1", OwnerDID: "did:plc:alice", Public: true, AllowAllCrew: false, UpdatedAt: now}, + } + if err := BatchUpsertCaptainRecords(d, records); err != nil { + t.Fatalf("upsert: %v", err) + } + if got := countRows(t, d, `SELECT COUNT(*) FROM hold_captain_records`); got != 1 { + t.Errorf("count = %d, want 1", got) + } +} + +func TestBatchUpsertCrewMembers(t *testing.T) { + d := setupBatchTestDB(t) + + members := []CrewMember{ + {HoldDID: "did:web:hold1", MemberDID: "did:plc:alice", Rkey: "rkey1", Role: "owner"}, + } + if err := BatchUpsertCrewMembers(d, members); err != nil { + t.Fatalf("upsert: %v", err) + } + // Update the rkey: triggers the ON CONFLICT path. + members[0].Rkey = "rkey2" + if err := BatchUpsertCrewMembers(d, members); err != nil { + t.Fatalf("update: %v", err) + } + var rkey string + if err := d.QueryRow(`SELECT rkey FROM hold_crew_members WHERE hold_did=? AND member_did=?`, + "did:web:hold1", "did:plc:alice").Scan(&rkey); err != nil { + t.Fatalf("select: %v", err) + } + if rkey != "rkey2" { + t.Errorf("rkey = %q, want rkey2", rkey) + } +} + +func TestBatchEmptySlices(t *testing.T) { + d := setupBatchTestDB(t) + // Every batch function must tolerate an empty input slice without erroring. + if _, err := BatchInsertManifests(d, nil); err != nil { + t.Errorf("manifests: %v", err) + } + if err := BatchInsertLayers(d, nil); err != nil { + t.Errorf("layers: %v", err) + } + if err := BatchInsertManifestReferences(d, nil); err != nil { + t.Errorf("refs: %v", err) + } + if err := BatchUpsertTags(d, nil); err != nil { + t.Errorf("tags: %v", err) + } + if err := BatchUpsertStars(d, nil); err != nil { + t.Errorf("stars: %v", err) + } + if err := BatchUpsertRepoPages(d, nil); err != nil { + t.Errorf("repo pages: %v", err) + } + if err := BatchUpsertDailyStats(d, nil); err != nil { + t.Errorf("daily: %v", err) + } + if err := BatchUpsertRepositoryStats(d, nil); err != nil { + t.Errorf("repo stats: %v", err) + } + if err := BatchUpsertCaptainRecords(d, nil); err != nil { + t.Errorf("captain: %v", err) + } + if err := BatchUpsertCrewMembers(d, nil); err != nil { + t.Errorf("crew: %v", err) + } + if err := BatchUpsertRepositoryAnnotations(d, nil); err != nil { + t.Errorf("annotations: %v", err) + } +} diff --git a/pkg/appview/db/conn.go b/pkg/appview/db/conn.go new file mode 100644 index 0000000..838e171 --- /dev/null +++ b/pkg/appview/db/conn.go @@ -0,0 +1,69 @@ +package db + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "strings" +) + +// poisonedTxSubstrings are error-message substrings emitted when go-libsql or the +// remote libsql server leaves a connection in a state that cannot safely be reused. +// Most come from Bunny Database killing a transaction that exceeded its server-side +// timeout; the follow-on COMMIT then sees the connection in a poisoned state. +var poisonedTxSubstrings = []string{ + "Transaction timed-out", + "no transaction is active", + "connection has reached an invalid state", + "invalid state, started with", +} + +// IsPoisonedTxErr reports whether err indicates the underlying connection is no +// longer usable for further statements. Callers should evict the connection from +// the pool when this returns true. +func IsPoisonedTxErr(err error) bool { + if err == nil { + return false + } + msg := err.Error() + for _, s := range poisonedTxSubstrings { + if strings.Contains(msg, s) { + return true + } + } + return false +} + +// ExecResilient borrows a dedicated connection from db, runs fn against it, and +// evicts the connection from the pool when fn returns a poisoned-transaction +// error. The connection is always released via Close. +// +// Poison eviction works by returning driver.ErrBadConn from within conn.Raw: +// database/sql treats that as a signal to discard the underlying driver conn +// rather than returning it to the idle pool. +// +// ExecResilient does NOT retry. Callers wrap the call in their own retry policy +// when that is desired (for example, a single retry on the live Jetstream path). +func ExecResilient(ctx context.Context, db *sql.DB, fn func(*sql.Conn) error) error { + conn, err := db.Conn(ctx) + if err != nil { + return err + } + defer conn.Close() + + execErr := fn(conn) + if IsPoisonedTxErr(execErr) { + // Discard the underlying driver conn so it never serves another caller. + // The Raw callback's return value is what triggers eviction; we ignore + // any error from Raw itself. + _ = conn.Raw(func(any) error { return driver.ErrBadConn }) + } + return execErr +} + +// ErrNoPoolConn is returned by ExecResilient when a connection cannot be +// obtained from the pool (e.g. context cancelled). It wraps the underlying +// pool error for callers that want to distinguish pool-exhaustion from +// statement-level errors. +var ErrNoPoolConn = errors.New("db: failed to acquire pool connection") diff --git a/pkg/appview/db/conn_test.go b/pkg/appview/db/conn_test.go new file mode 100644 index 0000000..26ef248 --- /dev/null +++ b/pkg/appview/db/conn_test.go @@ -0,0 +1,28 @@ +package db + +import ( + "errors" + "testing" +) + +func TestIsPoisonedTxErr(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"unrelated", errors.New("disk full"), false}, + {"bunny timeout", errors.New("Remote SQlite failure: `2:0:Transaction timed-out`"), true}, + {"no active tx", errors.New("Remote SQlite failure: `3:1:cannot commit - no transaction is active`"), true}, + {"init state", errors.New("error code = 2: Error executing statement: connection has reached an invalid state, started with Init"), true}, + {"just invalid state", errors.New("generic failure: invalid state, started with Query"), true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := IsPoisonedTxErr(c.err); got != c.want { + t.Errorf("IsPoisonedTxErr(%v) = %v, want %v", c.err, got, c.want) + } + }) + } +} diff --git a/pkg/appview/db/jetstream_cursor.go b/pkg/appview/db/jetstream_cursor.go new file mode 100644 index 0000000..b38a635 --- /dev/null +++ b/pkg/appview/db/jetstream_cursor.go @@ -0,0 +1,33 @@ +package db + +import ( + "database/sql" + "errors" +) + +// GetJetstreamCursor returns the last persisted Jetstream cursor (time_us). +// Returns 0 when no cursor has been saved yet (e.g. fresh database). +func GetJetstreamCursor(db DBTX) (int64, error) { + var cursor int64 + err := db.QueryRow(`SELECT cursor FROM jetstream_cursor WHERE id = 1`).Scan(&cursor) + if errors.Is(err, sql.ErrNoRows) { + return 0, nil + } + if err != nil { + return 0, err + } + return cursor, nil +} + +// SaveJetstreamCursor writes the given cursor to the singleton jetstream_cursor row. +// Idempotent — safe to call on every tick. +func SaveJetstreamCursor(db DBTX, cursor int64) error { + _, err := db.Exec(` + INSERT INTO jetstream_cursor (id, cursor, updated_at) + VALUES (1, ?, CURRENT_TIMESTAMP) + ON CONFLICT(id) DO UPDATE SET + cursor = excluded.cursor, + updated_at = excluded.updated_at + `, cursor) + return err +} diff --git a/pkg/appview/db/jetstream_cursor_test.go b/pkg/appview/db/jetstream_cursor_test.go new file mode 100644 index 0000000..ee5d48b --- /dev/null +++ b/pkg/appview/db/jetstream_cursor_test.go @@ -0,0 +1,50 @@ +package db + +import ( + "fmt" + "strings" + "testing" +) + +func TestJetstreamCursor_RoundTrip(t *testing.T) { + safeName := strings.ReplaceAll(t.Name(), "/", "_") + d, err := InitDB(fmt.Sprintf("file:%s?mode=memory&cache=shared", safeName), LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + d.SetMaxOpenConns(1) + defer d.Close() + + // Fresh DB: no persisted cursor. + got, err := GetJetstreamCursor(d) + if err != nil { + t.Fatalf("get empty: %v", err) + } + if got != 0 { + t.Errorf("initial cursor = %d, want 0", got) + } + + // Save → read. + if err := SaveJetstreamCursor(d, 1234567890); err != nil { + t.Fatalf("save: %v", err) + } + got, err = GetJetstreamCursor(d) + if err != nil { + t.Fatalf("get after save: %v", err) + } + if got != 1234567890 { + t.Errorf("cursor = %d, want 1234567890", got) + } + + // Overwrite with newer value. + if err := SaveJetstreamCursor(d, 9999999999); err != nil { + t.Fatalf("save 2: %v", err) + } + got, err = GetJetstreamCursor(d) + if err != nil { + t.Fatalf("get 2: %v", err) + } + if got != 9999999999 { + t.Errorf("cursor after overwrite = %d, want 9999999999", got) + } +} diff --git a/pkg/appview/db/migrations/0022_create_jetstream_cursor.yaml b/pkg/appview/db/migrations/0022_create_jetstream_cursor.yaml new file mode 100644 index 0000000..1adfcee --- /dev/null +++ b/pkg/appview/db/migrations/0022_create_jetstream_cursor.yaml @@ -0,0 +1,7 @@ +description: Persist Jetstream cursor so reconnects resume from last processed event +query: | + CREATE TABLE IF NOT EXISTS jetstream_cursor ( + id INTEGER PRIMARY KEY CHECK (id = 1), + cursor INTEGER NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ); diff --git a/pkg/appview/jetstream/backfill_batch.go b/pkg/appview/jetstream/backfill_batch.go new file mode 100644 index 0000000..5975db9 --- /dev/null +++ b/pkg/appview/jetstream/backfill_batch.go @@ -0,0 +1,518 @@ +package jetstream + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "strings" + "time" + + "atcr.io/pkg/appview/db" + "atcr.io/pkg/atproto" +) + +// batchManifests decodes all manifest records for a repo and writes them as +// a small set of multi-row INSERTs: one per table (manifests, layers, +// manifest_references, repository_annotations). This replaces the previous +// per-record chunked-transaction loop, which exceeded Bunny Database's +// remote transaction timeout once chunks grew large. +// +// Returns the number of manifest records that were successfully decoded and +// included in the batch. Decode/validation failures are logged and skipped. +func (b *BackfillWorker) batchManifests(ctx context.Context, did string, records []atproto.Record) (int, error) { + if len(records) == 0 { + return 0, nil + } + + type decoded struct { + manifestRecord atproto.ManifestRecord + manifest db.Manifest + } + + decodedRecords := make([]decoded, 0, len(records)) + for i := range records { + r := &records[i] + var mr atproto.ManifestRecord + if err := json.Unmarshal(r.Value, &mr); err != nil { + slog.Warn("Backfill skipping invalid manifest record", "uri", r.URI, "error", err) + continue + } + if mr.Digest == "" || mr.Repository == "" { + slog.Warn("Backfill skipping manifest with missing fields", "uri", r.URI) + continue + } + + // Resolve holdDID the same way the single-record path does. + holdDID := mr.HoldDID + if holdDID == "" && mr.HoldEndpoint != "" { + if resolved, err := atproto.ResolveHoldDID(ctx, mr.HoldEndpoint); err == nil { + holdDID = resolved + } + } + + isList := len(mr.Manifests) > 0 + artifactType := "container-image" + if !isList && mr.Config != nil { + artifactType = db.GetArtifactType(mr.Config.MediaType) + } + + m := db.Manifest{ + DID: did, + Repository: mr.Repository, + Digest: mr.Digest, + MediaType: mr.MediaType, + SchemaVersion: mr.SchemaVersion, + HoldEndpoint: holdDID, + ArtifactType: artifactType, + CreatedAt: mr.CreatedAt, + } + if !isList && mr.Config != nil { + m.ConfigDigest = mr.Config.Digest + m.ConfigSize = mr.Config.Size + } + if mr.Subject != nil { + m.SubjectDigest = mr.Subject.Digest + } + decodedRecords = append(decodedRecords, decoded{mr, m}) + } + + if len(decodedRecords) == 0 { + return 0, nil + } + + // Phase 1: upsert all manifests in one batch, fetch ids. + manifests := make([]db.Manifest, len(decodedRecords)) + for i, d := range decodedRecords { + manifests[i] = d.manifest + } + ids, err := db.BatchInsertManifests(b.db, manifests) + if err != nil { + return 0, fmt.Errorf("batch insert manifests: %w", err) + } + + // Phase 2: derive layers, references, and annotations using the returned ids. + var ( + layerRows []db.Layer + refRows []db.ManifestReference + ) + + // For annotations, we keep only the newest manifest per (did, repo) with a + // non-empty annotation set. Matches reconcileAnnotations semantics at + // backfill.go:573. + type newest struct { + createdAt time.Time + annotations map[string]string + } + newestByRepo := make(map[string]newest) + + for _, d := range decodedRecords { + mid, ok := ids[db.ManifestKey(did, d.manifest.Repository, d.manifest.Digest)] + if !ok { + // BatchInsertManifests did not return an id for this row — either the + // row was constraint-rejected or the SELECT missed it. Skip its + // dependent rows rather than inserting with id 0. + slog.Warn("Backfill manifest missing id after batch insert", + "did", did, "repository", d.manifest.Repository, "digest", d.manifest.Digest) + continue + } + + if len(d.manifestRecord.Manifests) > 0 { + for i, ref := range d.manifestRecord.Manifests { + var pa, po, pv, pov string + if ref.Platform != nil { + pa = ref.Platform.Architecture + po = ref.Platform.OS + pv = ref.Platform.Variant + pov = ref.Platform.OSVersion + } + isAttestation := false + if refType, ok := ref.Annotations["vnd.docker.reference.type"]; ok { + isAttestation = refType == "attestation-manifest" + } + refRows = append(refRows, db.ManifestReference{ + ManifestID: mid, + Digest: ref.Digest, + MediaType: ref.MediaType, + Size: ref.Size, + PlatformArchitecture: pa, + PlatformOS: po, + PlatformVariant: pv, + PlatformOSVersion: pov, + IsAttestation: isAttestation, + ReferenceIndex: i, + }) + } + } else { + for i, layer := range d.manifestRecord.Layers { + layerRows = append(layerRows, db.Layer{ + ManifestID: mid, + Digest: layer.Digest, + MediaType: layer.MediaType, + Size: layer.Size, + LayerIndex: i, + Annotations: layer.Annotations, + }) + } + } + + if hasNonEmpty(d.manifestRecord.Annotations) { + key := d.manifest.Repository + prev, ok := newestByRepo[key] + if !ok || d.manifestRecord.CreatedAt.After(prev.createdAt) { + newestByRepo[key] = newest{d.manifestRecord.CreatedAt, d.manifestRecord.Annotations} + } + } + } + + if err := db.BatchInsertLayers(b.db, layerRows); err != nil { + return 0, err + } + if err := db.BatchInsertManifestReferences(b.db, refRows); err != nil { + return 0, err + } + + // Flatten annotations into AnnotationRows. + var annotationRows []db.AnnotationRow + for repo, n := range newestByRepo { + for k, v := range n.annotations { + if v == "" { + continue + } + annotationRows = append(annotationRows, db.AnnotationRow{ + DID: did, + Repository: repo, + Key: k, + Value: v, + }) + } + } + if err := db.BatchUpsertRepositoryAnnotations(b.db, annotationRows); err != nil { + return 0, err + } + + slog.Info("Backfill batch manifests", + "did", did, + "manifests", len(manifests), + "layers", len(layerRows), + "references", len(refRows), + "annotations", len(annotationRows)) + + return len(decodedRecords), nil +} + +func hasNonEmpty(m map[string]string) bool { + for _, v := range m { + if v != "" { + return true + } + } + return false +} + +// batchTags decodes tag records and writes them in one multi-row upsert. +func (b *BackfillWorker) batchTags(did string, records []atproto.Record) (int, error) { + tags := make([]db.Tag, 0, len(records)) + for i := range records { + r := &records[i] + var tr atproto.TagRecord + if err := json.Unmarshal(r.Value, &tr); err != nil { + slog.Warn("Backfill skipping invalid tag record", "uri", r.URI, "error", err) + continue + } + digest, err := tr.GetManifestDigest() + if err != nil { + slog.Warn("Backfill skipping tag record without digest", "uri", r.URI, "error", err) + continue + } + if tr.Repository == "" || tr.Tag == "" { + continue + } + tags = append(tags, db.Tag{ + DID: did, + Repository: tr.Repository, + Tag: tr.Tag, + Digest: digest, + CreatedAt: tr.UpdatedAt, + }) + } + if err := db.BatchUpsertTags(b.db, tags); err != nil { + return 0, err + } + slog.Info("Backfill batch tags", "did", did, "rows", len(tags)) + return len(tags), nil +} + +// batchStars decodes star records and writes them in one multi-row upsert. +// Ensures star subject owners exist as users first (FK requirement). +func (b *BackfillWorker) batchStars(ctx context.Context, did string, records []atproto.Record) (int, error) { + stars := make([]db.StarInput, 0, len(records)) + ownerDIDs := make(map[string]struct{}) + + for i := range records { + r := &records[i] + var sr atproto.StarRecord + if err := json.Unmarshal(r.Value, &sr); err != nil { + slog.Warn("Backfill skipping invalid star record", "uri", r.URI, "error", err) + continue + } + owner, repo, err := sr.GetSubjectDIDAndRepository() + if err != nil { + slog.Warn("Backfill skipping star with bad subject", "uri", r.URI, "error", err) + continue + } + ownerDIDs[owner] = struct{}{} + stars = append(stars, db.StarInput{ + StarrerDID: did, + OwnerDID: owner, + Repository: repo, + CreatedAt: sr.CreatedAt, + }) + } + + // Ensure every star subject has a users row (FK to users.did on stars). + // These calls are idempotent and cached, so repeated owners cost nothing. + for owner := range ownerDIDs { + if err := b.processor.EnsureUserExists(ctx, owner); err != nil { + slog.Warn("Backfill failed to ensure star subject user", "owner_did", owner, "error", err) + } + } + + if err := db.BatchUpsertStars(b.db, stars); err != nil { + return 0, err + } + slog.Info("Backfill batch stars", "did", did, "rows", len(stars)) + return len(stars), nil +} + +// batchRepoPages decodes repo page records and writes them in one upsert. +func (b *BackfillWorker) batchRepoPages(did string, records []atproto.Record) (int, error) { + pages := make([]db.RepoPage, 0, len(records)) + for i := range records { + r := &records[i] + var pr atproto.RepoPageRecord + if err := json.Unmarshal(r.Value, &pr); err != nil { + slog.Warn("Backfill skipping invalid repo page", "uri", r.URI, "error", err) + continue + } + if pr.Repository == "" { + continue + } + avatarCID := "" + if pr.Avatar != nil && pr.Avatar.Ref.Link != "" { + avatarCID = pr.Avatar.Ref.Link + } + pages = append(pages, db.RepoPage{ + DID: did, + Repository: pr.Repository, + Description: pr.Description, + AvatarCID: avatarCID, + UserEdited: pr.UserEdited, + CreatedAt: pr.CreatedAt, + UpdatedAt: pr.UpdatedAt, + }) + } + if err := db.BatchUpsertRepoPages(b.db, pages); err != nil { + return 0, err + } + slog.Info("Backfill batch repo pages", "did", did, "rows", len(pages)) + return len(pages), nil +} + +// batchDailyStats decodes daily stats records and writes them in one upsert. +// Ensures every distinct owner exists as a user first (FK requirement). +func (b *BackfillWorker) batchDailyStats(ctx context.Context, holdDID string, records []atproto.Record) (int, error) { + stats := make([]db.DailyStats, 0, len(records)) + ownerDIDs := make(map[string]struct{}) + + for i := range records { + r := &records[i] + var dr atproto.DailyStatsRecord + if err := json.Unmarshal(r.Value, &dr); err != nil { + slog.Warn("Backfill skipping invalid daily stats", "uri", r.URI, "error", err) + continue + } + if dr.OwnerDID == "" || dr.Repository == "" || dr.Date == "" { + continue + } + ownerDIDs[dr.OwnerDID] = struct{}{} + stats = append(stats, db.DailyStats{ + DID: dr.OwnerDID, + Repository: dr.Repository, + Date: dr.Date, + PullCount: int(dr.PullCount), + PushCount: int(dr.PushCount), + }) + } + + for owner := range ownerDIDs { + if err := b.processor.EnsureUserExists(ctx, owner); err != nil { + slog.Warn("Backfill failed to ensure daily stats owner user", "owner_did", owner, "error", err) + } + } + + if err := db.BatchUpsertDailyStats(b.db, stats); err != nil { + return 0, err + } + slog.Info("Backfill batch daily stats", "hold_did", holdDID, "rows", len(stats)) + return len(stats), nil +} + +// batchStats updates the in-memory stats cache from a hold's stats records, +// then flushes the aggregated view of every touched (owner, repo) to the +// repository_stats table in a single multi-row upsert. Aggregation is across +// all holds known to the cache, preserving the single-record semantics. +func (b *BackfillWorker) batchStats(ctx context.Context, holdDID string, records []atproto.Record) (int, error) { + type key struct{ owner, repo string } + touched := make(map[key]struct{}) + ownerDIDs := make(map[string]struct{}) + + for i := range records { + r := &records[i] + var sr atproto.StatsRecord + if err := json.Unmarshal(r.Value, &sr); err != nil { + slog.Warn("Backfill skipping invalid stats record", "uri", r.URI, "error", err) + continue + } + if sr.OwnerDID == "" || sr.Repository == "" { + continue + } + + var lastPull, lastPush *time.Time + if sr.LastPull != "" { + if t, err := time.Parse(time.RFC3339, sr.LastPull); err == nil { + lastPull = &t + } + } + if sr.LastPush != "" { + if t, err := time.Parse(time.RFC3339, sr.LastPush); err == nil { + lastPush = &t + } + } + + b.processor.statsCache.Update(holdDID, sr.OwnerDID, sr.Repository, + sr.PullCount, sr.PushCount, lastPull, lastPush) + touched[key{sr.OwnerDID, sr.Repository}] = struct{}{} + ownerDIDs[sr.OwnerDID] = struct{}{} + } + + for owner := range ownerDIDs { + if err := b.processor.EnsureUserExists(ctx, owner); err != nil { + slog.Warn("Backfill failed to ensure stats owner user", "owner_did", owner, "error", err) + } + } + + // Build aggregated rows from the cache. + rows := make([]db.RepositoryStats, 0, len(touched)) + for k := range touched { + totalPull, totalPush, latestPull, latestPush := b.processor.statsCache.GetAggregated(k.owner, k.repo) + rows = append(rows, db.RepositoryStats{ + DID: k.owner, + Repository: k.repo, + PullCount: int(totalPull), + PushCount: int(totalPush), + LastPull: latestPull, + LastPush: latestPush, + }) + } + if err := db.BatchUpsertRepositoryStats(b.db, rows); err != nil { + return 0, err + } + slog.Info("Backfill batch stats", "hold_did", holdDID, "rows", len(rows)) + return len(rows), nil +} + +// batchCaptains decodes captain records and writes them in one upsert. +func (b *BackfillWorker) batchCaptains(holdDID string, records []atproto.Record) (int, error) { + captains := make([]db.HoldCaptainRecord, 0, len(records)) + now := time.Now() + for i := range records { + r := &records[i] + var cr atproto.CaptainRecord + if err := json.Unmarshal(r.Value, &cr); err != nil { + slog.Warn("Backfill skipping invalid captain record", "uri", r.URI, "error", err) + continue + } + if cr.Owner == "" || !strings.HasPrefix(cr.Owner, "did:") { + slog.Warn("Backfill skipping captain with invalid owner", "uri", r.URI) + continue + } + // Captain rkey is the hold DID (collections are stored on each hold's PDS, + // so record.URI already encodes the hold DID in the authority segment). + recordHoldDID := extractDIDFromURI(r.URI) + if recordHoldDID == "" { + recordHoldDID = holdDID + } + captains = append(captains, db.HoldCaptainRecord{ + HoldDID: recordHoldDID, + OwnerDID: cr.Owner, + Public: cr.Public, + AllowAllCrew: cr.AllowAllCrew, + DeployedAt: cr.DeployedAt, + Region: cr.Region, + Successor: cr.Successor, + UpdatedAt: now, + }) + } + if err := db.BatchUpsertCaptainRecords(b.db, captains); err != nil { + return 0, err + } + slog.Info("Backfill batch captains", "rows", len(captains)) + return len(captains), nil +} + +// batchCrew decodes crew records and writes them in one upsert. +func (b *BackfillWorker) batchCrew(holdDID string, records []atproto.Record) (int, error) { + members := make([]db.CrewMember, 0, len(records)) + for i := range records { + r := &records[i] + var cr atproto.CrewRecord + if err := json.Unmarshal(r.Value, &cr); err != nil { + slog.Warn("Backfill skipping invalid crew record", "uri", r.URI, "error", err) + continue + } + if cr.Member == "" || !strings.HasPrefix(cr.Member, "did:") { + slog.Warn("Backfill skipping crew with invalid member", "uri", r.URI) + continue + } + recordHoldDID := extractDIDFromURI(r.URI) + if recordHoldDID == "" { + recordHoldDID = holdDID + } + permsJSON := "" + if len(cr.Permissions) > 0 { + if b, err := json.Marshal(cr.Permissions); err == nil { + permsJSON = string(b) + } + } + rkey := extractRkeyFromURI(r.URI) + members = append(members, db.CrewMember{ + HoldDID: recordHoldDID, + MemberDID: cr.Member, + Rkey: rkey, + Role: cr.Role, + Permissions: permsJSON, + Tier: cr.Tier, + AddedAt: cr.AddedAt, + }) + } + if err := db.BatchUpsertCrewMembers(b.db, members); err != nil { + return 0, err + } + slog.Info("Backfill batch crew", "hold_did", holdDID, "rows", len(members)) + return len(members), nil +} + +// extractDIDFromURI pulls the DID authority segment out of an AT-URI. +// Format: at://did:…/collection/rkey → "did:…". +func extractDIDFromURI(uri string) string { + const prefix = "at://" + if !strings.HasPrefix(uri, prefix) { + return "" + } + rest := uri[len(prefix):] + if slash := strings.IndexByte(rest, '/'); slash >= 0 { + return rest[:slash] + } + return rest +} diff --git a/pkg/appview/src/css/main.css b/pkg/appview/src/css/main.css index 5bc107b..f7a1b43 100644 --- a/pkg/appview/src/css/main.css +++ b/pkg/appview/src/css/main.css @@ -653,7 +653,7 @@ } .sailor-typeahead-avatar { - @apply flex-shrink-0 w-9 h-9 rounded-full overflow-hidden; + @apply shrink-0 w-9 h-9 rounded-full overflow-hidden; @apply bg-base-300; } @@ -686,7 +686,7 @@ } .sailor-typeahead-selected .sailor-typeahead-clear { - @apply flex-shrink-0 w-8 h-8 rounded-full; + @apply shrink-0 w-8 h-8 rounded-full; @apply flex items-center justify-center; @apply text-xl leading-none text-base-content/60; @apply hover:bg-base-300 hover:text-base-content; diff --git a/pkg/appview/templates/components/repo-card.html b/pkg/appview/templates/components/repo-card.html index 014e8d2..b025523 100644 --- a/pkg/appview/templates/components/repo-card.html +++ b/pkg/appview/templates/components/repo-card.html @@ -43,7 +43,7 @@ {{ if .Description }} -

{{ .Description }}

+

{{ .Description }}

{{ end }}
{{ if eq .ArtifactType "helm-chart" }}