package server

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/mem9-ai/drive9/pkg/innerauth"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/tenant/token"
)

// ControlPlaneScope carries the IAM-resolved control-plane identity for the
// /v1/tokens surface. Tenant is nil for org-wide list; handlers that require an
// explicit tenant_id resolve and authorize it themselves.
type ControlPlaneScope struct {
	Identity       tenant.TiDBCloudAPIKeyIdentity
	Tenant         *meta.Tenant
	OrganizationID string
}

type controlPlaneScopeKeyType int

const controlPlaneScopeKey controlPlaneScopeKeyType = 1

const (
	// maxAPIKeyTTL is the wall-clock ceiling for explicit ttl_seconds on
	// generate/refresh. Compare against maxAPIKeyTTLSeconds (seconds) to
	// avoid int64 Duration overflow for huge client inputs.
	maxAPIKeyTTL        = 365 * 24 * time.Hour
	maxAPIKeyTTLSeconds = int64(maxAPIKeyTTL / time.Second)

	defaultTokenListLimit    = 50
	maxTokenListLimit        = 200
	defaultOrgTokenListLimit = 10
	maxOrgTokenListLimit     = 50
	// maxOrgListKeysPerTenant is the per-tenant enumeration cap for org-wide
	// list (§4.4). Call ListAPIKeys with this+1 to detect truncation.
	maxOrgListKeysPerTenant     = 500
	tokenManagePrincipalOwner   = "owner"
	tokenManagePrincipalControl = "control_plane"
)

// ControlPlaneScopeFromContext returns the control-plane scope injected by the
// tokens credential dispatcher, or nil when the request is Bearer-authenticated.
func ControlPlaneScopeFromContext(ctx context.Context) *ControlPlaneScope {
	s, _ := ctx.Value(controlPlaneScopeKey).(*ControlPlaneScope)
	return s
}

func withControlPlaneScope(ctx context.Context, scope *ControlPlaneScope) context.Context {
	return context.WithValue(ctx, controlPlaneScopeKey, scope)
}

// tokensCredentialDispatcher classifies /v1/tokens requests by credential shape
// and routes them through either tenantAuthMiddleware (Bearer) or
// controlPlaneTokenAuth (TiDB Cloud public/private key). Both paths converge on next.
func (s *Server) tokensCredentialDispatcher(cfg Config, next http.Handler) http.Handler {
	bearer := tenantAuthMiddleware(cfg.Meta, cfg.Pool, s.events, cfg.TokenSecret, next)
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if subject := InnerAuthSubjectFromContext(r.Context()); subject != nil {
			organizationID := strconv.FormatUint(subject.OrganizationID, 10)
			scope := &ControlPlaneScope{
				Identity: tenant.TiDBCloudAPIKeyIdentity{
					OrganizationID: organizationID,
				},
				OrganizationID: organizationID,
			}
			next.ServeHTTP(w, r.WithContext(withControlPlaneScope(r.Context(), scope)))
			return
		}
		hasControl, peekErr := peekTokenCredentials(w, r)
		if peekErr != nil {
			writeControlPlaneStatusError(w, http.StatusBadRequest, peekErr.Error())
			return
		}
		hasBearer := bearerToken(r) != ""
		switch {
		case hasBearer && hasControl:
			writeControlPlaneStatusError(w, http.StatusBadRequest, "ambiguous credentials")
			return
		case hasBearer:
			bearer.ServeHTTP(w, r)
			return
		case hasControl:
			cps, ok := s.controlPlaneTokenAuth(r.Context(), w, r)
			if !ok {
				return
			}
			next.ServeHTTP(w, r.WithContext(withControlPlaneScope(r.Context(), cps)))
			return
		default:
			writeControlPlaneStatusError(w, http.StatusUnauthorized, "missing or malformed Authorization header")
		}
	})
}

// peekTokenCredentials detects control-plane credentials without consuming the
// body permanently. Header-first: body sniffing is skipped when headers match.
// Malformed JSON does not force the control-plane branch.
func peekTokenCredentials(w http.ResponseWriter, r *http.Request) (hasControlCred bool, err error) {
	headerControl := strings.TrimSpace(r.Header.Get(quotaPublicKeyHeader)) != "" ||
		strings.TrimSpace(r.Header.Get(quotaPrivateKeyHeader)) != ""
	if headerControl {
		return true, nil
	}

	switch r.Method {
	case http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete:
	default:
		return false, nil
	}
	if r.Body == nil {
		return false, nil
	}

	limited := http.MaxBytesReader(w, r.Body, maxCredentialProvisionBodyBytes)
	buf, readErr := io.ReadAll(limited)
	// Always restore a readable body for downstream handlers.
	r.Body = io.NopCloser(bytes.NewReader(buf))
	if readErr != nil {
		var maxBytesErr *http.MaxBytesError
		if errors.As(readErr, &maxBytesErr) || errors.Is(readErr, io.ErrUnexpectedEOF) {
			return false, fmt.Errorf("request body too large")
		}
		// MaxBytesReader may wrap the oversize error as a plain error string.
		if strings.Contains(strings.ToLower(readErr.Error()), "request body too large") ||
			strings.Contains(strings.ToLower(readErr.Error()), "http: request body too large") {
			return false, fmt.Errorf("request body too large")
		}
		return false, nil
	}
	if int64(len(buf)) > maxCredentialProvisionBodyBytes {
		// Beyond the MaxBytesReader ceiling — treat as oversize.
		return false, fmt.Errorf("request body too large")
	}
	if len(bytes.TrimSpace(buf)) == 0 {
		return false, nil
	}

	var raw struct {
		PublicKey  string `json:"public_key"`
		PrivateKey string `json:"private_key"`
	}
	if jsonErr := json.Unmarshal(buf, &raw); jsonErr != nil {
		// Lenient: body may be Bearer request content; not an error.
		return false, nil
	}
	if strings.TrimSpace(raw.PublicKey) != "" || strings.TrimSpace(raw.PrivateKey) != "" {
		return true, nil
	}
	return false, nil
}

// controlPlaneTokenAuth resolves IAM identity only. Tenant authorization is
// performed inside each handler after business decoding produces tenant_id.
//
// Credential extraction: header keys first. Body public_key/private_key is only
// decoded when headers are incomplete. peekTokenCredentials already rewound the
// body when it body-sniffed control-plane credentials; we do not re-buffer when
// headers alone are enough. When the body must be read for credentials, it is
// rewound so handlers can re-decode business fields.
func (s *Server) controlPlaneTokenAuth(ctx context.Context, w http.ResponseWriter, r *http.Request) (*ControlPlaneScope, bool) {
	headerPub := strings.TrimSpace(r.Header.Get(quotaPublicKeyHeader))
	headerPriv := strings.TrimSpace(r.Header.Get(quotaPrivateKeyHeader))
	var bodyPub, bodyPriv string
	if headerPub == "" || headerPriv == "" {
		// Body may carry credentials (and business fields). Buffer once and restore.
		if r.Body != nil && (r.Method == http.MethodPost || r.Method == http.MethodPut ||
			r.Method == http.MethodPatch || r.Method == http.MethodDelete) {
			buf, _ := io.ReadAll(io.LimitReader(r.Body, maxCredentialProvisionBodyBytes))
			r.Body = io.NopCloser(bytes.NewReader(buf))
			var body struct {
				PublicKey  string `json:"public_key"`
				PrivateKey string `json:"private_key"`
			}
			_ = json.Unmarshal(buf, &body)
			bodyPub, bodyPriv = body.PublicKey, body.PrivateKey
		}
	}
	cred, err := adminCredentials(bodyPub, bodyPriv, r)
	if err != nil {
		status, msg := clientFacingErrorResponse(http.StatusBadRequest, err.Error(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
		return nil, false
	}
	identity, err := s.resolveTiDBCloudIdentity(ctx, cred, "tokens")
	if err != nil {
		if errors.Is(err, errTiDBCloudIAMNotEnabled) {
			writeControlPlaneError(w, http.StatusNotFound, "not_enabled", "token management not enabled")
			return nil, false
		}
		writeQuotaCredentialError(w, ctx, err, "tokens")
		return nil, false
	}
	return &ControlPlaneScope{
		Identity:       *identity,
		OrganizationID: identity.OrganizationID,
	}, true
}

type generateTokenRequest struct {
	TenantID   string `json:"tenant_id"`
	KeyName    string `json:"key_name"`
	TTLSeconds *int64 `json:"ttl_seconds"`
	PublicKey  string `json:"public_key"`
	PrivateKey string `json:"private_key"`
}

type generateTokenResponse struct {
	Token     string     `json:"token"`
	TokenID   string     `json:"token_id"`
	TenantID  string     `json:"tenant_id"`
	KeyName   string     `json:"key_name"`
	ScopeKind string     `json:"scope_kind"`
	Status    string     `json:"status"`
	IssuedAt  time.Time  `json:"issued_at"`
	ExpiresAt *time.Time `json:"expires_at"`
}

type tokenListItem struct {
	TokenID              string          `json:"token_id"`
	TenantID             string          `json:"tenant_id"`
	KeyName              string          `json:"key_name"`
	ScopeKind            string          `json:"scope_kind"`
	Status               string          `json:"status"`
	Expired              bool            `json:"expired"`
	IssuedByProvider     string          `json:"issued_by_provider,omitempty"`
	IssuedBySubjectKey   string          `json:"issued_by_subject_key,omitempty"`
	IssuedByMetadataJSON json.RawMessage `json:"issued_by_metadata_json,omitempty"`
	IssuedAt             time.Time       `json:"issued_at"`
	ExpiresAt            *time.Time      `json:"expires_at"`
	RevokedAt            *time.Time      `json:"revoked_at"`
	CreatedAt            time.Time       `json:"created_at"`
	UpdatedAt            time.Time       `json:"updated_at"`
}

type tokenListResponse struct {
	Tokens     []tokenListItem `json:"tokens"`
	NextOffset *int            `json:"next_offset,omitempty"`
	// TruncatedTenants is set only on control-plane org-wide list when a
	// tenant had more than maxOrgListKeysPerTenant keys. Callers should fall
	// back to single-tenant GET /v1/tokens?tenant_id=... for those tenants.
	TruncatedTenants []string `json:"truncated_tenants,omitempty"`
}

type tokenStatusTransitionRequest struct {
	TenantID   string `json:"tenant_id"`
	PublicKey  string `json:"public_key"`
	PrivateKey string `json:"private_key"`
}

type tokenStatusTransitionResponse struct {
	TokenID  string `json:"token_id"`
	TenantID string `json:"tenant_id"`
	Status   string `json:"status"`
}

type tokenDeleteRequest struct {
	TenantID   string `json:"tenant_id"`
	PublicKey  string `json:"public_key"`
	PrivateKey string `json:"private_key"`
}

type controlPlaneTokenMutationResponse struct {
	TokenID  string `json:"token_id"`
	TenantID string `json:"tenant_id"`
	Status   string `json:"status"`
}

type controlPlaneRevokeByAPIKeyRequest struct {
	APIKey     string `json:"api_key"`
	TenantID   string `json:"tenant_id"`
	PublicKey  string `json:"public_key"`
	PrivateKey string `json:"private_key"`
}

func (s *Server) handleTokenGenerate(w http.ResponseWriter, r *http.Request, cps *ControlPlaneScope) {
	start := time.Now()
	var req generateTokenRequest
	// Body is optional: header-only control-plane credentials may carry
	// tenant_id solely via query, with key_name/ttl_seconds defaulted.
	if err := decodeJSONBody(w, r, &req, false); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	tenantID, ok := resolveExplicitTenantID(w, r, req.TenantID, true)
	if !ok {
		return
	}
	keyName := strings.TrimSpace(req.KeyName)
	if keyName == "" {
		keyName = "default"
	}
	if len(keyName) > 64 {
		writeControlPlaneStatusError(w, http.StatusBadRequest, "key_name must be at most 64 bytes")
		return
	}

	var expiresAt time.Time
	if req.TTLSeconds != nil {
		if *req.TTLSeconds == 0 {
			writeControlPlaneStatusError(w, http.StatusBadRequest, "ttl_seconds must be positive when provided")
			return
		}
		if *req.TTLSeconds < 0 {
			writeControlPlaneStatusError(w, http.StatusBadRequest, "ttl_seconds must be positive when provided")
			return
		}
		// Compare in seconds to avoid int64 Duration overflow (~292y).
		if *req.TTLSeconds > maxAPIKeyTTLSeconds {
			writeControlPlaneStatusError(w, http.StatusBadRequest, "ttl_seconds exceeds maximum allowed")
			return
		}
		expiresAt = time.Now().UTC().Truncate(time.Second).Add(time.Duration(*req.TTLSeconds) * time.Second)
	}

	t, orgID, ok := s.controlPlaneAuthorizeTenant(w, r, cps, tenantID)
	if !ok {
		metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "auth_error")
		return
	}
	if !controlPlaneTenantAllowsGenerate(w, t) {
		metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "tenant_state")
		return
	}
	setRequestMetricTenant(r.Context(), t.ID, "", t.Provider, orgID, classifyTenantRequest(r))

	source := controlPlaneTokenIssueSource(r.Context(), cps)
	rawToken, apiKeyID, rec, err := s.buildOwnerAPIKey(r.Context(), t.ID, keyName, 0, expiresAt, source)
	if err != nil {
		metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "error")
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to issue token")
		return
	}
	if err := s.meta.InsertAPIKeyWithLimit(r.Context(), rec, meta.DefaultMaxActiveAPIKeys); err != nil {
		if errors.Is(err, meta.ErrAPIKeyLimitExceeded) {
			logger.Info(r.Context(), "server_event", eventFields(r.Context(), "token_generate",
				"tenant_id", t.ID, "organization_id", orgID,
				"principal", tokenManagePrincipalControl, "result", "limit_exceeded",
				"duration_ms", time.Since(start).Milliseconds())...)
			metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "limit_exceeded")
			writeControlPlaneStatusError(w, http.StatusConflict, "api key limit exceeded")
			return
		}
		if errors.Is(err, meta.ErrDuplicate) {
			metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "conflict")
			writeControlPlaneStatusError(w, http.StatusConflict, "token already exists")
			return
		}
		if errors.Is(err, meta.ErrNotFound) {
			metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "tenant_state")
			writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
			return
		}
		if errors.Is(err, meta.ErrTenantNotActive) {
			// Concurrent status change after precheck; re-read for accurate HTTP shape.
			metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "tenant_state")
			live, getErr := s.meta.GetTenant(r.Context(), t.ID)
			writeGenerateTenantNotActiveRecovery(w, r, live, getErr)
			return
		}
		metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "error")
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to persist token")
		return
	}

	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "token_generate",
		"tenant_id", t.ID, "token_id", apiKeyID, "organization_id", orgID,
		"principal", tokenManagePrincipalControl, "result", "ok",
		"duration_ms", time.Since(start).Milliseconds())...)
	metricEvent(r.Context(), "api_key_manage", "action", "generate", "principal", tokenManagePrincipalControl, "result", "ok")

	writeJSON(w, http.StatusCreated, generateTokenResponse{
		Token:     rawToken,
		TokenID:   apiKeyID,
		TenantID:  t.ID,
		KeyName:   keyName,
		ScopeKind: string(meta.APIKeyScopeKindOwner),
		Status:    string(meta.APIKeyActive),
		IssuedAt:  rec.IssuedAt.UTC(),
		ExpiresAt: rec.ExpiresAt,
	})
}

func controlPlaneTokenIssueSource(ctx context.Context, cps *ControlPlaneScope) apiKeyIssueSource {
	if subject := InnerAuthSubjectFromContext(ctx); subject != nil {
		metadataJSON, _ := json.Marshal(map[string]string{"credential_kind": string(subject.Kind)})
		subjectKey := "api_key:" + subject.PublicKey
		if subject.Kind == innerauth.CredentialJWT {
			subjectKey = "user:" + strconv.FormatUint(subject.UserID, 10)
		}
		return apiKeyIssueSource{
			Provider:     "tidbcloud",
			SubjectKey:   subjectKey,
			MetadataJSON: metadataJSON,
		}
	}
	metadataJSON, _ := json.Marshal(map[string]string{"role": cps.Identity.Role})
	return apiKeyIssueSource{
		Provider:     "tidbcloud",
		SubjectKey:   "org:" + cps.Identity.OrganizationID,
		MetadataJSON: metadataJSON,
	}
}

func (s *Server) handleTokenList(w http.ResponseWriter, r *http.Request) {
	start := time.Now()
	cps := ControlPlaneScopeFromContext(r.Context())
	scope := ScopeFromContext(r.Context())

	includeExpired, ok := parseIncludeExpiredQuery(w, r)
	if !ok {
		return
	}
	offset, limit := parseTokenListPagination(r, cps != nil && strings.TrimSpace(r.URL.Query().Get("tenant_id")) == "" && scope == nil)

	if cps != nil {
		queryTenantID := strings.TrimSpace(r.URL.Query().Get("tenant_id"))
		if queryTenantID == "" {
			// Org-wide list: include_expired is not supported.
			if includeExpired {
				writeControlPlaneStatusError(w, http.StatusBadRequest, "include_expired is not supported for org-wide list")
				return
			}
			s.handleTokenListOrgWide(w, r, cps, offset, limit, start)
			return
		}
		t, orgID, authOK := s.controlPlaneAuthorizeTenant(w, r, cps, queryTenantID)
		if !authOK {
			return
		}
		if !controlPlaneTenantAllowsList(w, t) {
			return
		}
		setRequestMetricTenant(r.Context(), t.ID, "", t.Provider, orgID, classifyTenantRequest(r))
		s.writeTokenListForTenant(w, r, t.ID, includeExpired, offset, limit, tokenManagePrincipalControl, start)
		return
	}

	// Owner own-tenant list.
	if scope == nil {
		writeControlPlaneStatusError(w, http.StatusUnauthorized, "missing tenant scope")
		return
	}
	if errTenantID := dataPlaneTenantIDConflict(w, r, scope.TenantID, ""); errTenantID != nil {
		return
	}
	s.writeTokenListForTenant(w, r, scope.TenantID, includeExpired, offset, limit, tokenManagePrincipalOwner, start)
}

func (s *Server) handleTokenListOrgWide(w http.ResponseWriter, r *http.Request, cps *ControlPlaneScope, offset, limit int, start time.Time) {
	tenants, err := s.meta.ListTenantsByTiDBCloudOrganization(r.Context(), cps.Identity.OrganizationID, offset, limit)
	if err != nil {
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to list tenants")
		return
	}
	rawPageSize := len(tenants)
	items := make([]tokenListItem, 0)
	truncated := make([]string, 0)
	for _, t := range tenants {
		switch t.Status {
		case meta.TenantPending, meta.TenantProvisioning, meta.TenantDeleting:
			continue
		case meta.TenantFailed, meta.TenantSuspended, meta.TenantActive:
			// included
		default:
			continue
		}
		keys, listErr := s.meta.ListAPIKeys(r.Context(), t.ID, false, 0, maxOrgListKeysPerTenant+1)
		if listErr != nil {
			status := backendErrorStatus(r.Context(), listErr)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, listErr), "failed to list tokens")
			return
		}
		if len(keys) > maxOrgListKeysPerTenant {
			truncated = append(truncated, t.ID)
			keys = keys[:maxOrgListKeysPerTenant]
		}
		for _, k := range keys {
			items = append(items, apiKeyToListItem(k))
		}
	}
	resp := tokenListResponse{Tokens: items}
	if len(truncated) > 0 {
		resp.TruncatedTenants = truncated
	}
	if rawPageSize == limit {
		next := offset + limit
		resp.NextOffset = &next
	}
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "token_list",
		"organization_id", cps.Identity.OrganizationID,
		"principal", tokenManagePrincipalControl, "result", "ok",
		"duration_ms", time.Since(start).Milliseconds())...)
	// Org-wide list has no single-tenant requestMetricScope (tenant stays
	// "unknown"), so the context-derived tidbcloud_org_id label is suppressed.
	// Pass the org id explicitly; it cannot collide with a context value here.
	metricEvent(r.Context(), "api_key_manage",
		"action", "list", "principal", tokenManagePrincipalControl, "result", "ok",
		"tidbcloud_org_id", cps.Identity.OrganizationID)
	writeJSON(w, http.StatusOK, resp)
}

func (s *Server) writeTokenListForTenant(w http.ResponseWriter, r *http.Request, tenantID string, includeExpired bool, offset, limit int, principal string, start time.Time) {
	keys, err := s.meta.ListAPIKeys(r.Context(), tenantID, includeExpired, offset, limit)
	if err != nil {
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to list tokens")
		return
	}
	items := make([]tokenListItem, 0, len(keys))
	for _, k := range keys {
		items = append(items, apiKeyToListItem(k))
	}
	resp := tokenListResponse{Tokens: items}
	if len(keys) == limit {
		next := offset + limit
		resp.NextOffset = &next
	}
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "token_list",
		"tenant_id", tenantID, "principal", principal, "result", "ok",
		"duration_ms", time.Since(start).Milliseconds())...)
	metricEvent(r.Context(), "api_key_manage", "action", "list", "principal", principal, "result", "ok")
	writeJSON(w, http.StatusOK, resp)
}

func (s *Server) handleTokenStatusTransition(w http.ResponseWriter, r *http.Request, tokenID string, activate bool) {
	start := time.Now()
	cps := ControlPlaneScopeFromContext(r.Context())
	scope := ScopeFromContext(r.Context())

	var req tokenStatusTransitionRequest
	// Body is optional for owner; required tenant_id for control plane.
	if r.Body != nil {
		// Use strict decoding for new endpoints.
		if err := decodeJSONBody(w, r, &req, false); err != nil {
			writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
			return
		}
	}

	var tenantID string
	var principal string
	if cps != nil {
		var ok bool
		tenantID, ok = resolveExplicitTenantID(w, r, req.TenantID, true)
		if !ok {
			return
		}
		t, orgID, authOK := s.controlPlaneAuthorizeTenant(w, r, cps, tenantID)
		if !authOK {
			return
		}
		if !controlPlaneTenantAllowsStatusMutation(w, t) {
			return
		}
		setRequestMetricTenant(r.Context(), t.ID, tokenID, t.Provider, orgID, classifyTenantRequest(r))
		tenantID = t.ID
		principal = tokenManagePrincipalControl
	} else {
		if scope == nil {
			writeControlPlaneStatusError(w, http.StatusUnauthorized, "missing tenant scope")
			return
		}
		tenantID = scope.TenantID
		if conflict := dataPlaneTenantIDConflict(w, r, scope.TenantID, req.TenantID); conflict != nil {
			return
		}
		// Owner may only activate/deactivate fs_scoped targets.
		key, err := s.meta.GetAPIKey(r.Context(), tenantID, tokenID)
		if err != nil {
			if errors.Is(err, meta.ErrNotFound) {
				writeControlPlaneStatusError(w, http.StatusNotFound, "token not found")
				return
			}
			status := backendErrorStatus(r.Context(), err)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to load token")
			return
		}
		if key.ScopeKind != meta.APIKeyScopeKindFS {
			writeControlPlaneStatusError(w, http.StatusForbidden, "owner key cannot activate or deactivate owner keys")
			return
		}
		principal = tokenManagePrincipalOwner
	}

	// Pre-check for activating an expired key.
	if activate {
		key, err := s.meta.GetAPIKey(r.Context(), tenantID, tokenID)
		if err != nil {
			if errors.Is(err, meta.ErrNotFound) {
				writeControlPlaneStatusError(w, http.StatusNotFound, "token not found")
				return
			}
			status := backendErrorStatus(r.Context(), err)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to load token")
			return
		}
		if key.ExpiresAt != nil && !key.ExpiresAt.After(time.Now().UTC()) {
			writeControlPlaneStatusError(w, http.StatusConflict, "key already expired; cannot be revived; please generate a new key")
			return
		}
	}

	from, to := meta.APIKeyDisabled, meta.APIKeyActive
	action := "activate"
	if !activate {
		from, to = meta.APIKeyActive, meta.APIKeyDisabled
		action = "deactivate"
	}
	current, changed, err := s.meta.SetAPIKeyStatus(r.Context(), tenantID, tokenID, from, to)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			writeControlPlaneStatusError(w, http.StatusNotFound, "token not found")
			return
		}
		if errors.Is(err, meta.ErrAPIKeyTerminal) {
			writeControlPlaneStatusError(w, http.StatusConflict, "api key is terminal")
			return
		}
		if errors.Is(err, meta.ErrAPIKeyExpired) {
			writeControlPlaneStatusError(w, http.StatusConflict, "key already expired; cannot be revived; please generate a new key")
			return
		}
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to update token status")
		return
	}

	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "token_"+action,
		"tenant_id", tenantID, "token_id", tokenID, "principal", principal,
		"changed", changed, "result", "ok",
		"duration_ms", time.Since(start).Milliseconds())...)
	metricEvent(r.Context(), "api_key_manage", "action", action, "principal", principal, "result", "ok")

	writeJSON(w, http.StatusOK, tokenStatusTransitionResponse{
		TokenID:  tokenID,
		TenantID: tenantID,
		Status:   string(current),
	})
}

func (s *Server) handleTokenDelete(w http.ResponseWriter, r *http.Request, cps *ControlPlaneScope, tokenID string) {
	start := time.Now()
	var req tokenDeleteRequest
	if err := decodeJSONBody(w, r, &req, false); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	tenantID, ok := resolveExplicitTenantID(w, r, req.TenantID, true)
	if !ok {
		return
	}
	t, orgID, authOK := s.controlPlaneAuthorizeTenant(w, r, cps, tenantID)
	if !authOK {
		return
	}
	if !controlPlaneTenantAllowsDelete(w, t) {
		return
	}
	setRequestMetricTenant(r.Context(), t.ID, tokenID, t.Provider, orgID, classifyTenantRequest(r))

	if err := s.meta.RevokeAPIKey(r.Context(), t.ID, tokenID); err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			writeControlPlaneStatusError(w, http.StatusNotFound, "token not found or already revoked")
			return
		}
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to revoke token")
		return
	}

	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "token_delete",
		"tenant_id", t.ID, "token_id", tokenID, "organization_id", orgID,
		"principal", tokenManagePrincipalControl, "result", "ok",
		"duration_ms", time.Since(start).Milliseconds())...)
	metricEvent(r.Context(), "api_key_manage", "action", "delete", "principal", tokenManagePrincipalControl, "result", "ok")

	writeJSON(w, http.StatusOK, controlPlaneTokenMutationResponse{
		TokenID:  tokenID,
		TenantID: t.ID,
		Status:   string(meta.APIKeyRevoked),
	})
}

func (s *Server) handleTokenRevokeByAPIKey(w http.ResponseWriter, r *http.Request, cps *ControlPlaneScope) {
	start := time.Now()
	var req controlPlaneRevokeByAPIKeyRequest
	if err := decodeJSONBody(w, r, &req, true); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	apiKey := strings.TrimSpace(req.APIKey)
	if apiKey == "" {
		writeControlPlaneStatusError(w, http.StatusBadRequest, "api_key is required")
		return
	}
	resolved, err := s.meta.ResolveByAPIKeyHash(r.Context(), token.HashToken(apiKey))
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			writeControlPlaneStatusError(w, http.StatusNotFound, "token not found or already revoked")
			return
		}
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to revoke token")
		return
	}
	// Optional tenant_id must match the resolved row (anti-enumeration → 404).
	if bodyTenant := strings.TrimSpace(req.TenantID); bodyTenant != "" && bodyTenant != resolved.APIKey.TenantID {
		writeControlPlaneStatusError(w, http.StatusNotFound, "token not found or already revoked")
		return
	}
	if queryTenant := strings.TrimSpace(r.URL.Query().Get("tenant_id")); queryTenant != "" && queryTenant != resolved.APIKey.TenantID {
		writeControlPlaneStatusError(w, http.StatusNotFound, "token not found or already revoked")
		return
	}

	// Resolve org belonging. Expected absence / cross-org denial collapse to
	// 404 (anti-enumeration); metadb/backend lookup failures map to 5xx per §4.8.
	t, orgID, authErr := s.resolveTiDBCloudAuthorizedTenant(r.Context(), resolved.APIKey.TenantID, cps.Identity.OrganizationID)
	if authErr != nil {
		if writeControlPlaneTokenAuthzError(w, r, authErr, "token not found or already revoked") {
			return
		}
		return
	}
	if t == nil || t.Status == meta.TenantDeleting || t.Status == meta.TenantDeleted {
		writeControlPlaneStatusError(w, http.StatusNotFound, "token not found or already revoked")
		return
	}
	switch t.Status {
	case meta.TenantPending, meta.TenantProvisioning:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return
	}
	setRequestMetricTenant(r.Context(), resolved.APIKey.TenantID, resolved.APIKey.ID, t.Provider, orgID, classifyTenantRequest(r))

	if err := s.meta.RevokeAPIKey(r.Context(), resolved.APIKey.TenantID, resolved.APIKey.ID); err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			writeControlPlaneStatusError(w, http.StatusNotFound, "token not found or already revoked")
			return
		}
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "failed to revoke token")
		return
	}

	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "token_delete",
		"tenant_id", resolved.APIKey.TenantID, "token_id", resolved.APIKey.ID,
		"organization_id", orgID, "principal", tokenManagePrincipalControl, "result", "ok",
		"duration_ms", time.Since(start).Milliseconds())...)
	metricEvent(r.Context(), "api_key_manage", "action", "delete", "principal", tokenManagePrincipalControl, "result", "ok")

	writeJSON(w, http.StatusOK, controlPlaneTokenMutationResponse{
		TokenID:  resolved.APIKey.ID,
		TenantID: resolved.APIKey.TenantID,
		Status:   string(meta.APIKeyRevoked),
	})
}

// controlPlaneAuthorizeTenant resolves and authorizes a tenant for the control
// plane identity. On failure it writes the HTTP response and returns ok=false.
func (s *Server) controlPlaneAuthorizeTenant(w http.ResponseWriter, r *http.Request, cps *ControlPlaneScope, tenantID string) (*meta.Tenant, string, bool) {
	t, orgID, err := s.resolveTiDBCloudAuthorizedTenant(r.Context(), tenantID, cps.Identity.OrganizationID)
	if err != nil {
		if writeControlPlaneTokenAuthzErrorExplicitTenant(w, r, err, "tenant not found") {
			return nil, "", false
		}
		return nil, "", false
	}
	if t.Status == meta.TenantDeleting {
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
		return nil, "", false
	}
	return t, orgID, true
}

// writeControlPlaneTokenAuthzError maps authz errors for plaintext-key revoke:
// cross-org denial collapses to 404 (anti-enumeration — the key itself is the
// address; a 403 would confirm the key exists under another org).
func writeControlPlaneTokenAuthzError(w http.ResponseWriter, r *http.Request, err error, notFoundMsg string) bool {
	return mapControlPlaneTokenAuthzError(w, r, err, notFoundMsg, false)
}

// writeControlPlaneTokenAuthzErrorExplicitTenant maps authz errors for
// endpoints with an explicit tenant_id (generate/list/activate/deactivate/
// delete): the caller already named the tenant, so cross-org denial is 403
// (§4.8); absence cases stay 404.
func writeControlPlaneTokenAuthzErrorExplicitTenant(w http.ResponseWriter, r *http.Request, err error, notFoundMsg string) bool {
	return mapControlPlaneTokenAuthzError(w, r, err, notFoundMsg, true)
}

func mapControlPlaneTokenAuthzError(w http.ResponseWriter, r *http.Request, err error, notFoundMsg string, explicitTenant bool) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, errTokensProviderUnsupported) {
		writeControlPlaneError(w, http.StatusConflict, controlPlaneErrorCode(http.StatusConflict, err), "only tidb_cloud_native / tidb_cloud_native_shared are supported")
		return true
	}
	if explicitTenant && errors.Is(err, tenant.ErrQuotaPermissionDenied) {
		writeControlPlaneError(w, http.StatusForbidden, controlPlaneErrorCode(http.StatusForbidden, err), "tenant does not belong to the caller's organization")
		return true
	}
	// Uniform anti-enumeration: missing tenant / binding / shared DB / free pool
	// (and, for revoke-by-key, cross-org denial) → 404 with the SAME generic
	// code, so the code cannot leak whether the tenant exists or belongs to
	// the caller's organization.
	if errors.Is(err, tenant.ErrQuotaBackendNotFound) ||
		errors.Is(err, tenant.ErrQuotaPermissionDenied) ||
		errors.Is(err, errTenantOrgBindingNotFound) ||
		errors.Is(err, errTenantSharedDBNotFound) {
		writeControlPlaneError(w, http.StatusNotFound, "not_found", notFoundMsg)
		return true
	}
	// Lookup/backend failures → quota/client-facing 5xx machinery (§4.8).
	writeQuotaCredentialError(w, r.Context(), err, "tokens")
	return true
}

// writeGenerateTenantNotActiveRecovery writes the response for the generate
// insert-side tenant-status race. Every branch writes exactly one non-2xx
// response: an active re-read means the status flapped and the insert was
// rejected — that is a retryable conflict, never an empty 200.
func writeGenerateTenantNotActiveRecovery(w http.ResponseWriter, r *http.Request, live *meta.Tenant, getErr error) {
	switch {
	case getErr == nil && live != nil && live.Status != meta.TenantActive:
		// Precise shape for the observed state (403 suspended / 409 not-ready /
		// 404 deleting|deleted).
		_ = controlPlaneTenantAllowsGenerate(w, live)
	case getErr == nil && live != nil:
		// Flapped back to active: nothing was persisted; the client must retry.
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant state changed concurrently; retry")
	case getErr == nil || errors.Is(getErr, meta.ErrNotFound):
		// Tenant gone (live == nil is defensive; GetTenant reports ErrNotFound).
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
	default:
		// Metadb re-read failure → 5xx per §4.8, never 404.
		writeQuotaCredentialError(w, r.Context(), getErr, "tokens")
	}
}

func controlPlaneTenantAllowsGenerate(w http.ResponseWriter, t *meta.Tenant) bool {
	switch t.Status {
	case meta.TenantActive:
		return true
	case meta.TenantSuspended:
		writeControlPlaneStatusError(w, http.StatusForbidden, "tenant is suspended")
		return false
	case meta.TenantFailed, meta.TenantPending, meta.TenantProvisioning:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	case meta.TenantDeleting, meta.TenantDeleted:
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
		return false
	default:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	}
}

func controlPlaneTenantAllowsStatusMutation(w http.ResponseWriter, t *meta.Tenant) bool {
	switch t.Status {
	case meta.TenantActive, meta.TenantSuspended:
		return true
	case meta.TenantFailed:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	case meta.TenantPending, meta.TenantProvisioning:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	case meta.TenantDeleting, meta.TenantDeleted:
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
		return false
	default:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	}
}

func controlPlaneTenantAllowsDelete(w http.ResponseWriter, t *meta.Tenant) bool {
	switch t.Status {
	case meta.TenantActive, meta.TenantSuspended, meta.TenantFailed:
		return true
	case meta.TenantPending, meta.TenantProvisioning:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	case meta.TenantDeleting, meta.TenantDeleted:
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
		return false
	default:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	}
}

func controlPlaneTenantAllowsList(w http.ResponseWriter, t *meta.Tenant) bool {
	switch t.Status {
	case meta.TenantActive, meta.TenantSuspended, meta.TenantFailed:
		return true
	case meta.TenantPending, meta.TenantProvisioning:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	case meta.TenantDeleting, meta.TenantDeleted:
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
		return false
	default:
		writeControlPlaneStatusError(w, http.StatusConflict, "tenant is not ready")
		return false
	}
}

func resolveExplicitTenantID(w http.ResponseWriter, r *http.Request, bodyTenantID string, required bool) (string, bool) {
	bodyTenantID = strings.TrimSpace(bodyTenantID)
	queryTenantID := strings.TrimSpace(r.URL.Query().Get("tenant_id"))
	switch {
	case bodyTenantID != "" && queryTenantID != "" && bodyTenantID != queryTenantID:
		writeControlPlaneStatusError(w, http.StatusBadRequest, "tenant_id body and query disagree")
		return "", false
	case bodyTenantID != "":
		return bodyTenantID, true
	case queryTenantID != "":
		return queryTenantID, true
	case required:
		writeControlPlaneStatusError(w, http.StatusBadRequest, "tenant_id is required")
		return "", false
	default:
		return "", true
	}
}

// dataPlaneTenantIDConflict returns a non-nil error after writing 403 if a
// request tenant_id disagrees with the Bearer token's tenant.
func dataPlaneTenantIDConflict(w http.ResponseWriter, r *http.Request, scopeTenantID, bodyTenantID string) error {
	bodyTenantID = strings.TrimSpace(bodyTenantID)
	queryTenantID := strings.TrimSpace(r.URL.Query().Get("tenant_id"))
	if bodyTenantID != "" && queryTenantID != "" && bodyTenantID != queryTenantID {
		writeControlPlaneStatusError(w, http.StatusBadRequest, "tenant_id body and query disagree")
		return fmt.Errorf("tenant_id conflict")
	}
	for _, candidate := range []string{bodyTenantID, queryTenantID} {
		if candidate != "" && candidate != scopeTenantID {
			writeControlPlaneStatusError(w, http.StatusForbidden, "tenant_id does not match authenticated tenant")
			return fmt.Errorf("tenant_id mismatch")
		}
	}
	return nil
}

func parseIncludeExpiredQuery(w http.ResponseWriter, r *http.Request) (bool, bool) {
	raw := strings.TrimSpace(r.URL.Query().Get("include_expired"))
	if raw == "" {
		return false, true
	}
	switch strings.ToLower(raw) {
	case "1", "true":
		return true, true
	default:
		writeControlPlaneStatusError(w, http.StatusBadRequest, "include_expired must be 1 or true")
		return false, false
	}
}

func parseTokenListPagination(r *http.Request, orgWide bool) (offset, limit int) {
	defaultLimit := defaultTokenListLimit
	maxLimit := maxTokenListLimit
	if orgWide {
		defaultLimit = defaultOrgTokenListLimit
		maxLimit = maxOrgTokenListLimit
	}
	offset = 0
	if raw := strings.TrimSpace(r.URL.Query().Get("offset")); raw != "" {
		if v, err := strconv.Atoi(raw); err == nil && v > 0 {
			offset = v
		}
		// negative / non-numeric → clamped to 0
	}
	limit = defaultLimit
	if raw := strings.TrimSpace(r.URL.Query().Get("limit")); raw != "" {
		if v, err := strconv.Atoi(raw); err == nil && v > 0 {
			limit = v
			if limit > maxLimit {
				limit = maxLimit
			}
		}
		// non-numeric / ≤0 → default
	}
	return offset, limit
}

func apiKeyToListItem(k meta.APIKey) tokenListItem {
	expired := k.ExpiresAt != nil && !k.ExpiresAt.After(time.Now().UTC())
	item := tokenListItem{
		TokenID:            k.ID,
		TenantID:           k.TenantID,
		KeyName:            k.KeyName,
		ScopeKind:          string(k.ScopeKind),
		Status:             string(k.Status),
		Expired:            expired,
		IssuedByProvider:   k.IssuedByProvider,
		IssuedBySubjectKey: k.IssuedBySubjectKey,
		IssuedAt:           k.IssuedAt.UTC(),
		ExpiresAt:          k.ExpiresAt,
		RevokedAt:          k.RevokedAt,
		CreatedAt:          k.CreatedAt.UTC(),
		UpdatedAt:          k.UpdatedAt.UTC(),
	}
	if len(k.IssuedByMetadataJSON) > 0 {
		item.IssuedByMetadataJSON = json.RawMessage(k.IssuedByMetadataJSON)
	}
	return item
}