package server

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

	"go.uber.org/zap"

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

const (
	defaultAdminTenantPageSize     = 10
	maxAdminTenantPageSize         = 100
	maxAdminTenantLabelEntries     = 30
	maxKubernetesLabelNameLength   = 63
	maxKubernetesLabelPrefixLength = 253
	maxKubernetesLabelValueLength  = 63

	invalidAdminTenantDisplayNameMessage = "display_name must be empty or 4-64 characters using letters, numbers, or hyphens, and must start and end with a letter or number"
	tooManyAdminTenantLabelsMessage      = "label must contain at most 30 entries"
	invalidAdminTenantLabelKeyMessage    = "label keys must be Kubernetes qualified names: an optional lowercase DNS prefix of at most 253 bytes and '/', followed by 1-63 bytes using letters, numbers, '-', '_' or '.', starting and ending with a letter or number"
	invalidAdminTenantLabelValueMessage  = "label values must be empty or at most 63 bytes using letters, numbers, '-', '_' or '.', and must start and end with a letter or number"
)

var (
	adminTenantDisplayNameRegexp      = regexp.MustCompile(`^[A-Za-z0-9][-A-Za-z0-9]{2,62}[A-Za-z0-9]$`)
	kubernetesLabelNameRegexp         = regexp.MustCompile(`^[A-Za-z0-9](?:[-A-Za-z0-9_.]*[A-Za-z0-9])?$`)
	kubernetesLabelDNSSubdomainRegexp = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\.[a-z0-9](?:[-a-z0-9]*[a-z0-9])?)*$`)
)

type adminTenantCreateRequest struct {
	PublicKey   string            `json:"public_key"`
	PrivateKey  string            `json:"private_key"`
	DisplayName string            `json:"display_name"`
	Label       map[string]string `json:"label"`
	quotaFields
}

type adminTenantQuotaRequest struct {
	PublicKey  string `json:"public_key"`
	PrivateKey string `json:"private_key"`
	quotaFields
}

type adminTenantResponse struct {
	TenantID    string                   `json:"tenant_id"`
	DisplayName string                   `json:"display_name"`
	Label       map[string]string        `json:"label"`
	Status      string                   `json:"status"`
	Kind        string                   `json:"kind"`
	Quota       *adminTenantQuotaSummary `json:"quota"`
}

type adminTenantQuotaSummary struct {
	Config quotaConfigResponse `json:"config"`
	Usage  quotaUsageResponse  `json:"usage"`
}

type adminTenantCreateResponse struct {
	TenantID      string            `json:"tenant_id"`
	DisplayName   string            `json:"display_name"`
	Label         map[string]string `json:"label"`
	APIKey        string            `json:"api_key"`
	Status        string            `json:"status"`
	CloudProvider string            `json:"cloud_provider,omitempty"`
	Region        string            `json:"region,omitempty"`
}

type adminTenantListResponse struct {
	Tenants  []adminTenantResponse `json:"tenants"`
	Page     int                   `json:"page"`
	PageSize int                   `json:"page_size"`
	NextPage int                   `json:"next_page,omitempty"`
}

type adminQuotaResponse struct {
	TenantID string              `json:"tenant_id"`
	Status   string              `json:"status"`
	Config   quotaConfigResponse `json:"config"`
	Usage    quotaUsageResponse  `json:"usage"`
}

type adminTenantDeleteResponse struct {
	TenantID string `json:"tenant_id"`
	Status   string `json:"status"`
}

func (s *Server) adminTenantsRootHandler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !s.adminTenantAPIEnabled() {
			writeControlPlaneError(w, http.StatusNotFound, "not_enabled", "admin tenant API not enabled")
			return
		}
		switch r.Method {
		case http.MethodPost:
			s.handleAdminTenantCreate(w, r)
		case http.MethodGet:
			s.handleAdminTenantList(w, r)
		default:
			writeControlPlaneStatusError(w, http.StatusMethodNotAllowed, "method not allowed")
		}
	})
}

func (s *Server) adminTenantsItemHandler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !s.adminTenantAPIEnabled() {
			writeControlPlaneError(w, http.StatusNotFound, "not_enabled", "admin tenant API not enabled")
			return
		}
		parts := splitAdminTenantPath(r.URL.Path)
		if len(parts) == 0 {
			writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
			return
		}
		tenantID := strings.TrimSpace(parts[0])
		if tenantID == "" {
			writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
			return
		}
		if len(parts) == 1 {
			switch r.Method {
			case http.MethodGet:
				s.handleAdminTenantGet(w, r, tenantID)
			case http.MethodDelete:
				s.handleAdminTenantDelete(w, r, tenantID)
			default:
				writeControlPlaneStatusError(w, http.StatusMethodNotAllowed, "method not allowed")
			}
			return
		}
		if len(parts) == 2 && parts[1] == "quota" {
			switch r.Method {
			case http.MethodPost:
				s.handleAdminTenantQuotaSet(w, r, tenantID)
			default:
				writeControlPlaneStatusError(w, http.StatusMethodNotAllowed, "method not allowed")
			}
			return
		}
		if len(parts) == 2 && parts[1] == "object-namespace" {
			s.handleAdminTenantObjectNamespace(w, r, tenantID)
			return
		}
		if len(parts) == 2 && parts[1] == "embedding-config" {
			switch r.Method {
			case http.MethodGet:
				s.handleAdminTenantEmbeddingConfigGet(w, r, tenantID)
			case http.MethodPut:
				s.handleAdminTenantEmbeddingConfigPut(w, r, tenantID)
			default:
				writeControlPlaneStatusError(w, http.StatusMethodNotAllowed, "method not allowed")
			}
			return
		}
		if len(parts) == 3 && parts[1] == "extract-config" {
			if parts[2] != meta.ExtractMediaTypeImage && parts[2] != meta.ExtractMediaTypeAudio && parts[2] != meta.ExtractMediaTypeVideo {
				writeControlPlaneStatusError(w, http.StatusBadRequest, fmt.Sprintf("extract config for %q is not supported", parts[2]))
				return
			}
			switch r.Method {
			case http.MethodGet:
				s.handleAdminTenantExtractConfigGet(w, r, tenantID, parts[2])
			case http.MethodPut:
				s.handleAdminTenantExtractConfigPut(w, r, tenantID, parts[2])
			default:
				writeControlPlaneStatusError(w, http.StatusMethodNotAllowed, "method not allowed")
			}
			return
		}
		writeControlPlaneStatusError(w, http.StatusNotFound, "not found")
	})
}

func (s *Server) adminTenantAPIEnabled() bool {
	if s.meta == nil || s.pool == nil || s.provisioner == nil || len(s.tokenSecret) == 0 {
		return false
	}
	provider, err := tenant.NormalizeProvider(s.provisioner.ProviderType())
	if err != nil || (provider != tenant.ProviderTiDBCloudNative && provider != tenant.ProviderTiDBCloudNativeShared) {
		return false
	}
	_, ok := s.provisioner.(tenant.TiDBCloudAPIKeyIdentityResolver)
	return ok
}

func (s *Server) handleAdminTenantCreate(w http.ResponseWriter, r *http.Request) {
	var req adminTenantCreateRequest
	if err := decodeJSONBody(w, r, &req, true); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	cred, err := controlPlaneCredentials(r, req.PublicKey, req.PrivateKey)
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	quotaReq := quotaRequest{
		TenantID:    "pending",
		quotaFields: req.quotaFields,
	}
	var quotaOpt *quotaRequest
	if quotaReq.anySet() {
		if err := s.validateQuotaSetRequestWithoutSpendingLimit(quotaReq); err != nil {
			writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
			return
		}
		quotaOpt = &quotaReq
	}
	access, err := s.authorizeTiDBCloudAdminPlanAccess(r.Context(), cred, "admin_tenant_create")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "create tenant")
		return
	}
	metadata := normalizeTenantMetadata(req.DisplayName, req.Label)
	if err := validateAdminTenantMetadata(metadata); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	exists, err := s.meta.TenantDisplayNameExistsInTiDBCloudOrganization(r.Context(), access.OrganizationID, metadata.DisplayName)
	if err != nil {
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "tenant display name lookup failed")
		return
	}
	if exists {
		writeControlPlaneStatusError(w, http.StatusConflict, "display_name conflicts with an existing tenant display name or tenant ID in organization")
		return
	}
	poolClaimStarted := time.Now()
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_claim_started", "provider", tenant.ProviderTiDBCloudNative, "quota_requested", quotaOpt != nil)...)
	res, pool, claimed, sharedPoolMatched, err := s.claimAdminTenantFromPoolWithAccess(r.Context(), cred, quotaOpt, access, metadata)
	if err != nil {
		logger.Error(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_claim_failed", "provider", tenant.ProviderTiDBCloudNative, "duration_ms", durationMillis(poolClaimStarted), "error", err)...)
		writeProvisionTenantError(w, newPoolClaimError(err))
		return
	} else if claimed {
		logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_claim_accepted", "tenant_id", res.TenantID, "provider", res.Provider, "pool_id", pool.PoolID, "organization_id", res.OrganizationID, "duration_ms", durationMillis(poolClaimStarted), "status", res.Status)...)
		setRequestMetricTenant(r.Context(), res.TenantID, res.APIKeyID, res.Provider, res.OrganizationID, classifyTenantRequest(r))
		if res.Status == meta.TenantProvisioning {
			s.startProvisionedTenantSchemaInit(r.Context(), res)
		}
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusAccepted)
		_ = json.NewEncoder(w).Encode(adminTenantCreateResponse{
			TenantID:      res.TenantID,
			DisplayName:   effectiveDisplayName(res.TenantID, metadata.DisplayName),
			Label:         normalizedTenantLabel(metadata.Label),
			APIKey:        res.APIKey,
			Status:        string(res.Status),
			CloudProvider: res.CloudProvider,
			Region:        res.Region,
		})
		logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_create_accepted", "tenant_id", res.TenantID, "provider", res.Provider, "pool_id", pool.PoolID, "organization_id", res.OrganizationID, "duration_ms", durationMillis(poolClaimStarted))...)
		return
	}
	provider := ""
	if sharedPoolMatched {
		provider = tenant.ProviderTiDBCloudNativeShared
		quotaOpt = quotaRequestForShared(quotaOpt)
	}
	logProvider := tenant.ProviderTiDBCloudNative
	if provider != "" {
		logProvider = provider
	}
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_claim_missed", "provider", logProvider, "duration_ms", durationMillis(poolClaimStarted))...)
	res, err = s.provisionTenant(r.Context(), provisionTenantOptions{
		KeyName:               "default",
		TokenVersion:          1,
		CredentialProvisioner: &cred,
		TiDBCloudAccess:       access,
		Quota:                 quotaOpt,
		Provider:              provider,
		Metadata:              metadata,
	})
	if err != nil {
		var pe *provisionTenantError
		if errors.As(err, &pe) {
			writeProvisionTenantError(w, pe)
			return
		}
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to provision tenant")
		return
	}
	setRequestMetricTenant(r.Context(), res.TenantID, res.APIKeyID, res.Provider, res.OrganizationID, classifyTenantRequest(r))
	s.startProvisionedTenantSchemaInit(r.Context(), res)
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	_ = json.NewEncoder(w).Encode(adminTenantCreateResponse{
		TenantID:      res.TenantID,
		DisplayName:   effectiveDisplayName(res.TenantID, metadata.DisplayName),
		Label:         normalizedTenantLabel(metadata.Label),
		APIKey:        res.APIKey,
		Status:        string(res.Status),
		CloudProvider: res.CloudProvider,
		Region:        res.Region,
	})
}

func (s *Server) handleAdminTenantList(w http.ResponseWriter, r *http.Request) {
	cred, err := controlPlaneCredentials(r, "", "")
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	organizationID, err := s.authorizeTiDBCloudAdminOrganization(r.Context(), cred, "admin_tenant_list")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "list tenants")
		return
	}
	pageSize, page, offset, err := adminPagination(r)
	if err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	filter, err := adminTenantListFilter(r)
	if err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	rows, err := s.meta.ListTenantsByTiDBCloudOrganizationFiltered(r.Context(), organizationID, filter, offset, pageSize+1)
	if err != nil {
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "tenant list failed")
		return
	}
	nextPage := 0
	if len(rows) > pageSize {
		rows = rows[:pageSize]
		nextPage = page + 1
	}
	tenantIDs := make([]string, len(rows))
	for i := range rows {
		tenantIDs[i] = rows[i].ID
	}
	quotaSnapshots, err := s.meta.ListTenantQuotaSnapshots(r.Context(), tenantIDs)
	if err != nil {
		logger.Warn(r.Context(), "admin_tenant_quota_list_failed", zap.Error(err))
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "quota lookup failed")
		return
	}
	out := make([]adminTenantResponse, 0, len(rows))
	for i := range rows {
		row := &rows[i]
		snapshot := quotaSnapshots[row.ID]
		quota := adminTenantQuotaSummaryFromValues(row.Provider, snapshot.Config, snapshot.Usage)
		out = append(out, s.adminTenantResponse(row, quota))
	}
	writeJSON(w, http.StatusOK, adminTenantListResponse{Tenants: out, Page: page, PageSize: pageSize, NextPage: nextPage})
}

func (s *Server) handleAdminTenantGet(w http.ResponseWriter, r *http.Request, tenantID string) {
	cred, err := controlPlaneCredentials(r, "", "")
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	organizationID, err := s.authorizeTiDBCloudAdminOrganization(r.Context(), cred, "admin_tenant_get")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "get tenant")
		return
	}
	t, ok := s.authorizedAdminTenant(w, r, tenantID, organizationID)
	if !ok {
		return
	}
	quota, err := s.loadAdminTenantQuotaSummary(r.Context(), t)
	if err != nil {
		logger.Warn(r.Context(), "admin_tenant_quota_lookup_failed", zap.String("tenant_id", t.ID), zap.Error(err))
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "quota lookup failed")
		return
	}
	writeJSON(w, http.StatusOK, s.adminTenantResponse(t, quota))
}

func (s *Server) handleAdminTenantQuotaSet(w http.ResponseWriter, r *http.Request, tenantID string) {
	var req adminTenantQuotaRequest
	if err := decodeJSONBody(w, r, &req, true); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	cred, err := controlPlaneCredentials(r, req.PublicKey, req.PrivateKey)
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	quotaReq := quotaRequest{
		TenantID:    tenantID,
		quotaFields: req.quotaFields,
	}
	if err := s.validateQuotaSetRequestWithoutSpendingLimit(quotaReq); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	access, err := s.authorizeTiDBCloudQuotaMutation(r.Context(), cred, "admin_tenant_quota_set")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "update tenant quota")
		return
	}
	t, ok := s.authorizedAdminTenant(w, r, tenantID, access.OrganizationID)
	if !ok {
		return
	}
	if err := s.rejectQuotaSetForTenantStatus(t); err != nil {
		writeControlPlaneStatusError(w, http.StatusConflict, err.Error())
		return
	}
	var applyErr error
	if t.Provider == tenant.ProviderTiDBCloudNativeShared {
		quotaReq = *quotaRequestForShared(&quotaReq)
		applyErr = s.applySharedQuotaSet(r.Context(), t, quotaReq)
	} else {
		if err := validateTiDBCloudSpendingLimit(quotaReq.TiDBCloudSpendingLimit); err != nil {
			writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
			return
		}
		applyErr = s.applyQuotaSet(r.Context(), t, cred, quotaReq)
	}
	if applyErr != nil {
		writeQuotaSetError(w, r.Context(), applyErr, "update")
		return
	}
	s.writeAdminQuotaResponse(w, r, t)
}

func (s *Server) handleAdminTenantDelete(w http.ResponseWriter, r *http.Request, tenantID string) {
	var raw struct {
		PublicKey  string `json:"public_key"`
		PrivateKey string `json:"private_key"`
	}
	if err := decodeJSONBody(w, r, &raw, false); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	cred, err := controlPlaneCredentials(r, raw.PublicKey, raw.PrivateKey)
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	organizationID, err := s.authorizeTiDBCloudAdminOrganization(r.Context(), cred, "admin_tenant_delete")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "delete tenant")
		return
	}
	t, ok := s.authorizedAdminTenant(w, r, tenantID, organizationID)
	if !ok {
		return
	}
	if t.Kind != meta.TenantKindLive {
		writeControlPlaneStatusError(w, http.StatusConflict, "only live tenants can be deleted")
		return
	}
	if t.Status == meta.TenantDeleted {
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
		return
	}
	if t.Provider == tenant.ProviderTiDBCloudNativeShared {
		s.handleSharedTenantDeleteWithStatusWriter(w, r, t, func(w http.ResponseWriter, status meta.TenantStatus) {
			writeJSON(w, http.StatusAccepted, adminTenantDeleteResponse{TenantID: t.ID, Status: string(status)})
		})
		return
	}
	if t.StorageNamespaceID != "" {
		hasFork, err := s.meta.NamespaceHasNonDeletedFork(r.Context(), t.StorageNamespaceID)
		if err != nil {
			writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to check tenant forks")
			return
		}
		if hasFork {
			writeControlPlaneStatusError(w, http.StatusConflict, "tenant has non-deleted forks")
			return
		}
	}
	retryingUnconfirmedDelete := false
	clusterExists := true
	if t.Status == meta.TenantDeleting {
		armed, err := s.meta.TenantDeleteJobArmed(r.Context(), t.ID)
		if err != nil {
			writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to check tenant delete cleanup")
			return
		}
		if armed {
			s.clearLocalTenantMetrics(t.ID)
			_ = s.meta.RevokeTenantAPIKeys(r.Context(), t.ID)
			writeJSON(w, http.StatusAccepted, adminTenantDeleteResponse{TenantID: t.ID, Status: string(meta.TenantDeleting)})
			return
		}
		retryingUnconfirmedDelete = true
	}
	if !retryingUnconfirmedDelete {
		if verifier, ok := s.provisioner.(tenant.ClusterAccessVerifier); ok {
			var err error
			clusterExists, err = verifier.VerifyClusterAccess(r.Context(), clusterInfoFromTenant(t), cred)
			if err != nil {
				writeAdminTenantDeleteAccessError(w, r.Context(), err)
				return
			}
		} else {
			writeControlPlaneError(w, http.StatusNotFound, "not_enabled", "tenant delete not enabled")
			return
		}
	}
	// Same durability contract as the owner-facing delete path: cluster-backed
	// tenants persist the deleting mark before deprovision and create the
	// cleanup job only after it succeeds. The work runs on a detached context
	// so a client disconnect cannot abort it halfway.
	workCtx := s.detachedDeleteWorkContext(r.Context())
	updated, err := s.markTenantDeletingForCleanup(workCtx, t)
	if err != nil {
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to mark tenant deleting")
		return
	}
	if !updated {
		writeJSON(w, http.StatusAccepted, adminTenantDeleteResponse{TenantID: t.ID, Status: string(meta.TenantDeleting)})
		return
	}
	if err := s.drainTenantBackend(workCtx, t.ID); err != nil {
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to drain tenant backend")
		return
	}
	if clusterExists {
		if err := s.deprovisionTenantCluster(workCtx, t, cred); err != nil {
			status, msg := clientFacingErrorResponse(http.StatusBadGateway, "delete tenant cluster failed", err)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
			return
		}
	}
	_ = s.meta.AbortActiveUploadReservations(workCtx, t.ID)
	status, err := s.enqueueTenantDeleteJob(workCtx, t)
	if err != nil {
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to enqueue tenant delete cleanup")
		return
	}
	_ = s.meta.RevokeTenantAPIKeys(workCtx, t.ID)
	writeJSON(w, http.StatusAccepted, adminTenantDeleteResponse{TenantID: t.ID, Status: string(status)})
}

func writeAdminTenantDeleteAccessError(w http.ResponseWriter, ctx context.Context, err error) {
	logger.Warn(ctx, "tidbcloud_admin_tenant_delete_access_failed", zap.Error(err))
	status, msg := clientFacingErrorResponse(
		http.StatusBadGateway,
		"verify TiDB Cloud cluster access before tenant delete failed",
		err,
	)
	writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
}

func (s *Server) authorizedAdminTenant(w http.ResponseWriter, r *http.Request, tenantID, organizationID string) (*meta.Tenant, bool) {
	t, orgID, err := s.resolveTiDBCloudAuthorizedTenant(r.Context(), tenantID, organizationID)
	if err != nil {
		switch {
		case errors.Is(err, errTenantOrgBindingNotFound):
			writeControlPlaneError(w, http.StatusNotFound, controlPlaneErrorCode(http.StatusNotFound, err), "tenant tidbcloud organization binding not found")
			return nil, false
		case errors.Is(err, errTenantSharedDBNotFound):
			writeControlPlaneError(w, http.StatusNotFound, controlPlaneErrorCode(http.StatusNotFound, err), "tenant shared DB pool not found")
			return nil, false
		case errors.Is(err, tenant.ErrQuotaBackendNotFound):
			writeControlPlaneError(w, http.StatusNotFound, controlPlaneErrorCode(http.StatusNotFound, err), "tenant not found")
			return nil, false
		case errors.Is(err, tenant.ErrQuotaPermissionDenied):
			writeControlPlaneError(w, http.StatusForbidden, controlPlaneErrorCode(http.StatusForbidden, err), "TiDB Cloud API key organization does not match tenant")
			return nil, false
		case errors.Is(err, errTokensProviderUnsupported):
			writeControlPlaneError(w, http.StatusConflict, controlPlaneErrorCode(http.StatusConflict, err), "admin tenant API is only supported for tidb_cloud_native or tidb_cloud_native_shared tenants")
			return nil, false
		case errors.Is(err, errTenantPoolMembershipLookup):
			writeControlPlaneError(w, http.StatusInternalServerError, controlPlaneErrorCode(http.StatusInternalServerError, err), "tenant pool membership lookup failed")
			return nil, false
		case errors.Is(err, errTenantSharedDBLookup):
			status := backendErrorStatus(r.Context(), err)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "tenant shared DB pool lookup failed")
			return nil, false
		case errors.Is(err, errTenantOrgBindingLookup):
			status := backendErrorStatus(r.Context(), err)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "tenant organization binding lookup failed")
			return nil, false
		default:
			status := backendErrorStatus(r.Context(), err)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "tenant lookup failed")
			return nil, false
		}
	}
	setRequestMetricTenant(r.Context(), t.ID, "", t.Provider, orgID, classifyTenantRequest(r))
	return t, true
}

func (s *Server) adminTenantResponse(t *meta.Tenant, quota *adminTenantQuotaSummary) adminTenantResponse {
	return adminTenantResponse{
		TenantID:    t.ID,
		DisplayName: effectiveDisplayName(t.ID, t.DisplayName),
		Label:       normalizedTenantLabel(t.Label),
		Status:      string(t.Status),
		Kind:        string(t.Kind),
		Quota:       quota,
	}
}

func normalizeTenantMetadata(displayName string, label map[string]string) meta.TenantMetadata {
	return meta.TenantMetadata{DisplayName: displayName, Label: normalizedTenantLabel(label)}
}

func validateAdminTenantMetadata(metadata meta.TenantMetadata) error {
	if metadata.DisplayName != "" && !adminTenantDisplayNameRegexp.MatchString(metadata.DisplayName) {
		return errors.New(invalidAdminTenantDisplayNameMessage)
	}
	if len(metadata.Label) > maxAdminTenantLabelEntries {
		return errors.New(tooManyAdminTenantLabelsMessage)
	}
	for key := range metadata.Label {
		if !isKubernetesLabelKey(key) {
			return errors.New(invalidAdminTenantLabelKeyMessage)
		}
	}
	for _, value := range metadata.Label {
		if !isKubernetesLabelValue(value) {
			return errors.New(invalidAdminTenantLabelValueMessage)
		}
	}
	return nil
}

func isKubernetesLabelKey(key string) bool {
	parts := strings.Split(key, "/")
	if len(parts) > 2 {
		return false
	}
	name := parts[len(parts)-1]
	if len(name) == 0 || len(name) > maxKubernetesLabelNameLength || !kubernetesLabelNameRegexp.MatchString(name) {
		return false
	}
	if len(parts) == 1 {
		return true
	}
	prefix := parts[0]
	return len(prefix) > 0 && len(prefix) <= maxKubernetesLabelPrefixLength && kubernetesLabelDNSSubdomainRegexp.MatchString(prefix)
}

func isKubernetesLabelValue(value string) bool {
	return value == "" || (len(value) <= maxKubernetesLabelValueLength && kubernetesLabelNameRegexp.MatchString(value))
}

func normalizedTenantLabel(label map[string]string) map[string]string {
	if label == nil {
		return map[string]string{}
	}
	out := make(map[string]string, len(label))
	for key, value := range label {
		out[key] = value
	}
	return out
}

func effectiveDisplayName(tenantID, displayName string) string {
	if displayName = strings.TrimSpace(displayName); displayName != "" {
		return displayName
	}
	return tenantID
}

func adminTenantListFilter(r *http.Request) (meta.TenantListFilter, error) {
	filter := meta.TenantListFilter{DisplayNameContains: r.URL.Query().Get("display_name")}
	raw := r.URL.Query().Get("label")
	if raw == "" {
		return filter, nil
	}
	key, value, ok := strings.Cut(raw, "==")
	if !ok {
		return meta.TenantListFilter{}, fmt.Errorf("label must use key==value format")
	}
	if !isKubernetesLabelKey(key) {
		return meta.TenantListFilter{}, errors.New(invalidAdminTenantLabelKeyMessage)
	}
	if !isKubernetesLabelValue(value) {
		return meta.TenantListFilter{}, errors.New(invalidAdminTenantLabelValueMessage)
	}
	filter.LabelKey = key
	filter.LabelValue = value
	return filter, nil
}

func (s *Server) loadAdminTenantQuotaSummary(ctx context.Context, t *meta.Tenant) (*adminTenantQuotaSummary, error) {
	cfg, err := s.meta.GetQuotaConfig(ctx, t.ID)
	if err != nil {
		return nil, fmt.Errorf("quota config lookup failed: %w", err)
	}
	usage, err := s.meta.GetQuotaUsage(ctx, t.ID)
	if err != nil {
		return nil, fmt.Errorf("quota usage lookup failed: %w", err)
	}
	return adminTenantQuotaSummaryFromValues(t.Provider, *cfg, *usage), nil
}

func adminTenantQuotaSummaryFromValues(provider string, cfg meta.QuotaConfig, usage meta.QuotaUsage) *adminTenantQuotaSummary {
	return &adminTenantQuotaSummary{
		Config: quotaConfigResponseFromValues(provider, cfg),
		Usage: quotaUsageResponse{
			StorageBytes:   usage.StorageBytes,
			ReservedBytes:  usage.ReservedBytes,
			FileCount:      usage.FileCount,
			MediaFileCount: usage.MediaFileCount,
			VideoFileCount: usage.VideoFileCount,
		},
	}
}

func (s *Server) writeAdminQuotaResponse(w http.ResponseWriter, r *http.Request, t *meta.Tenant) {
	quota, err := s.loadAdminTenantQuotaSummary(r.Context(), t)
	if err != nil {
		logger.Warn(r.Context(), "admin_tenant_quota_lookup_failed", zap.String("tenant_id", t.ID), zap.Error(err))
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "quota lookup failed")
		return
	}
	writeJSON(w, http.StatusOK, adminQuotaResponse{
		TenantID: t.ID,
		Status:   string(t.Status),
		Config:   quota.Config,
		Usage:    quota.Usage,
	})
}

func adminCredentials(publicKey, privateKey string, r *http.Request) (tenant.CredentialProvisionRequest, error) {
	if strings.TrimSpace(publicKey) == "" {
		publicKey = r.Header.Get(quotaPublicKeyHeader)
	}
	if strings.TrimSpace(privateKey) == "" {
		privateKey = r.Header.Get(quotaPrivateKeyHeader)
	}
	req := quotaRequest{
		TenantID:   "admin",
		PublicKey:  strings.TrimSpace(publicKey),
		PrivateKey: strings.TrimSpace(privateKey),
	}
	return quotaCredentials(req)
}

func adminPagination(r *http.Request) (pageSize, page, offset int, err error) {
	pageSize = defaultAdminTenantPageSize
	page = 1
	if raw := strings.TrimSpace(r.URL.Query().Get("page_size")); raw != "" {
		v, parseErr := strconv.Atoi(raw)
		if parseErr != nil || v <= 0 {
			return 0, 0, 0, fmt.Errorf("page_size must be a positive integer")
		}
		if v > maxAdminTenantPageSize {
			v = maxAdminTenantPageSize
		}
		pageSize = v
	}
	if raw := strings.TrimSpace(r.URL.Query().Get("page")); raw != "" {
		v, parseErr := strconv.Atoi(raw)
		if parseErr != nil || v <= 0 {
			return 0, 0, 0, fmt.Errorf("page must be a positive integer")
		}
		page = v
	}
	maxOffsetPage := 1 + int(^uint(0)>>1)/pageSize
	if page > maxOffsetPage {
		return 0, 0, 0, fmt.Errorf("page is too large")
	}
	offset = (page - 1) * pageSize
	return pageSize, page, offset, nil
}

func decodeJSONBody(w http.ResponseWriter, r *http.Request, out any, requireBody bool) error {
	if r.Body == nil {
		if requireBody {
			return fmt.Errorf("request body is required")
		}
		return nil
	}
	dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxCredentialProvisionBodyBytes))
	dec.DisallowUnknownFields()
	if err := dec.Decode(out); err != nil {
		if errors.Is(err, io.EOF) && !requireBody {
			return nil
		}
		return fmt.Errorf("invalid JSON body: %w", err)
	}
	var extra struct{}
	if err := dec.Decode(&extra); !errors.Is(err, io.EOF) {
		return fmt.Errorf("invalid JSON body: trailing data")
	}
	return nil
}

func writeJSON(w http.ResponseWriter, status int, body any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(body)
}

func writeAdminTiDBCloudError(w http.ResponseWriter, ctx context.Context, err error, action string) {
	if errors.Is(err, tenant.ErrTiDBCloudFreeTenantPoolForbidden) {
		writeControlPlaneError(w, http.StatusPaymentRequired, "free_tenant_pool_forbidden", freeTenantPoolForbiddenMessage())
		return
	}
	if errors.Is(err, tenant.ErrTiDBCloudFreeQuotaMutationForbidden) {
		writeControlPlaneError(w, http.StatusPaymentRequired, "free_quota_mutation_forbidden", freeQuotaMutationForbiddenMessage())
		return
	}
	if isTiDBCloudBillingLookupError(err) {
		status, msg := tiDBCloudBillingErrorResponse(err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
		return
	}
	if errors.Is(err, tenant.ErrQuotaBackendNotFound) {
		writeControlPlaneError(w, http.StatusNotFound, "quota_backend_not_found", quotaBackendNotFoundMessage)
		return
	}
	logger.Warn(ctx, "tidbcloud_admin_failed", zap.String("action", action), zap.Error(err))
	status, msg := clientFacingErrorResponse(http.StatusBadGateway, "tidbcloud admin "+action+" failed", err)
	writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
}