From 16f354b7b99f493d53dc55e7dc460f23615b9007 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Tue, 21 Oct 2025 20:23:14 -0500 Subject: [PATCH] fix some backfill and db queries --- .tangled/workflows/tests.yml | 8 ++ pkg/appview/db/models.go | 38 ++++--- pkg/appview/db/queries.go | 71 +++++++++++- pkg/appview/db/tag_delete_test.go | 171 +++++++++++++++++++++++++++++ pkg/appview/jetstream/backfill.go | 22 +++- pkg/appview/jetstream/worker.go | 10 +- pkg/atproto/manifest_store.go | 19 ++-- pkg/atproto/manifest_store_test.go | 14 +-- pkg/atproto/tag_store.go | 6 +- pkg/atproto/tag_store_test.go | 6 +- pkg/auth/oauth/client.go | 2 +- 11 files changed, 317 insertions(+), 50 deletions(-) create mode 100644 pkg/appview/db/tag_delete_test.go diff --git a/.tangled/workflows/tests.yml b/.tangled/workflows/tests.yml index 46ca750..d2acadd 100644 --- a/.tangled/workflows/tests.yml +++ b/.tangled/workflows/tests.yml @@ -10,8 +10,16 @@ dependencies: nixpkgs: - gcc - go + - curl steps: + - name: Download and Generate + environment: + CGO_ENABLED: 1 + command: | + go mod download + go generate ./... + - name: Run Tests environment: CGO_ENABLED: 1 diff --git a/pkg/appview/db/models.go b/pkg/appview/db/models.go index b81489c..f14a590 100644 --- a/pkg/appview/db/models.go +++ b/pkg/appview/db/models.go @@ -13,23 +13,27 @@ type User struct { // Manifest represents an OCI manifest stored in the cache type Manifest struct { - ID int64 - DID string - Repository string - Digest string - HoldEndpoint string - SchemaVersion int - MediaType string - ConfigDigest string - ConfigSize int64 - CreatedAt time.Time - Title string - Description string - SourceURL string - DocumentationURL string - Licenses string - IconURL string - ReadmeURL string + ID int64 + DID string + Repository string + Digest string + HoldEndpoint string + SchemaVersion int + MediaType string + ConfigDigest string + ConfigSize int64 + CreatedAt time.Time + Title string + Description string + SourceURL string + DocumentationURL string + Licenses string + IconURL string + ReadmeURL string + PlatformOS string // UNUSED: Reserved for future use, always NULL + PlatformArchitecture string // UNUSED: Reserved for future use, always NULL + PlatformVariant string // UNUSED: Reserved for future use, always NULL + PlatformOSVersion string // UNUSED: Reserved for future use, always NULL } // Layer represents a layer in a manifest diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 9cd7206..384cbf3 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -534,8 +534,9 @@ func DeleteTagsNotInList(db *sql.DB, did string, keepTags []struct{ Repository, // InsertManifest inserts or updates a manifest record // Uses UPSERT to update labels/annotations if manifest already exists +// Returns the manifest ID (works correctly for both insert and update) func InsertManifest(db *sql.DB, manifest *Manifest) (int64, error) { - result, err := db.Exec(` + _, err := db.Exec(` INSERT INTO manifests (did, repository, digest, hold_endpoint, schema_version, media_type, config_digest, config_size, created_at, @@ -564,7 +565,18 @@ func InsertManifest(db *sql.DB, manifest *Manifest) (int64, error) { return 0, err } - return result.LastInsertId() + // Query for the ID (works for both insert and update) + var id int64 + err = db.QueryRow(` + SELECT id FROM manifests + WHERE did = ? AND repository = ? AND digest = ? + `, manifest.DID, manifest.Repository, manifest.Digest).Scan(&id) + + if err != nil { + return 0, fmt.Errorf("failed to get manifest ID after upsert: %w", err) + } + + return id, nil } // InsertLayer inserts a new layer record @@ -597,8 +609,8 @@ func DeleteTag(db *sql.DB, did, repository, tag string) error { } // GetTagsWithPlatforms returns all tags for a repository with platform information -// For multi-arch tags, includes all platforms from manifest_references -// For single-arch tags, includes the platform info +// 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) func GetTagsWithPlatforms(db *sql.DB, did, repository string) ([]TagWithPlatforms, error) { rows, err := db.Query(` SELECT @@ -648,7 +660,7 @@ func GetTagsWithPlatforms(db *sql.DB, did, repository string) ([]TagWithPlatform tagOrder = append(tagOrder, tagKey) } - // Add platform info if present + // Add platform info if present (only for multi-arch manifest lists) if platformOS != "" || platformArch != "" { tagMap[tagKey].Platforms = append(tagMap[tagKey].Platforms, PlatformInfo{ OS: platformOS, @@ -908,6 +920,55 @@ func GetTopLevelManifests(db *sql.DB, did, repository string, limit, offset int) manifests = append(manifests, m) } + // Fetch platform details for multi-arch manifests AFTER closing the main query + for i := range manifests { + if manifests[i].IsManifestList { + platformRows, err := db.Query(` + SELECT + mr.platform_os, + mr.platform_architecture, + mr.platform_variant, + mr.platform_os_version + FROM manifest_references mr + WHERE mr.manifest_id = ? + ORDER BY mr.reference_index + `, manifests[i].ID) + + if err != nil { + return nil, err + } + + manifests[i].Platforms = []PlatformInfo{} + for platformRows.Next() { + var p PlatformInfo + var os, arch, variant, osVersion sql.NullString + + if err := platformRows.Scan(&os, &arch, &variant, &osVersion); err != nil { + platformRows.Close() + return nil, err + } + + if os.Valid { + p.OS = os.String + } + if arch.Valid { + p.Architecture = arch.String + } + if variant.Valid { + p.Variant = variant.String + } + if osVersion.Valid { + p.OSVersion = osVersion.String + } + + manifests[i].Platforms = append(manifests[i].Platforms, p) + } + platformRows.Close() + + manifests[i].PlatformCount = len(manifests[i].Platforms) + } + } + return manifests, nil } diff --git a/pkg/appview/db/tag_delete_test.go b/pkg/appview/db/tag_delete_test.go new file mode 100644 index 0000000..28892c0 --- /dev/null +++ b/pkg/appview/db/tag_delete_test.go @@ -0,0 +1,171 @@ +package db + +import ( + "testing" + "time" + + "atcr.io/pkg/atproto" +) + +// TestTagDeleteRoundTrip tests the full flow of creating and deleting tags +// This simulates what Jetstream does: encode repo/tag to rkey, then decode and delete +func TestTagDeleteRoundTrip(t *testing.T) { + // Create in-memory test database + db, err := InitDB(":memory:") + if err != nil { + t.Fatalf("Failed to init database: %v", err) + } + defer db.Close() + + // Insert test user + testUser := &User{ + DID: "did:plc:test123", + Handle: "testuser.bsky.social", + PDSEndpoint: "https://test.pds.example.com", + Avatar: "", + LastSeen: time.Now(), + } + if err := UpsertUser(db, testUser); err != nil { + t.Fatalf("Failed to insert user: %v", err) + } + + // Test cases covering different tag patterns + testCases := []struct { + name string + repository string + tag string + expectRoundTrip bool // Some cases can't round-trip due to encoding limitations + }{ + { + name: "simple tag", + repository: "test-image", + tag: "latest", + expectRoundTrip: true, + }, + { + name: "tag with hyphen (like latest-amd64)", + repository: "test-image", + tag: "latest-amd64", + expectRoundTrip: true, + }, + { + name: "tag with hyphen (like latest-arm64)", + repository: "test-image", + tag: "latest-arm64", + expectRoundTrip: true, + }, + { + name: "tag with version", + repository: "myapp", + tag: "v1.0.0", + expectRoundTrip: true, + }, + { + name: "repository with underscore", + repository: "my_repo", + tag: "latest", + expectRoundTrip: true, + }, + { + name: "both with underscores (known limitation)", + repository: "my_repo", + tag: "my_tag", + expectRoundTrip: false, // Cannot round-trip: underscore is the separator + }, + { + name: "repository with multiple hyphens", + repository: "multi-part-name", + tag: "test-build", + expectRoundTrip: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + // Step 1: Insert tag using UpsertTag (simulates tag creation) + tag := &Tag{ + DID: testUser.DID, + Repository: tc.repository, + Tag: tc.tag, + Digest: "sha256:abc123def456", + CreatedAt: time.Now(), + } + if err := UpsertTag(db, tag); err != nil { + t.Fatalf("Failed to upsert tag: %v", err) + } + + // Step 2: Verify tag was created + var count int + err := db.QueryRow(` + SELECT COUNT(*) FROM tags + WHERE did = ? AND repository = ? AND tag = ? + `, testUser.DID, tc.repository, tc.tag).Scan(&count) + if err != nil { + t.Fatalf("Failed to count tags: %v", err) + } + if count != 1 { + t.Fatalf("Expected 1 tag after insert, got %d", count) + } + + // Step 3: Simulate Jetstream delete flow + // This is what happens in processTag when operation == "delete" + // The rkey comes from ATProto, we need to parse it back to repo/tag + + // First, let's see what the rkey would be (this is how tags are stored in ATProto) + rkey := atproto.RepositoryTagToRKey(tc.repository, tc.tag) + t.Logf("RKey for %s:%s = %s", tc.repository, tc.tag, rkey) + + // Then parse it back (this is what Jetstream does) + parsedRepo, parsedTag := atproto.RKeyToRepositoryTag(rkey) + t.Logf("Parsed back: repository=%s, tag=%s", parsedRepo, parsedTag) + + // Verify round-trip (skip for known limitations) + if tc.expectRoundTrip { + if parsedRepo != tc.repository { + t.Errorf("Repository round-trip failed: stored=%s, parsed=%s", tc.repository, parsedRepo) + } + if parsedTag != tc.tag { + t.Errorf("Tag round-trip failed: stored=%s, parsed=%s", tc.tag, parsedTag) + } + + // Step 4: Delete using parsed values (like Jetstream does) + if err := DeleteTag(db, testUser.DID, parsedRepo, parsedTag); err != nil { + t.Fatalf("Failed to delete tag: %v", err) + } + + // Step 5: Verify tag was deleted + err = db.QueryRow(` + SELECT COUNT(*) FROM tags + WHERE did = ? AND repository = ? AND tag = ? + `, testUser.DID, tc.repository, tc.tag).Scan(&count) + if err != nil { + t.Fatalf("Failed to count tags after delete: %v", err) + } + if count != 0 { + // This is the bug! Tag wasn't deleted + t.Errorf("Expected 0 tags after delete, got %d (tag still exists!)", count) + + // Debug: show what's actually in the database + rows, err := db.Query(` + SELECT repository, tag FROM tags WHERE did = ? + `, testUser.DID) + if err != nil { + t.Logf("Failed to query remaining tags: %v", err) + } else { + t.Logf("Remaining tags in database:") + for rows.Next() { + var repo, tag string + rows.Scan(&repo, &tag) + t.Logf(" - repository=%s, tag=%s", repo, tag) + } + rows.Close() + } + } + } else { + // Known limitation: skip delete test for non-round-trippable cases + t.Logf("Skipping delete test - known limitation: %s != %s or %s != %s", + tc.repository, parsedRepo, tc.tag, parsedTag) + } + }) + } +} diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index b997ed5..0bb4321 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -336,14 +336,28 @@ func (b *BackfillWorker) processManifestRecord(did string, record *atproto.Recor manifest.ConfigSize = manifestRecord.Config.Size } - // Insert manifest + // Platform info is only stored for multi-arch images in manifest_references table + // Single-arch images don't need platform display (it's obvious) + + // Insert manifest (or get existing ID if already exists) manifestID, err := db.InsertManifest(b.db, manifest) if err != nil { - // Skip if already exists + // If manifest already exists, get its ID so we can still insert references/layers if strings.Contains(err.Error(), "UNIQUE constraint failed") { - return nil + // Query for existing manifest ID + var existingID int64 + err := b.db.QueryRow(` + SELECT id FROM manifests + WHERE did = ? AND repository = ? AND digest = ? + `, manifest.DID, manifest.Repository, manifest.Digest).Scan(&existingID) + + if err != nil { + return fmt.Errorf("failed to get existing manifest ID: %w", err) + } + manifestID = existingID + } else { + return fmt.Errorf("failed to insert manifest: %w", err) } - return fmt.Errorf("failed to insert manifest: %w", err) } if isManifestList { diff --git a/pkg/appview/jetstream/worker.go b/pkg/appview/jetstream/worker.go index 26c503c..6162e1b 100644 --- a/pkg/appview/jetstream/worker.go +++ b/pkg/appview/jetstream/worker.go @@ -545,7 +545,15 @@ func (w *Worker) processTag(commit *CommitEvent) error { if commit.Operation == "delete" { // Delete tag - decode rkey back to repository and tag repo, tag := atproto.RKeyToRepositoryTag(commit.RKey) - return db.DeleteTag(w.db, commit.DID, repo, tag) + fmt.Printf("Jetstream: Deleting tag: did=%s, repository=%s, tag=%s (from rkey=%s)\n", + commit.DID, repo, tag, commit.RKey) + if err := db.DeleteTag(w.db, commit.DID, repo, tag); err != nil { + fmt.Printf("Jetstream: ERROR deleting tag: %v\n", err) + return err + } + fmt.Printf("Jetstream: Successfully deleted tag: did=%s, repository=%s, tag=%s\n", + commit.DID, repo, tag) + return nil } // Parse tag record diff --git a/pkg/atproto/manifest_store.go b/pkg/atproto/manifest_store.go index e43fa47..9ce7f84 100644 --- a/pkg/atproto/manifest_store.go +++ b/pkg/atproto/manifest_store.go @@ -185,7 +185,7 @@ func (s *ManifestStore) Put(ctx context.Context, manifest distribution.Manifest, if tagOpt, ok := option.(distribution.WithTagOption); ok { tag := tagOpt.Tag tagRecord := NewTagRecord(s.client.DID(), s.repository, tag, dgst.String()) - tagRKey := repositoryTagToRKey(s.repository, tag) + tagRKey := RepositoryTagToRKey(s.repository, tag) _, err = s.client.PutRecord(ctx, TagCollection, tagRKey, tagRecord) if err != nil { return "", fmt.Errorf("failed to store tag in ATProto: %w", err) @@ -209,21 +209,22 @@ func digestToRKey(dgst digest.Digest) string { return dgst.Encoded() } -// repositoryTagToRKey converts a repository and tag to an ATProto record key +// RepositoryTagToRKey converts a repository and tag to an ATProto record key // ATProto record keys must match: ^[a-zA-Z0-9._~-]{1,512}$ -func repositoryTagToRKey(repository, tag string) string { +func RepositoryTagToRKey(repository, tag string) string { // Combine repository and tag to create a unique key - // Replace invalid characters: slashes become dashes + // Replace invalid characters: slashes become tildes (~) + // We use tilde instead of dash to avoid ambiguity with repository names that contain hyphens key := fmt.Sprintf("%s_%s", repository, tag) - // Replace / with - (slash not allowed in rkeys) - key = strings.ReplaceAll(key, "/", "-") + // Replace / with ~ (slash not allowed in rkeys, tilde is allowed and unlikely in repo names) + key = strings.ReplaceAll(key, "/", "~") return key } // RKeyToRepositoryTag converts an ATProto record key back to repository and tag -// This is the inverse of repositoryTagToRKey +// This is the inverse of RepositoryTagToRKey // Note: If the tag contains underscores, this will split on the LAST underscore func RKeyToRepositoryTag(rkey string) (repository, tag string) { // Find the last underscore to split repository and tag @@ -236,8 +237,8 @@ func RKeyToRepositoryTag(rkey string) (repository, tag string) { repository = rkey[:lastUnderscore] tag = rkey[lastUnderscore+1:] - // Convert dashes back to slashes in repository - repository = strings.ReplaceAll(repository, "-", "/") + // Convert tildes back to slashes in repository (tilde was used to encode slashes) + repository = strings.ReplaceAll(repository, "~", "/") return repository, tag } diff --git a/pkg/atproto/manifest_store_test.go b/pkg/atproto/manifest_store_test.go index 188cb78..0b6fea3 100644 --- a/pkg/atproto/manifest_store_test.go +++ b/pkg/atproto/manifest_store_test.go @@ -152,7 +152,7 @@ func TestRepositoryTagToRKey(t *testing.T) { name: "repo with namespace", repository: "org/myapp", tag: "v1.0.0", - want: "org-myapp_v1.0.0", + want: "org~myapp_v1.0.0", }, { name: "tag with underscore", @@ -164,15 +164,15 @@ func TestRepositoryTagToRKey(t *testing.T) { name: "deep namespace", repository: "a/b/c/myapp", tag: "prod", - want: "a-b-c-myapp_prod", + want: "a~b~c~myapp_prod", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got := repositoryTagToRKey(tt.repository, tt.tag) + got := RepositoryTagToRKey(tt.repository, tt.tag) if got != tt.want { - t.Errorf("repositoryTagToRKey() = %v, want %v", got, tt.want) + t.Errorf("RepositoryTagToRKey() = %v, want %v", got, tt.want) } }) } @@ -194,7 +194,7 @@ func TestRKeyToRepositoryTag(t *testing.T) { }, { name: "namespaced repo", - rkey: "org-myapp_v1.0.0", + rkey: "org~myapp_v1.0.0", wantRepository: "org/myapp", wantTag: "v1.0.0", }, @@ -206,7 +206,7 @@ func TestRKeyToRepositoryTag(t *testing.T) { }, { name: "deep namespace", - rkey: "a-b-c-myapp_prod", + rkey: "a~b~c~myapp_prod", wantRepository: "a/b/c/myapp", wantTag: "prod", }, @@ -247,7 +247,7 @@ func TestRepositoryTagRoundTrip(t *testing.T) { for _, tt := range tests { t.Run(tt.repository+":"+tt.tag, func(t *testing.T) { - rkey := repositoryTagToRKey(tt.repository, tt.tag) + rkey := RepositoryTagToRKey(tt.repository, tt.tag) gotRepo, gotTag := RKeyToRepositoryTag(rkey) if gotRepo != tt.repository { diff --git a/pkg/atproto/tag_store.go b/pkg/atproto/tag_store.go index 4d74709..771f78e 100644 --- a/pkg/atproto/tag_store.go +++ b/pkg/atproto/tag_store.go @@ -27,7 +27,7 @@ func NewTagStore(client *Client, repository string) *TagStore { // Get retrieves the descriptor for a tag func (s *TagStore) Get(ctx context.Context, tag string) (distribution.Descriptor, error) { // Build record key - rkey := repositoryTagToRKey(s.repository, tag) + rkey := RepositoryTagToRKey(s.repository, tag) // Fetch tag record from ATProto record, err := s.client.GetRecord(ctx, TagCollection, rkey) @@ -65,7 +65,7 @@ func (s *TagStore) Tag(ctx context.Context, tag string, desc distribution.Descri tagRecord := NewTagRecord(s.client.DID(), s.repository, tag, desc.Digest.String()) // Store in ATProto - rkey := repositoryTagToRKey(s.repository, tag) + rkey := RepositoryTagToRKey(s.repository, tag) _, err := s.client.PutRecord(ctx, TagCollection, rkey, tagRecord) if err != nil { return fmt.Errorf("failed to store tag in ATProto: %w", err) @@ -76,7 +76,7 @@ func (s *TagStore) Tag(ctx context.Context, tag string, desc distribution.Descri // Untag removes a tag func (s *TagStore) Untag(ctx context.Context, tag string) error { - rkey := repositoryTagToRKey(s.repository, tag) + rkey := RepositoryTagToRKey(s.repository, tag) return s.client.DeleteRecord(ctx, TagCollection, rkey) } diff --git a/pkg/atproto/tag_store_test.go b/pkg/atproto/tag_store_test.go index 24ff8f7..eb46593 100644 --- a/pkg/atproto/tag_store_test.go +++ b/pkg/atproto/tag_store_test.go @@ -67,7 +67,7 @@ func TestTagStore_Get(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // Verify query parameters query := r.URL.Query() - rkey := repositoryTagToRKey("myapp", tt.tag) + rkey := RepositoryTagToRKey("myapp", tt.tag) if query.Get("rkey") != rkey { t.Errorf("rkey = %v, want %v", query.Get("rkey"), rkey) } @@ -240,7 +240,7 @@ func TestTagStore_Tag(t *testing.T) { json.NewDecoder(r.Body).Decode(&body) // Verify rkey - expectedRKey := repositoryTagToRKey("myapp", tt.tag) + expectedRKey := RepositoryTagToRKey("myapp", tt.tag) if body["rkey"] != expectedRKey { t.Errorf("rkey = %v, want %v", body["rkey"], expectedRKey) } @@ -340,7 +340,7 @@ func TestTagStore_Untag(t *testing.T) { var body map[string]interface{} json.NewDecoder(r.Body).Decode(&body) - expectedRKey := repositoryTagToRKey("myapp", tt.tag) + expectedRKey := RepositoryTagToRKey("myapp", tt.tag) if body["rkey"] != expectedRKey { t.Errorf("rkey = %v, want %v", body["rkey"], expectedRKey) } diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index e7e5ad3..1289329 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -133,7 +133,7 @@ func GetDefaultScopes(did string, testMode bool) []string { // OCI artifact manifests (for cosign signatures, SBOMs, attestations) "blob:application/vnd.cncf.oras.artifact.manifest.v1+json", } - + scopes = append(scopes, fmt.Sprintf("rpc:com.atproto.repo.getRecord?aud=%s", "*")) // Add repo scopes