at-container-registry/pkg/appview/middleware/registry.go

package middleware

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/distribution/distribution/v3"
	"github.com/distribution/distribution/v3/registry/api/errcode"
	v2 "github.com/distribution/distribution/v3/registry/api/v2"
	registrymw "github.com/distribution/distribution/v3/registry/middleware/registry"
	"github.com/distribution/distribution/v3/registry/storage/driver"
	"github.com/distribution/reference"

	"atcr.io/pkg/appview/db"
	"atcr.io/pkg/appview/readme"
	"atcr.io/pkg/appview/storage"
	"atcr.io/pkg/atproto"
	"atcr.io/pkg/auth"
	"atcr.io/pkg/auth/oauth"
	"atcr.io/pkg/auth/token"
)

// authMethodKey is the context key for storing auth method from JWT
const authMethodKey contextKey = "auth.method"

// pullerDIDKey is the context key for storing the authenticated user's DID from JWT
const pullerDIDKey contextKey = "puller.did"

// hasPushScopeKey is the context key for storing whether the JWT has push scope
const hasPushScopeKey contextKey = "token.has_push_scope"

// validationCacheEntry stores a validated service token with expiration
type validationCacheEntry struct {
	serviceToken string
	validUntil   time.Time
	err          error         // Cached error for fast-fail
	mu           sync.Mutex    // Per-entry lock to serialize cache population
	inFlight     bool          // True if another goroutine is fetching the token
	done         chan struct{} // Closed when fetch completes
}

// validationCache provides request-level caching for service tokens
// This prevents concurrent layer uploads from racing on OAuth/DPoP requests
type validationCache struct {
	mu      sync.RWMutex
	entries map[string]*validationCacheEntry // key: "did:holdDID"
}

// newValidationCache creates a new validation cache
func newValidationCache() *validationCache {
	return &validationCache{
		entries: make(map[string]*validationCacheEntry),
	}
}

// getOrFetch retrieves a service token from cache or fetches it
// Multiple concurrent requests for the same DID:holdDID will share the fetch operation
func (vc *validationCache) getOrFetch(ctx context.Context, cacheKey string, fetchFunc func() (string, error)) (string, error) {
	// Fast path: check cache with read lock
	vc.mu.RLock()
	entry, exists := vc.entries[cacheKey]
	vc.mu.RUnlock()

	if exists {
		// Entry exists, check if it's still valid
		entry.mu.Lock()

		// If another goroutine is fetching, wait for it
		if entry.inFlight {
			done := entry.done
			entry.mu.Unlock()

			select {
			case <-done:
				// Fetch completed, check result
				entry.mu.Lock()
				defer entry.mu.Unlock()

				if entry.err != nil {
					return "", entry.err
				}
				if time.Now().Before(entry.validUntil) {
					return entry.serviceToken, nil
				}
				// Fall through to refetch
			case <-ctx.Done():
				return "", ctx.Err()
			}
		} else {
			// Check if cached token is still valid
			if entry.err != nil && time.Now().Before(entry.validUntil) {
				// Return cached error (fast-fail)
				entry.mu.Unlock()
				return "", entry.err
			}
			if entry.err == nil && time.Now().Before(entry.validUntil) {
				// Return cached token
				token := entry.serviceToken
				entry.mu.Unlock()
				return token, nil
			}
			entry.mu.Unlock()
		}
	}

	// Slow path: need to fetch token
	vc.mu.Lock()
	entry, exists = vc.entries[cacheKey]
	if !exists {
		// Create new entry
		entry = &validationCacheEntry{
			inFlight: true,
			done:     make(chan struct{}),
		}
		vc.entries[cacheKey] = entry
	}
	vc.mu.Unlock()

	// Lock the entry to perform fetch
	entry.mu.Lock()

	// Double-check: another goroutine may have fetched while we waited
	if !entry.inFlight {
		if entry.err != nil && time.Now().Before(entry.validUntil) {
			err := entry.err
			entry.mu.Unlock()
			return "", err
		}
		if entry.err == nil && time.Now().Before(entry.validUntil) {
			token := entry.serviceToken
			entry.mu.Unlock()
			return token, nil
		}
	}

	// Mark as in-flight and create fresh done channel for this fetch
	// IMPORTANT: Always create a new channel - a closed channel is not nil
	entry.done = make(chan struct{})
	entry.inFlight = true
	done := entry.done
	entry.mu.Unlock()

	// Perform the fetch (outside the lock to allow other operations)
	serviceToken, err := fetchFunc()

	// Update the entry with result
	entry.mu.Lock()
	entry.inFlight = false

	if err != nil {
		// Cache errors for 5 seconds (fast-fail for subsequent requests)
		entry.err = err
		entry.validUntil = time.Now().Add(5 * time.Second)
		entry.serviceToken = ""
	} else {
		// Cache token for 45 seconds (covers typical Docker push operation)
		entry.err = nil
		entry.serviceToken = serviceToken
		entry.validUntil = time.Now().Add(45 * time.Second)
	}

	// Signal completion to waiting goroutines
	close(done)
	entry.mu.Unlock()

	return serviceToken, err
}

// LabelChecker checks whether content has been taken down via ATProto labels.
type LabelChecker interface {
	IsTakenDown(did, repository string) (bool, error)
}

// UserPrefsCache reads and writes the appview's local copy of the two sailor
// profile fields the registry hot path needs. It is kept current by the
// Jetstream processor and prefilled by the startup backfill; the middleware
// only writes to it on the one-shot fallback for a user it has never seen.
// Implemented by db.HoldDIDDB.
type UserPrefsCache interface {
	GetUserHoldPrefs(did string) (db.UserHoldPrefs, error)
	CacheUserHoldPrefs(did, handle, pdsEndpoint, holdDID string, autoRemoveUntagged bool) error
}

// Global variables for initialization only
// These are set by main.go during startup and copied into NamespaceResolver instances.
// After initialization, request handling uses the NamespaceResolver's instance fields.
var (
	globalRefresher          *oauth.Refresher
	globalDatabase           storage.HoldDIDLookup
	globalAuthorizer         auth.HoldAuthorizer
	globalWebhookDispatcher  storage.PushWebhookDispatcher
	globalManifestRefChecker storage.ManifestReferenceChecker
	globalLabelChecker       LabelChecker
	globalUserPrefs          UserPrefsCache
)

// SetGlobalRefresher sets the OAuth refresher instance during initialization
// Must be called before the registry starts serving requests
func SetGlobalRefresher(refresher *oauth.Refresher) {
	globalRefresher = refresher
}

// SetGlobalDatabase sets the database instance during initialization
// Must be called before the registry starts serving requests
func SetGlobalDatabase(database storage.HoldDIDLookup) {
	globalDatabase = database
}

// SetGlobalUserPrefs sets the cached sailor-profile preference store during
// initialization. Must be called before the registry starts serving requests.
// Leaving it nil is safe but costs a live profile fetch on every request.
func SetGlobalUserPrefs(prefs UserPrefsCache) {
	globalUserPrefs = prefs
}

// SetGlobalManifestRefChecker sets the manifest reference checker during initialization
func SetGlobalManifestRefChecker(checker storage.ManifestReferenceChecker) {
	globalManifestRefChecker = checker
}

// SetGlobalLabelChecker sets the label checker instance during initialization
func SetGlobalLabelChecker(checker LabelChecker) {
	globalLabelChecker = checker
}

// SetGlobalAuthorizer sets the authorizer instance during initialization
// Must be called before the registry starts serving requests
func SetGlobalAuthorizer(authorizer auth.HoldAuthorizer) {
	globalAuthorizer = authorizer
}

// SetGlobalWebhookDispatcher sets the push webhook dispatcher during initialization
// Must be called before the registry starts serving requests
func SetGlobalWebhookDispatcher(dispatcher storage.PushWebhookDispatcher) {
	globalWebhookDispatcher = dispatcher
}

// GetGlobalAuthorizer returns the global authorizer instance
// Used by components that need to clear denial cache (e.g., EnsureCrewMembership)
func GetGlobalAuthorizer() auth.HoldAuthorizer {
	return globalAuthorizer
}

func init() {
	// Register the name resolution middleware
	if err := registrymw.Register("atproto-resolver", initATProtoResolver); err != nil {
		panic("failed to register atproto-resolver middleware: " + err.Error())
	}
}

// NamespaceResolver wraps a namespace and resolves names
type NamespaceResolver struct {
	distribution.Namespace
	defaultHoldDID      string                           // Default hold DID (e.g., "did:web:hold01.atcr.io")
	baseURL             string                           // Base URL for error messages (e.g., "https://atcr.io")
	fallbackUnreachable bool                             // Fall back to the default hold when the user's hold is unreachable (testmode builds)
	refresher           *oauth.Refresher                 // OAuth session manager (copied from global on init)
	database            storage.HoldDIDLookup            // Database for hold DID lookups (copied from global on init)
	authorizer          auth.HoldAuthorizer              // Hold authorization (copied from global on init)
	webhookDispatcher   storage.PushWebhookDispatcher    // Push webhook dispatcher (copied from global on init)
	manifestRefChecker  storage.ManifestReferenceChecker // Manifest reference checker (copied from global on init)
	validationCache     *validationCache                 // Request-level service token cache
	readmeFetcher       *readme.Fetcher                  // README fetcher for repo pages
	userPrefs           UserPrefsCache                   // Cached sailor profile preferences (copied from global on init)
}

// initATProtoResolver initializes the name resolution middleware
func initATProtoResolver(ctx context.Context, ns distribution.Namespace, _ driver.StorageDriver, options map[string]any) (distribution.Namespace, error) {
	// Get default hold DID from config (required)
	// Expected format: "did:web:hold01.atcr.io"
	defaultHoldDID := ""
	if holdDID, ok := options["default_hold_did"].(string); ok {
		defaultHoldDID = holdDID
	}

	// Get base URL from config (for error messages)
	baseURL := ""
	if url, ok := options["base_url"].(string); ok {
		baseURL = url
	}

	// Copy shared services from globals into the instance
	// This avoids accessing globals during request handling
	return &NamespaceResolver{
		Namespace:           ns,
		defaultHoldDID:      defaultHoldDID,
		baseURL:             baseURL,
		fallbackUnreachable: atproto.TestModeBuild,
		refresher:           globalRefresher,
		database:            globalDatabase,
		authorizer:          globalAuthorizer,
		webhookDispatcher:   globalWebhookDispatcher,
		manifestRefChecker:  globalManifestRefChecker,
		validationCache:     newValidationCache(),
		readmeFetcher:       readme.NewFetcher(),
		userPrefs:           globalUserPrefs,
	}, nil
}

// authErrorMessage creates a user-friendly auth error with login URL
func (nr *NamespaceResolver) authErrorMessage(message string) error {
	loginURL := fmt.Sprintf("%s/auth/oauth/login", nr.baseURL)
	fullMessage := fmt.Sprintf("%s - please re-authenticate at %s", message, loginURL)
	return errcode.ErrorCodeUnauthorized.WithMessage(fullMessage)
}

// noHoldConfiguredError reports that the appview itself has no hold to route
// to. Unlike the other failures in Repository(), this one really is our fault,
// so it keeps a 5xx, but it still has to be coded: an uncoded error is dropped
// by distribution's type switch and served as `500 {}`, which tells the
// operator reading the client's output nothing at all.
func noHoldConfiguredError() errcode.Error {
	return errcode.ErrorCodeUnknown.WithMessage(
		"registry is misconfigured: no hold service is available, set default_hold_did in the appview middleware config")
}

// holdResolutionError classifies a hold-URL resolution failure into an OCI
// error code. Everything returned from Repository() must be an errcode.Error:
// distribution's dispatcher (registry/handlers/app.go) type-switches on the
// error and has no default branch, so an uncoded error leaves the error list
// empty and errcode.ServeJSON emits a bodyless `500 {}` that clients retry.
//
// The split matters as much as the coding. A DNS blip or a PLC outage is
// genuinely transient, so it stays a retryable 503 UNAVAILABLE. A hold DID
// that can never resolve (a malformed identifier, or one naming a host the
// identity directory rejects outright, such as did:web:localhost%3A8080 left
// in a sailor profile) is stored user data that no retry can fix, so it
// terminates the client with 404 NAME_UNKNOWN instead of sending it round the
// retry loop three more times.
//
// The message names the offending hold DID. That is not a disclosure: the
// value lives in the owner's world-readable sailor profile record, and without
// it the user has no way to know which setting to fix.
func holdResolutionError(holdDID string, err error) error {
	if errors.Is(err, atproto.ErrHoldDIDPermanent) {
		slog.Debug("Hold DID is permanently unresolvable",
			"component", "registry/middleware", "holdDID", holdDID, "error", err)
		return errcode.Error{
			Code: v2.ErrorCodeNameUnknown,
			Message: fmt.Sprintf(
				"repository name not known to registry: the storage hold configured for this repository, %s, cannot be resolved. The owner should update defaultHold in their sailor profile",
				holdDID),
		}
	}

	slog.Warn("Hold DID resolution failed",
		"component", "registry/middleware", "holdDID", holdDID, "error", err)
	return errcode.Error{
		Code: errcode.ErrorCodeUnavailable,
		Message: fmt.Sprintf(
			"could not resolve the storage hold %s for this repository, please retry",
			holdDID),
	}
}

// Repository resolves the repository name and delegates to underlying namespace
// Handles names like:
// - atcr.io/alice/myimage → resolve alice to DID
// - atcr.io/did:plc:xyz123/myimage → use DID directly
func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Named) (distribution.Repository, error) {
	// Extract the first part of the name (username or DID)
	repoPath := name.Name()
	parts := strings.SplitN(repoPath, "/", 2)

	if len(parts) < 2 {
		// Must be a coded error: distribution's type switch in handlers/app.go
		// drops anything that is not an errcode.Error and serves a bodyless
		// 500, which clients treat as retryable. A name with no owner
		// component is malformed input, so NAME_INVALID is the right answer.
		return nil, errcode.Error{
			Code:    v2.ErrorCodeNameInvalid,
			Message: fmt.Sprintf("repository name must include an owner: %s", repoPath),
		}
	}

	identityStr := parts[0]
	imageName := parts[1]

	// Support hyphen-encoded DIDs in image paths (e.g., did-plc-abc123/repo:tag)
	// OCI reference grammar doesn't allow colons in path components, so DIDs must
	// be encoded with hyphens instead: did:plc:abc123 → did-plc-abc123
	if decoded, ok := auth.DecodeDIDFromHyphens(identityStr); ok {
		identityStr = decoded
	}

	// Resolve identity to DID, handle, and PDS endpoint.
	//
	// A bare error here reaches distribution as ErrorCodeUnknown and becomes a
	// 500. That is wrong on its own terms — an unresolvable name is a bad request,
	// not a server fault — and anonymous pull makes it reachable by any
	// unauthenticated caller, so `/v2/<garbage>/x/manifests/y` would answer 500 to
	// the internet and send clients that retry 5xx into a retry loop. NAME_UNKNOWN
	// is the OCI-correct answer and terminates the client immediately.
	did, handle, pdsEndpoint, err := atproto.ResolveIdentity(ctx, identityStr)
	if err != nil {
		slog.Debug("Identity resolution failed",
			"component", "registry/middleware", "identity", identityStr, "error", err)
		return nil, errcode.Error{
			Code:    v2.ErrorCodeNameUnknown,
			Message: fmt.Sprintf("repository name not known to registry: %s", identityStr),
		}
	}

	slog.Debug("Resolved identity", "component", "registry/middleware", "did", did, "pds", pdsEndpoint, "handle", handle)

	// Check for takedown labels before proceeding
	if globalLabelChecker != nil {
		if taken, _ := globalLabelChecker.IsTakenDown(did, imageName); taken {
			return nil, errcode.Error{
				Code:    errcode.ErrorCodeDenied,
				Message: "this repository has been removed due to a policy violation",
			}
		}
	}

	// Query for hold DID - either user's hold or default hold service
	// Also returns the cached profile preferences (e.g. AutoRemoveUntagged)
	holdDID, prefs := nr.findHoldDIDAndPrefs(ctx, did, handle, pdsEndpoint)
	if holdDID == "" {
		// A fatal configuration error: the registry cannot function without a
		// hold service, so a 5xx is honest here. It still has to be a coded
		// error, or distribution serves it as an empty {} body with no clue
		// for the operator reading the client's output.
		slog.Error("No hold DID configured",
			"component", "registry/middleware", "ownerDID", did)
		return nil, noHoldConfiguredError()
	}

	// Single-hop hold migration: check if this hold has declared a successor
	holdDID = nr.resolveSuccessor(ctx, holdDID)

	// Resolve hold DID to HTTP URL via identity directory (cached 24h)
	holdURL, err := atproto.ResolveHoldURL(ctx, holdDID)
	if err != nil {
		return nil, holdResolutionError(holdDID, err)
	}

	// Crew reconciliation moved to the auth-phase push gate
	// (pkg/appview/authgate). The JWT carries authorization for its short
	// lifetime, so /v2/* doesn't need to re-check membership per blob. The
	// hold-side requireBlobWriteAccess middleware remains as the eventual
	// stopgap if reconciliation hasn't propagated yet.

	// Get service token for hold authentication (only if authenticated)
	// Use validation cache to prevent concurrent requests from racing on OAuth/DPoP
	// Route based on auth method from JWT token
	// IMPORTANT: Use PULLER's DID/PDS for service token, not owner's!
	// The puller (authenticated user) needs to authenticate to the hold service.
	var serviceToken string
	authMethod, _ := ctx.Value(authMethodKey).(string)
	pullerDID, _ := ctx.Value(pullerDIDKey).(string)
	hasPushScope, _ := ctx.Value(hasPushScopeKey).(bool)
	var pullerPDSEndpoint string

	// Only fetch service token if user is authenticated
	// Unauthenticated requests (like /v2/ ping) should not trigger token fetching
	if authMethod != "" && pullerDID != "" {
		// Resolve puller's PDS endpoint for service token request
		_, _, pullerPDSEndpoint, err = atproto.ResolveIdentity(ctx, pullerDID)
		if err != nil {
			slog.Warn("Failed to resolve puller's PDS, falling back to anonymous access",
				"component", "registry/middleware",
				"pullerDID", pullerDID,
				"error", err)
			// Continue without service token - hold will decide if anonymous access is allowed
		} else {
			// Create cache key: "pullerDID:holdDID"
			cacheKey := fmt.Sprintf("%s:%s", pullerDID, holdDID)

			// Fetch service token through validation cache
			// This ensures only ONE request per pullerDID:holdDID pair fetches the token
			// Concurrent requests will wait for the first request to complete
			var fetchErr error
			serviceToken, fetchErr = nr.validationCache.getOrFetch(ctx, cacheKey, func() (string, error) {
				if authMethod == token.AuthMethodAppPassword {
					// App-password flow: use Bearer token authentication
					slog.Debug("Using app-password flow for service token",
						"component", "registry/middleware",
						"pullerDID", pullerDID,
						"cacheKey", cacheKey)

					token, err := auth.GetOrFetchServiceTokenWithAppPassword(ctx, pullerDID, holdDID, pullerPDSEndpoint)
					if err != nil {
						slog.Error("Failed to get service token with app-password",
							"component", "registry/middleware",
							"pullerDID", pullerDID,
							"holdDID", holdDID,
							"pullerPDSEndpoint", pullerPDSEndpoint,
							"denial_reason", "service_token_app_password_failed",
							"error", err)
						return "", err
					}
					return token, nil
				} else if nr.refresher != nil {
					// OAuth flow: use DPoP authentication
					slog.Debug("Using OAuth flow for service token",
						"component", "registry/middleware",
						"pullerDID", pullerDID,
						"cacheKey", cacheKey)

					token, err := auth.GetOrFetchServiceToken(ctx, nr.refresher, pullerDID, holdDID, pullerPDSEndpoint)
					if err != nil {
						slog.Error("Failed to get service token with OAuth",
							"component", "registry/middleware",
							"pullerDID", pullerDID,
							"holdDID", holdDID,
							"pullerPDSEndpoint", pullerPDSEndpoint,
							"denial_reason", "service_token_oauth_failed",
							"error", err)
						return "", err
					}
					return token, nil
				}
				return "", fmt.Errorf("no authentication method available")
			})

			// Handle errors from cached fetch
			if fetchErr != nil {
				errMsg := fetchErr.Error()

				// Check for app-password specific errors
				if authMethod == token.AuthMethodAppPassword {
					if strings.Contains(errMsg, "expired or invalid") || strings.Contains(errMsg, "no app-password") {
						return nil, nr.authErrorMessage("App-password authentication failed. Please re-authenticate with: docker login")
					}
				}

				// Check for OAuth specific errors
				if strings.Contains(errMsg, "OAuth session") || strings.Contains(errMsg, "OAuth validation") {
					return nil, nr.authErrorMessage("OAuth session expired or invalidated by PDS. Your session has been cleared")
				}

				// Generic service token error
				return nil, nr.authErrorMessage(fmt.Sprintf("Failed to obtain storage credentials: %v", fetchErr))
			}
		}
	} else {
		slog.Debug("Skipping service token fetch for unauthenticated request",
			"component", "registry/middleware",
			"ownerDID", did)
	}

	// Create a new reference with identity/image format
	// Use the resolved handle (not raw DID) to ensure the name is valid per OCI reference grammar.
	// DIDs contain colons which are illegal in reference path components.
	// This transforms: did-plc-abc123/myimage -> alice.bsky.social/myimage
	canonicalName := fmt.Sprintf("%s/%s", handle, imageName)
	ref, err := reference.ParseNamed(canonicalName)
	if err != nil {
		return nil, errcode.Error{
			Code:    v2.ErrorCodeNameInvalid,
			Message: fmt.Sprintf("invalid image name: %s", imageName),
		}
	}

	// Delegate to underlying namespace with modified name
	repo, err := nr.Namespace.Repository(ctx, ref)
	if err != nil {
		return nil, err
	}

	// Create ATProto client for manifest/tag operations
	// Pulls: ATProto records are public, no auth needed
	// Pushes: Need auth, but puller must be owner anyway
	var atprotoClient *atproto.Client

	if pullerDID == did {
		// Puller is owner - may need auth for pushes
		if authMethod == token.AuthMethodOAuth && nr.refresher != nil {
			atprotoClient = atproto.NewClientWithSessionProvider(pdsEndpoint, did, nr.refresher)
		} else if authMethod == token.AuthMethodAppPassword {
			accessToken, _ := auth.GetGlobalTokenCache().Get(did)
			atprotoClient = atproto.NewClient(pdsEndpoint, did, accessToken)
		} else {
			atprotoClient = atproto.NewClient(pdsEndpoint, did, "")
		}
	} else {
		// Puller != owner - reads only, no auth needed
		atprotoClient = atproto.NewClient(pdsEndpoint, did, "")
	}

	// IMPORTANT: Use only the image name (not identity/image) for ATProto storage
	// ATProto records are scoped to the user's DID, so we don't need the identity prefix
	// Example: "evan.jarrett.net/debian" -> store as "debian"
	repositoryName := imageName

	// Default auth method to OAuth if not already set (backward compatibility with old tokens)
	if authMethod == "" {
		authMethod = token.AuthMethodOAuth
	}

	// Anonymous request to a private hold: refuse here rather than deeper in
	// the stack.
	//
	// captain.Public is the only thing that admits a reader with no identity.
	// The hold enforces that too, but a denial raised from the blob store
	// cannot reach the client intact: distribution's blobHandler.GetBlob maps
	// everything except ErrBlobUnknown to ErrorCodeUnknown, so a 401 leaves
	// here as a 500 — misreporting an auth failure as a server fault and
	// sending clients that retry 5xx into a loop. An errcode.Error returned
	// from Repository() is passed through verbatim by the registry app, so the
	// client gets a real 401 and BearerChallenge can attach WWW-Authenticate,
	// which is what makes Docker prompt for credentials.
	//
	// Fail open on a lookup error: the hold is the enforcing authority, and a
	// transient failure here should not break anonymous pulls of public
	// images.
	if pullerDID == "" && nr.authorizer != nil && holdDID != "" {
		allowed, authErr := nr.authorizer.CheckReadAccess(ctx, holdDID, "")
		if authErr != nil {
			slog.Warn("Anonymous read check failed, deferring to hold",
				"holdDID", holdDID, "repository", repositoryName, "error", authErr)
		} else if !allowed {
			slog.Debug("Anonymous read denied: hold is not public",
				"holdDID", holdDID, "repository", repositoryName)
			return nil, errcode.ErrorCodeUnauthorized.WithMessage("authentication required")
		}
	}

	// Create routing repository - routes manifests to ATProto, blobs to hold service
	// The registry is stateless - no local storage is used
	// Bundle all context into a single RegistryContext struct
	//
	// NOTE: We create a fresh RoutingRepository on every request (no caching) because:
	// 1. Each layer upload is a separate HTTP request (possibly different process)
	// 2. OAuth sessions can be refreshed/invalidated between requests
	// 3. The refresher already caches sessions efficiently (in-memory + DB)
	// 4. Caching the repository with a stale ATProtoClient causes refresh token errors
	registryCtx := &storage.RegistryContext{
		DID:                did,
		Handle:             handle,
		HoldDID:            holdDID,
		HoldURL:            holdURL,
		PDSEndpoint:        pdsEndpoint,
		Repository:         repositoryName,
		ServiceToken:       serviceToken, // Cached service token from puller's PDS
		ATProtoClient:      atprotoClient,
		AuthMethod:         authMethod,        // Auth method from JWT token
		PullerDID:          pullerDID,         // Authenticated user making the request
		PullerPDSEndpoint:  pullerPDSEndpoint, // Puller's PDS for service token refresh
		HasPushScope:       hasPushScope,      // Whether JWT has push scope (for pull stats filtering)
		Anonymous:          pullerDID == "",   // No puller identity: hold decides via captain.Public
		AutoRemoveUntagged: prefs.AutoRemoveUntagged,
		Database:           nr.database,
		Authorizer:         nr.authorizer,
		Refresher:          nr.refresher,
		ReadmeFetcher:      nr.readmeFetcher,
		WebhookDispatcher:  nr.webhookDispatcher,
		ManifestRefChecker: nr.manifestRefChecker,
	}

	return storage.NewRoutingRepository(repo, registryCtx), nil
}

// Repositories delegates to underlying namespace
func (nr *NamespaceResolver) Repositories(ctx context.Context, repos []string, last string) (int, error) {
	return nr.Namespace.Repositories(ctx, repos, last)
}

// Blobs delegates to underlying namespace
func (nr *NamespaceResolver) Blobs() distribution.BlobEnumerator {
	return nr.Namespace.Blobs()
}

// BlobStatter delegates to underlying namespace
func (nr *NamespaceResolver) BlobStatter() distribution.BlobStatter {
	return nr.Namespace.BlobStatter()
}

// holdPrefs carries the only two sailor profile fields the registry hot path
// reads. The full profile record is not needed here, and fetching it was the
// last per-request PDS round trip on the push path that had nothing to do with
// moving bytes.
type holdPrefs struct {
	// AutoRemoveUntagged is whether a tag overwrite deletes the manifest that
	// lost its last tag.
	AutoRemoveUntagged bool
}

// findHoldDIDAndPrefs determines which hold DID to use for blob storage and
// returns the owner's cached profile preferences.
//
// The answer comes from the local `users` row, which Jetstream keeps current
// (ProcessSailorProfile) and the startup backfill prefills. That row is read
// once per request instead of doing a com.atproto.repo.getRecord against the
// owner's PDS on every HEAD, POST, PATCH, PUT and GET under /v2/, which on a
// ten-layer push was forty-odd round trips.
//
// Priority order is unchanged:
// 1. The user's cached defaultHold (if set)
// 2. AppView's default hold DID
//
// Returns a hold DID (e.g., "did:web:hold01.atcr.io"), or empty string if none
// configured anywhere.
func (nr *NamespaceResolver) findHoldDIDAndPrefs(ctx context.Context, did, handle, pdsEndpoint string) (string, holdPrefs) {
	if nr.userPrefs != nil {
		prefs, err := nr.userPrefs.GetUserHoldPrefs(did)
		if err != nil {
			slog.Warn("Failed to read cached hold preferences, falling back to a live profile fetch",
				"component", "registry/middleware", "did", did, "error", err)
		} else if prefs.Found && prefs.AutoRemoveUntagged.Valid {
			// Both fields are known locally. The cached defaultHold was already
			// normalized to a DID by whoever wrote it, so no URL-to-DID
			// migration is needed on this path.
			return nr.applyTestModeFallback(ctx, prefs.DefaultHoldDID),
				holdPrefs{AutoRemoveUntagged: prefs.AutoRemoveUntagged.Bool}
		}
	}

	// The row is missing, or auto_remove_untagged is still NULL ("never
	// learned"). Learn it from the PDS once and write it down.
	return nr.learnHoldPrefs(ctx, did, handle, pdsEndpoint)
}

// learnHoldPrefs fetches the sailor profile live, exactly once per user, and
// caches what it finds so no later request has to.
//
// This one mechanism covers every way the local cache can have nothing to say:
// the minutes after a deploy while the startup backfill fills a new column, a
// brand-new user with no row at all, and a user the backfill has not reached.
func (nr *NamespaceResolver) learnHoldPrefs(ctx context.Context, did, handle, pdsEndpoint string) (string, holdPrefs) {
	// Unauthenticated client: the sailor profile is a public record.
	client := atproto.NewClient(pdsEndpoint, did, "")

	profile, err := storage.GetProfile(ctx, client)
	if err != nil {
		// A network failure is not an answer. Serve this request from the
		// appview default and write nothing, so a transient error does not get
		// cached as the user's preference and silence the fallback forever.
		slog.Warn("Failed to read profile, using default hold for this request",
			"component", "registry/middleware", "did", did, "error", err)
		return nr.defaultHoldDID, holdPrefs{}
	}

	// A missing profile (404) is an answer: no custom hold, no auto-remove.
	// It is cached like any other, so the fallback does not repeat for a user
	// who has never written a profile record.
	holdDID, autoRemove := "", false
	if profile != nil {
		holdDID = profile.DefaultHold
		autoRemove = profile.AutoRemoveUntagged
	}

	slog.Debug("Sailor profile not cached locally, fetched live and caching",
		"component", "registry/middleware", "did", did,
		"default_hold", holdDID, "auto_remove_untagged", autoRemove,
		"profile_exists", profile != nil)

	if nr.userPrefs != nil {
		if err := nr.userPrefs.CacheUserHoldPrefs(did, handle, pdsEndpoint, holdDID, autoRemove); err != nil {
			slog.Warn("Failed to cache hold preferences; the live fetch will repeat",
				"component", "registry/middleware", "did", did, "error", err)
		}
	}

	return nr.applyTestModeFallback(ctx, holdDID), holdPrefs{AutoRemoveUntagged: autoRemove}
}

// applyTestModeFallback turns a user's chosen hold into the hold to actually
// use. An empty choice means the appview default. With fallbackUnreachable set
// (testmode builds) a chosen hold that is not answering also falls back, so a
// developer whose local hold is down can still push.
func (nr *NamespaceResolver) applyTestModeFallback(ctx context.Context, userHoldDID string) string {
	if userHoldDID == "" {
		return nr.defaultHoldDID
	}
	if nr.fallbackUnreachable && !nr.isHoldReachable(ctx, userHoldDID) {
		slog.Debug("User's defaultHold unreachable, falling back to default",
			"component", "registry/middleware/testmode", "default_hold", userHoldDID)
		return nr.defaultHoldDID
	}
	return userHoldDID
}

// resolveSuccessor checks if a hold has declared a successor and returns it.
// Single-hop only — does not follow chains. Returns the original holdDID if
// no successor is set or if the captain record can't be fetched.
func (nr *NamespaceResolver) resolveSuccessor(ctx context.Context, holdDID string) string {
	if nr.authorizer == nil {
		return holdDID
	}
	captain, err := nr.authorizer.GetCaptainRecord(ctx, holdDID)
	if err != nil {
		return holdDID
	}
	if captain != nil && captain.Successor != "" {
		slog.Info("Hold successor redirect",
			"component", "registry/middleware",
			"from", holdDID,
			"to", captain.Successor)
		return captain.Successor
	}
	return holdDID
}

// isHoldReachable checks if a hold service is reachable
// Used in test mode to fallback to default hold when user's hold is unavailable
func (nr *NamespaceResolver) isHoldReachable(ctx context.Context, holdDID string) bool {
	holdURL, err := atproto.ResolveHoldURL(ctx, holdDID)
	if err != nil {
		slog.Debug("Cannot resolve hold URL for reachability check", "component", "registry/middleware", "holdDID", holdDID, "error", err)
		return false
	}

	testURL := holdURL + "/.well-known/did.json"
	client := atproto.NewClient("", "", "")
	_, err = client.FetchDIDDocument(ctx, testURL)
	return err == nil
}

// ExtractAuthMethod is an HTTP middleware that extracts the auth method and puller DID from the JWT Authorization header
// and stores them in the request context for later use by the registry middleware.
// Also stores the HTTP method for routing decisions (GET/HEAD = pull, PUT/POST = push).
func ExtractAuthMethod(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()

		// Store HTTP method in context for routing decisions
		// This is used by routing_repository.go to distinguish pull (GET/HEAD) from push (PUT/POST)
		ctx = context.WithValue(ctx, storage.HTTPRequestMethod, r.Method)

		// Extract Authorization header
		authHeader := r.Header.Get("Authorization")
		if authHeader != "" {
			// Parse "Bearer <token>" format
			parts := strings.SplitN(authHeader, " ", 2)
			if len(parts) == 2 && strings.ToLower(parts[0]) == "bearer" {
				tokenString := parts[1]

				// Extract auth method from JWT (does not validate - just parses)
				authMethod := token.ExtractAuthMethod(tokenString)
				if authMethod != "" {
					// Store in context for registry middleware
					ctx = context.WithValue(ctx, authMethodKey, authMethod)
				}

				// Extract puller DID (Subject) from JWT
				// This is the authenticated user's DID, used for service token requests
				pullerDID := token.ExtractSubject(tokenString)
				if pullerDID != "" {
					ctx = context.WithValue(ctx, pullerDIDKey, pullerDID)
				}

				// Extract access scopes from JWT to detect push-scoped tokens
				// Used to distinguish real pulls from manifest GETs during push/imagetools flows
				access := token.ExtractAccess(tokenString)
				if token.HasPushScope(access) {
					ctx = context.WithValue(ctx, hasPushScopeKey, true)
				}

				slog.Debug("Extracted auth info from JWT",
					"component", "registry/middleware",
					"authMethod", authMethod,
					"pullerDID", pullerDID,
					"hasPushScope", token.HasPushScope(access),
					"httpMethod", r.Method)
			}
		}

		r = r.WithContext(ctx)
		next.ServeHTTP(w, r)
	})
}

// retryAfterResponseWriter wraps http.ResponseWriter and, on the first
// WriteHeader call, injects a Retry-After header if the status is 429 and
// a retry-after duration was recorded in the request context.
type retryAfterResponseWriter struct {
	http.ResponseWriter
	carrier     *storage.RetryAfterCarrier
	wroteHeader bool
}

func (w *retryAfterResponseWriter) WriteHeader(code int) {
	if !w.wroteHeader {
		w.wroteHeader = true
		if code == http.StatusTooManyRequests {
			if d := w.carrier.Duration(); d > 0 {
				// Round up to whole seconds; minimum of 1 to avoid 0-second hints.
				secs := int64(d.Seconds())
				if d%time.Second != 0 {
					secs++
				}
				if secs < 1 {
					secs = 1
				}
				w.Header().Set("Retry-After", fmt.Sprintf("%d", secs))
			}
		}
	}
	w.ResponseWriter.WriteHeader(code)
}

func (w *retryAfterResponseWriter) Write(b []byte) (int, error) {
	if !w.wroteHeader {
		// Implicit 200 — still fire WriteHeader so flag flips.
		w.WriteHeader(http.StatusOK)
	}
	return w.ResponseWriter.Write(b)
}

// RetryAfterMiddleware installs a per-request RetryAfterCarrier in the
// request context and wraps the response writer so deeper handlers (e.g.,
// the manifest store, when an upstream PDS returns 429) can cause a
// Retry-After header to be emitted on 429 responses.
func RetryAfterMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		carrier := storage.NewRetryAfterCarrier()
		ctx := context.WithValue(r.Context(), storage.RetryAfterContextKey, carrier)
		wrapped := &retryAfterResponseWriter{ResponseWriter: w, carrier: carrier}
		next.ServeHTTP(wrapped, r.WithContext(ctx))
	})
}