From fb5e09489e1e7dabba6e593c24bddd79aab025f8 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Fri, 10 Oct 2025 17:04:23 -0500 Subject: [PATCH] fix tag and manifest deletion --- deploy/.env.prod.template | 20 +++++++--------- pkg/appview/db/queries.go | 27 ++++++++++++++++++--- pkg/appview/jetstream/backfill.go | 7 ++++++ pkg/appview/jetstream/worker.go | 39 +++++++------------------------ pkg/atproto/manifest_store.go | 20 ++++++++++++++++ 5 files changed, 68 insertions(+), 45 deletions(-) diff --git a/deploy/.env.prod.template b/deploy/.env.prod.template index 584bc8a..c2f62e6 100644 --- a/deploy/.env.prod.template +++ b/deploy/.env.prod.template @@ -49,26 +49,24 @@ STORAGE_DRIVER=s3 AWS_ACCESS_KEY_ID= AWS_SECRET_ACCESS_KEY= -# S3 Region +# S3 Region (for distribution S3 driver) # UpCloud regions: us-chi1, us-nyc1, de-fra1, uk-lon1, sg-sin1, etc. # Default: us-chi1 -AWS_REGION=us-chi1 +S3_REGION=us-chi1 # S3 Bucket Name # Create this bucket in UpCloud Object Storage # Example: atcr-blobs S3_BUCKET=atcr -# S3 Endpoint (for custom domain or UpCloud endpoint) -# If using custom domain (blobs.atcr.io): -# S3_ENDPOINT=https://blobs.atcr.io -# If using UpCloud default endpoint: -# S3_ENDPOINT=https://s3.us-chi1.upcloudobjects.com +# S3 Endpoint +# Get this from UpCloud Console → Storage → Object Storage → Your bucket → "S3 endpoint" +# Format: https://[bucket-id].upcloudobjects.com +# Example: https://6vmss.upcloudobjects.com # -# IMPORTANT: If using custom domain, create CNAME: -# blobs.atcr.io → [bucket].us-chi1.upcloudobjects.com -# (with Cloudflare proxy DISABLED - gray cloud) -S3_ENDPOINT=https://blobs.atcr.io +# NOTE: Use the bucket-specific endpoint, NOT a custom domain +# Custom domains break presigned URL generation +S3_ENDPOINT=https://6vmss.upcloudobjects.com # S3 Region Endpoint (alternative to S3_ENDPOINT) # Use this if your S3 driver requires region-specific endpoint format diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 561ec8a..2d9c359 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -535,10 +535,16 @@ func DeleteTag(db *sql.DB, did, repository, tag string) error { } // 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 { - _, err := db.Exec(` - DELETE FROM manifests WHERE did = ? AND repository = ? AND digest = ? - `, did, repository, digest) + var err error + if repository == "" { + // Delete by DID + digest only (used when repository is unknown, e.g., Jetstream DELETE events) + _, err = db.Exec(`DELETE FROM manifests WHERE did = ? AND digest = ?`, did, digest) + } else { + // Delete specific manifest + _, err = db.Exec(`DELETE FROM manifests WHERE did = ? AND repository = ? AND digest = ?`, did, repository, digest) + } return err } @@ -1017,6 +1023,21 @@ func GetStarsForDID(db *sql.DB, starrerDID string) (map[string]time.Time, error) return stars, rows.Err() } +// 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 { + _, err := db.Exec(` + DELETE FROM tags + WHERE did = ? + AND NOT EXISTS ( + SELECT 1 FROM manifests + WHERE manifests.did = tags.did + AND manifests.digest = tags.digest + ) + `, did) + return err +} + // 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 { diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index 6fdaf5c..9821805 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -198,6 +198,13 @@ func (b *BackfillWorker) backfillRepo(ctx context.Context, did, collection strin fmt.Printf("WARNING: Failed to reconcile deletions for %s: %v\n", did, err) } + // After processing manifests, clean up orphaned tags (tags pointing to non-existent manifests) + if collection == atproto.ManifestCollection { + if err := db.CleanupOrphanedTags(b.db, did); err != nil { + fmt.Printf("WARNING: Failed to cleanup orphaned tags for %s: %v\n", did, err) + } + } + return recordCount, nil } diff --git a/pkg/appview/jetstream/worker.go b/pkg/appview/jetstream/worker.go index 155c0bd..a271316 100644 --- a/pkg/appview/jetstream/worker.go +++ b/pkg/appview/jetstream/worker.go @@ -6,7 +6,6 @@ import ( "encoding/json" "fmt" "net/url" - "strings" "sync" "time" @@ -418,10 +417,13 @@ func (w *Worker) processManifest(commit *CommitEvent) error { } if commit.Operation == "delete" { - // Delete manifest - repo := extractRepoFromRKey(commit.RKey) + // Delete manifest - rkey is just the digest, repository is not encoded digest := commit.RKey - return db.DeleteManifest(w.db, commit.DID, repo, digest) + if err := db.DeleteManifest(w.db, commit.DID, "", digest); err != nil { + return err + } + // Clean up any orphaned tags pointing to this manifest + return db.CleanupOrphanedTags(w.db, commit.DID) } // Parse manifest record @@ -497,13 +499,8 @@ func (w *Worker) processTag(commit *CommitEvent) error { } if commit.Operation == "delete" { - // Delete tag - parts := strings.Split(commit.RKey, "/") - if len(parts) < 2 { - return fmt.Errorf("invalid tag rkey: %s", commit.RKey) - } - repo := strings.Join(parts[:len(parts)-1], "/") - tag := parts[len(parts)-1] + // Delete tag - decode rkey back to repository and tag + repo, tag := atproto.RKeyToRepositoryTag(commit.RKey) return db.DeleteTag(w.db, commit.DID, repo, tag) } @@ -606,23 +603,3 @@ type AccountInfo struct { Status string `json:"status,omitempty"` } -// Helper functions - -func extractRepoFromRKey(rkey string) string { - // RKey format: or / - // For manifest, it's just the digest - parts := strings.Split(rkey, "/") - if len(parts) > 1 { - return parts[0] - } - return "" -} - -func calculateManifestSize(manifest *atproto.ManifestRecord) int64 { - var total int64 - total += manifest.Config.Size - for _, layer := range manifest.Layers { - total += layer.Size - } - return total -} diff --git a/pkg/atproto/manifest_store.go b/pkg/atproto/manifest_store.go index 1eed37d..895cb32 100644 --- a/pkg/atproto/manifest_store.go +++ b/pkg/atproto/manifest_store.go @@ -207,6 +207,26 @@ func repositoryTagToRKey(repository, tag string) string { return key } +// RKeyToRepositoryTag converts an ATProto record key back to repository and tag +// 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 + lastUnderscore := strings.LastIndex(rkey, "_") + if lastUnderscore == -1 { + // No underscore found - treat entire string as tag with empty repository + return "", rkey + } + + repository = rkey[:lastUnderscore] + tag = rkey[lastUnderscore+1:] + + // Convert dashes back to slashes in repository + repository = strings.ReplaceAll(repository, "-", "/") + + return repository, tag +} + // GetLastFetchedHoldEndpoint returns the hold endpoint from the most recently fetched manifest // This is used by the routing repository to cache the hold for blob requests func (s *ManifestStore) GetLastFetchedHoldEndpoint() string {