package server

import (
	"context"
	"crypto/subtle"
	"errors"
	"net/http"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/pathutil"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/tenant/token"
	"github.com/mem9-ai/drive9/pkg/tenantctx"
	"github.com/mem9-ai/drive9/pkg/vault"
	"go.uber.org/zap"
)

type scopeKey int

const tenantScopeKey scopeKey = iota

type TenantScope struct {
	TenantID           string
	APIKeyID           string
	TokenVersion       int
	Provider           string
	TiDBCloudOrgID     string
	Backend            *backend.Dat9Backend
	JournalPermissions map[string]bool
	ScopeKind          meta.APIKeyScopeKind
	IsScoped           bool
	FSScopes           []FSScope
	FSPseudoRoots      []string
}

const (
	JournalPermissionCreate        = "journal:create"
	JournalPermissionAppend        = "journal:append"
	JournalPermissionRead          = "journal:read"
	JournalPermissionFind          = "journal:find"
	JournalPermissionVerify        = "journal:verify"
	JournalPermissionSeal          = "journal:seal"
	JournalPermissionArtifactWrite = "journal:artifact:write"
	JournalPermissionAdmin         = "journal:admin"
	JournalPermissionSourceGateway = "journal:source:gateway_observed"
	JournalPermissionSourceServer  = "journal:source:server_observed"
	JournalPermissionSourceImport  = "journal:source:imported"
)

const statusClientClosedRequest = 499

func isQuotaError(err error) bool {
	if err == nil {
		return false
	}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) && mysqlErr.Number == 1105 {
		return strings.Contains(strings.ToLower(mysqlErr.Message), "quota")
	}
	msg := strings.ToLower(err.Error())
	idx := strings.Index(msg, "error 1105")
	return idx >= 0 && strings.Contains(msg[idx:], "quota")
}

func isSchemaMigrationError(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "ensure tidb auto-embedding schema") ||
		strings.Contains(msg, "validate tidb auto-embedding schema") ||
		strings.Contains(msg, "migrate split tables") ||
		strings.Contains(msg, "detect legacy files table") ||
		strings.Contains(msg, "meta schema contract mismatch") ||
		strings.Contains(msg, "tidb schema contract mismatch")
}

func sanitizeClientError(err error) string {
	if err == nil {
		return "backend unavailable"
	}
	if errors.Is(err, datastore.ErrDirectoryNotEmpty) {
		return err.Error()
	}
	if errors.Is(err, pathutil.ErrInvalidPath) {
		// Client-supplied path failed validation; surface the reason so the
		// caller can fix the request (e.g. "path contains backslash").
		return err.Error()
	}
	if errors.Is(err, datastore.ErrDeleteIncomplete) {
		return "recursive delete in progress, retry to resume"
	}
	if isQuotaError(err) {
		return "tenant usage quota exceeded"
	}
	if isSchemaMigrationError(err) {
		return "tenant metadata error"
	}
	return "backend unavailable"
}

func backendErrorStatus(ctx context.Context, err error) int {
	if isClientCanceled(ctx, err) {
		return statusClientClosedRequest
	}
	if errors.Is(err, pathutil.ErrInvalidPath) {
		return http.StatusBadRequest
	}
	if errors.Is(err, backend.ErrAppendLogUnsupported) {
		return http.StatusBadRequest
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return http.StatusGatewayTimeout
	}
	if errors.Is(err, datastore.ErrDeleteIncomplete) {
		return http.StatusServiceUnavailable
	}
	if errors.Is(err, datastore.ErrDirectoryNotEmpty) {
		return http.StatusConflict
	}
	if errors.Is(err, datastore.ErrContentLayoutConflict) {
		return http.StatusConflict
	}
	if isQuotaError(err) {
		return http.StatusPaymentRequired
	}
	if isSchemaMigrationError(err) {
		return http.StatusBadGateway
	}
	return http.StatusInternalServerError
}

func writeBackendError(w http.ResponseWriter, r *http.Request, err error) {
	if errors.Is(err, backend.ErrAppendLogUnsupported) {
		errJSONCode(w, backendErrorStatus(r.Context(), err), "append_log_unsupported", backend.ErrAppendLogUnsupported.Error())
		return
	}
	if errors.Is(err, datastore.ErrDeleteIncomplete) {
		// A bounded recursive delete ran out of budget; it is resumable, so tell
		// the client when to re-issue the same request to continue the sweep.
		w.Header().Set("Retry-After", "2")
	}
	if errors.Is(err, pathutil.ErrInvalidPath) {
		// 400s stay out of 5xx alerts by design; keep a dedicated tenant-
		// attributed event so invalid-path storms remain perceivable.
		metricEvent(r.Context(), "fs_invalid_path")
	}
	errJSON(w, backendErrorStatus(r.Context(), err), sanitizeClientError(err))
}

// logBackendError logs a failed backend operation with the level derived
// from the same classification that picks the HTTP status: client-caused
// failures (any sub-500 mapping) log at Warn, cancellations at Info, and
// only genuine server-side faults at Error — so Error-level *_failed log
// alerting means "server fault". Handlers call this instead of growing
// per-endpoint invalid-path branches. The "error" field is appended here.
func logBackendError(ctx context.Context, event string, err error, kv ...any) {
	kv = append(kv, "error", err)
	switch status := backendErrorStatus(ctx, err); {
	case status >= http.StatusInternalServerError:
		logger.Error(ctx, "server_event", eventFields(ctx, event, kv...)...)
	case status == statusClientClosedRequest:
		logger.Info(ctx, "server_event", eventFields(ctx, event, kv...)...)
	default:
		logger.Warn(ctx, "server_event", eventFields(ctx, event, kv...)...)
	}
}

func ScopeFromContext(ctx context.Context) *TenantScope {
	s, _ := ctx.Value(tenantScopeKey).(*TenantScope)
	return s
}

func (s *TenantScope) HasJournalPermission(permission string) bool {
	return s != nil && (s.JournalPermissions[permission] || s.JournalPermissions[JournalPermissionAdmin])
}

func ownerJournalPermissions() map[string]bool {
	perms := make(map[string]bool, len(ownerJournalPermissionList()))
	for _, permission := range ownerJournalPermissionList() {
		perms[permission] = true
	}
	return perms
}

func ownerJournalPermissionList() []string {
	return []string{
		JournalPermissionCreate,
		JournalPermissionAppend,
		JournalPermissionRead,
		JournalPermissionFind,
		JournalPermissionVerify,
		JournalPermissionSeal,
		JournalPermissionArtifactWrite,
		JournalPermissionAdmin,
		JournalPermissionSourceGateway,
		JournalPermissionSourceServer,
		JournalPermissionSourceImport,
	}
}

func journalPermissionsFromClaims(claims *token.Claims) map[string]bool {
	if claims == nil || len(claims.JournalPermissions) == 0 {
		return ownerJournalPermissions()
	}
	perms := make(map[string]bool, len(claims.JournalPermissions))
	for _, permission := range claims.JournalPermissions {
		permission = strings.TrimSpace(permission)
		if permission == "" {
			continue
		}
		perms[permission] = true
	}
	return perms
}

func journalSourcePermission(source string) string {
	switch source {
	case "gateway_observed":
		return JournalPermissionSourceGateway
	case "server_observed":
		return JournalPermissionSourceServer
	case "imported":
		return JournalPermissionSourceImport
	default:
		return ""
	}
}

func withScope(ctx context.Context, scope *TenantScope) context.Context {
	ctx = context.WithValue(ctx, tenantScopeKey, scope)
	if scope == nil {
		return ctx
	}
	return tenantctx.WithTenantID(ctx, scope.TenantID)
}

func authPhaseMs(start time.Time) float64 {
	return durationMs(time.Since(start))
}

func durationMs(d time.Duration) float64 {
	return float64(d.Microseconds()) / 1000.0
}

func isClientCanceled(ctx context.Context, err error) bool {
	return errors.Is(err, context.Canceled) || errors.Is(ctx.Err(), context.Canceled)
}

func logAuthClientCanceled(ctx context.Context, event string, kv ...any) {
	logger.Info(ctx, "server_event", eventFields(ctx, event, kv...)...)
	metricEvent(ctx, "auth", "result", "client_canceled")
}

func writeClientCanceled(w http.ResponseWriter) {
	w.WriteHeader(statusClientClosedRequest)
}

func tenantAuthMiddleware(metaStore *meta.Store, pool *tenant.Pool, events *eventBuses, tokenSecret []byte, next http.Handler) http.Handler {
	return tenantAuthMiddlewareWithFSScopeLoader(metaStore, pool, events, tokenSecret, metaStore, next)
}

type tenantBackendAcquirer func(context.Context, *meta.Tenant) (*backend.Dat9Backend, func(), error)

type tenantFSScopeLoader interface {
	ListAPIKeyFSScopes(ctx context.Context, tenantID, apiKeyID string) ([]meta.APIKeyFSScope, error)
}

func tenantAuthMiddlewareWithFSScopeLoader(metaStore *meta.Store, pool *tenant.Pool, events *eventBuses, tokenSecret []byte, fsScopeLoader tenantFSScopeLoader, next http.Handler) http.Handler {
	return tenantAuthMiddlewareWithFSScopeLoaderAndAcquirer(metaStore, pool, events, tokenSecret, fsScopeLoader, pool.Acquire, next)
}

func tenantAuthMiddlewareWithFSScopeLoaderAndAcquirer(metaStore *meta.Store, pool *tenant.Pool, events *eventBuses, tokenSecret []byte, fsScopeLoader tenantFSScopeLoader, acquire tenantBackendAcquirer, next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		authStart := time.Now()
		tok := bearerToken(r)
		if tok == "" {
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_missing_token")...)
			metricEvent(r.Context(), "auth", "result", "missing_token")
			errJSON(w, http.StatusUnauthorized, "missing or malformed Authorization header")
			return
		}

		resolveStart := time.Now()
		resolved, err := metaStore.ResolveByAPIKeyHash(r.Context(), token.HashToken(tok))
		resolveDurationMs := authPhaseMs(resolveStart)
		if err != nil {
			if errors.Is(err, meta.ErrNotFound) {
				logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_key_not_found")...)
				metricEvent(r.Context(), "auth", "result", "key_not_found")
				errJSON(w, http.StatusUnauthorized, "invalid API key")
				return
			}
			if isClientCanceled(r.Context(), err) {
				logAuthClientCanceled(r.Context(), "auth_client_canceled")
				writeClientCanceled(w)
				return
			}
			logger.Error(r.Context(), "server_event", eventFields(r.Context(), "auth_backend_unavailable", "error", err)...)
			metricEvent(r.Context(), "auth", "result", "meta_backend_error")
			errJSON(w, backendErrorStatus(r.Context(), err), "auth backend unavailable")
			return
		}

		if subtle.ConstantTimeCompare([]byte(token.HashToken(tok)), []byte(resolved.APIKey.JWTHash)) != 1 {
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_hash_mismatch", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID)...)
			metricEvent(r.Context(), "auth", "result", "hash_mismatch")
			errJSON(w, http.StatusUnauthorized, "invalid API key")
			return
		}
		if resolved.APIKey.Status != meta.APIKeyActive {
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_key_inactive", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "status", resolved.APIKey.Status)...)
			metricEvent(r.Context(), "auth", "result", "key_inactive")
			errJSON(w, http.StatusUnauthorized, "invalid API key")
			return
		}

		decryptStart := time.Now()
		plain, err := poolDecryptToken(r.Context(), pool, resolved.APIKey.JWTCiphertext)
		decryptDurationMs := authPhaseMs(decryptStart)
		if err != nil {
			if isClientCanceled(r.Context(), err) {
				logAuthClientCanceled(r.Context(), "auth_decrypt_client_canceled", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID)
				writeClientCanceled(w)
				return
			}
			logger.Error(r.Context(), "server_event", eventFields(r.Context(), "auth_decrypt_failed", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "error", err)...)
			metricEvent(r.Context(), "auth", "result", "decrypt_failed")
			errJSON(w, backendErrorStatus(r.Context(), err), "auth backend unavailable")
			return
		}
		if subtle.ConstantTimeCompare([]byte(tok), plain) != 1 {
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_token_cipher_mismatch", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID)...)
			metricEvent(r.Context(), "auth", "result", "cipher_mismatch")
			errJSON(w, http.StatusUnauthorized, "invalid API key")
			return
		}

		verifyStart := time.Now()
		claims, err := token.ParseAndVerifyToken(tokenSecret, tok)
		verifyDurationMs := authPhaseMs(verifyStart)
		if err != nil {
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_token_invalid", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "error", err)...)
			metricEvent(r.Context(), "auth", "result", "token_invalid")
			errJSON(w, http.StatusUnauthorized, "invalid API key")
			return
		}
		if claims.TenantID != resolved.Tenant.ID || claims.TokenVersion != resolved.APIKey.TokenVersion {
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_claims_mismatch", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "claim_tenant", claims.TenantID, "claim_version", claims.TokenVersion)...)
			metricEvent(r.Context(), "auth", "result", "claims_mismatch")
			errJSON(w, http.StatusUnauthorized, "invalid API key")
			return
		}

		if isTenantDeleteRequest(r) || isForkDeleteRequest(r) {
			handleDeleteNoAcquireAuth(w, r, pool, next, resolved, claims)
			return
		}

		switch resolved.Tenant.Status {
		case meta.TenantActive:
		case meta.TenantPending:
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_pending", "tenant_id", resolved.Tenant.ID)...)
			metricEvent(r.Context(), "tenant_status", "status", string(meta.TenantPending))
			errJSONRetryable(w, "tenant provisioning is pending")
			return
		case meta.TenantProvisioning:
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_provisioning", "tenant_id", resolved.Tenant.ID)...)
			metricEvent(r.Context(), "tenant_status", "status", string(meta.TenantProvisioning))
			errJSONRetryable(w, "tenant is provisioning")
			return
		case meta.TenantFailed:
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_failed", "tenant_id", resolved.Tenant.ID)...)
			metricEvent(r.Context(), "tenant_status", "status", string(meta.TenantFailed))
			errJSON(w, http.StatusForbidden, "tenant provisioning failed")
			return
		case meta.TenantSuspended, meta.TenantDeleting, meta.TenantDeleted:
			pool.Invalidate(resolved.Tenant.ID)
			// Tear down this pod's leftover SSE streams for the tenant:
			// closeBus is per-pod, so the delete path cannot reach streams
			// established on peer pods — any later request landing here does
			// it instead.
			events.closeBus(r.Context(), resolved.Tenant.ID)
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_blocked", "tenant_id", resolved.Tenant.ID, "status", resolved.Tenant.Status)...)
			metricEvent(r.Context(), "tenant_status", "status", string(resolved.Tenant.Status))
			errJSON(w, http.StatusForbidden, "tenant is unavailable")
			return
		default:
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_unavailable", "tenant_id", resolved.Tenant.ID, "status", resolved.Tenant.Status)...)
			metricEvent(r.Context(), "tenant_status", "status", string(resolved.Tenant.Status))
			errJSON(w, http.StatusForbidden, "tenant is unavailable")
			return
		}
		scopeKind := resolved.APIKey.ScopeKind
		if scopeKind == "" {
			scopeKind = meta.APIKeyScopeKindOwner
		}
		isScoped := false
		var fsScopes []FSScope
		var fsPseudoRoots []string
		switch scopeKind {
		case meta.APIKeyScopeKindOwner:
		case meta.APIKeyScopeKindFS:
			isScoped = true
			rows, err := fsScopeLoader.ListAPIKeyFSScopes(r.Context(), resolved.Tenant.ID, resolved.APIKey.ID)
			if err != nil {
				if isClientCanceled(r.Context(), err) {
					logAuthClientCanceled(r.Context(), "auth_fs_scope_load_client_canceled", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID)
					writeClientCanceled(w)
					return
				}
				logger.Error(r.Context(), "server_event", eventFields(r.Context(), "auth_fs_scope_load_failed", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "error", err)...)
				metricEvent(r.Context(), "auth", "result", "fs_scope_load_failed")
				errJSON(w, backendErrorStatus(r.Context(), err), "auth backend unavailable")
				return
			}
			policy, err := fsPolicyFromMeta(rows)
			if err != nil {
				logger.Error(r.Context(), "server_event", eventFields(r.Context(), "auth_fs_scope_invalid", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "error", err)...)
				metricEvent(r.Context(), "auth", "result", "fs_scope_invalid")
				errJSON(w, backendErrorStatus(r.Context(), err), "invalid API key scope policy")
				return
			}
			fsScopes = policy.Scopes
			fsPseudoRoots = policy.PseudoRoots
		default:
			logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_scope_kind_unsupported", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "scope_kind", scopeKind)...)
			metricEvent(r.Context(), "auth", "result", "scope_kind_unsupported")
			errJSON(w, http.StatusForbidden, "unsupported API key scope kind")
			return
		}

		acquireStart := time.Now()
		b, release, err := acquire(r.Context(), &resolved.Tenant)
		acquireDurationMs := authPhaseMs(acquireStart)
		// Request-path acquire: this is the expected hot-path user-DB access,
		// driven by real traffic. Record so scan-detection alerts can compare
		// periodic-path rates against this baseline.
		metrics.RecordTenantOperationWithOrg(resolved.Tenant.ID, resolved.TiDBCloudOrgID, "user_db_access", "auth_acquire", metrics.ResultForError(err), time.Since(acquireStart))
		if err != nil {
			if isClientCanceled(r.Context(), err) {
				logAuthClientCanceled(r.Context(), "backend_load_client_canceled", "tenant_id", resolved.Tenant.ID)
				writeClientCanceled(w)
				return
			}
			logger.Error(r.Context(), "server_event", eventFields(r.Context(), "backend_load_failed", "tenant_id", resolved.Tenant.ID, "error", err)...)
			metricEvent(r.Context(), "tenant_backend", "result", "load_failed")
			writeBackendError(w, r, err)
			return
		}
		defer release()
		defer func() { recordTenantHTTPRequestAtCompletion(r.Context()) }()
		metricEvent(r.Context(), "auth", "result", "ok")
		totalDuration := time.Since(authStart)
		logger.InfoOpenPoolTiming(r.Context(), "tenant_auth_timing", totalDuration,
			zap.String("path", r.URL.Path),
			zap.String("method", r.Method),
			zap.String("tenant_id", resolved.Tenant.ID),
			zap.String("api_key_id", resolved.APIKey.ID),
			zap.String("tidbcloud_org_id", b.TiDBCloudOrgID()),
			zap.Float64("resolve_api_key_hash_ms", resolveDurationMs),
			zap.Float64("decrypt_token_ms", decryptDurationMs),
			zap.Float64("verify_token_ms", verifyDurationMs),
			zap.Float64("pool_acquire_ms", acquireDurationMs),
			zap.Float64("total_ms", durationMs(totalDuration)),
		)

		scope := &TenantScope{
			TenantID:           resolved.Tenant.ID,
			APIKeyID:           resolved.APIKey.ID,
			TokenVersion:       resolved.APIKey.TokenVersion,
			Provider:           resolved.Tenant.Provider,
			TiDBCloudOrgID:     b.TiDBCloudOrgID(),
			Backend:            b,
			JournalPermissions: journalPermissionsFromClaims(claims),
			ScopeKind:          scopeKind,
			IsScoped:           isScoped,
			FSScopes:           fsScopes,
			FSPseudoRoots:      fsPseudoRoots,
		}
		setRequestMetricScope(r.Context(), scope, classifyTenantRequest(r))
		next.ServeHTTP(w, r.WithContext(withScope(r.Context(), scope)))
	})
}

func isTenantDeleteRequest(r *http.Request) bool {
	return r.Method == http.MethodDelete && r.URL.Path == "/v1/tenant"
}

func isForkDeleteRequest(r *http.Request) bool {
	return r.Method == http.MethodDelete && r.URL.Path == "/v1/fork"
}

func handleDeleteNoAcquireAuth(w http.ResponseWriter, r *http.Request, pool *tenant.Pool, next http.Handler, resolved *meta.TenantWithAPIKey, claims *token.Claims) {
	switch resolved.Tenant.Status {
	case meta.TenantActive, meta.TenantFailed, meta.TenantDeleting, meta.TenantDeleted:
	case meta.TenantPending:
		logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_pending", "tenant_id", resolved.Tenant.ID)...)
		metricEvent(r.Context(), "tenant_status", "status", string(meta.TenantPending))
		errJSONRetryable(w, "tenant provisioning is pending")
		return
	case meta.TenantProvisioning:
		logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_provisioning", "tenant_id", resolved.Tenant.ID)...)
		metricEvent(r.Context(), "tenant_status", "status", string(meta.TenantProvisioning))
		errJSONRetryable(w, "tenant is provisioning")
		return
	case meta.TenantSuspended:
		pool.Invalidate(resolved.Tenant.ID)
		logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_blocked", "tenant_id", resolved.Tenant.ID, "status", resolved.Tenant.Status)...)
		metricEvent(r.Context(), "tenant_status", "status", string(resolved.Tenant.Status))
		errJSON(w, http.StatusForbidden, "tenant is unavailable")
		return
	default:
		logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "tenant_unavailable", "tenant_id", resolved.Tenant.ID, "status", resolved.Tenant.Status)...)
		metricEvent(r.Context(), "tenant_status", "status", string(resolved.Tenant.Status))
		errJSON(w, http.StatusForbidden, "tenant is unavailable")
		return
	}

	scopeKind := resolved.APIKey.ScopeKind
	if scopeKind == "" {
		scopeKind = meta.APIKeyScopeKindOwner
	}
	isScoped := false
	switch scopeKind {
	case meta.APIKeyScopeKindOwner:
	case meta.APIKeyScopeKindFS:
		isScoped = true
	default:
		logger.Warn(r.Context(), "server_event", eventFields(r.Context(), "auth_scope_kind_unsupported", "tenant_id", resolved.Tenant.ID, "api_key_id", resolved.APIKey.ID, "scope_kind", scopeKind)...)
		metricEvent(r.Context(), "auth", "result", "scope_kind_unsupported")
		errJSON(w, http.StatusForbidden, "unsupported API key scope kind")
		return
	}

	scope := &TenantScope{
		TenantID:           resolved.Tenant.ID,
		APIKeyID:           resolved.APIKey.ID,
		TokenVersion:       resolved.APIKey.TokenVersion,
		Provider:           resolved.Tenant.Provider,
		TiDBCloudOrgID:     resolved.TiDBCloudOrgID,
		JournalPermissions: journalPermissionsFromClaims(claims),
		ScopeKind:          scopeKind,
		IsScoped:           isScoped,
	}
	setRequestMetricTenantForAuthStatus(r.Context(), scope.TenantID, scope.APIKeyID, scope.Provider, scope.TiDBCloudOrgID, resolved.Tenant.Status, classifyTenantRequest(r))
	next.ServeHTTP(w, r.WithContext(withScope(r.Context(), scope)))
}

// capabilityAuthMiddleware returns a handler that resolves the tenant backend
// from a capability token's tenant_id claim. It does NOT do full token verification
// (signature, TTL, revocation) — that is handled by handleVaultRead itself.
// This middleware only exists to load the correct tenant DB so the handler can
// access vault tables.
func (s *Server) capabilityAuthMiddleware(metaStore *meta.Store, pool *tenant.Pool) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		tok := bearerToken(r)
		if tok == "" {
			errJSON(w, http.StatusUnauthorized, "missing capability token")
			return
		}

		// Peek at claims to get tenant_id. We only need the payload, not
		// full HMAC verification (handleVaultRead does that).
		// Support both legacy cap tokens (vault_ prefix) and grant tokens (vt_ prefix).
		tenantID, err := peekTokenTenantID(tok)
		if err != nil {
			errJSON(w, http.StatusUnauthorized, "invalid capability token")
			return
		}

		// Look up tenant and load backend. Use a uniform 401 for all
		// failure modes to avoid leaking tenant existence or status to
		// unauthenticated callers (tenant_id is from unverified peek).
		tenant, err := metaStore.GetTenant(r.Context(), tenantID)
		if err != nil || tenant.Status != meta.TenantActive {
			errJSON(w, http.StatusUnauthorized, "invalid capability token")
			return
		}

		capAcquireStart := time.Now()
		b, release, err := pool.Acquire(r.Context(), tenant)
		// Capability-token request-path acquire: same baseline signal as the
		// API-key path above, but for the /v1/vault/read surface.
		authAcquireOrgID := ""
		if b != nil {
			authAcquireOrgID = b.TiDBCloudOrgID()
		}
		metrics.RecordTenantOperationWithOrg(tenantID, authAcquireOrgID, "user_db_access", "auth_acquire", metrics.ResultForError(err), time.Since(capAcquireStart))
		if err != nil {
			logger.Error(r.Context(), "server_event", eventFields(r.Context(), "capability_backend_load_failed", "tenant_id", tenantID, "error", err)...)
			errJSON(w, backendErrorStatus(r.Context(), err), "backend unavailable")
			return
		}
		defer release()
		defer func() { recordTenantHTTPRequestAtCompletion(r.Context()) }()

		scope := &TenantScope{TenantID: tenantID, Provider: tenant.Provider, TiDBCloudOrgID: b.TiDBCloudOrgID(), Backend: b}
		setRequestMetricScope(r.Context(), scope, classifyTenantRequest(r))
		sub := strings.TrimPrefix(r.URL.Path, "/v1/vault/read")
		s.handleVaultRead(w, r.WithContext(withScope(r.Context(), scope)), sub)
	})
}

// peekTokenTenantID extracts tenant_id from either a legacy cap token (vault_
// prefix) or a grant token (vt_ prefix). Used by capabilityAuthMiddleware for
// pre-auth tenant routing.
func peekTokenTenantID(raw string) (string, error) {
	if strings.HasPrefix(raw, "vt_") {
		return vault.PeekGrantTenantID(raw)
	}
	return vault.PeekCapTokenTenantID(raw)
}

func poolDecryptToken(ctx context.Context, pool *tenant.Pool, cipher []byte) ([]byte, error) {
	// Decrypt is tenant-independent and uses pool encryptor shared for API key storage.
	// Keep this helper to avoid exposing raw encryptor in handlers.
	return pool.Decrypt(ctx, cipher)
}

func bearerToken(r *http.Request) string {
	h := r.Header.Get("Authorization")
	if h == "" {
		h = r.Header.Get("X-API-Key")
		if h != "" {
			return strings.TrimSpace(h)
		}
		return ""
	}
	const prefix = "Bearer "
	if !strings.HasPrefix(h, prefix) {
		return ""
	}
	return strings.TrimSpace(h[len(prefix):])
}