at-container-registry/pkg/hold/gc/gc.go

package gc

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"maps"
	"net"
	"net/http"
	"net/url"
	"regexp"
	"strings"
	"sync"
	"syscall"
	"time"

	"atcr.io/pkg/atproto"
	"atcr.io/pkg/hold/pds"
	"atcr.io/pkg/s3"
	"github.com/bluesky-social/indigo/atproto/syntax"
)

// maxPreviewItems caps per-category detail slices to prevent memory/HTML bloat
const maxPreviewItems = 10000

// OrphanedRecordDetail holds info about a single orphaned record. Collection
// names which record type it is — layer records carry a digest, media type,
// and size; scan and image-config records leave those empty.
type OrphanedRecordDetail struct {
	Collection string `json:"collection"`
	Rkey       string `json:"rkey"`
	// CID identifies the exact record revision that was judged orphaned. Scan
	// and image-config rkeys are derived from the manifest digest and are reused
	// on re-push, so the rkey alone does not identify what we looked at; the
	// delete path re-reads this to confirm the slot still holds the same
	// revision. Empty for layer records, whose rkeys are unique per write.
	CID         string `json:"cid,omitempty"`
	Digest      string `json:"digest"`
	ManifestURI string `json:"manifestUri"`
	UserDID     string `json:"userDid"`
	MediaType   string `json:"mediaType"`
	Size        int64  `json:"size"`
}

// auxOrphanState is the analysis state auxRecordOrphaned consults. Grouped into
// a struct rather than passed positionally because three of the four fields are
// same-shaped maps that would otherwise be trivial to transpose at a call site,
// and transposing knownDigests with fetchedUsers would silently widen what the
// sweep deletes.
type auxOrphanState struct {
	// knownManifests are the manifests live on their owners' PDSes, by AT-URI.
	knownManifests map[string]*manifestInfo
	// knownDigests are the manifest digests any successfully fetched user still
	// holds, across all users.
	knownDigests map[string]bool
	// digestOwners maps a manifest digest to every DID known to have pushed it
	// to this hold, derived from the hold's own layer records. It is "every
	// owner we can still see": a user whose layer records were collected in an
	// earlier run no longer appears, which is self-consistent, since those
	// records are only collected once that user's manifest is already gone.
	digestOwners map[string]map[string]bool
	// fetchedUsers are the DIDs whose PDS answered completely this run.
	fetchedUsers map[string]bool

	// now is the wall clock for this analysis pass, held once so every record
	// in the pass is judged against the same instant. The zero value means
	// time.Now(), which is what unit tests that don't care about the clock get.
	now time.Time
	// orphanSince carries forward, per auxRecordKey slot, the first time an
	// earlier pass in this process observed that slot orphaned. Reading it is
	// what makes grace measure orphanhood rather than record age.
	orphanSince map[string]time.Time
	// orphanObserved collects this pass's observations, keyed the same way, for
	// the caller to carry into the next pass. A nil map means "carry nothing
	// forward", which fails safe: no slot accrues age, so nothing is collected.
	orphanObserved map[string]time.Time
}

// auxRecordKey identifies one auxiliary record slot for the orphan clock.
//
// Deliberately collection+rkey with no CID: the clock times how long the SLOT
// has been orphaned, and a rewrite that leaves it orphaned — a rescan
// restamping io.atcr.hold.scan — must not restart it. A rewrite that makes the
// slot live again needs no special handling here; the orphan test stops
// observing it and commit drops the entry.
func auxRecordKey(collection, rkey string) string {
	return collection + "/" + rkey
}

// graceElapsed records that recordKey looks orphaned as of this pass, and
// reports whether it has looked that way for at least gcRecordGracePeriod.
//
// Every uncertain input starts the clock at now rather than expiring it: an
// unrecorded slot (first pass, or first pass after a restart), a zero time, and
// a time in the future all mean "we cannot show this has been orphaned for a
// full grace period", so the record survives at least one more pass.
//
// Call this LAST, after a record has already failed every other test. Observing
// a slot that is not actually orphaned accrues age against it, and the entry
// would then already be stale the moment its manifest genuinely disappeared.
func (s auxOrphanState) graceElapsed(recordKey string) bool {
	now := s.now
	if now.IsZero() {
		now = time.Now()
	}

	since, ok := s.orphanSince[recordKey]
	if !ok || since.IsZero() || since.After(now) {
		since = now
	}
	if s.orphanObserved != nil {
		s.orphanObserved[recordKey] = since
	}
	return now.Sub(since) >= gcRecordGracePeriod
}

// auxOrphanClock is the process-lifetime memory behind that calculation: which
// auxiliary record slots looked orphaned, and since when.
//
// It is in-memory only, so a restart forgets every observation and each
// surviving orphan starts its grace period over. That is the conservative
// direction — a restart delays collection and never advances it — and it is
// why commit replaces rather than merges.
type auxOrphanClock struct {
	mu    sync.Mutex
	since map[string]time.Time
}

// snapshot copies the recorded first-observation times for one analysis pass.
func (c *auxOrphanClock) snapshot() map[string]time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()

	out := make(map[string]time.Time, len(c.since))
	maps.Copy(out, c.since)
	return out
}

// commit replaces the recorded times with a completed pass's observations.
//
// Replacing rather than merging is the point: a slot the pass did not observe
// orphaned loses its accrued age. That covers the record coming back to life,
// the record being deleted, and the record's owner being unreachable — the last
// of which is why this is a replace. An owner we could not reach is an owner we
// cannot judge, and the sweep's standing rule is that an unreachable PDS never
// contributes to a deletion, not even by letting a clock run in the background.
//
// Only call this for a pass that completed. A pass that failed partway has
// observed only a prefix of the collection, and committing it would reset the
// clock on everything past that point.
func (c *auxOrphanClock) commit(observed map[string]time.Time) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.since = observed
}

// recordDigestOwner notes that the DID in manifestURI owns a manifest at that
// URI's digest. Called for every layer record the sweep walks, which is what
// makes the ownership set free to build: layer records are required for storage
// accounting and billing, so they exist for anything a user is charged for.
func recordDigestOwner(owners map[string]map[string]bool, manifestURI string) {
	parts := parseATURI(manifestURI)
	if parts == nil || parts.Collection != atproto.ManifestCollection {
		return
	}
	digest := "sha256:" + parts.Rkey
	if owners[digest] == nil {
		owners[digest] = make(map[string]bool)
	}
	owners[digest][parts.DID] = true
}

// orphanRef is the minimal address needed to delete an orphaned record. CID
// pins the revision that was judged (see OrphanedRecordDetail.CID); it is empty
// for layer records, which have unique per-write rkeys and need no such check.
type orphanRef struct {
	Collection string
	Rkey       string
	CID        string
	// ManifestURI is the manifest the record belongs to, used at delete time to
	// check the manifest has not reappeared. Empty for layer records.
	ManifestURI string
}

// manifestsWithLayerRecords returns the set of manifest AT-URIs the hold
// currently holds at least one layer record for.
//
// A push writes layer records to this hold, so their presence means the image
// is here now — regardless of what a previous scan concluded. That makes this a
// re-check of orphanhood rather than of record identity, and unlike consulting
// the owner's PDS it is entirely local.
//
// A manifest whose layer records are themselves orphaned but not yet collected
// also lands in this set, which only defers the aux record to the next run.
// Erring toward one more cycle of a leak is the right trade against deleting a
// live user's scan and image config.
func (gc *GarbageCollector) manifestsWithLayerRecords(ctx context.Context) (map[string]bool, error) {
	recordsIndex := gc.pds.RecordsIndex()
	if recordsIndex == nil {
		return nil, fmt.Errorf("records index not available")
	}

	live := make(map[string]bool)
	cursor := ""
	for {
		records, nextCursor, err := recordsIndex.ListRecords(atproto.LayerCollection, 1000, cursor, true)
		if err != nil {
			return nil, fmt.Errorf("list layer records: %w", err)
		}
		for _, rec := range records {
			layer, err := gc.decodeLayerRecord(ctx, rec)
			if err != nil {
				// Undecodable: we cannot tell which manifest it belongs to.
				// Skipping only costs protection, never causes a deletion.
				gc.logger.Warn("Failed to decode layer record while checking live manifests",
					"rkey", rec.Rkey, "error", err)
				continue
			}
			live[layer.Manifest] = true
		}
		if nextCursor == "" {
			break
		}
		cursor = nextCursor
	}
	return live, nil
}

// OrphanedBlobDetail holds info about a single orphaned blob in S3
type OrphanedBlobDetail struct {
	Digest string `json:"digest"`
	Size   int64  `json:"size"`
}

// MissingRecordDetail holds info about a layer record that should exist but doesn't
type MissingRecordDetail struct {
	Digest      string `json:"digest"`
	ManifestURI string `json:"manifestUri"`
	UserDID     string `json:"userDid"`
	MediaType   string `json:"mediaType"`
	Size        int64  `json:"size"`
}

// GCPreview contains detailed analysis results (no mutations performed)
type GCPreview struct {
	OrphanedRecords []OrphanedRecordDetail `json:"orphanedRecords"`
	OrphanedBlobs   []OrphanedBlobDetail   `json:"orphanedBlobs"`
	MissingRecords  []MissingRecordDetail  `json:"missingRecords"`

	TotalLayerRecords int           `json:"totalLayerRecords"`
	TotalBlobs        int           `json:"totalBlobs"`
	ReferencedBlobs   int           `json:"referencedBlobs"`
	UsersChecked      int           `json:"usersChecked"`
	ManifestsChecked  int           `json:"manifestsChecked"`
	Reconcile         bool          `json:"reconcile"`
	Duration          time.Duration `json:"duration"`
}

// BackfillConfigCandidate identifies one manifest that's missing its image
// config record on the hold.
type BackfillConfigCandidate struct {
	ManifestURI string `json:"manifestUri"`
	UserDID     string `json:"userDid"`
	Digest      string `json:"digest"`
}

// BackfillConfigsPreview is the dry-run output for the image-config backfill.
// No PDS writes or S3 fetches happen during preview — we only check which
// manifest digests already have an io.atcr.hold.image.config record.
type BackfillConfigsPreview struct {
	Missing          []BackfillConfigCandidate `json:"missing"`
	MissingCount     int                       `json:"missingCount"`
	PresentCount     int                       `json:"presentCount"`
	ManifestsChecked int                       `json:"manifestsChecked"`
	UsersAffected    int                       `json:"usersAffected"`
	Duration         time.Duration             `json:"duration"`
}

// TakedownGate is the interface the GC uses to consult the labeler cache when
// computing reachability. Defined here (rather than imported) to keep the GC
// package free of any direct dependency on the labeler package.
//
// IsTakenDown returns the takedown's creation timestamp and a boolean
// indicating whether the manifest URI is currently under takedown (either by
// per-manifest label or via a user-level label on its DID).
type TakedownGate interface {
	IsTakenDown(manifestURI string) (cts time.Time, ok bool)
}

// Option configures optional GC behavior.
type Option func(*GarbageCollector)

// WithTakedownCache wires a takedown gate (typically the hold's labeler cache)
// and a grace window. When set, analyzeRecords protects blobs of taken-down
// manifests from collection until grace expires, and skips reconciliation of
// their layer records so the labeler-driven purge isn't undone.
//
// Passing a nil gate or a non-positive window leaves the GC behaving as before.
func WithTakedownCache(gate TakedownGate, graceWindow time.Duration) Option {
	return func(gc *GarbageCollector) {
		gc.takedownGate = gate
		gc.takedownGrace = graceWindow
	}
}

// GarbageCollector handles cleanup of orphaned blobs from storage
type GarbageCollector struct {
	pds    *pds.HoldPDS
	s3     *s3.S3Service
	cfg    Config
	logger *slog.Logger

	// takedownGate, if non-nil, is consulted in analyzeRecords to gate blob
	// reachability and skip reconcile for taken-down manifests.
	takedownGate  TakedownGate
	takedownGrace time.Duration

	// stopCh signals the background goroutine to stop
	stopCh chan struct{}
	// wg tracks the background goroutine
	wg sync.WaitGroup

	// mu protects running state and last results
	mu      sync.Mutex
	running bool

	// Last results (for admin panel display)
	lastPreview           *GCPreview
	lastPreviewAt         time.Time
	lastResult            *GCResult
	lastResultAt          time.Time
	lastBackfillPreview   *BackfillConfigsPreview
	lastBackfillPreviewAt time.Time

	// Progress tracking for background operations
	phase         string // "manifests", "records", "blobs", "deleting", "complete", "error"
	progressMsg   string // e.g. "Fetching manifests (35/69 users)"
	operationType string // "preview", "run", "reconcile", "delete-records", "delete-blobs"
	lastError     error

	// predecessorCache caches holdDID → "is this a predecessor of our hold?"
	// A predecessor is a hold whose captain record has a successor label set.
	//
	// Only definitive answers belong here. The cache is never reset, so a false
	// recorded from an unreachable hold would outlive the outage and unreference
	// that hold's blobs on every subsequent run — deleting content this hold is
	// still serving on the predecessor's behalf.
	predecessorCache map[string]bool

	// auxOrphans times how long each scan / image-config record slot has looked
	// orphaned, across analysis passes. See auxOrphanClock: it is the grace
	// clock for those two collections, and it lives only as long as the process.
	auxOrphans auxOrphanClock

	// predecessorUnresolved holds the DIDs whose predecessor status could not be
	// determined during the current analysis. It exists only so that one
	// unreachable hold costs a single 5s timeout per run rather than one per
	// manifest, and it is cleared at the start of every analysis so a hold that
	// was down once is re-checked next time instead of being written off.
	predecessorUnresolved map[string]bool
}

// GCResult contains statistics from a GC run
type GCResult struct {
	BlobsDeleted      int64         `json:"blobs_deleted"`
	BytesReclaimed    int64         `json:"bytes_reclaimed"`
	RecordsDeleted    int64         `json:"records_deleted"`
	OrphanedRecords   int64         `json:"orphaned_records"`
	OrphanedBlobs     int64         `json:"orphaned_blobs"`
	ReferencedBlobs   int64         `json:"referenced_blobs"`
	RecordsReconciled int64         `json:"records_reconciled"`
	RecordsSkipped    int64         `json:"records_skipped"`
	ManifestsChecked  int64         `json:"manifests_checked"`
	UsersChecked      int64         `json:"users_checked"`
	Duration          time.Duration `json:"duration"`
}

// manifestInfo holds a parsed manifest fetched from a user's PDS
// manifestClaim is what a manifest's hold field tells us about ownership. The
// domain genuinely has three states, and collapsing it into a bool is what put
// phantom layer records on hold01: "we could not reach the hold" is not the same
// statement as either "ours" or "not ours", and the two questions this drives —
// keep the blobs, and adopt the manifest — want different answers for it.
type manifestClaim int

const (
	// claimNotOurs: the hold answered and this manifest is another hold's, or
	// there is no hold to ask. Ignore it entirely.
	claimNotOurs manifestClaim = iota
	// claimOurs: this hold's own manifest, or one belonging to a confirmed
	// predecessor. Adopt it and reference its blobs.
	claimOurs
	// claimUnknown: the hold did not answer, so ownership is unknowable. Don't
	// delete, but do not adopt. Its blobs stay referenced because an outage must
	// never make content deletable, while the manifest stays out of
	// knownManifests because adopting it reports every layer as a missing layer
	// record and lets reconcileMissingRecords write io.atcr.hold.layer records
	// claiming another hold's content.
	//
	// This is not always transient. A dev push leaves manifests naming
	// did:web:localhost:8080 or an RFC1918 address, which can never resolve from
	// a server, so this state can be permanent for a given hold.
	claimUnknown
)

func (c manifestClaim) String() string {
	switch c {
	case claimOurs:
		return "ours"
	case claimUnknown:
		return "unknown"
	default:
		return "not-ours"
	}
}

type manifestInfo struct {
	URI     string                  // AT-URI of the manifest
	UserDID string                  // DID of the user who owns it
	Record  *atproto.ManifestRecord // Parsed manifest data
	// ProtectOnly marks a claimUnknown manifest: reference its blobs, but do not
	// adopt it. See the ProtectOnly branch in analyzeRecords.
	ProtectOnly bool
}

// analysisResult holds intermediate data from record analysis, shared between Run and Preview
type analysisResult struct {
	referenced       map[string]bool
	orphanedRefs     []orphanRef            // collection+rkey for deletion in Run
	orphanedDetails  []OrphanedRecordDetail // details for display in Preview
	missingDetails   []MissingRecordDetail  // details for creation in Run / display in Preview
	usersChecked     int64
	manifestsChecked int64
	totalRecords     int
}

// NewGarbageCollector creates a new GC instance. Optional behavior (such as
// the labeler-aware takedown gate) is configured via Option arguments.
func NewGarbageCollector(holdPDS *pds.HoldPDS, s3svc *s3.S3Service, cfg Config, opts ...Option) *GarbageCollector {
	gc := &GarbageCollector{
		pds:                   holdPDS,
		s3:                    s3svc,
		cfg:                   cfg,
		logger:                slog.Default().With("component", "gc"),
		stopCh:                make(chan struct{}),
		predecessorCache:      make(map[string]bool),
		predecessorUnresolved: make(map[string]bool),
	}
	for _, opt := range opts {
		opt(gc)
	}
	return gc
}

// isManifestTakenDown reports whether the labeler cache (if any) currently
// holds a takedown for this manifest URI, returning the takedown's cts so the
// caller can decide whether the grace window has elapsed.
func (gc *GarbageCollector) isManifestTakenDown(manifestURI string) (time.Time, bool) {
	if gc.takedownGate == nil {
		return time.Time{}, false
	}
	return gc.takedownGate.IsTakenDown(manifestURI)
}

// takedownExpired reports whether a takedown's cts is older than the
// configured grace window. With a non-positive window every takedown is
// considered expired immediately (i.e. blobs are never protected).
func (gc *GarbageCollector) takedownExpired(cts time.Time) bool {
	if gc.takedownGrace <= 0 {
		return true
	}
	return time.Since(cts) > gc.takedownGrace
}

// tryStart attempts to mark GC as running. Returns false if already running.
func (gc *GarbageCollector) tryStart() bool {
	gc.mu.Lock()
	defer gc.mu.Unlock()
	if gc.running {
		return false
	}
	gc.running = true
	return true
}

// finish marks GC as no longer running
func (gc *GarbageCollector) finish() {
	gc.mu.Lock()
	gc.running = false
	gc.mu.Unlock()
}

// setProgress updates the progress fields (thread-safe).
func (gc *GarbageCollector) setProgress(phase, msg, opType string) {
	gc.mu.Lock()
	gc.phase = phase
	gc.progressMsg = msg
	gc.operationType = opType
	if phase != "error" {
		gc.lastError = nil
	}
	gc.mu.Unlock()
}

// GCProgress holds a snapshot of the current GC operation progress.
type GCProgress struct {
	Phase         string // "manifests", "records", "blobs", "deleting", "complete", "error"
	Message       string
	OperationType string // "preview", "run", "reconcile", "delete-records", "delete-blobs", "backfill-configs", "backfill-configs-preview"
	Running       bool
	Error         string
}

// GetProgress returns the current progress state.
func (gc *GarbageCollector) GetProgress() GCProgress {
	gc.mu.Lock()
	defer gc.mu.Unlock()
	p := GCProgress{
		Phase:         gc.phase,
		Message:       gc.progressMsg,
		OperationType: gc.operationType,
		Running:       gc.running,
	}
	if gc.lastError != nil {
		p.Error = gc.lastError.Error()
	}
	return p
}

// startBackground is the common pattern for launching a GC operation in the background.
// Returns false if already running.
func (gc *GarbageCollector) startBackground(opType, initialPhase, initialMsg string, fn func(ctx context.Context) error) bool {
	if !gc.tryStart() {
		return false
	}
	gc.setProgress(initialPhase, initialMsg, opType)
	go func() {
		defer gc.finish()
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
		defer cancel()

		if err := fn(ctx); err != nil {
			gc.mu.Lock()
			gc.phase = "error"
			gc.progressMsg = err.Error()
			gc.lastError = err
			gc.mu.Unlock()
			gc.logger.Error("GC operation failed", "type", opType, "error", err)
			return
		}
		gc.setProgress("complete", "", opType)
	}()
	return true
}

// StartPreview launches a GC preview in the background.
func (gc *GarbageCollector) StartPreview() bool {
	return gc.startBackground("preview", "manifests", "Starting preview...", func(ctx context.Context) error {
		_, err := gc.doPreview(ctx)
		return err
	})
}

// StartRun launches a full GC run in the background.
func (gc *GarbageCollector) StartRun() bool {
	return gc.startBackground("run", "manifests", "Starting GC run...", func(ctx context.Context) error {
		_, err := gc.doRun(ctx)
		return err
	})
}

// StartReconcile launches record reconciliation in the background.
func (gc *GarbageCollector) StartReconcile() bool {
	return gc.startBackground("reconcile", "deleting", "Reconciling missing records...", func(ctx context.Context) error {
		_, err := gc.doReconcile(ctx)
		return err
	})
}

// StartDeleteRecords launches orphaned record deletion in the background.
func (gc *GarbageCollector) StartDeleteRecords() bool {
	return gc.startBackground("delete-records", "deleting", "Deleting orphaned records...", func(ctx context.Context) error {
		_, err := gc.doDeleteOrphanedRecords(ctx)
		return err
	})
}

// StartDeleteBlobs launches orphaned blob deletion in the background.
func (gc *GarbageCollector) StartDeleteBlobs() bool {
	return gc.startBackground("delete-blobs", "manifests", "Analyzing records before blob deletion...", func(ctx context.Context) error {
		_, err := gc.doDeleteOrphanedBlobs(ctx)
		return err
	})
}

// Start begins the GC background process with a nightly ticker
func (gc *GarbageCollector) Start(ctx context.Context) {
	if !gc.cfg.Enabled {
		gc.logger.Info("GC disabled")
		return
	}

	gc.wg.Go(func() {

		ticker := time.NewTicker(gcInterval)
		defer ticker.Stop()

		for {
			select {
			case <-gc.stopCh:
				gc.logger.Info("GC background process stopped")
				return
			case <-ctx.Done():
				gc.logger.Info("GC context cancelled")
				return
			case <-ticker.C:
				gc.logger.Info("Running nightly GC")
				result, err := gc.Run(ctx)
				if err != nil {
					gc.logger.Warn("Nightly GC skipped or failed", "error", err)
				} else {
					gc.logResult(result)
				}
			}
		}
	})

	gc.logger.Info("GC background process started", "interval", gcInterval)
}

// Stop gracefully stops the GC background process
func (gc *GarbageCollector) Stop() {
	close(gc.stopCh)
	gc.wg.Wait()
}

// Run executes a single GC cycle
func (gc *GarbageCollector) Run(ctx context.Context) (*GCResult, error) {
	if !gc.tryStart() {
		return nil, fmt.Errorf("GC operation already in progress")
	}
	defer gc.finish()
	return gc.doRun(ctx)
}

// doRun is the internal implementation of Run (caller must hold the running lock).
func (gc *GarbageCollector) doRun(ctx context.Context) (*GCResult, error) {
	start := time.Now()
	result := &GCResult{}

	gc.logger.Info("Starting GC run")

	// Phase 1: Analyze records (build referenced set, find orphans, identify missing)
	analysis, err := gc.analyzeRecords(ctx)
	if err != nil {
		return nil, fmt.Errorf("phase 1 (analyze records) failed: %w", err)
	}

	result.OrphanedRecords = int64(len(analysis.orphanedRefs))
	result.UsersChecked = analysis.usersChecked
	result.ManifestsChecked = analysis.manifestsChecked

	gc.logger.Info("Phase 1 complete",
		"referenced", len(analysis.referenced),
		"orphanedRecords", len(analysis.orphanedRefs),
		"missingRecords", len(analysis.missingDetails))

	// Reconcile: create missing layer records
	gc.setProgress("deleting", "Reconciling missing records...", "run")
	if len(analysis.missingDetails) > 0 {
		gc.reconcileMissingRecords(ctx, analysis.missingDetails, result)
	}

	// Phase 2: Delete orphaned layer records
	gc.setProgress("deleting", "Deleting orphaned records...", "run")
	if err := gc.deleteOrphanedRecords(ctx, analysis.orphanedRefs, result); err != nil {
		gc.logger.Error("Phase 2 (delete orphaned records) failed", "error", err)
		// Continue to phase 3 - we can still clean up blobs
	}

	// Phase 3: Walk storage and delete unreferenced blobs
	gc.setProgress("blobs", "Deleting orphaned blobs...", "run")
	if err := gc.deleteOrphanedBlobs(ctx, analysis.referenced, result); err != nil {
		return nil, fmt.Errorf("phase 3 (delete orphaned blobs) failed: %w", err)
	}

	result.Duration = time.Since(start)
	result.ReferencedBlobs = int64(len(analysis.referenced))

	// Store last result for admin panel
	gc.mu.Lock()
	gc.lastResult = result
	gc.lastResultAt = time.Now()
	gc.mu.Unlock()

	return result, nil
}

// Preview runs read-only analysis and returns detailed results for the admin panel.
// It never creates, deletes, or modifies any data.
func (gc *GarbageCollector) Preview(ctx context.Context) (*GCPreview, error) {
	if !gc.tryStart() {
		return nil, fmt.Errorf("GC operation already in progress")
	}
	defer gc.finish()
	return gc.doPreview(ctx)
}

// doPreview is the internal implementation of Preview (caller must hold the running lock).
func (gc *GarbageCollector) doPreview(ctx context.Context) (*GCPreview, error) {
	start := time.Now()

	gc.logger.Info("Starting GC preview")

	// Phase 1: Analyze records
	analysis, err := gc.analyzeRecords(ctx)
	if err != nil {
		return nil, fmt.Errorf("analyze records: %w", err)
	}

	// Phase 2: Walk S3 to find orphaned blobs (read-only)
	gc.setProgress("blobs", "Walking S3 storage...", "preview")
	orphanedBlobs, totalBlobs, err := gc.scanOrphanedBlobDetails(ctx, analysis.referenced)
	if err != nil {
		return nil, fmt.Errorf("scan orphaned blobs: %w", err)
	}

	preview := &GCPreview{
		OrphanedRecords:   analysis.orphanedDetails,
		OrphanedBlobs:     orphanedBlobs,
		MissingRecords:    analysis.missingDetails,
		TotalLayerRecords: analysis.totalRecords,
		TotalBlobs:        totalBlobs,
		ReferencedBlobs:   len(analysis.referenced),
		UsersChecked:      int(analysis.usersChecked),
		ManifestsChecked:  int(analysis.manifestsChecked),
		Reconcile:         true,
		Duration:          time.Since(start),
	}

	gc.logger.Info("GC preview complete",
		"orphanedRecords", len(preview.OrphanedRecords),
		"orphanedBlobs", len(preview.OrphanedBlobs),
		"missingRecords", len(preview.MissingRecords),
		"referencedBlobs", preview.ReferencedBlobs,
		"duration", preview.Duration)

	// Store for admin panel
	gc.mu.Lock()
	gc.lastPreview = preview
	gc.lastPreviewAt = time.Now()
	gc.mu.Unlock()

	return preview, nil
}

// Reconcile creates missing layer records without deleting anything.
// Requires a prior Preview() to identify missing records.
func (gc *GarbageCollector) Reconcile(ctx context.Context) (*GCResult, error) {
	if !gc.tryStart() {
		return nil, fmt.Errorf("GC operation already in progress")
	}
	defer gc.finish()
	return gc.doReconcile(ctx)
}

// doReconcile is the internal implementation of Reconcile (caller must hold the running lock).
func (gc *GarbageCollector) doReconcile(ctx context.Context) (*GCResult, error) {
	gc.mu.Lock()
	preview := gc.lastPreview
	gc.mu.Unlock()

	if preview == nil {
		return nil, fmt.Errorf("no preview available — run Scan first")
	}
	if len(preview.MissingRecords) == 0 {
		return &GCResult{}, nil
	}

	start := time.Now()
	result := &GCResult{}

	gc.logger.Info("Starting reconciliation", "missingRecords", len(preview.MissingRecords))
	gc.reconcileMissingRecords(ctx, preview.MissingRecords, result)
	result.Duration = time.Since(start)

	gc.mu.Lock()
	gc.lastResult = result
	gc.lastResultAt = time.Now()
	gc.mu.Unlock()

	return result, nil
}

// DeleteOrphanedRecords deletes layer records whose manifests no longer exist.
// Requires a prior Preview() to identify orphaned records.
func (gc *GarbageCollector) DeleteOrphanedRecords(ctx context.Context) (*GCResult, error) {
	if !gc.tryStart() {
		return nil, fmt.Errorf("GC operation already in progress")
	}
	defer gc.finish()
	return gc.doDeleteOrphanedRecords(ctx)
}

// doDeleteOrphanedRecords is the internal implementation (caller must hold the running lock).
func (gc *GarbageCollector) doDeleteOrphanedRecords(ctx context.Context) (*GCResult, error) {
	gc.mu.Lock()
	preview := gc.lastPreview
	previewAt := gc.lastPreviewAt
	gc.mu.Unlock()

	if preview == nil {
		return nil, fmt.Errorf("no preview available — run Scan first")
	}
	if age := time.Since(previewAt); age > maxPreviewAgeForDelete {
		return nil, fmt.Errorf("preview is %s old (limit %s) — run Scan again before deleting",
			age.Round(time.Minute), maxPreviewAgeForDelete)
	}
	if len(preview.OrphanedRecords) == 0 {
		return &GCResult{}, nil
	}

	start := time.Now()
	result := &GCResult{
		OrphanedRecords: int64(len(preview.OrphanedRecords)),
	}

	refs := make([]orphanRef, len(preview.OrphanedRecords))
	for i, r := range preview.OrphanedRecords {
		refs[i] = orphanRef{Collection: r.Collection, Rkey: r.Rkey, CID: r.CID, ManifestURI: r.ManifestURI}
	}

	gc.logger.Info("Deleting orphaned records", "count", len(refs))
	if err := gc.deleteOrphanedRecords(ctx, refs, result); err != nil {
		return nil, fmt.Errorf("delete orphaned records: %w", err)
	}
	result.Duration = time.Since(start)

	gc.mu.Lock()
	gc.lastResult = result
	gc.lastResultAt = time.Now()
	gc.mu.Unlock()

	return result, nil
}

// DeleteOrphanedBlobs walks S3 and deletes blobs not referenced by any manifest.
// Runs a fresh analysis to build the current referenced set (reflects any reconciliation
// done since the last preview).
func (gc *GarbageCollector) DeleteOrphanedBlobs(ctx context.Context) (*GCResult, error) {
	if !gc.tryStart() {
		return nil, fmt.Errorf("GC operation already in progress")
	}
	defer gc.finish()
	return gc.doDeleteOrphanedBlobs(ctx)
}

// doDeleteOrphanedBlobs is the internal implementation (caller must hold the running lock).
func (gc *GarbageCollector) doDeleteOrphanedBlobs(ctx context.Context) (*GCResult, error) {
	start := time.Now()
	result := &GCResult{}

	gc.logger.Info("Starting orphaned blob deletion (fresh analysis)")

	// Fresh analysis so the referenced set includes any records reconciled since preview
	analysis, err := gc.analyzeRecords(ctx)
	if err != nil {
		return nil, fmt.Errorf("analyze records: %w", err)
	}

	result.ReferencedBlobs = int64(len(analysis.referenced))

	gc.setProgress("blobs", "Deleting orphaned blobs...", "delete-blobs")
	if err := gc.deleteOrphanedBlobs(ctx, analysis.referenced, result); err != nil {
		return nil, fmt.Errorf("delete orphaned blobs: %w", err)
	}
	result.Duration = time.Since(start)

	gc.mu.Lock()
	gc.lastResult = result
	gc.lastResultAt = time.Now()
	gc.mu.Unlock()

	return result, nil
}

// analyzeRecords performs Phase 1 analysis: builds referenced set, finds orphaned records,
// and identifies missing layer records. Pure analysis — no mutations.
// Discovers users, fetches manifests, scans records, identifies missing records.
func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult, error) {
	result := &analysisResult{
		referenced: make(map[string]bool),
	}

	// Start each analysis with a clean slate of unresolved holds, so a hold that
	// was unreachable last night gets another chance tonight. Definitive answers
	// in predecessorCache are kept — those do not go stale within a process.
	gc.predecessorUnresolved = make(map[string]bool)

	recordsIndex := gc.pds.RecordsIndex()
	if recordsIndex == nil {
		return nil, fmt.Errorf("records index not available")
	}

	// Step 1: Discover all user DIDs
	userDIDs, err := gc.discoverUserDIDs(ctx)
	if err != nil {
		return nil, fmt.Errorf("discover user DIDs: %w", err)
	}
	gc.logger.Info("Discovered user DIDs", "count", len(userDIDs))

	// Step 2: Fetch manifests from each user's PDS
	knownManifests := make(map[string]*manifestInfo)
	// knownDigests is the set of manifest digests any fetched user still holds,
	// across ALL users. Scan and image-config records are keyed by digest with no
	// DID, so they are shared between users pushing identical images and must
	// survive as long as any one of those users still has the manifest.
	knownDigests := make(map[string]bool)
	// digestOwners is built from the hold's own layer records in step 3 below,
	// giving every DID that pushed a given digest here — the co-owners a shared
	// scan/image-config record actually serves.
	digestOwners := make(map[string]map[string]bool)
	fetchedUsers := make(map[string]bool)

	totalUsers := len(userDIDs)
	for _, did := range userDIDs {
		result.usersChecked++
		gc.setProgress("manifests", fmt.Sprintf("Fetching manifests (%d/%d users)", result.usersChecked, totalUsers), gc.operationType)

		// Resolve PDS endpoint (needed for both manifests and optional profile/tag fetches)
		pdsEndpoint, err := atproto.ResolveDIDToPDS(ctx, did)
		if err != nil {
			gc.logger.Warn("Failed to resolve PDS for user, treating their records as referenced",
				"did", did, "error", err)
			continue
		}

		manifests, err := gc.fetchUserManifestsFromEndpoint(ctx, did, pdsEndpoint, gc.pds.DID())
		if err != nil {
			gc.logger.Warn("Failed to fetch manifests for user, treating their records as referenced",
				"did", did, "error", err)
			continue
		}

		// If user opted into auto-remove-untagged, filter out untagged manifests
		// so their layers are treated as orphaned and cleaned up
		if profile, _ := gc.fetchUserProfile(ctx, pdsEndpoint, did); profile != nil && profile.AutoRemoveUntagged {
			taggedDigests, tagErr := gc.fetchUserTags(ctx, pdsEndpoint, did)
			if tagErr != nil {
				gc.logger.Warn("Failed to fetch tags for auto-remove, keeping all manifests referenced",
					"did", did, "error", tagErr)
			} else {
				before := len(manifests)
				manifests = gc.filterUntaggedManifests(manifests, taggedDigests)
				if filtered := before - len(manifests); filtered > 0 {
					gc.logger.Info("Filtered untagged manifests for auto-remove user",
						"did", did, "filtered", filtered, "remaining", len(manifests))
				}
			}
		}

		fetchedUsers[did] = true
		for _, m := range manifests {
			result.manifestsChecked++

			// Labeler-aware reachability:
			//   - in-grace takedown: blobs stay referenced (so a reversal can
			//     restore content), but the manifest is NOT added to
			//     knownManifests so reconcileMissingRecords won't recreate the
			//     layer records the labeler subscriber just purged.
			//   - past-grace takedown: skip entirely — digests fall out of
			//     the referenced set and become eligible for blob GC.
			if cts, taken := gc.isManifestTakenDown(m.URI); taken {
				if !gc.takedownExpired(cts) {
					for _, layer := range m.Record.Layers {
						result.referenced[layer.Digest] = true
					}
					if m.Record.Config != nil && m.Record.Config.Digest != "" {
						result.referenced[m.Record.Config.Digest] = true
					}
					gc.logger.Debug("Manifest under in-grace takedown: blobs protected, reconcile skipped",
						"manifest", m.URI, "cts", cts)
				} else {
					gc.logger.Debug("Manifest takedown past grace: orphaning blobs",
						"manifest", m.URI, "cts", cts)
				}
				continue
			}

			// Ownership was indefinite: the manifest names a hold we could not
			// reach, so we cannot say it is ours. Protect its blobs exactly as an
			// in-grace takedown does, but do not adopt it. Adopting would report
			// every layer as a missing layer record and let reconcileMissingRecords
			// write io.atcr.hold.layer records claiming another hold's content.
			if m.ProtectOnly {
				for _, layer := range m.Record.Layers {
					result.referenced[layer.Digest] = true
				}
				if m.Record.Config != nil && m.Record.Config.Digest != "" {
					result.referenced[m.Record.Config.Digest] = true
				}
				gc.logger.Debug("Manifest hold unreachable: blobs protected, ownership not claimed",
					"manifest", m.URI, "holdDid", m.Record.HoldDID)
				continue
			}

			knownManifests[m.URI] = m
			if d := extractDigestFromManifestURI(m.URI); d != "" {
				knownDigests[d] = true
			}

			// Add all layer digests to referenced set
			for _, layer := range m.Record.Layers {
				result.referenced[layer.Digest] = true
			}
			// Add config digest to referenced set (config blobs are in S3 but
			// don't get layer records — just protect them from deletion)
			if m.Record.Config != nil && m.Record.Config.Digest != "" {
				result.referenced[m.Record.Config.Digest] = true
			}
		}
	}

	gc.logger.Info("Fetched manifests from user PDS instances",
		"knownManifests", len(knownManifests),
		"fetchedUsers", len(fetchedUsers))

	// Step 3: Scan existing layer records to find orphans and build coveredPairs
	gc.setProgress("records", "Scanning layer records...", gc.operationType)
	coveredPairs := make(map[string]bool)

	cursor := ""
	batchSize := 1000

	for {
		records, nextCursor, err := recordsIndex.ListRecords(atproto.LayerCollection, batchSize, cursor, true)
		if err != nil {
			return nil, fmt.Errorf("failed to list layer records: %w", err)
		}

		for _, rec := range records {
			result.totalRecords++

			layer, err := gc.decodeLayerRecord(ctx, rec)
			if err != nil {
				gc.logger.Warn("Failed to decode layer record", "rkey", rec.Rkey, "error", err)
				continue
			}

			// Track this (manifest, digest) pair as covered
			pairKey := layer.Manifest + "|" + layer.Digest
			coveredPairs[pairKey] = true

			// Note the manifest's owner. This MUST stay above every continue
			// below: an owner missing from the set is an owner the aux sweep
			// will not wait for, which is how a co-owner's live records get
			// deleted. The one owner we cannot record is a layer record that
			// failed to decode above — there is no manifest URI to read — which
			// is logged as a warning.
			recordDigestOwner(digestOwners, layer.Manifest)

			// Too young to judge: the manifest may still be in flight. Keep
			// the record and protect its blob until the record matures.
			recordTime := tidToTime(rec.Rkey)
			if time.Since(recordTime) < gcRecordGracePeriod {
				result.referenced[layer.Digest] = true
				continue
			}

			// Check if this layer's manifest is known
			if _, known := knownManifests[layer.Manifest]; known {
				result.referenced[layer.Digest] = true
			} else {
				// Manifest not in our fetched set — check if the user's PDS was unreachable
				parts := parseATURI(layer.Manifest)
				if parts != nil && !fetchedUsers[parts.DID] {
					// User's PDS was unreachable — safe default: assume referenced
					result.referenced[layer.Digest] = true
				} else {
					// User's PDS was reachable but manifest not found — orphaned
					result.orphanedRefs = append(result.orphanedRefs, orphanRef{
						Collection: atproto.LayerCollection,
						Rkey:       rec.Rkey,
					})
					if len(result.orphanedDetails) < maxPreviewItems {
						result.orphanedDetails = append(result.orphanedDetails, OrphanedRecordDetail{
							Collection:  atproto.LayerCollection,
							Rkey:        rec.Rkey,
							Digest:      layer.Digest,
							ManifestURI: layer.Manifest,
							UserDID:     layer.UserDID,
							MediaType:   layer.MediaType,
							Size:        layer.Size,
						})
					}
					gc.logger.Debug("Found orphaned layer record",
						"rkey", rec.Rkey,
						"digest", layer.Digest,
						"manifest", layer.Manifest)
				}
			}
		}

		if nextCursor == "" {
			break
		}
		cursor = nextCursor

		if result.totalRecords%10000 == 0 {
			gc.logger.Info("Phase 1 progress", "processed", result.totalRecords)
		}
	}

	gc.logger.Info("Scanned layer records", "total", result.totalRecords, "coveredPairs", len(coveredPairs))

	// Step 3b: Scan scan and image-config records. These are keyed by manifest
	// digest rather than TID, so the only question is whether their manifest
	// still exists. Without this sweep they survive forever when a user deletes
	// a manifest record directly on their PDS — the purgeManifest XRPC only
	// fires on appview-driven deletes.
	//
	// Their blob references are NOT part of the referenced set this sweep
	// builds. io.atcr.hold.scan carries sbomBlob and vulnReportBlob, but those
	// live in the hold's own PDS blob store at /repos/<safe-did>/blobs/<cid>,
	// while the blob sweep walks the registry prefix /docker/registry/v2/blobs.
	// The two spaces do not overlap, so deleting a scan record here neither
	// frees its SBOM nor risks orphaning a blob this sweep could delete. That
	// blob space is currently never collected at all; a sweep for it would have
	// to read these two fields before the record goes.
	gc.setProgress("records", "Scanning scan and image config records...", gc.operationType)
	auxState := auxOrphanState{
		knownManifests: knownManifests,
		knownDigests:   knownDigests,
		digestOwners:   digestOwners,
		fetchedUsers:   fetchedUsers,
		now:            time.Now(),
		orphanSince:    gc.auxOrphans.snapshot(),
		orphanObserved: make(map[string]time.Time),
	}
	for _, collection := range []string{atproto.ScanCollection, atproto.ImageConfigCollection} {
		if err := gc.scanAuxRecords(ctx, collection, auxState, result); err != nil {
			// Deliberately not committing: a pass that stopped partway saw only
			// part of the collection, and committing it would reset the grace
			// clock on every slot it never reached.
			return nil, fmt.Errorf("scan %s records: %w", collection, err)
		}
	}
	// Both collections walked end to end, so this pass's observations are a
	// complete picture and can replace the previous one.
	gc.auxOrphans.commit(auxState.orphanObserved)

	// Step 4: Identify missing layer records (uncovered manifest+layer pairs)
	for _, m := range knownManifests {
		for _, layer := range m.Record.Layers {
			pairKey := m.URI + "|" + layer.Digest
			if coveredPairs[pairKey] {
				continue
			}
			if len(result.missingDetails) < maxPreviewItems {
				result.missingDetails = append(result.missingDetails, MissingRecordDetail{
					Digest:      layer.Digest,
					ManifestURI: m.URI,
					UserDID:     m.UserDID,
					MediaType:   layer.MediaType,
					Size:        layer.Size,
				})
			}
		}
	}

	return result, nil
}

// scanAuxRecords walks one auxiliary manifest collection (scan or image
// config) and appends orphans to result. A record is orphaned when its
// manifest is absent from the owning user's PDS and that PDS was reachable —
// the same test the layer sweep applies, so an unreachable PDS never causes a
// deletion.
//
// Grace comes from state's orphan clock, not from the record. These
// collections use deterministic digest rkeys rather than TIDs, so there is no
// write time in the rkey to read, and the timestamps in the body cannot stand
// in for one: io.atcr.hold.scan's scannedAt is restamped by every rescan, so on
// a hold whose rescan_interval is shorter than gcRecordGracePeriod it never
// ages past grace and the record is immortal. The clock times orphanhood
// instead, which is what grace was always meant to measure.
func (gc *GarbageCollector) scanAuxRecords(
	ctx context.Context,
	collection string,
	state auxOrphanState,
	result *analysisResult,
) error {
	recordsIndex := gc.pds.RecordsIndex()
	if recordsIndex == nil {
		return fmt.Errorf("records index not available")
	}

	cursor := ""
	scanned, orphaned := 0, 0
	for {
		records, nextCursor, err := recordsIndex.ListRecords(collection, 1000, cursor, true)
		if err != nil {
			return fmt.Errorf("list records: %w", err)
		}

		for _, rec := range records {
			scanned++
			result.totalRecords++

			manifestURI, writtenAt, recCID, err := gc.decodeAuxRecord(ctx, collection, rec)
			if err != nil {
				gc.logger.Warn("Failed to decode record",
					"collection", collection, "rkey", rec.Rkey, "error", err)
				continue
			}

			parts, isOrphan := auxRecordOrphaned(auxRecordKey(collection, rec.Rkey), manifestURI, state)
			if !isOrphan {
				continue
			}

			orphaned++
			result.orphanedRefs = append(result.orphanedRefs, orphanRef{
				Collection:  collection,
				Rkey:        rec.Rkey,
				CID:         recCID,
				ManifestURI: manifestURI,
			})
			if len(result.orphanedDetails) < maxPreviewItems {
				result.orphanedDetails = append(result.orphanedDetails, OrphanedRecordDetail{
					Collection:  collection,
					Rkey:        rec.Rkey,
					CID:         recCID,
					ManifestURI: manifestURI,
					UserDID:     parts.DID,
				})
			}
			// writtenAt is the record body's own timestamp (scannedAt /
			// createdAt). It is diagnostic only — a scan record's is restamped
			// by every rescan, which is exactly why it no longer gates grace.
			gc.logger.Debug("Found orphaned record",
				"collection", collection, "rkey", rec.Rkey, "manifest", manifestURI,
				"recordWrittenAt", writtenAt,
				"orphanedSince", state.orphanSince[auxRecordKey(collection, rec.Rkey)])
		}

		if nextCursor == "" {
			break
		}
		cursor = nextCursor
	}

	gc.logger.Info("Scanned auxiliary manifest records",
		"collection", collection, "scanned", scanned, "orphaned", orphaned)
	return nil
}

// auxRecordOrphaned decides whether a scan or image-config record should be
// collected, given the slot it occupies and the manifest it names. It returns
// the parsed AT-URI alongside the verdict so the caller can record the owning
// DID.
//
// Every uncertain case resolves to "keep": a record whose manifest URI is
// unparseable, one whose owning PDS was unreachable this run, one a co-owner
// might still need, and one that has not yet looked orphaned for a full grace
// period. Only a record whose every reachable owner demonstrably lacks the
// manifest, sustained across grace, is orphaned.
//
// The grace test comes last, and deliberately so. It both reads and advances
// the orphan clock, so it must only ever see records that have already failed
// every other test — a record judged live must not accrue orphan age.
func auxRecordOrphaned(recordKey, manifestURI string, state auxOrphanState) (*atURIParts, bool) {
	if _, known := state.knownManifests[manifestURI]; known {
		return nil, false
	}
	parts := parseATURI(manifestURI)
	if parts == nil || !state.fetchedUsers[parts.DID] {
		return nil, false
	}

	// Scan and image-config records are keyed by manifest digest alone
	// (atproto.ScanRecordKey), with no DID, so every user who pushes the
	// identical image shares ONE record — and its body names only whoever wrote
	// last. The checks above only establish that THAT user no longer has the
	// manifest, which says nothing about the record's other owners. Deleting on
	// that basis strips the vulnerability scan and the layer
	// history/env/entrypoint from every co-owner's still-live image.
	digest := extractDigestFromManifestURI(manifestURI)

	// An owner whose PDS we could not reach this run cannot be judged, so the
	// record has to stay. This is the same principle the rest of the sweep
	// follows — an unreachable PDS never causes a deletion — extended from the
	// record's named user to everyone the record actually serves.
	for did := range state.digestOwners[digest] {
		if !state.fetchedUsers[did] {
			return nil, false
		}
	}

	// Every owner was reachable; keep the record if any of them still holds a
	// manifest at this digest.
	if state.knownDigests[digest] {
		return nil, false
	}

	// The record looks orphaned right now. Grace runs from the first pass that
	// saw it this way — not from any timestamp inside the record, which a
	// rescan would reset — so a slot that has only just started looking
	// orphaned survives to be re-judged next pass.
	if !state.graceElapsed(recordKey) {
		return nil, false
	}
	return parts, true
}

// decodeAuxRecord reads a scan or image-config record from the carstore and
// returns the manifest AT-URI it belongs to along with its creation time.
// It also returns the record's CID, which the delete path uses to confirm the
// rkey still holds this exact revision rather than one written since.
func (gc *GarbageCollector) decodeAuxRecord(ctx context.Context, collection string, rec pds.Record) (string, time.Time, string, error) {
	recordPath := rec.Collection + "/" + rec.Rkey
	recCID, recBytes, err := gc.pds.GetRecordBytes(ctx, recordPath)
	if err != nil {
		return "", time.Time{}, "", fmt.Errorf("get record bytes: %w", err)
	}
	uri, ts, err := decodeAuxRecordBytes(collection, *recBytes)
	return uri, ts, recCID.String(), err
}

// auxRecordUnchanged reports whether collection/rkey still holds the exact
// revision that was judged orphaned.
//
// This catches a re-push rewriting the slot: image-config records are upserted
// unconditionally on every push (pkg/hold/oci/xrpc.go), so a re-push always
// changes their CID. A missing record, an unreadable one, or a ref carrying no
// recorded CID all report false, so the delete is skipped rather than performed
// on something we did not examine.
//
// It is NOT sufficient on its own. A re-push does not necessarily rewrite the
// SCAN record — scan-on-push is tier-gated, and the discovery loop skips any
// manifest that already has a scan record (pkg/hold/pds/scan_broadcaster.go) —
// so a scan record's CID survives a re-push unchanged. Record identity is not
// orphanhood; see manifestsWithLayerRecords for the check that covers it.
func (gc *GarbageCollector) auxRecordUnchanged(ctx context.Context, collection, rkey, wantCID string) bool {
	if wantCID == "" {
		return false
	}
	gotCID, _, err := gc.pds.GetRecordBytes(ctx, collection+"/"+rkey)
	if err != nil {
		return false
	}
	return gotCID.String() == wantCID
}

// decodeAuxRecordBytes decodes raw CBOR for one of the auxiliary manifest
// collections. An unparseable timestamp yields the zero time rather than an
// error, which auxRecordOrphaned treats as in-grace, so a record with a
// malformed date is kept instead of collected.
func decodeAuxRecordBytes(collection string, data []byte) (string, time.Time, error) {
	var manifestURI, timestamp string
	switch collection {
	case atproto.ScanCollection:
		var scan atproto.ScanRecord
		if err := scan.UnmarshalCBOR(bytes.NewReader(data)); err != nil {
			return "", time.Time{}, fmt.Errorf("unmarshal CBOR: %w", err)
		}
		manifestURI, timestamp = scan.Manifest, scan.ScannedAt
	case atproto.ImageConfigCollection:
		var cfg atproto.ImageConfigRecord
		if err := cfg.UnmarshalCBOR(bytes.NewReader(data)); err != nil {
			return "", time.Time{}, fmt.Errorf("unmarshal CBOR: %w", err)
		}
		manifestURI, timestamp = cfg.Manifest, cfg.CreatedAt
	default:
		return "", time.Time{}, fmt.Errorf("unsupported collection %q", collection)
	}

	parsed, err := time.Parse(time.RFC3339, timestamp)
	if err != nil {
		return manifestURI, time.Time{}, nil
	}
	return manifestURI, parsed, nil
}

// blobPastGrace reports whether a blob is old enough to delete, based on its
// own S3 modification time rather than on any record that references it.
//
// Records now age out faster than blobs, so a blob can outlive every record
// naming it. Its age is the only thing left to judge it by. A listing that
// reports no modification time yields the zero time, which is treated as
// "unknown age" and keeps the blob.
func blobPastGrace(lastModified time.Time) bool {
	if lastModified.IsZero() {
		return false
	}
	return time.Since(lastModified) >= gcBlobGracePeriod
}

// scanOrphanedBlobDetails walks S3 and returns details of unreferenced blobs.
// Read-only — no deletions. Returns orphaned blob details and total blob count.
func (gc *GarbageCollector) scanOrphanedBlobDetails(ctx context.Context, referenced map[string]bool) ([]OrphanedBlobDetail, int, error) {
	var orphaned []OrphanedBlobDetail
	totalBlobs := 0
	blobsPath := "/docker/registry/v2/blobs"

	err := gc.s3.WalkBlobs(ctx, blobsPath, func(key string, size int64, lastModified time.Time) error {
		if !strings.HasSuffix(key, "/data") {
			return nil
		}

		digest := extractDigestFromPath(key)
		if digest == "" {
			return nil
		}

		totalBlobs++

		if !referenced[digest] && blobPastGrace(lastModified) {
			if len(orphaned) < maxPreviewItems {
				orphaned = append(orphaned, OrphanedBlobDetail{
					Digest: digest,
					Size:   size,
				})
			}
		}

		return nil
	})

	if err != nil {
		return nil, 0, fmt.Errorf("walk storage failed: %w", err)
	}

	return orphaned, totalBlobs, nil
}

// reconcileBatchSize is the number of layer records per repo commit.
// Batching reduces firehose events from N to N/batchSize.
const reconcileBatchSize = 200

// reconcileMissingRecords creates layer records for manifest+layer pairs that are missing.
// Records are batched into single commits to avoid flooding relays with firehose events.
func (gc *GarbageCollector) reconcileMissingRecords(ctx context.Context, missing []MissingRecordDetail, result *GCResult) {
	for i := 0; i < len(missing); i += reconcileBatchSize {
		end := min(i+reconcileBatchSize, len(missing))
		chunk := missing[i:end]

		records := make([]*atproto.LayerRecord, 0, len(chunk))
		for _, m := range chunk {
			records = append(records, atproto.NewLayerRecord(
				m.Digest,
				m.Size,
				m.MediaType,
				m.UserDID,
				m.ManifestURI,
			))
		}

		created, err := gc.pds.BatchCreateLayerRecords(ctx, records)
		if err != nil {
			gc.logger.Error("Failed to create reconciled layer batch",
				"batchStart", i,
				"batchSize", len(chunk),
				"error", err)
			continue
		}

		result.RecordsReconciled += int64(created)
		gc.logger.Info("Reconciliation progress",
			"created", result.RecordsReconciled,
			"total", len(missing),
			"batch", len(chunk))

		// Small delay between batches as a courtesy to relays
		time.Sleep(100 * time.Millisecond)
	}

	if result.RecordsReconciled > 0 {
		gc.logger.Info("Reconciliation complete", "recordsCreated", result.RecordsReconciled)
	}
}

// StartBackfillConfigsPreview launches a dry-run scan that classifies every
// manifest URI referenced from layer records as either already having an
// image config record or missing one. No PDS or S3 writes happen.
func (gc *GarbageCollector) StartBackfillConfigsPreview() bool {
	return gc.startBackground("backfill-configs-preview", "records", "Scanning for manifests missing image config records...", func(ctx context.Context) error {
		_, err := gc.doBackfillConfigsPreview(ctx)
		return err
	})
}

// StartBackfillConfigs launches image config backfill in the background.
// Creates io.atcr.hold.image.config records for manifests that don't have one yet
// by fetching OCI config blobs from S3.
func (gc *GarbageCollector) StartBackfillConfigs() bool {
	return gc.startBackground("backfill-configs", "records", "Scanning for manifests missing image config records...", func(ctx context.Context) error {
		_, err := gc.doBackfillConfigs(ctx)
		return err
	})
}

// scanBackfillCandidates walks every layer record, dedupes the manifest URIs
// they reference, and bucket each one as already-present or missing an image
// config record. Returns missing candidates and the count of present.
//
// opType is the GC operationType used for progress messages so this helper
// can serve both the preview and the run.
func (gc *GarbageCollector) scanBackfillCandidates(ctx context.Context, opType string) (missing []BackfillConfigCandidate, presentCount int, err error) {
	recordsIndex := gc.pds.RecordsIndex()
	if recordsIndex == nil {
		return nil, 0, fmt.Errorf("records index not available")
	}

	manifestURIs := make(map[string]bool)
	cursor := ""
	totalScanned := 0
	for {
		records, nextCursor, listErr := recordsIndex.ListRecords(atproto.LayerCollection, 1000, cursor, true)
		if listErr != nil {
			return nil, 0, fmt.Errorf("list layer records: %w", listErr)
		}
		for _, rec := range records {
			totalScanned++
			layer, decodeErr := gc.decodeLayerRecord(ctx, rec)
			if decodeErr != nil {
				continue
			}
			manifestURIs[layer.Manifest] = true
		}
		if nextCursor == "" {
			break
		}
		cursor = nextCursor
	}

	gc.logger.Info("Found unique manifests from layer records",
		"manifests", len(manifestURIs),
		"layersScanned", totalScanned)

	processed := 0
	for manifestURI := range manifestURIs {
		processed++
		gc.setProgress("records",
			fmt.Sprintf("Checking image configs (%d/%d manifests)...", processed, len(manifestURIs)),
			opType)

		aturi, parseErr := syntax.ParseATURI(manifestURI)
		if parseErr != nil {
			gc.logger.Warn("Invalid manifest URI", "uri", manifestURI, "error", parseErr)
			continue
		}
		manifestDigest := "sha256:" + aturi.RecordKey().String()

		if _, _, getErr := gc.pds.GetImageConfigRecord(ctx, manifestDigest); getErr == nil {
			presentCount++
			continue
		}
		missing = append(missing, BackfillConfigCandidate{
			ManifestURI: manifestURI,
			UserDID:     aturi.Authority().String(),
			Digest:      manifestDigest,
		})
	}
	return missing, presentCount, nil
}

// doBackfillConfigsPreview runs scanBackfillCandidates and stores the result
// for the admin UI to display. The full missing slice is kept in memory but
// rendering is capped via maxPreviewItems in the template layer.
func (gc *GarbageCollector) doBackfillConfigsPreview(ctx context.Context) (*BackfillConfigsPreview, error) {
	start := time.Now()

	missing, presentCount, err := gc.scanBackfillCandidates(ctx, "backfill-configs-preview")
	if err != nil {
		return nil, err
	}

	users := make(map[string]struct{}, len(missing))
	for _, c := range missing {
		users[c.UserDID] = struct{}{}
	}

	missingCount := len(missing)
	display := missing
	if len(display) > maxPreviewItems {
		display = display[:maxPreviewItems]
	}

	preview := &BackfillConfigsPreview{
		Missing:          display,
		MissingCount:     missingCount,
		PresentCount:     presentCount,
		ManifestsChecked: missingCount + presentCount,
		UsersAffected:    len(users),
		Duration:         time.Since(start),
	}

	gc.mu.Lock()
	gc.lastBackfillPreview = preview
	gc.lastBackfillPreviewAt = time.Now()
	gc.mu.Unlock()

	gc.logger.Info("Image config backfill preview complete",
		"missing", missingCount,
		"present", presentCount,
		"usersAffected", preview.UsersAffected,
		"duration", preview.Duration)
	return preview, nil
}

// doBackfillConfigs creates image config records for manifests that are missing them.
func (gc *GarbageCollector) doBackfillConfigs(ctx context.Context) (*GCResult, error) {
	start := time.Now()

	missing, presentCount, err := gc.scanBackfillCandidates(ctx, "backfill-configs")
	if err != nil {
		return nil, err
	}

	result := &GCResult{RecordsSkipped: int64(presentCount)}
	created := int64(0)
	httpClient := &http.Client{Timeout: 30 * time.Second}

	for i, candidate := range missing {
		gc.setProgress("records",
			fmt.Sprintf("Backfilling configs (%d/%d missing)...", i+1, len(missing)),
			"backfill-configs")

		userDID := candidate.UserDID
		manifestRkey := strings.TrimPrefix(candidate.Digest, "sha256:")
		manifestURI := candidate.ManifestURI
		manifestDigest := candidate.Digest

		pdsEndpoint, err := atproto.ResolveDIDToPDS(ctx, userDID)
		if err != nil {
			gc.logger.Warn("Failed to resolve PDS for backfill", "did", userDID, "error", err)
			continue
		}

		reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
			pdsEndpoint,
			url.QueryEscape(userDID),
			url.QueryEscape(atproto.ManifestCollection),
			url.QueryEscape(manifestRkey))

		req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil)
		if err != nil {
			continue
		}
		resp, err := httpClient.Do(req)
		if err != nil {
			gc.logger.Warn("Failed to fetch manifest for backfill", "uri", manifestURI, "error", err)
			continue
		}
		if resp.StatusCode != http.StatusOK {
			resp.Body.Close()
			continue
		}

		var envelope struct {
			Value json.RawMessage `json:"value"`
		}
		if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
			resp.Body.Close()
			continue
		}
		resp.Body.Close()

		var manifest atproto.ManifestRecord
		if err := json.Unmarshal(envelope.Value, &manifest); err != nil {
			continue
		}

		if manifest.Config == nil || manifest.Config.Digest == "" {
			continue
		}

		configBytes, err := gc.s3.GetBytes(ctx, s3.BlobPath(manifest.Config.Digest))
		if err != nil {
			gc.logger.Warn("Failed to fetch config blob", "digest", manifest.Config.Digest, "error", err)
			continue
		}

		configRecord := atproto.NewImageConfigRecord(manifestURI, string(configBytes))
		if _, _, err := gc.pds.CreateImageConfigRecord(ctx, configRecord, manifestDigest); err != nil {
			gc.logger.Warn("Failed to create image config record", "manifest", manifestURI, "error", err)
			continue
		}
		created++
		time.Sleep(200 * time.Millisecond) // throttle firehose events
	}

	result.RecordsReconciled = created
	result.Duration = time.Since(start)

	gc.mu.Lock()
	gc.lastResult = result
	gc.lastResultAt = time.Now()
	gc.mu.Unlock()

	gc.logger.Info("Image config backfill complete",
		"created", created,
		"skipped", result.RecordsSkipped)
	return result, nil
}

// discoverUserDIDs returns all DIDs that may have manifests referencing this hold.
// Union of: captain owner + crew members + distinct DIDs from layer records.
func (gc *GarbageCollector) discoverUserDIDs(ctx context.Context) ([]string, error) {
	seen := make(map[string]bool)

	// Captain owner DID
	_, captain, err := gc.pds.GetCaptainRecord(ctx)
	if err != nil {
		gc.logger.Warn("Failed to get captain record", "error", err)
	} else if captain.Owner != "" {
		seen[captain.Owner] = true
	}

	// Crew member DIDs
	crew, err := gc.pds.ListCrewMembers(ctx)
	if err != nil {
		gc.logger.Warn("Failed to list crew members", "error", err)
	} else {
		for _, m := range crew {
			if m.Record.Member != "" {
				seen[m.Record.Member] = true
			}
		}
	}

	// Distinct DIDs from existing layer records (catches non-crew pushers on public holds)
	recordsIndex := gc.pds.RecordsIndex()
	if recordsIndex != nil {
		dids, err := recordsIndex.DistinctDIDs(atproto.LayerCollection)
		if err != nil {
			gc.logger.Warn("Failed to get distinct DIDs from records", "error", err)
		} else {
			for _, did := range dids {
				seen[did] = true
			}
		}
	}

	dids := make([]string, 0, len(seen))
	for did := range seen {
		dids = append(dids, did)
	}
	return dids, nil
}

// fetchUserProfile fetches the sailor profile from a user's PDS.
// Returns nil without error if the profile doesn't exist.
func (gc *GarbageCollector) fetchUserProfile(ctx context.Context, pdsEndpoint, userDID string) (*atproto.SailorProfileRecord, error) {
	client := &http.Client{Timeout: 10 * time.Second}

	reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self",
		pdsEndpoint,
		url.QueryEscape(userDID),
		url.QueryEscape(atproto.SailorProfileCollection))

	req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil)
	if err != nil {
		return nil, fmt.Errorf("create request: %w", err)
	}

	resp, err := client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("http request: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusBadRequest || resp.StatusCode == http.StatusNotFound {
		return nil, nil // No profile
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("getRecord returned status %d", resp.StatusCode)
	}

	var envelope struct {
		Value json.RawMessage `json:"value"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
		return nil, fmt.Errorf("decode response: %w", err)
	}

	var profile atproto.SailorProfileRecord
	if err := json.Unmarshal(envelope.Value, &profile); err != nil {
		return nil, fmt.Errorf("unmarshal profile: %w", err)
	}

	return &profile, nil
}

// GC walks a user's PDS to decide which blobs are still referenced, and a
// failed walk is treated as "assume everything is referenced" — safe, but it
// pins that user's storage for the whole run. Finding 16 measured the cost:
// every DID GC classified unreachable-but-healthy had failed only one or two
// runs out of six, and replaying the identical calls afterwards returned 200
// in 45-680 ms with no rate limiting. These are ordinary blips on small
// self-hosted PDSes, amplified into a full-DID skip by a single-shot fetch.
const (
	gcListAttempts = 3
	gcListBackoff  = 500 * time.Millisecond
)

// retryableFetchErr reports whether a transport error is worth another attempt.
// The distinction matters in both directions: retrying a blip recovers a
// healthy user's storage, but retrying a host that is genuinely gone only slows
// the run and makes a dead PDS look alive for longer. DNS failure, TLS failure
// and connection refused are stable facts about an endpoint and repeat
// identically, so they are never retried. Timeouts and resets are the transient
// shapes. Anything unrecognised is treated as permanent, which errs toward the
// current behaviour rather than toward hammering.
func retryableFetchErr(err error) bool {
	if err == nil {
		return false
	}
	var dnsErr *net.DNSError
	if errors.As(err, &dnsErr) {
		return false
	}
	var certErr *tls.CertificateVerificationError
	if errors.As(err, &certErr) {
		return false
	}
	var recordErr tls.RecordHeaderError
	if errors.As(err, &recordErr) {
		return false
	}
	if errors.Is(err, syscall.ECONNREFUSED) {
		return false
	}
	var netErr net.Error
	if errors.As(err, &netErr) && netErr.Timeout() {
		return true
	}
	return errors.Is(err, syscall.ECONNRESET) || errors.Is(err, io.ErrUnexpectedEOF)
}

// retryableStatus reports whether an HTTP status deserves another attempt.
// 5xx and 429 are the server saying "not now"; a 4xx is the server saying
// "not ever", and repeating it would be pointless.
func retryableStatus(status int) bool {
	return status == http.StatusTooManyRequests || status >= 500
}

// listRecordsPage performs one listRecords GET and decodes it into out,
// retrying transient failures with linear backoff. The caller's context still
// short-circuits everything: a cancelled run stops immediately rather than
// sleeping through its remaining attempts.
func (gc *GarbageCollector) listRecordsPage(ctx context.Context, client *http.Client, reqURL string, out any) error {
	var lastErr error

	for attempt := 1; attempt <= gcListAttempts; attempt++ {
		if attempt > 1 {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(time.Duration(attempt-1) * gcListBackoff):
			}
		}

		req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil)
		if err != nil {
			return fmt.Errorf("create request: %w", err)
		}

		resp, err := client.Do(req)
		if err != nil {
			lastErr = fmt.Errorf("http request: %w", err)
			if ctx.Err() != nil || !retryableFetchErr(err) {
				return lastErr
			}
			slog.Debug("GC listRecords attempt failed, retrying",
				"component", "gc", "url", reqURL, "attempt", attempt, "error", err)
			continue
		}

		if resp.StatusCode != http.StatusOK {
			resp.Body.Close()
			lastErr = fmt.Errorf("listRecords returned status %d", resp.StatusCode)
			if !retryableStatus(resp.StatusCode) {
				return lastErr
			}
			slog.Debug("GC listRecords attempt returned retryable status",
				"component", "gc", "url", reqURL, "attempt", attempt, "status", resp.StatusCode)
			continue
		}

		err = json.NewDecoder(resp.Body).Decode(out)
		resp.Body.Close()
		if err != nil {
			// A truncated body is a transport-shaped failure, so it is worth
			// one more attempt rather than writing the whole DID off.
			lastErr = fmt.Errorf("decode response: %w", err)
			if ctx.Err() != nil || !retryableFetchErr(err) {
				return lastErr
			}
			slog.Debug("GC listRecords decode failed, retrying",
				"component", "gc", "url", reqURL, "attempt", attempt, "error", err)
			continue
		}

		return nil
	}

	return fmt.Errorf("after %d attempts: %w", gcListAttempts, lastErr)
}

// fetchUserTags fetches all tag records from a user's PDS.
// Returns a map of digest → true for all tagged manifest digests.
func (gc *GarbageCollector) fetchUserTags(ctx context.Context, pdsEndpoint, userDID string) (map[string]bool, error) {
	tagged := make(map[string]bool)
	cursor := ""
	client := &http.Client{Timeout: 30 * time.Second}

	for {
		reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100",
			pdsEndpoint,
			url.QueryEscape(userDID),
			url.QueryEscape(atproto.TagCollection))
		if cursor != "" {
			reqURL += "&cursor=" + cursor
		}

		var listResult struct {
			Records []struct {
				Value json.RawMessage `json:"value"`
			} `json:"records"`
			Cursor string `json:"cursor,omitempty"`
		}

		if err := gc.listRecordsPage(ctx, client, reqURL, &listResult); err != nil {
			return nil, err
		}

		for _, rec := range listResult.Records {
			var tag atproto.TagRecord
			if err := json.Unmarshal(rec.Value, &tag); err != nil {
				continue
			}
			if d, err := tag.GetManifestDigest(); err == nil {
				tagged[d] = true
			}
		}

		if listResult.Cursor == "" {
			break
		}
		cursor = listResult.Cursor
	}

	return tagged, nil
}

// filterUntaggedManifests removes untagged manifests from the list, preserving
// manifest list children that are referenced by tagged manifest lists.
func (gc *GarbageCollector) filterUntaggedManifests(manifests []*manifestInfo, taggedDigests map[string]bool) []*manifestInfo {
	// Build a set of digests that are children of tagged manifest lists
	childDigests := make(map[string]bool)
	for _, m := range manifests {
		// Check if this manifest is a manifest list AND is tagged
		mDigest := extractDigestFromManifestURI(m.URI)
		if mDigest == "" || !taggedDigests[mDigest] {
			continue
		}
		// This is a tagged manifest list — protect its children
		for _, ref := range m.Record.Manifests {
			childDigests[ref.Digest] = true
		}
	}

	// Filter: keep manifests that are tagged, or are children of tagged manifest lists
	var kept []*manifestInfo
	for _, m := range manifests {
		mDigest := extractDigestFromManifestURI(m.URI)
		if mDigest == "" {
			kept = append(kept, m)
			continue
		}
		if taggedDigests[mDigest] || childDigests[mDigest] {
			kept = append(kept, m)
		}
	}
	return kept
}

// extractDigestFromManifestURI extracts the digest from a manifest AT-URI.
// URI format: at://did:plc:xxx/io.atcr.manifest/{encoded_digest}
// Returns the full digest (e.g., "sha256:abc123...") or empty string.
func extractDigestFromManifestURI(uri string) string {
	parts := parseATURI(uri)
	if parts == nil || parts.Collection != atproto.ManifestCollection {
		return ""
	}
	// The rkey is the digest without the "sha256:" prefix
	return "sha256:" + parts.Rkey
}

// fetchUserManifestsFromEndpoint fetches manifests from a specific PDS endpoint.
func (gc *GarbageCollector) fetchUserManifestsFromEndpoint(ctx context.Context, userDID, pdsEndpoint, holdDID string) ([]*manifestInfo, error) {
	var manifests []*manifestInfo
	cursor := ""
	client := &http.Client{Timeout: 30 * time.Second}

	for {
		reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100",
			pdsEndpoint,
			url.QueryEscape(userDID),
			url.QueryEscape(atproto.ManifestCollection))
		if cursor != "" {
			reqURL += "&cursor=" + cursor
		}

		var listResult struct {
			Records []struct {
				URI   string          `json:"uri"`
				CID   string          `json:"cid"`
				Value json.RawMessage `json:"value"`
			} `json:"records"`
			Cursor string `json:"cursor,omitempty"`
		}

		if err := gc.listRecordsPage(ctx, client, reqURL, &listResult); err != nil {
			return nil, err
		}

		for _, rec := range listResult.Records {
			var manifest atproto.ManifestRecord
			if err := json.Unmarshal(rec.Value, &manifest); err != nil {
				gc.logger.Warn("Failed to parse manifest record",
					"uri", rec.URI, "error", err)
				continue
			}

			// The fail-open decision lives here, in the open, rather than inside
			// the classifier: an unknown claim is carried so its blobs stay
			// referenced, but marked so it is never adopted as ours.
			switch gc.classifyManifest(ctx, &manifest, holdDID) {
			case claimOurs:
				manifests = append(manifests, &manifestInfo{
					URI:     rec.URI,
					UserDID: userDID,
					Record:  &manifest,
				})
			case claimUnknown:
				manifests = append(manifests, &manifestInfo{
					URI:         rec.URI,
					UserDID:     userDID,
					Record:      &manifest,
					ProtectOnly: true,
				})
			case claimNotOurs:
				// Definitively another hold's: nothing to keep and nothing to protect.
			}
		}

		if listResult.Cursor == "" {
			break
		}
		cursor = listResult.Cursor
	}

	return manifests, nil
}

// classifyManifest reports what a manifest's hold field says about ownership,
// via HoldDID, legacy HoldEndpoint, or a predecessor hold that has been migrated.
// It states what it found and applies no policy; the caller decides what an
// unknown claim is worth. See manifestClaim.
func (gc *GarbageCollector) classifyManifest(ctx context.Context, manifest *atproto.ManifestRecord, holdDID string) manifestClaim {
	manifestHoldDID := manifest.HoldDID

	// Direct match
	if manifestHoldDID == holdDID {
		return claimOurs
	}

	// Legacy: resolve holdEndpoint to DID.
	//
	// A resolve failure here reports claimNotOurs rather than claimUnknown,
	// preserving long-standing behaviour: this path predates the predecessor
	// check and fails closed. That is arguably the same shape error in the other
	// direction — an unresolvable endpoint drops the blobs — but it is a distinct
	// bug with a distinct blast radius and wants its own change.
	if manifestHoldDID == "" && manifest.HoldEndpoint != "" {
		resolved, err := atproto.ResolveHoldDID(ctx, manifest.HoldEndpoint)
		if err != nil {
			gc.logger.Debug("Failed to resolve hold DID from legacy endpoint", "holdEndpoint", manifest.HoldEndpoint, "error", err)
			return claimNotOurs
		}
		manifestHoldDID = resolved
		if manifestHoldDID == holdDID {
			return claimOurs
		}
	}

	if manifestHoldDID == "" {
		return claimNotOurs
	}

	// Check if the manifest's hold is a predecessor (has a successor label set)
	return gc.classifyPredecessorHold(ctx, manifestHoldDID)
}

// classifyPredecessorHold reports whether holdDID is a predecessor of this hold,
// by fetching its captain record and checking for a successor label that names
// us. Definitive answers are cached to avoid repeated network calls; an
// unreachable hold is reported as claimUnknown and deliberately not cached as a
// negative, since caching it would make one five-second blip permanent for the
// life of the process.
//
// This reports what it found. It does not decide what an unknown claim earns —
// that policy lives at the call site in fetchUserManifestsFromEndpoint.
func (gc *GarbageCollector) classifyPredecessorHold(ctx context.Context, holdDID string) manifestClaim {
	if gc.predecessorCache == nil {
		gc.predecessorCache = make(map[string]bool)
	}

	if cachedPredecessor, cached := gc.predecessorCache[holdDID]; cached {
		if cachedPredecessor {
			return claimOurs
		}
		return claimNotOurs
	}

	// Already unreachable earlier in this run. Answer the same way without
	// paying another timeout; the next run starts fresh and re-checks.
	if gc.predecessorUnresolved[holdDID] {
		return claimUnknown
	}

	isPredecessor, definitive := gc.checkPredecessor(ctx, holdDID)
	if !definitive {
		if gc.predecessorUnresolved == nil {
			gc.predecessorUnresolved = make(map[string]bool)
		}
		gc.predecessorUnresolved[holdDID] = true
		gc.logger.Warn("GC: predecessor status unresolved, keeping hold's blobs referenced but not adopting its manifests",
			"holdDID", holdDID)
		return claimUnknown
	}

	gc.predecessorCache[holdDID] = isPredecessor
	if isPredecessor {
		return claimOurs
	}
	return claimNotOurs
}

// checkPredecessor fetches a hold's captain record to check if it has a successor label
// (meaning the hold has been migrated/retired and its blobs are served by this hold).
//
// The second return value reports whether the answer is definitive. It is false whenever
// the hold could not be reached or its reply could not be understood: those are cases
// where the hold may well be a predecessor and we simply cannot tell. Callers must not
// read an inconclusive result as "not a predecessor" — a single five-second blip would
// otherwise drop a live predecessor's entire blob set out of the referenced set, and
// those blobs are long past the grace period that protects recent content.
//
// A non-200 is treated as inconclusive rather than a negative on the same reasoning: a
// reachable service that cannot produce its own captain record is malfunctioning, not
// answering. Over-protecting an unrelated hold merely leaves some blobs unreclaimed.
func (gc *GarbageCollector) checkPredecessor(ctx context.Context, holdDID string) (isPredecessor, definitive bool) {
	fetchCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()

	holdURL, err := atproto.ResolveHoldURL(fetchCtx, holdDID)
	if err != nil {
		gc.logger.Debug("GC: failed to resolve predecessor hold URL",
			"holdDID", holdDID, "error", err)
		return false, false
	}

	return gc.checkPredecessorAt(fetchCtx, holdDID, holdURL)
}

// checkPredecessorAt is checkPredecessor with the hold's base URL already resolved,
// split out so the fetch-and-parse half can be exercised against a local server.
// It carries the same contract: the second return value is false whenever the answer
// is inconclusive rather than negative.
func (gc *GarbageCollector) checkPredecessorAt(ctx context.Context, holdDID, holdURL string) (isPredecessor, definitive bool) {
	recordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self",
		holdURL,
		url.QueryEscape(holdDID),
		url.QueryEscape(atproto.CaptainCollection),
	)

	req, err := http.NewRequestWithContext(ctx, "GET", recordURL, nil)
	if err != nil {
		return false, false
	}

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		gc.logger.Debug("GC: failed to fetch predecessor captain record",
			"holdDID", holdDID, "error", err)
		return false, false
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		gc.logger.Debug("GC: predecessor captain record fetch returned non-200",
			"holdDID", holdDID, "status", resp.StatusCode)
		return false, false
	}

	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if err != nil {
		gc.logger.Debug("GC: failed to read predecessor captain record",
			"holdDID", holdDID, "error", err)
		return false, false
	}

	var envelope struct {
		Value json.RawMessage `json:"value"`
	}
	if err := json.Unmarshal(body, &envelope); err != nil {
		gc.logger.Debug("GC: failed to parse predecessor captain envelope",
			"holdDID", holdDID, "error", err)
		return false, false
	}

	var captain atproto.CaptainRecord
	if err := json.Unmarshal(envelope.Value, &captain); err != nil {
		gc.logger.Debug("GC: failed to parse predecessor captain record",
			"holdDID", holdDID, "error", err)
		return false, false
	}

	// The hold answered and declares no successor. This is the one negative we
	// are entitled to cache.
	if captain.Successor == "" {
		return false, true
	}

	// A successor label alone is not enough: it has to name us. A hold that
	// retired into some third hold is that hold's predecessor, not ours, and its
	// manifests are not a reason to keep blobs referenced here.
	// scan_broadcaster.go makes exactly this comparison, and the two are meant
	// to agree.
	ourHoldDID := gc.ourHoldDID()
	if ourHoldDID == "" {
		// We cannot say who we are, so we cannot say the successor is not us.
		// Keep the older, permissive answer: over-protecting leaks blobs,
		// guessing the other way deletes them.
		gc.logger.Warn("GC: own hold DID unknown, treating any successor label as pointing at us",
			"holdDID", holdDID, "successor", captain.Successor)
		return true, true
	}

	if captain.Successor != ourHoldDID {
		gc.logger.Info("GC: hold has a successor, but it is not us",
			"holdDID", holdDID, "successor", captain.Successor, "ourHoldDID", ourHoldDID)
		return false, true
	}

	gc.logger.Info("GC: discovered predecessor hold (successor points at us)",
		"holdDID", holdDID, "successor", captain.Successor)
	return true, true
}

// ourHoldDID reports this hold's own DID, or "" when it cannot be determined.
func (gc *GarbageCollector) ourHoldDID() string {
	if gc.pds == nil {
		return ""
	}
	return gc.pds.DID()
}

// deleteOrphanedRecords removes layer records whose manifests no longer exist
func (gc *GarbageCollector) deleteOrphanedRecords(ctx context.Context, refs []orphanRef, result *GCResult) error {
	// Manifests the hold currently holds layer records for. Built once, only if
	// there is an aux record to delete, and used to confirm at delete time that
	// the manifest has not come back since it was judged orphaned. Purely local
	// — no PDS round trip — because a push writes layer records here.
	var liveManifests map[string]bool
	for _, ref := range refs {
		if ref.Collection != "" && ref.Collection != atproto.LayerCollection {
			var err error
			if liveManifests, err = gc.manifestsWithLayerRecords(ctx); err != nil {
				return fmt.Errorf("checking which manifests still have layer records: %w", err)
			}
			break
		}
	}

	skipped := 0
	for _, ref := range refs {
		// An empty collection means the ref came from a preview taken before
		// the sweep covered scan and image-config records; those were all
		// layer records.
		collection := ref.Collection
		if collection == "" {
			collection = atproto.LayerCollection
		}

		if collection == atproto.LayerCollection {
			if err := gc.pds.DeleteLayerRecord(ctx, ref.Rkey); err != nil {
				gc.logger.Error("Failed to delete layer record", "rkey", ref.Rkey, "error", err)
				continue
			}
		} else {
			// Confirm the slot still holds the revision we judged. Without this,
			// a re-push between the scan and the delete would have us destroy a
			// live record that merely inherited the same digest-derived rkey.
			if !gc.auxRecordUnchanged(ctx, collection, ref.Rkey, ref.CID) {
				gc.logger.Info("Skipping record: changed or gone since it was scanned",
					"collection", collection, "rkey", ref.Rkey, "scannedCID", ref.CID)
				skipped++
				continue
			}
			// The manifest has layer records again, so it is back on this hold
			// and the record is in use. Covers the case the CID check cannot: a
			// re-push does not rewrite the scan record, so its CID still matches
			// even though the image is live again.
			if ref.ManifestURI != "" && liveManifests[ref.ManifestURI] {
				gc.logger.Info("Skipping record: manifest has layer records again",
					"collection", collection, "rkey", ref.Rkey, "manifest", ref.ManifestURI)
				skipped++
				continue
			}
			deleted, err := gc.pds.DeleteManifestAuxRecord(ctx, collection, ref.Rkey)
			if err != nil {
				gc.logger.Error("Failed to delete record",
					"collection", collection, "rkey", ref.Rkey, "error", err)
				continue
			}
			if !deleted {
				// Already gone (e.g. a concurrent purge) — not an error, but
				// don't count it as work this run did.
				continue
			}
		}

		result.RecordsDeleted++
		gc.logger.Debug("Deleted orphaned record", "collection", collection, "rkey", ref.Rkey)
	}

	gc.logger.Info("Phase 2 complete",
		"orphaned", len(refs),
		"deleted", result.RecordsDeleted,
		"skippedChangedSinceScan", skipped)

	return nil
}

// deleteOrphanedBlobs walks storage and deletes blobs not in the referenced set
func (gc *GarbageCollector) deleteOrphanedBlobs(ctx context.Context, referenced map[string]bool, result *GCResult) error {
	blobsPath := "/docker/registry/v2/blobs"

	err := gc.s3.WalkBlobs(ctx, blobsPath, func(key string, size int64, lastModified time.Time) error {
		// Only process data files
		if !strings.HasSuffix(key, "/data") {
			return nil
		}

		// Extract digest from path
		digest := extractDigestFromPath(key)
		if digest == "" {
			return nil
		}

		// Check if referenced by any layer record
		if referenced[digest] {
			return nil
		}

		// Unreferenced, but young enough that we'd rather keep paying for it
		// than risk deleting content still being pushed. Records for this
		// blob may already be gone; blob pruning runs on its own clock.
		if !blobPastGrace(lastModified) {
			return nil
		}

		result.OrphanedBlobs++

		if err := gc.s3.Delete(ctx, key); err != nil {
			gc.logger.Error("Failed to delete blob", "path", key, "error", err)
			return nil // Continue with other blobs
		}
		result.BlobsDeleted++
		result.BytesReclaimed += size
		gc.logger.Debug("Deleted orphaned blob",
			"digest", digest,
			"size", size)

		return nil
	})

	if err != nil {
		return fmt.Errorf("walk storage failed: %w", err)
	}

	gc.logger.Info("Phase 3 complete",
		"orphanedBlobs", result.OrphanedBlobs,
		"deleted", result.BlobsDeleted,
		"reclaimed", result.BytesReclaimed)

	return nil
}

// decodeLayerRecord reads and decodes a layer record from the PDS
func (gc *GarbageCollector) decodeLayerRecord(ctx context.Context, rec pds.Record) (*atproto.LayerRecord, error) {
	// Get the record from the repo
	recordPath := rec.Collection + "/" + rec.Rkey
	_, recBytes, err := gc.pds.GetRecordBytes(ctx, recordPath)
	if err != nil {
		return nil, fmt.Errorf("get record bytes: %w", err)
	}

	// Decode the layer record
	var layer atproto.LayerRecord
	if err := layer.UnmarshalCBOR(bytes.NewReader(*recBytes)); err != nil {
		return nil, fmt.Errorf("unmarshal CBOR: %w", err)
	}

	return &layer, nil
}

// atURIParts contains parsed components of an AT-URI
type atURIParts struct {
	DID        string
	Collection string
	Rkey       string
}

// parseATURI parses an AT-URI into its components
// Format: at://did:plc:xxx/collection/rkey
func parseATURI(uri string) *atURIParts {
	if !strings.HasPrefix(uri, "at://") {
		return nil
	}

	// Remove at:// prefix
	path := strings.TrimPrefix(uri, "at://")

	// Split by /
	parts := strings.SplitN(path, "/", 3)
	if len(parts) != 3 {
		return nil
	}

	return &atURIParts{
		DID:        parts[0],
		Collection: parts[1],
		Rkey:       parts[2],
	}
}

// tidToTime extracts the timestamp from a TID (Timestamp ID)
// TIDs are 13-character base32 encoded timestamps with counter
func tidToTime(tid string) time.Time {
	// TIDs are base32-sortable timestamps
	// Use indigo's syntax package for proper parsing
	t, err := syntax.ParseTID(tid)
	if err != nil {
		// Return zero time - will be older than grace period
		return time.Time{}
	}
	return t.Time()
}

// extractDigestFromPath extracts a digest from a storage path
// Path format: /docker/registry/v2/blobs/{algorithm}/{xx}/{hash}/data
// Returns: {algorithm}:{hash}
func extractDigestFromPath(path string) string {
	// Match pattern: /blobs/{alg}/{xx}/{hash}/data
	re := regexp.MustCompile(`/blobs/([^/]+)/[^/]+/([^/]+)/data$`)
	matches := re.FindStringSubmatch(path)
	if len(matches) != 3 {
		return ""
	}
	return matches[1] + ":" + matches[2]
}

// logResult logs the GC result in a structured format
func (gc *GarbageCollector) logResult(result *GCResult) {
	gc.logger.Info("GC run complete",
		"duration", result.Duration,
		"referencedBlobs", result.ReferencedBlobs,
		"orphanedRecords", result.OrphanedRecords,
		"recordsDeleted", result.RecordsDeleted,
		"recordsReconciled", result.RecordsReconciled,
		"manifestsChecked", result.ManifestsChecked,
		"usersChecked", result.UsersChecked,
		"orphanedBlobs", result.OrphanedBlobs,
		"blobsDeleted", result.BlobsDeleted,
		"bytesReclaimed", result.BytesReclaimed)

	// Also log as JSON for easier parsing
	resultJSON, _ := json.Marshal(result)
	gc.logger.Debug("GC result JSON", "result", string(resultJSON))
}

// LastPreview returns the most recent preview result and when it was generated
func (gc *GarbageCollector) LastPreview() (*GCPreview, time.Time) {
	gc.mu.Lock()
	defer gc.mu.Unlock()
	return gc.lastPreview, gc.lastPreviewAt
}

// LastResult returns the most recent GC run result and when it was generated
func (gc *GarbageCollector) LastResult() (*GCResult, time.Time) {
	gc.mu.Lock()
	defer gc.mu.Unlock()
	return gc.lastResult, gc.lastResultAt
}

// LastBackfillPreview returns the most recent image-config backfill preview
// and when it was generated.
func (gc *GarbageCollector) LastBackfillPreview() (*BackfillConfigsPreview, time.Time) {
	gc.mu.Lock()
	defer gc.mu.Unlock()
	return gc.lastBackfillPreview, gc.lastBackfillPreviewAt
}

// IsRunning returns whether a GC operation is currently in progress
func (gc *GarbageCollector) IsRunning() bool {
	gc.mu.Lock()
	defer gc.mu.Unlock()
	return gc.running
}

// GetConfig returns the current GC configuration
func (gc *GarbageCollector) GetConfig() Config {
	return gc.cfg
}