package server

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"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/metrics"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/tenant/token"
)

type adminTenantPoolRequest struct {
	PublicKey  string `json:"public_key"`
	PrivateKey string `json:"private_key"`
	PoolSize   *int   `json:"pool_size,omitempty"`
}

type adminTenantPoolResponse struct {
	PoolID         string                `json:"pool_id"`
	OrganizationID string                `json:"organization_id,omitempty"`
	PoolSize       int                   `json:"pool_size"`
	FreeSize       int                   `json:"free_size"`
	Status         adminTenantPoolStatus `json:"status"`
}

type adminTenantPoolHTTPError struct {
	status  int
	message string
	// code optionally carries a machine-readable control-plane code (e.g. a
	// free-tier sentinel). When empty the writer derives a generic
	// status-based code.
	code string
}

type tenantPoolResumeJob struct {
	rerun atomic.Bool
}

type tenantPoolWorkGate struct {
	mu          sync.Mutex
	running     bool
	rerun       bool
	scheduled   bool
	nextAllowed time.Time
}

type tenantPoolReplenishDecision struct {
	start         bool
	scheduleAfter time.Duration
}

type tenantPoolWorkerStarter func(context.Context, func(context.Context)) bool

type adminTenantPoolStatus string

const adminTenantPoolStatusCreating adminTenantPoolStatus = "creating"

const adminTenantPoolMetricsComponent = "admin_tenant_pool"

const tenantPoolClaimCASRetryLimit = 8

const (
	tenantPoolReplenishMinInterval       = time.Second
	tenantPoolPendingResumeEmptyInterval = 15 * time.Second
)

func managedSharedDBRefillTenantCount(missing, maxTenants, batchSize int) int {
	if missing <= 0 {
		return 0
	}
	if maxTenants <= 0 {
		maxTenants = defaultManagedSharedDBMaxTenants
	}
	maxInt := int(^uint(0) >> 1)
	limit := maxInt
	if batchSize <= 0 {
		batchSize = DefaultManagedSharedDBCloudBatchSize
	}
	if maxTenants <= maxInt/batchSize {
		limit = maxTenants * batchSize
	}
	return min(missing, limit)
}

func retryTenantPoolClaimCAS[T any](attempt func() (T, error)) (T, error) {
	var zero T
	for range tenantPoolClaimCASRetryLimit {
		result, err := attempt()
		if err == nil {
			return result, nil
		}
		if !errors.Is(err, meta.ErrNotFound) {
			return zero, err
		}
	}
	return zero, nil
}

type tenantPoolClaimSelection struct {
	sharedResult *provisionTenantResult
	native       *meta.TenantWithTiDBCloudOrgBinding
}

func (e *adminTenantPoolHTTPError) Error() string {
	return e.message
}

func adminTenantPoolError(status int, message string) error {
	return &adminTenantPoolHTTPError{status: status, message: message}
}

// adminTenantPoolErrorWithCode builds a tenant-pool error carrying an
// explicit machine-readable control-plane code (e.g. a free-tier sentinel)
// instead of the generic status-derived fallback.
func adminTenantPoolErrorWithCode(status int, code, message string) error {
	return &adminTenantPoolHTTPError{status: status, code: code, message: message}
}

func writeAdminTenantPoolError(w http.ResponseWriter, r *http.Request, err error) {
	var httpErr *adminTenantPoolHTTPError
	if errors.As(err, &httpErr) {
		code := httpErr.code
		if code == "" {
			code = controlPlaneErrorCode(httpErr.status, nil)
		}
		writeControlPlaneError(w, httpErr.status, code, httpErr.message)
		return
	}
	writeBackendError(w, r, err)
}

func (s *Server) adminTenantPoolHandler() 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.handleAdminTenantPoolCreate(w, r)
		case http.MethodGet:
			s.handleAdminTenantPoolGet(w, r)
		case http.MethodPatch:
			s.handleAdminTenantPoolUpdate(w, r)
		case http.MethodDelete:
			s.handleAdminTenantPoolDelete(w, r)
		default:
			writeControlPlaneStatusError(w, http.StatusMethodNotAllowed, "method not allowed")
		}
	})
}

func (s *Server) handleAdminTenantPoolCreate(w http.ResponseWriter, r *http.Request) {
	var req adminTenantPoolRequest
	if err := decodeJSONBody(w, r, &req, true); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	if req.PoolSize == nil || *req.PoolSize <= 0 {
		writeControlPlaneStatusError(w, http.StatusBadRequest, "pool_size must be positive")
		return
	}
	if err := s.validateTenantPoolSize(*req.PoolSize); 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
	}
	organizationID, err := s.authorizeTiDBCloudPaidTenantPoolOrganization(r.Context(), cred, "admin_tenant_pool_create")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "create tenant pool")
		return
	}
	createStarted := time.Now()
	metricResult := "ok"
	defer func() {
		metrics.RecordOperation(adminTenantPoolMetricsComponent, "create", metricResult, time.Since(createStarted))
	}()
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_create_requested",
		"provider", tenant.ProviderTiDBCloudNative,
		"pool_size", *req.PoolSize)...)
	createLock := s.tenantPoolCreateLock(cred)
	createLock.Lock()
	defer createLock.Unlock()
	if err := s.meta.WithTenantPoolLock(r.Context(), tenantPoolCreateDatabaseLockKey(cred), func(ctx context.Context) error {
		stageStarted := time.Now()
		orgID := organizationID
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_create_org_lookup_done",
			"provider", tenant.ProviderTiDBCloudNative,
			"organization_id", orgID,
			"duration_ms", durationMillis(stageStarted))...)
		if orgID != "" {
			if _, err := s.meta.GetTenantPoolByOrganization(ctx, orgID); err == nil {
				metricResult = "conflict"
				writeControlPlaneStatusError(w, http.StatusConflict, "tenant pool already exists for organization")
				return nil
			} else if !errors.Is(err, meta.ErrNotFound) {
				metricResult = "error"
				writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "tenant pool lookup failed")
				return nil
			}
		}
		poolID := token.NewID()
		now := time.Now().UTC()
		stageStarted = time.Now()
		if err := s.meta.CreateTenantPool(ctx, &meta.TenantPool{
			PoolID:         poolID,
			OrganizationID: orgID,
			Size:           *req.PoolSize,
			Status:         meta.TenantPoolActive,
			CreatedAt:      now,
			UpdatedAt:      now,
		}); err != nil {
			if errors.Is(err, meta.ErrDuplicate) {
				metricResult = "conflict"
				writeControlPlaneStatusError(w, http.StatusConflict, "tenant pool already exists for organization")
				return nil
			}
			metricResult = "error"
			writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to persist tenant pool")
			return nil
		}
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_create_pool_persisted",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", poolID,
			"organization_id", orgID,
			"pool_size", *req.PoolSize,
			"status", meta.TenantPoolActive,
			"duration_ms", durationMillis(stageStarted))...)
		stageStarted = time.Now()
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_create_free_tenants_started",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", poolID,
			"organization_id", orgID,
			"requested_count", *req.PoolSize)...)
		results, err := s.createFreePoolTenants(ctx, poolID, *req.PoolSize, cred, nil)
		if err != nil {
			logger.Error(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_create_free_tenants_failed",
				"provider", tenant.ProviderTiDBCloudNative,
				"pool_id", poolID,
				"organization_id", orgID,
				"requested_count", *req.PoolSize,
				"duration_ms", durationMillis(stageStarted),
				"error", err)...)
			s.cleanupCreatedPoolTenants(ctx, results, cred, "create_free_tenants_error")
			s.deleteTenantPoolMetadata(ctx, poolID, "create_free_tenants_error")
			metricResult = "cluster_error"
			status, msg := clientFacingErrorResponse(http.StatusBadGateway, "create tenant pool failed", err)
			writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
			return nil
		}
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_create_free_tenants_done",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", poolID,
			"organization_id", orgID,
			"requested_count", *req.PoolSize,
			"created_count", len(results),
			"duration_ms", durationMillis(stageStarted))...)
		if orgID == "" {
			orgID = firstResultOrganizationID(results)
			if orgID != "" {
				stageStarted = time.Now()
				if err := s.meta.UpdateTenantPoolOrganization(ctx, poolID, orgID); err != nil {
					s.cleanupCreatedPoolTenants(ctx, results, cred, "update_pool_org_error")
					s.deleteTenantPoolMetadata(ctx, poolID, "update_pool_org_error")
					if errors.Is(err, meta.ErrDuplicate) {
						metricResult = "conflict"
						writeControlPlaneStatusError(w, http.StatusConflict, "tenant pool already exists for organization")
						return nil
					}
					metricResult = "error"
					writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "failed to update tenant pool organization")
					return nil
				}
				logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_create_org_persisted",
					"provider", tenant.ProviderTiDBCloudNative,
					"pool_id", poolID,
					"organization_id", orgID,
					"duration_ms", durationMillis(stageStarted))...)
			}
		}
		for _, res := range results {
			s.startProvisionedTenantSchemaInit(ctx, res)
		}
		freeSize := 0
		slotSize := len(results)
		if orgID != "" {
			if n, err := s.meta.CountFreeTenantPoolBindings(ctx, orgID); err == nil {
				freeSize = n
			} else {
				logger.Warn(ctx, "admin_tenant_pool_create_free_size_lookup_failed",
					zap.String("pool_id", poolID),
					zap.String("organization_id", orgID),
					zap.Error(err))
			}
		}
		s.recordTenantPoolCapacity(poolID, orgID, *req.PoolSize, freeSize)
		writeJSON(w, http.StatusAccepted, adminTenantPoolResponse{
			PoolID:         poolID,
			OrganizationID: orgID,
			PoolSize:       *req.PoolSize,
			FreeSize:       freeSize,
			Status:         adminTenantPoolDisplayStatus(meta.TenantPoolActive, freeSize, slotSize),
		})
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_create_done",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", poolID,
			"organization_id", orgID,
			"pool_size", *req.PoolSize,
			"free_size", freeSize,
			"slot_size", slotSize,
			"status", adminTenantPoolDisplayStatus(meta.TenantPoolActive, freeSize, slotSize),
			"duration_ms", durationMillis(createStarted))...)
		return nil
	}); err != nil {
		logger.Error(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_create_failed",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_size", *req.PoolSize,
			"duration_ms", durationMillis(createStarted),
			"error", err)...)
		metricResult = adminTenantPoolMetricResult(err)
		writeAdminTenantPoolError(w, r, err)
		return
	}
}

func (s *Server) handleAdminTenantPoolGet(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_pool_get")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "get tenant pool")
		return
	}
	pool, ok := s.authorizedTenantPool(w, r, organizationID)
	if !ok {
		return
	}
	freeSize, slotSize, err := s.tenantPoolSizes(r.Context(), pool.OrganizationID)
	if err != nil {
		writeBackendError(w, r, err)
		return
	}
	s.recordTenantPoolCapacity(pool.PoolID, pool.OrganizationID, pool.Size, freeSize)
	writeJSON(w, http.StatusOK, adminTenantPoolResponse{
		PoolID:         pool.PoolID,
		OrganizationID: pool.OrganizationID,
		PoolSize:       pool.Size,
		FreeSize:       freeSize,
		Status:         adminTenantPoolDisplayStatus(pool.Status, freeSize, slotSize),
	})
}

func (s *Server) handleAdminTenantPoolUpdate(w http.ResponseWriter, r *http.Request) {
	var req adminTenantPoolRequest
	if err := decodeJSONBody(w, r, &req, true); err != nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, err.Error())
		return
	}
	if req.PoolSize != nil && *req.PoolSize <= 0 {
		writeControlPlaneStatusError(w, http.StatusBadRequest, "pool_size must be positive")
		return
	}
	if req.PoolSize == nil {
		writeControlPlaneStatusError(w, http.StatusBadRequest, "pool_size is required")
		return
	}
	if err := s.validateTenantPoolSize(*req.PoolSize); 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
	}
	organizationID, err := s.authorizeTiDBCloudPaidTenantPoolOrganization(r.Context(), cred, "admin_tenant_pool_update")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "update tenant pool")
		return
	}
	pool, ok := s.authorizedTenantPool(w, r, organizationID)
	if !ok {
		return
	}
	updateStarted := time.Now()
	metricResult := "ok"
	defer func() {
		metrics.RecordOperation(adminTenantPoolMetricsComponent, "update", metricResult, time.Since(updateStarted))
	}()
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_update_requested",
		"provider", tenant.ProviderTiDBCloudNative,
		"pool_id", pool.PoolID,
		"organization_id", pool.OrganizationID,
		"current_pool_size", pool.Size,
		"target_pool_size", *req.PoolSize,
		"pool_status", pool.Status)...)
	lock := s.tenantPoolLock(pool.PoolID)
	lock.Lock()
	defer lock.Unlock()
	workCtx := s.detachedDeleteWorkContext(r.Context())

	var out adminTenantPoolResponse
	if err := s.meta.WithTenantPoolLock(workCtx, pool.PoolID, func(ctx context.Context) error {
		if pool, err = s.meta.GetTenantPoolByID(ctx, pool.PoolID); err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "tenant pool lookup failed")
		}
		if pool.Status != meta.TenantPoolActive {
			return adminTenantPoolError(http.StatusConflict, "tenant pool is not active")
		}
		if err := s.retryUnconfirmedFreePoolTenantDeletes(ctx, pool.OrganizationID, pool.PoolID, cred); err != nil {
			return adminTenantPoolError(http.StatusBadGateway, "retry tenant pool cleanup failed")
		}
		freeSize, err := s.meta.CountFreeTenantPoolBindings(ctx, pool.OrganizationID)
		if err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "tenant pool free size lookup failed")
		}
		slotSize, err := s.meta.CountTenantPoolFreeSlots(ctx, pool.OrganizationID)
		if err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "tenant pool slot size lookup failed")
		}
		targetSize := *req.PoolSize
		if targetSize > slotSize {
			stageStarted := time.Now()
			growCount := targetSize - slotSize
			logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_grow_started",
				"provider", tenant.ProviderTiDBCloudNative,
				"pool_id", pool.PoolID,
				"organization_id", pool.OrganizationID,
				"current_pool_size", pool.Size,
				"target_pool_size", targetSize,
				"free_size", freeSize,
				"slot_size", slotSize,
				"grow_count", growCount)...)
			var results []*provisionTenantResult
			if s.defaultTenantProvider == tenant.ProviderTiDBCloudNativeShared && targetSize > pool.Size {
				results, err = s.createFreeSharedPoolTenantsWithResize(ctx, pool.PoolID, growCount, cred, nil,
					&meta.ManagedSharedDBPoolWaveResize{ExpectedSize: pool.Size, TargetSize: targetSize})
			} else {
				results, err = s.createFreePoolTenants(ctx, pool.PoolID, growCount, cred, nil)
			}
			if err != nil {
				logger.Error(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_grow_failed",
					"provider", tenant.ProviderTiDBCloudNative,
					"pool_id", pool.PoolID,
					"organization_id", pool.OrganizationID,
					"target_pool_size", targetSize,
					"grow_count", growCount,
					"duration_ms", durationMillis(stageStarted),
					"error", err)...)
				status, msg := clientFacingErrorResponse(http.StatusBadGateway, "grow tenant pool failed", err)
				return adminTenantPoolErrorWithCode(status, controlPlaneErrorCode(status, err), msg)
			}
			for _, res := range results {
				s.startProvisionedTenantSchemaInit(ctx, res)
			}
			logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_grow_done",
				"provider", tenant.ProviderTiDBCloudNative,
				"pool_id", pool.PoolID,
				"organization_id", pool.OrganizationID,
				"target_pool_size", targetSize,
				"grow_count", growCount,
				"created_count", len(results),
				"duration_ms", durationMillis(stageStarted))...)
		} else if targetSize < slotSize {
			stageStarted := time.Now()
			shrinkCount := slotSize - targetSize
			logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_shrink_started",
				"provider", tenant.ProviderTiDBCloudNative,
				"pool_id", pool.PoolID,
				"organization_id", pool.OrganizationID,
				"current_pool_size", pool.Size,
				"target_pool_size", targetSize,
				"free_size", freeSize,
				"slot_size", slotSize,
				"shrink_count", shrinkCount)...)
			deleted, err := s.deleteNewestFreePoolTenants(ctx, pool.PoolID, pool.OrganizationID, shrinkCount, cred, false)
			if err != nil {
				if actualSlotSize, countErr := s.meta.CountTenantPoolFreeSlots(ctx, pool.OrganizationID); countErr == nil {
					if updateErr := s.meta.UpdateTenantPoolSize(ctx, pool.PoolID, actualSlotSize); updateErr != nil {
						logger.Warn(ctx, "admin_tenant_pool_shrink_partial_size_update_failed", zap.String("pool_id", pool.PoolID), zap.Int("slot_size", actualSlotSize), zap.Error(updateErr))
					}
				} else {
					logger.Warn(ctx, "admin_tenant_pool_shrink_partial_count_failed", zap.String("pool_id", pool.PoolID), zap.Error(countErr))
				}
				logger.Error(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_shrink_failed",
					"provider", tenant.ProviderTiDBCloudNative,
					"pool_id", pool.PoolID,
					"organization_id", pool.OrganizationID,
					"target_pool_size", targetSize,
					"shrink_count", shrinkCount,
					"deleted_count", deleted,
					"duration_ms", durationMillis(stageStarted),
					"error", err)...)
				status, msg := clientFacingErrorResponse(http.StatusBadGateway, "shrink tenant pool failed", err)
				return adminTenantPoolErrorWithCode(status, controlPlaneErrorCode(status, err), msg)
			}
			logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_shrink_done",
				"provider", tenant.ProviderTiDBCloudNative,
				"pool_id", pool.PoolID,
				"organization_id", pool.OrganizationID,
				"target_pool_size", targetSize,
				"shrink_count", shrinkCount,
				"deleted_count", deleted,
				"duration_ms", durationMillis(stageStarted))...)
		} else {
			logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_noop",
				"provider", tenant.ProviderTiDBCloudNative,
				"pool_id", pool.PoolID,
				"organization_id", pool.OrganizationID,
				"current_pool_size", pool.Size,
				"target_pool_size", targetSize,
				"free_size", freeSize,
				"slot_size", slotSize)...)
		}
		stageStarted := time.Now()
		if err := s.meta.UpdateTenantPoolSize(ctx, pool.PoolID, targetSize); err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "failed to update tenant pool")
		}
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_update_size_persisted",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", pool.PoolID,
			"organization_id", pool.OrganizationID,
			"target_pool_size", targetSize,
			"duration_ms", durationMillis(stageStarted))...)
		if freeSize, err = s.meta.CountFreeTenantPoolBindings(ctx, pool.OrganizationID); err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "tenant pool free size lookup failed")
		}
		if slotSize, err = s.meta.CountTenantPoolFreeSlots(ctx, pool.OrganizationID); err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "tenant pool slot size lookup failed")
		}
		out = adminTenantPoolResponse{
			PoolID:         pool.PoolID,
			OrganizationID: pool.OrganizationID,
			PoolSize:       targetSize,
			FreeSize:       freeSize,
			Status:         adminTenantPoolDisplayStatus(pool.Status, freeSize, slotSize),
		}
		s.recordTenantPoolCapacity(pool.PoolID, pool.OrganizationID, targetSize, freeSize)
		return nil
	}); err != nil {
		metricResult = adminTenantPoolMetricResult(err)
		logger.Error(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_update_failed",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", pool.PoolID,
			"organization_id", pool.OrganizationID,
			"current_pool_size", pool.Size,
			"target_pool_size", *req.PoolSize,
			"duration_ms", durationMillis(updateStarted),
			"error", err)...)
		writeAdminTenantPoolError(w, r, err)
		return
	}
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_update_done",
		"provider", tenant.ProviderTiDBCloudNative,
		"pool_id", out.PoolID,
		"organization_id", out.OrganizationID,
		"target_pool_size", out.PoolSize,
		"free_size", out.FreeSize,
		"status", out.Status,
		"duration_ms", durationMillis(updateStarted))...)
	writeJSON(w, http.StatusAccepted, out)
}

func (s *Server) validateTenantPoolSize(size int) error {
	maxSize := s.tenantPoolMaxSize
	if maxSize <= 0 {
		maxSize = DefaultTenantPoolMaxSize
	}
	if size > maxSize {
		return fmt.Errorf("pool_size %d exceeds maximum %d", size, maxSize)
	}
	return nil
}

func adminTenantPoolDisplayStatus(status meta.TenantPoolStatus, freeSize, slotSize int) adminTenantPoolStatus {
	if status == meta.TenantPoolActive && freeSize == 0 && slotSize > 0 {
		return adminTenantPoolStatusCreating
	}
	return adminTenantPoolStatus(status)
}

func adminTenantPoolMetricResult(err error) string {
	if err == nil {
		return "ok"
	}
	var httpErr *adminTenantPoolHTTPError
	if errors.As(err, &httpErr) && httpErr.status == http.StatusBadGateway {
		return "cluster_error"
	}
	return metrics.ResultForError(err)
}

func (s *Server) handleAdminTenantPoolDelete(w http.ResponseWriter, r *http.Request) {
	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_pool_delete")
	if err != nil {
		writeAdminTiDBCloudError(w, r.Context(), err, "delete tenant pool")
		return
	}
	pool, ok := s.authorizedTenantPool(w, r, organizationID)
	if !ok {
		return
	}
	deleteStarted := time.Now()
	metricResult := "ok"
	defer func() {
		metrics.RecordOperation(adminTenantPoolMetricsComponent, "delete", metricResult, time.Since(deleteStarted))
	}()
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_delete_requested",
		"provider", tenant.ProviderTiDBCloudNative,
		"pool_id", pool.PoolID,
		"organization_id", pool.OrganizationID,
		"pool_size", pool.Size,
		"pool_status", pool.Status)...)
	lock := s.tenantPoolLock(pool.PoolID)
	lock.Lock()
	defer lock.Unlock()

	// The free-tenant purge below can take far longer than any sane client
	// timeout. Detach it from the request context so a client disconnect
	// cannot abort the purge halfway and leave the pool stuck in deleting.
	// Deriving from forkWorkerCtx keeps the purge cancelable on server
	// shutdown, and the leader reconciler
	// (reconcileDeletingTenantPoolsWithCtx) finishes pools whose delete was
	// interrupted either way.
	workCtx := s.detachedDeleteWorkContext(r.Context())

	var out adminTenantPoolResponse
	if err := s.meta.WithTenantPoolLock(workCtx, pool.PoolID, func(ctx context.Context) error {
		stageStarted := time.Now()
		if err := s.meta.UpdateTenantPoolStatus(ctx, pool.PoolID, meta.TenantPoolDeleting); err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "failed to mark tenant pool deleting")
		}
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_delete_pool_marked_deleting",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", pool.PoolID,
			"organization_id", pool.OrganizationID,
			"status", meta.TenantPoolDeleting,
			"duration_ms", durationMillis(stageStarted))...)
		if err := s.retryUnconfirmedFreePoolTenantDeletes(ctx, pool.OrganizationID, pool.PoolID, cred); err != nil {
			status, msg := clientFacingErrorResponse(http.StatusBadGateway, "retry tenant pool cleanup failed", err)
			return adminTenantPoolErrorWithCode(status, controlPlaneErrorCode(status, err), msg)
		}
		stageStarted = time.Now()
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_delete_free_tenants_started",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", pool.PoolID,
			"organization_id", pool.OrganizationID)...)
		deleted, err := s.deleteNewestFreePoolTenants(ctx, pool.PoolID, pool.OrganizationID, 0, cred, true)
		if err != nil {
			status, msg := clientFacingErrorResponse(http.StatusBadGateway, "delete tenant pool failed", err)
			return adminTenantPoolErrorWithCode(status, controlPlaneErrorCode(status, err), msg)
		}
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_delete_free_tenants_done",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", pool.PoolID,
			"organization_id", pool.OrganizationID,
			"deleted_free_tenants", deleted,
			"duration_ms", durationMillis(stageStarted))...)
		stageStarted = time.Now()
		if err := s.meta.DeleteTenantPoolAndDetachUsedMembers(ctx, pool.PoolID); err != nil {
			return adminTenantPoolError(http.StatusInternalServerError, "failed to delete tenant pool")
		}
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_delete_metadata_deleted",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", pool.PoolID,
			"organization_id", pool.OrganizationID,
			"duration_ms", durationMillis(stageStarted))...)
		out = adminTenantPoolResponse{
			PoolID:         pool.PoolID,
			OrganizationID: pool.OrganizationID,
			PoolSize:       pool.Size,
			FreeSize:       0,
			Status:         adminTenantPoolStatus(meta.TenantPoolDeleting),
		}
		return nil
	}); err != nil {
		logger.Error(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_delete_failed",
			"provider", tenant.ProviderTiDBCloudNative,
			"pool_id", pool.PoolID,
			"organization_id", pool.OrganizationID,
			"duration_ms", durationMillis(deleteStarted),
			"error", err)...)
		metricResult = adminTenantPoolMetricResult(err)
		writeAdminTenantPoolError(w, r, err)
		return
	}
	logger.Info(r.Context(), "server_event", eventFields(r.Context(), "admin_tenant_pool_delete_done",
		"provider", tenant.ProviderTiDBCloudNative,
		"pool_id", pool.PoolID,
		"organization_id", pool.OrganizationID,
		"pool_size", pool.Size,
		"free_size", 0,
		"status", meta.TenantPoolDeleting,
		"duration_ms", durationMillis(deleteStarted))...)
	s.recordTenantPoolCapacity(pool.PoolID, pool.OrganizationID, 0, 0)
	writeJSON(w, http.StatusAccepted, out)
}

func (s *Server) authorizedTenantPool(w http.ResponseWriter, r *http.Request, orgID string) (*meta.TenantPool, bool) {
	orgID = strings.TrimSpace(orgID)
	if orgID == "" {
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant pool not found")
		return nil, false
	}
	pool, err := s.meta.GetTenantPoolByOrganization(r.Context(), orgID)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			writeControlPlaneStatusError(w, http.StatusNotFound, "tenant pool not found")
			return nil, false
		}
		writeControlPlaneStatusError(w, backendErrorStatus(r.Context(), err), "tenant pool lookup failed")
		return nil, false
	}
	return pool, true
}

func (s *Server) tenantPoolSizes(ctx context.Context, organizationID string) (int, int, error) {
	freeSize, err := s.meta.CountFreeTenantPoolBindings(ctx, organizationID)
	if err != nil {
		return 0, 0, fmt.Errorf("tenant pool free size lookup failed")
	}
	slotSize, err := s.meta.CountTenantPoolFreeSlots(ctx, organizationID)
	if err != nil {
		return 0, 0, fmt.Errorf("tenant pool slot size lookup failed")
	}
	return freeSize, slotSize, nil
}

func (s *Server) recordTenantPoolCapacity(poolID, organizationID string, size, freeSize int) {
	if poolID == "" || organizationID == "" {
		return
	}
	if size < 0 {
		size = 0
	}
	if freeSize < 0 {
		freeSize = 0
	}
	metrics.RecordTenantPoolCapacity(poolID, organizationID, "size", float64(size))
	metrics.RecordTenantPoolCapacity(poolID, organizationID, "free", float64(freeSize))
}

func (s *Server) firstManagedOrganization(ctx context.Context, cred tenant.CredentialProvisionRequest) (string, error) {
	identity, err := s.resolveTiDBCloudIdentity(ctx, cred, "provision_org_resolve")
	if err != nil {
		return "", err
	}
	return identity.OrganizationID, nil
}

func (s *Server) createFreePoolTenants(ctx context.Context, poolID string, count int, cred tenant.CredentialProvisionRequest, quotaOpt *quotaRequest) ([]*provisionTenantResult, error) {
	if s.defaultTenantProvider == tenant.ProviderTiDBCloudNativeShared {
		return s.createFreeSharedPoolTenants(ctx, poolID, count, cred, quotaOpt)
	}
	manager, ok := s.provisioner.(tenant.TenantPoolClusterManager)
	if !ok {
		return nil, fmt.Errorf("tenant pool provisioning not enabled")
	}
	if count <= 0 {
		return []*provisionTenantResult{}, nil
	}
	provider := tenant.ProviderTiDBCloudNative
	tenantIDs := make([]string, 0, count)
	now := time.Now().UTC()
	stageStarted := time.Now()
	logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_free_tenants_insert_started",
		"provider", provider,
		"pool_id", poolID,
		"count", count)...)
	for i := 0; i < count; i++ {
		tenantID := token.NewID()
		if err := s.insertPendingPoolTenant(ctx, tenantID, provider, now); err != nil {
			s.cleanupPoolProvisionedClusters(ctx, nil, cred, tenantIDs, "insert_pending_tenant_error")
			return nil, err
		}
		tenantIDs = append(tenantIDs, tenantID)
	}
	logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_free_tenants_insert_done",
		"provider", provider,
		"pool_id", poolID,
		"count", len(tenantIDs),
		"duration_ms", durationMillis(stageStarted))...)
	var opts tenant.QuotaUpdateOptions
	opts.TenantPoolID = poolID
	stageStarted = time.Now()
	logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_batch_create_started",
		"provider", provider,
		"pool_id", poolID,
		"count", len(tenantIDs),
		"quota_requested", quotaOpt != nil)...)
	clusters, _, err := manager.BatchProvisionFreeClustersWithCredentialsAndQuota(ctx, tenantIDs, cred, opts)
	if err != nil && len(clusters) == 0 {
		logger.Error(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_batch_create_failed",
			"provider", provider,
			"pool_id", poolID,
			"count", len(tenantIDs),
			"duration_ms", durationMillis(stageStarted),
			"error", err)...)
		s.cleanupPoolProvisionedClusters(ctx, clusters, cred, tenantIDs, "batch_provision_error")
		for _, tenantID := range tenantIDs {
			s.markTenantPoolTenantFailed(ctx, tenantID, "batch_provision_error")
		}
		return nil, err
	}
	logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_batch_create_done",
		"provider", provider,
		"pool_id", poolID,
		"requested_count", len(tenantIDs),
		"cluster_count", len(clusters),
		"duration_ms", durationMillis(stageStarted),
		"has_error", err != nil)...)
	if err != nil {
		logger.Warn(ctx, "admin_tenant_pool_batch_metadata_incomplete",
			zap.String("pool_id", poolID),
			zap.Int("cluster_count", len(clusters)),
			zap.Error(err))
	}
	cleanupOnError := true
	defer func() {
		if cleanupOnError {
			s.cleanupPoolProvisionedClusters(ctx, clusters, cred, tenantIDs, "metadata_error")
		}
	}()
	results := make([]*provisionTenantResult, 0, len(clusters))
	pendingResume := make([]*tenant.ClusterInfo, 0, len(clusters))
	persistedTenants := make(map[string]struct{}, len(clusters))
	discardedTenants := make(map[string]struct{}, len(clusters))
	cloudProvider, region := provisioningCloudRegion(s.provisioner)
	poolOrgID := ""
	for _, cluster := range clusters {
		if cluster == nil {
			continue
		}
		tenantID := strings.TrimSpace(cluster.TenantID)
		if tenantID == "" {
			logger.Warn(ctx, "admin_tenant_pool_cluster_tenant_missing",
				zap.String("pool_id", poolID),
				zap.String("cluster_id", cluster.ClusterID))
			s.cleanupPoolProvisionedClusters(ctx, []*tenant.ClusterInfo{cluster}, cred, nil, "cluster_tenant_missing")
			continue
		}
		orgID := strings.TrimSpace(cluster.OrganizationID)
		if orgID == "" {
			if poolOrgID == "" {
				pool, lookupErr := s.meta.GetTenantPoolByID(ctx, poolID)
				if lookupErr != nil && !errors.Is(lookupErr, meta.ErrNotFound) {
					cleanupOnError = true
					return nil, lookupErr
				}
				if pool != nil {
					poolOrgID = strings.TrimSpace(pool.OrganizationID)
				}
			}
			orgID = poolOrgID
		}
		if orgID == "" {
			logger.Warn(ctx, "admin_tenant_pool_cluster_org_missing",
				zap.String("pool_id", poolID),
				zap.String("tenant_id", tenantID),
				zap.String("cluster_id", cluster.ClusterID))
			s.cleanupPoolProvisionedClusters(ctx, []*tenant.ClusterInfo{cluster}, cred, []string{tenantID}, "cluster_org_missing")
			discardedTenants[tenantID] = struct{}{}
			continue
		}
		cluster.OrganizationID = orgID
		persistedTenants[tenantID] = struct{}{}
		stageStarted = time.Now()
		if err := s.meta.UpsertTenantTiDBCloudOrgBinding(ctx, &meta.TenantTiDBCloudOrgBinding{
			TenantID:       tenantID,
			OrganizationID: orgID,
			ClusterID:      cluster.ClusterID,
			PoolID:         poolID,
			PoolStatus:     meta.TenantPoolBindingFree,
			CreatedAt:      now,
			UpdatedAt:      now,
		}); err != nil {
			cleanupOnError = true
			return nil, err
		}
		logProvisionStage(ctx, "admin_tenant_pool_free_tenant_binding_persisted", tenantID, provider, stageStarted,
			"pool_id", poolID,
			"organization_id", orgID,
			"cluster_id", cluster.ClusterID)
		res := &provisionTenantResult{
			TenantID:       tenantID,
			Status:         meta.TenantPending,
			Provider:       provider,
			CloudProvider:  cloudProvider,
			Region:         region,
			OrganizationID: orgID,
		}
		if poolClusterConnectionReady(cluster) {
			stageStarted = time.Now()
			if err := s.persistPoolTenantConnection(ctx, cluster, provider); err != nil {
				cleanupOnError = true
				return nil, err
			}
			if err := s.meta.UpdateTenantStatus(ctx, tenantID, meta.TenantProvisioning); err != nil {
				cleanupOnError = true
				return nil, err
			}
			res.Status = meta.TenantProvisioning
			dbtls := dbTLSForProvisionedTenant(provider)
			res.TenantDSN = tenantDSN(cluster.Username, cluster.Password, cluster.Host, cluster.Port, cluster.DBName, dbtls, provider)
			logProvisionStage(ctx, "admin_tenant_pool_free_tenant_connection_persisted", tenantID, provider, stageStarted,
				"pool_id", poolID,
				"organization_id", orgID,
				"cluster_id", cluster.ClusterID,
				"db_tls", dbtls,
				"status", meta.TenantProvisioning)
		} else {
			stageStarted = time.Now()
			if err := s.persistPoolTenantProvisionSeed(ctx, cluster, provider); err != nil {
				cleanupOnError = true
				return nil, err
			}
			pendingResume = append(pendingResume, cluster)
			logProvisionStage(ctx, "admin_tenant_pool_free_tenant_seed_persisted", tenantID, provider, stageStarted,
				"pool_id", poolID,
				"organization_id", orgID,
				"cluster_id", cluster.ClusterID,
				"status", meta.TenantPending)
		}
		if quotaOpt != nil {
			stageStarted = time.Now()
			quotaReq := *quotaOpt
			quotaReq.TenantID = tenantID
			if err := s.applyQuotaLocalConfig(ctx, tenantID, quotaReq); err != nil {
				cleanupOnError = true
				return nil, err
			}
			logProvisionStage(ctx, "admin_tenant_pool_free_tenant_quota_local_config_applied", tenantID, provider, stageStarted,
				"pool_id", poolID,
				"organization_id", orgID,
				"cluster_id", cluster.ClusterID)
		}
		results = append(results, res)
	}
	// A partial TiDB Cloud response can omit some requested tenant IDs. Those
	// pending local tenants have no recoverable cluster metadata, so fail them
	// while preserving tenants that were persisted above.
	for _, tenantID := range tenantIDs {
		if _, ok := persistedTenants[tenantID]; ok {
			continue
		}
		if _, ok := discardedTenants[tenantID]; ok {
			continue
		}
		s.markTenantPoolTenantFailed(ctx, tenantID, "missing_cluster_response")
	}
	if len(clusters) > 0 && len(results) == 0 {
		cleanupOnError = false
		return nil, fmt.Errorf("no tenant pool clusters could be persisted")
	}
	cleanupOnError = false
	s.startPoolClustersMetadataResume(ctx, poolID, pendingResume, cred)
	return results, nil
}

func (s *Server) createFreeSharedPoolTenants(ctx context.Context, poolID string, count int, _ tenant.CredentialProvisionRequest, quotaOpt *quotaRequest) ([]*provisionTenantResult, error) {
	return s.createFreeSharedPoolTenantsWithResize(ctx, poolID, count, tenant.CredentialProvisionRequest{}, quotaOpt, nil)
}

func (s *Server) createFreeSharedPoolTenantsWithResize(ctx context.Context, poolID string, count int, _ tenant.CredentialProvisionRequest, quotaOpt *quotaRequest, resize *meta.ManagedSharedDBPoolWaveResize) ([]*provisionTenantResult, error) {
	if count <= 0 {
		return []*provisionTenantResult{}, nil
	}
	if _, ok := s.provisioner.(tenant.SharedDBPoolProvisioner); !ok {
		return nil, fmt.Errorf("shared tenant pool provisioning not enabled")
	}
	sharedCred, err := s.sharedDBCloudCredentials()
	if err != nil {
		return nil, err
	}
	pool, err := s.meta.GetTenantPoolByID(ctx, poolID)
	if err != nil {
		return nil, err
	}
	organizationID := strings.TrimSpace(pool.OrganizationID)
	if organizationID == "" {
		return nil, fmt.Errorf("shared tenant pool organization is required")
	}
	partitions, err := s.stageSharedPoolTenantWave(ctx, poolID, organizationID, sharedCred, count, quotaOpt, resize)
	if err != nil {
		return nil, err
	}
	continuationIDs := make([]int64, 0, len(partitions))
	createIDs := make([]int64, 0, len(partitions))
	for _, partition := range partitions {
		continuationIDs = append(continuationIDs, partition.db.ID)
		if partition.created {
			createIDs = append(createIDs, partition.db.ID)
		}
	}
	sort.Slice(continuationIDs, func(i, j int) bool { return continuationIDs[i] < continuationIDs[j] })
	sort.Slice(createIDs, func(i, j int) bool { return createIDs[i] < createIDs[j] })
	_, provisionErr := s.provisionManagedSharedDBPoolsBatchWithCredentials(ctx, createIDs, sharedCred)
	// The whole logical reservation wave is durable before Cloud creation
	// starts, so allocators can only consume genuinely unreserved tail capacity.
	s.scheduleManagedSharedDBContinuations(ctx, continuationIDs)
	results := make([]*provisionTenantResult, 0, count)
	resultPoolIDs := make([]int64, 0, count)
	for _, partition := range partitions {
		for _, tenantID := range partition.tenantIDs {
			results = append(results, &provisionTenantResult{TenantID: tenantID, Status: meta.TenantPending,
				Provider: tenant.ProviderTiDBCloudNativeShared, CloudProvider: partition.db.CloudProvider,
				Region: partition.db.Region, OrganizationID: organizationID})
			resultPoolIDs = append(resultPoolIDs, partition.db.ID)
		}
	}
	continuationPoolStatuses := make(map[int64]meta.TenantStatus, len(continuationIDs))
	for _, dbID := range continuationIDs {
		sharedDB, err := s.meta.GetSharedDB(ctx, dbID)
		if err != nil {
			provisionErr = errors.Join(provisionErr, err)
			continue
		}
		status := meta.TenantPending
		switch sharedDB.Status {
		case meta.SharedDBStatusProvisioning:
			status = meta.TenantProvisioning
		case meta.SharedDBStatusActive:
			status = meta.TenantActive
		}
		continuationPoolStatuses[dbID] = status
	}
	for i, dbID := range resultPoolIDs {
		if status, ok := continuationPoolStatuses[dbID]; ok {
			results[i].Status = status
		}
	}
	return results, provisionErr
}

type sharedPoolTenantPartition struct {
	db        *meta.SharedDB
	tenantIDs []string
	created   bool
}

func (s *Server) stageSharedPoolTenantWave(ctx context.Context, poolID, organizationID string, cred tenant.CredentialProvisionRequest, count int, quotaOpt *quotaRequest, resize *meta.ManagedSharedDBPoolWaveResize) ([]sharedPoolTenantPartition, error) {
	if count <= 0 {
		return nil, nil
	}
	// The durable wave fills existing physical capacity before inserting a new
	// db_pool row. Capacity selection and each per-DB tenant_count increment are
	// batched under the same transaction as placements and memberships, avoiding
	// both one-physical-DB-per-refill fragmentation and per-tenant hot-row writes.
	remaining := count
	provisioningKey := sharedDBProvisioningKey(cred)
	maxTenants, _ := s.managedSharedDBPolicy()
	physicalCount := (count + maxTenants - 1) / maxTenants
	plans := make([]meta.ManagedSharedDBPoolWavePlan, 0, physicalCount)
	now := time.Now().UTC()
	for remaining > 0 {
		row, err := s.newManagedSharedDBPlan(ctx, organizationID, provisioningKey)
		if err != nil {
			return nil, err
		}
		assigned := min(remaining, maxTenants)
		members := make([]meta.ManagedSharedDBPoolWaveMember, 0, assigned)
		for range assigned {
			tenantID := token.NewID()
			quota, quotaErr := s.sharedTenantQuotaConfig(tenantID, provisionTenantOptions{Quota: quotaOpt})
			if quotaErr != nil {
				return nil, quotaErr
			}
			members = append(members, meta.ManagedSharedDBPoolWaveMember{
				Tenant: &meta.Tenant{ID: tenantID, Status: meta.TenantPending, DBPasswordCipher: []byte{}, DBTLS: true,
					Provider: tenant.ProviderTiDBCloudNativeShared, SchemaVersion: 1, CreatedAt: now, UpdatedAt: now},
				Membership: &meta.TenantPoolMembership{TenantID: tenantID, TiDBCloudOrganizationID: organizationID,
					PoolID: poolID, PoolStatus: meta.TenantPoolBindingFree, CreatedAt: now, UpdatedAt: now},
				Quota: quota,
			})
		}
		plans = append(plans, meta.ManagedSharedDBPoolWavePlan{DB: row, Members: members})
		remaining -= assigned
	}
	created, err := s.meta.CreateManagedSharedDBPoolTenantWaveWithResize(ctx, plans, resize)
	if err != nil {
		metrics.RecordSharedDBPoolWave(metrics.ResultForError(err), len(plans), count)
		return nil, err
	}
	createdPhysical := 0
	for _, result := range created {
		if result.Created {
			createdPhysical++
		}
	}
	metrics.RecordSharedDBPoolWave("ok", createdPhysical, count)
	partitions := make([]sharedPoolTenantPartition, 0, len(created))
	for _, result := range created {
		row, loadErr := s.meta.GetSharedDB(ctx, result.DBID)
		if loadErr != nil {
			return nil, loadErr
		}
		partitions = append(partitions, sharedPoolTenantPartition{db: row, tenantIDs: result.TenantIDs, created: result.Created})
	}
	return partitions, nil
}

func (s *Server) provisionManagedSharedDBPoolsBatch(ctx context.Context, dbIDs []int64) (string, error) {
	if len(dbIDs) == 0 {
		return "", nil
	}
	cred, err := s.sharedDBCloudCredentials()
	if err != nil {
		return "", err
	}
	return s.provisionManagedSharedDBPoolsBatchWithCredentials(ctx, dbIDs, cred)
}

func (s *Server) provisionManagedSharedDBPoolsBatchWithCredentials(ctx context.Context, dbIDs []int64, cred tenant.CredentialProvisionRequest) (string, error) {
	if len(dbIDs) == 0 {
		return "", nil
	}
	provisioner, ok := s.provisioner.(tenant.SharedDBPoolProvisioner)
	if !ok {
		return "", fmt.Errorf("provisioner does not support managed shared db pools")
	}
	type allocationGroup struct {
		dbIDs []int64
	}
	groupsByIdentity := make(map[string]*allocationGroup)
	groups := make([]*allocationGroup, 0)
	for _, dbID := range dbIDs {
		row, err := s.meta.GetSharedDB(ctx, dbID)
		if err != nil {
			return "", err
		}
		identity := sharedDBAllocationIdentity(row.TiDBCloudOrganizationID, row.ProvisioningKey)
		group := groupsByIdentity[identity]
		if group == nil {
			group = &allocationGroup{}
			groupsByIdentity[identity] = group
			groups = append(groups, group)
		}
		group.dbIDs = append(group.dbIDs, dbID)
	}
	type groupResult struct {
		organizationID string
		err            error
	}
	results := make([]groupResult, len(groups))
	var wg sync.WaitGroup
	for groupIndex := range groups {
		wg.Add(1)
		go func() {
			defer wg.Done()
			results[groupIndex].organizationID, results[groupIndex].err = s.provisionManagedSharedDBPoolAllocationGroup(
				ctx, provisioner, groups[groupIndex].dbIDs, cred)
		}()
	}
	wg.Wait()
	resolvedOrg := ""
	var groupErr error
	for _, result := range results {
		if resolvedOrg == "" {
			resolvedOrg = result.organizationID
		}
		groupErr = errors.Join(groupErr, result.err)
	}
	return resolvedOrg, groupErr
}

func (s *Server) provisionManagedSharedDBPoolAllocationGroup(ctx context.Context, provisioner tenant.SharedDBPoolProvisioner, dbIDs []int64, cred tenant.CredentialProvisionRequest) (string, error) {
	first, err := s.meta.GetSharedDB(ctx, dbIDs[0])
	if err != nil {
		return "", err
	}
	resolvedOrg := strings.TrimSpace(first.TiDBCloudOrganizationID)
	return resolvedOrg, s.provisionManagedSharedDBPoolsBatchForGroup(ctx, provisioner, dbIDs, cred)
}

func (s *Server) provisionManagedSharedDBPoolsBatchForGroup(ctx context.Context, provisioner tenant.SharedDBPoolProvisioner, dbIDs []int64, cred tenant.CredentialProvisionRequest) error {
	return s.meta.WithSharedDBPoolWorkClaims(ctx, dbIDs, func(claimCtx context.Context, ownedIDs []int64) error {
		return s.provisionManagedSharedDBPoolsBatchGroupClaimed(claimCtx, provisioner, ownedIDs, cred)
	})
}

func (s *Server) provisionManagedSharedDBPoolsBatchGroupClaimed(ctx context.Context, provisioner tenant.SharedDBPoolProvisioner, dbIDs []int64, cred tenant.CredentialProvisionRequest) error {
	rows := make(map[int64]*meta.SharedDB, len(dbIDs))
	loadRequests := make([]tenant.SharedDBPoolLoadRequest, 0, len(dbIDs))
	for _, dbID := range dbIDs {
		row, err := s.meta.GetSharedDB(ctx, dbID)
		if err != nil {
			return err
		}
		// Re-read after taking per-pool ownership. A concurrent direct
		// continuation may have persisted the cluster before this group
		// acquired its claim; such rows continue through metadata polling.
		if row.Status != meta.SharedDBStatusPending || strings.TrimSpace(row.ClusterID) != "" {
			continue
		}
		rows[dbID] = row
		loadRequests = append(loadRequests, tenant.SharedDBPoolLoadRequest{DBPoolID: dbID, DBPoolUUID: row.UUID})
	}
	if len(rows) == 0 {
		return nil
	}

	discovered, loadErr := loadManagedSharedDBPoolsBeforeCreate(ctx, provisioner, loadRequests, cred)
	adopted, persistErr := s.persistManagedSharedDBPoolCloudResults(ctx, discovered, rows)
	if loadErr != nil || persistErr != nil {
		// A failed lookup cannot prove that the Cloud POST did not already
		// succeed, so retry discovery instead of issuing another create.
		return errors.Join(loadErr, persistErr)
	}

	requests := make([]tenant.SharedDBPoolCreateRequest, 0, len(rows)-len(adopted))
	for _, dbID := range dbIDs {
		row := rows[dbID]
		if row == nil {
			continue
		}
		if _, ok := adopted[dbID]; ok {
			continue
		}
		plain, err := s.pool.Decrypt(ctx, row.PasswordCipher)
		if err != nil {
			return err
		}
		if row.SpendingLimit == nil {
			return fmt.Errorf("managed db pool %d has no spending target", dbID)
		}
		requests = append(requests, tenant.SharedDBPoolCreateRequest{DBPoolID: dbID, DBPoolUUID: row.UUID,
			CustomerOrganizationID: strings.TrimSpace(row.TiDBCloudOrganizationID), DatabaseName: row.Name,
			RootPassword: string(plain), SpendingLimitMonthly: *row.SpendingLimit})
	}
	if len(requests) == 0 {
		return nil
	}

	type batchResult struct {
		err error
	}
	batchSize := s.managedSharedDBCloudBatchSize
	if batchSize <= 0 {
		batchSize = DefaultManagedSharedDBCloudBatchSize
	}
	batchCount := (len(requests) + batchSize - 1) / batchSize
	batchResults := make([]batchResult, batchCount)
	batches := make([][]tenant.SharedDBPoolCreateRequest, 0, batchCount)
	for start := 0; start < len(requests); start += batchSize {
		end := min(start+batchSize, len(requests))
		batches = append(batches, append([]tenant.SharedDBPoolCreateRequest(nil), requests[start:end]...))
	}
	var wg sync.WaitGroup
	for batchIndex := range batches {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := ctx.Err(); err != nil {
				batchResults[batchIndex].err = err
				return
			}
			batchResults[batchIndex].err = s.provisionManagedSharedDBPoolsBatchChunkClaimed(ctx, provisioner, batches[batchIndex], rows, cred)
		}()
	}
	wg.Wait()
	var batchErr error
	for _, result := range batchResults {
		batchErr = errors.Join(batchErr, result.err)
	}
	return batchErr
}

func loadManagedSharedDBPoolsBeforeCreate(ctx context.Context, provisioner tenant.SharedDBPoolProvisioner, requests []tenant.SharedDBPoolLoadRequest, cred tenant.CredentialProvisionRequest) ([]*tenant.SharedDBPoolInfo, error) {
	if loader, ok := provisioner.(tenant.SharedDBPoolBatchLoader); ok {
		return loader.BatchLoadSharedDBPoolsWithCredentials(ctx, requests, cred)
	}
	loaded := make([]*tenant.SharedDBPoolInfo, 0, len(requests))
	for _, request := range requests {
		info, err := provisioner.LoadSharedDBPoolWithCredentials(ctx, request.DBPoolID, request.DBPoolUUID, request.ClusterID, cred)
		if err != nil {
			return loaded, err
		}
		if info != nil {
			loaded = append(loaded, info)
		}
	}
	return loaded, nil
}

func (s *Server) persistManagedSharedDBPoolCloudResults(ctx context.Context, results []*tenant.SharedDBPoolInfo, rows map[int64]*meta.SharedDB) (map[int64]struct{}, error) {
	persisted := make(map[int64]struct{}, len(results))
	var persistErr error
	for _, info := range results {
		if info == nil || rows[info.DBPoolID] == nil || rows[info.DBPoolID].UUID != info.DBPoolUUID {
			persistErr = errors.Join(persistErr, fmt.Errorf("shared db batch returned an unknown db pool"))
			continue
		}
		row := rows[info.DBPoolID]
		logicalOrganizationID := strings.TrimSpace(row.TiDBCloudOrganizationID)
		if logicalOrganizationID == "" {
			persistErr = errors.Join(persistErr, fmt.Errorf("managed shared db pool %d customer organization is required", info.DBPoolID))
			continue
		}
		if info.DBName == "" {
			info.DBName = row.Name
		}
		// Use sharedDBTLSMode (not public-cloud "true") so LocalClustersAPI
		// plaintext TiDB can complete schema ensure + free-tenant activation.
		if err := s.meta.UpdateManagedSharedDBPoolCloudResult(ctx, &meta.SharedDB{ID: info.DBPoolID,
			TiDBCloudOrganizationID: logicalOrganizationID, ClusterID: info.ClusterID, Host: info.Host,
			Port: info.Port, User: info.Username, PasswordCipher: row.PasswordCipher, Name: info.DBName,
			TLSMode: sharedDBTLSMode()}); err != nil {
			persistErr = errors.Join(persistErr, err)
			continue
		}
		persisted[info.DBPoolID] = struct{}{}
	}
	return persisted, persistErr
}

func (s *Server) provisionManagedSharedDBPoolsBatchChunkClaimed(ctx context.Context, provisioner tenant.SharedDBPoolProvisioner, requests []tenant.SharedDBPoolCreateRequest, rows map[int64]*meta.SharedDB, cred tenant.CredentialProvisionRequest) error {
	if err := s.acquireManagedSharedDBCloudBatchSlot(ctx); err != nil {
		return err
	}
	created, createErr := func() ([]*tenant.SharedDBPoolInfo, error) {
		defer s.releaseManagedSharedDBCloudBatchSlot()
		return provisioner.BatchProvisionSharedDBPoolsWithCredentials(ctx, requests, cred)
	}()
	if createErr != nil && len(created) == 0 {
		logger.Warn(ctx, "managed_shared_db_batch_create_failed",
			zap.Int("db_pool_count", len(requests)), zap.Error(createErr))
		return createErr
	}
	persistedIDs, persistErr := s.persistManagedSharedDBPoolCloudResults(ctx, created, rows)
	if createErr != nil {
		logger.Warn(ctx, "managed_shared_db_batch_create_partial",
			zap.Int("requested", len(requests)), zap.Int("persisted", len(persistedIDs)), zap.Error(createErr))
	}
	return errors.Join(createErr, persistErr)
}

func (s *Server) acquireManagedSharedDBCloudBatchSlot(ctx context.Context) error {
	s.managedSharedDBCloudBatchSlotsOnce.Do(func() {
		if s.managedSharedDBCloudBatchSlots == nil {
			s.managedSharedDBCloudBatchSlots = make(chan struct{}, defaultManagedSharedDBCloudBatchConcurrency)
		}
	})
	select {
	case s.managedSharedDBCloudBatchSlots <- struct{}{}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (s *Server) releaseManagedSharedDBCloudBatchSlot() {
	<-s.managedSharedDBCloudBatchSlots
}

func (s *Server) markTenantPoolTenantFailed(ctx context.Context, tenantID, reason string) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return
	}
	if err := s.meta.UpdateTenantStatus(ctx, tenantID, meta.TenantFailed); err != nil {
		logger.Warn(ctx, "admin_tenant_pool_mark_failed_failed",
			zap.String("tenant_id", tenantID),
			zap.String("reason", reason),
			zap.Error(err))
	}
}

func (s *Server) persistPoolTenantProvisionSeed(ctx context.Context, cluster *tenant.ClusterInfo, provider string) error {
	if cluster == nil {
		return fmt.Errorf("cluster info is required")
	}
	if strings.TrimSpace(cluster.TenantID) == "" {
		return fmt.Errorf("cluster tenant id is required")
	}
	if strings.TrimSpace(cluster.ClusterID) == "" {
		return fmt.Errorf("cluster id is required")
	}
	if strings.TrimSpace(cluster.Password) == "" {
		return fmt.Errorf("cluster password is required")
	}
	cipherPass, err := s.pool.Encrypt(ctx, []byte(cluster.Password))
	if err != nil {
		return err
	}
	return s.meta.UpdateTenantConnection(ctx, cluster.TenantID, &meta.Tenant{
		DBPasswordCipher: cipherPass,
		DBName:           cluster.DBName,
		DBTLS:            dbTLSForProvisionedTenant(provider),
		Provider:         provider,
		ClusterID:        cluster.ClusterID,
	})
}

func (s *Server) persistPoolTenantConnection(ctx context.Context, cluster *tenant.ClusterInfo, provider string) error {
	if cluster == nil {
		return fmt.Errorf("cluster info is required")
	}
	if !poolClusterConnectionReady(cluster) {
		return fmt.Errorf("cluster connection metadata is incomplete")
	}
	if strings.TrimSpace(cluster.Password) == "" {
		return fmt.Errorf("cluster password is required")
	}
	cipherPass, err := s.pool.Encrypt(ctx, []byte(cluster.Password))
	if err != nil {
		return err
	}
	return s.meta.UpdateTenantConnection(ctx, cluster.TenantID, &meta.Tenant{
		DBHost:           cluster.Host,
		DBPort:           cluster.Port,
		DBUser:           cluster.Username,
		DBPasswordCipher: cipherPass,
		DBName:           cluster.DBName,
		DBTLS:            dbTLSForProvisionedTenant(provider),
		Provider:         provider,
		ClusterID:        cluster.ClusterID,
	})
}

func poolClusterConnectionReady(cluster *tenant.ClusterInfo) bool {
	return cluster != nil &&
		strings.TrimSpace(cluster.TenantID) != "" &&
		strings.TrimSpace(cluster.ClusterID) != "" &&
		strings.TrimSpace(cluster.OrganizationID) != "" &&
		strings.TrimSpace(cluster.Host) != "" &&
		cluster.Port > 0 &&
		strings.TrimSpace(cluster.Username) != "" &&
		strings.TrimSpace(cluster.Password) != "" &&
		strings.TrimSpace(cluster.DBName) != ""
}

func (s *Server) startPoolClustersMetadataResume(ctx context.Context, poolID string, clusters []*tenant.ClusterInfo, cred tenant.CredentialProvisionRequest) {
	clusters = compactPoolResumeClusters(clusters)
	if len(clusters) == 0 {
		return
	}
	job := &tenantPoolResumeJob{}
	if poolID != "" {
		actual, loaded := s.tenantPoolResumeJobs.LoadOrStore(poolID, job)
		if loaded {
			if existing, ok := actual.(*tenantPoolResumeJob); ok {
				existing.rerun.Store(true)
			}
			logger.Info(ctx, "admin_tenant_pool_metadata_resume_skipped",
				zap.String("pool_id", poolID),
				zap.String("reason", "metadata_resume_already_running_rerun_requested"),
				zap.Int("cluster_count", len(clusters)))
			return
		}
	}
	clusterCopies := make([]*tenant.ClusterInfo, 0, len(clusters))
	for _, cluster := range clusters {
		copy := *cluster
		clusterCopies = append(clusterCopies, &copy)
	}
	s.startServerWorker(ctx, func(workerCtx context.Context) {
		if poolID != "" {
			defer s.tenantPoolResumeJobs.Delete(poolID)
		}
		workerCtx, cancel := context.WithTimeout(workerCtx, tenantPoolMetadataResumeWaitTimeout)
		defer cancel()

		for {
			started := time.Now()
			s.resumePoolClustersMetadataGroups(workerCtx, started, poolID, clusterCopies, cred)
			if poolID == "" || !job.rerun.Swap(false) {
				return
			}
			next, err := s.pendingTenantPoolResumeClusters(workerCtx, poolID, len(clusterCopies))
			if err != nil {
				logger.Warn(workerCtx, "admin_tenant_pool_metadata_resume_rerun_list_failed",
					zap.String("pool_id", poolID),
					zap.Error(err))
				return
			}
			clusterCopies = next
			if len(clusterCopies) == 0 {
				return
			}
		}
	})
}

func (s *Server) resumePoolClustersMetadataGroups(ctx context.Context, started time.Time, poolID string, clusters []*tenant.ClusterInfo, cred tenant.CredentialProvisionRequest) {
	waitStarted := time.Now()
	groups := poolMetadataResumeGroups(clusters, tenantPoolMetadataResumeGroupSize)
	if len(groups) == 0 {
		return
	}
	overallResult := "ok"
	var overallMu sync.Mutex
	recordGroupResult := func(result string) {
		overallMu.Lock()
		defer overallMu.Unlock()
		if adminTenantPoolMetadataResumeResultRank(result) > adminTenantPoolMetadataResumeResultRank(overallResult) {
			overallResult = result
		}
	}
	var wg sync.WaitGroup
	for i, group := range groups {
		groupIndex := i
		group := group
		wg.Add(1)
		go func() {
			defer wg.Done()
			groupStarted := time.Now()
			updated, err := s.waitForPoolClustersMetadata(ctx, group, cred)
			groupResult := metrics.ResultForError(err)
			groupDuration := time.Since(groupStarted)
			metrics.RecordOperation(adminTenantPoolMetricsComponent, "metadata_resume_group_wait", groupResult, groupDuration)
			metrics.RecordTenantPoolMetadataResumeWait(poolID, poolResumeOrganizationID(group), "group", groupResult, groupDuration)
			recordGroupResult(groupResult)
			if err != nil {
				logger.Warn(ctx, "admin_tenant_pool_metadata_resume_batch_failed",
					zap.String("pool_id", poolID),
					zap.Int("group_index", groupIndex),
					zap.Int("group_count", len(groups)),
					zap.Int("cluster_count", len(group)),
					zap.Strings("tenant_ids", poolResumeTenantIDs(group)),
					zap.Strings("cluster_ids", poolResumeClusterIDs(group)),
					zap.Error(err))
			}
			for _, cluster := range updated {
				s.completePoolClusterMetadataResume(ctx, started, cluster)
			}
		}()
	}
	wg.Wait()
	waitDuration := time.Since(waitStarted)
	metrics.RecordOperation(adminTenantPoolMetricsComponent, "metadata_resume_wait", overallResult, waitDuration)
	metrics.RecordTenantPoolMetadataResumeWait(poolID, poolResumeOrganizationID(clusters), "batch", overallResult, waitDuration)
}

func adminTenantPoolMetadataResumeResultRank(result string) int {
	switch result {
	case "ok":
		return 0
	case "canceled":
		return 1
	case "deadline_exceeded":
		return 2
	case "bad_conn":
		return 3
	case "error":
		return 4
	default:
		return 4
	}
}

func poolMetadataResumeGroups(clusters []*tenant.ClusterInfo, groupSize int) [][]*tenant.ClusterInfo {
	if groupSize <= 0 {
		groupSize = 10
	}
	groups := make([][]*tenant.ClusterInfo, 0, (len(clusters)+groupSize-1)/groupSize)
	for start := 0; start < len(clusters); start += groupSize {
		end := start + groupSize
		if end > len(clusters) {
			end = len(clusters)
		}
		groups = append(groups, clusters[start:end])
	}
	return groups
}

func poolResumeTenantIDs(clusters []*tenant.ClusterInfo) []string {
	out := make([]string, 0, len(clusters))
	for _, cluster := range clusters {
		if cluster == nil {
			continue
		}
		out = append(out, strings.TrimSpace(cluster.TenantID))
	}
	return out
}

func poolResumeClusterIDs(clusters []*tenant.ClusterInfo) []string {
	out := make([]string, 0, len(clusters))
	for _, cluster := range clusters {
		if cluster == nil {
			continue
		}
		out = append(out, strings.TrimSpace(cluster.ClusterID))
	}
	return out
}

func poolResumeOrganizationID(clusters []*tenant.ClusterInfo) string {
	for _, cluster := range clusters {
		if cluster == nil {
			continue
		}
		if orgID := strings.TrimSpace(cluster.OrganizationID); orgID != "" {
			return orgID
		}
	}
	return ""
}

func compactPoolResumeClusters(clusters []*tenant.ClusterInfo) []*tenant.ClusterInfo {
	out := clusters[:0]
	for _, cluster := range clusters {
		if cluster == nil || strings.TrimSpace(cluster.TenantID) == "" || strings.TrimSpace(cluster.ClusterID) == "" {
			continue
		}
		out = append(out, cluster)
	}
	return out
}

func (s *Server) waitForPoolClustersMetadata(ctx context.Context, clusters []*tenant.ClusterInfo, cred tenant.CredentialProvisionRequest) ([]*tenant.ClusterInfo, error) {
	if batchWaiter, ok := s.provisioner.(tenant.TenantPoolClusterMetadataBatchWaiter); ok {
		return batchWaiter.WaitForPoolClustersMetadata(ctx, clusters, cred)
	}
	waiter, ok := s.provisioner.(tenant.TenantPoolClusterMetadataWaiter)
	if !ok {
		return nil, fmt.Errorf("metadata waiter unavailable")
	}
	updated := make([]*tenant.ClusterInfo, 0, len(clusters))
	var errs []error
	for _, cluster := range clusters {
		got, err := waiter.WaitForPoolClusterMetadata(ctx, cluster, cred)
		if err != nil {
			errs = append(errs, fmt.Errorf("tenant %s cluster %s: %w", cluster.TenantID, cluster.ClusterID, err))
			continue
		}
		updated = append(updated, got)
	}
	return updated, errors.Join(errs...)
}

func (s *Server) completePoolClusterMetadataResume(ctx context.Context, started time.Time, updated *tenant.ClusterInfo) {
	ctx, cancel := s.tenantPoolMetadataResumePersistContext(ctx)
	defer cancel()

	if !poolClusterConnectionReady(updated) {
		if updated != nil {
			logger.Warn(ctx, "admin_tenant_pool_metadata_resume_incomplete",
				zap.String("tenant_id", updated.TenantID),
				zap.String("cluster_id", updated.ClusterID))
		}
		return
	}
	if err := s.persistPoolTenantConnection(ctx, updated, tenant.ProviderTiDBCloudNative); err != nil {
		logger.Warn(ctx, "admin_tenant_pool_metadata_resume_persist_failed",
			zap.String("tenant_id", updated.TenantID),
			zap.String("cluster_id", updated.ClusterID),
			zap.Error(err))
		return
	}
	statusUpdated, err := s.meta.UpdateTenantStatusIf(ctx, updated.TenantID, meta.TenantPending, meta.TenantProvisioning)
	if err != nil {
		logger.Warn(ctx, "admin_tenant_pool_metadata_resume_status_failed",
			zap.String("tenant_id", updated.TenantID),
			zap.String("cluster_id", updated.ClusterID),
			zap.Error(err))
		return
	}
	if !statusUpdated {
		logger.Info(ctx, "admin_tenant_pool_metadata_resume_status_skipped",
			zap.String("tenant_id", updated.TenantID),
			zap.String("cluster_id", updated.ClusterID),
			zap.String("reason", "status_changed"))
		return
	}
	cloudProvider, region := provisioningCloudRegion(s.provisioner)
	dbtls := dbTLSForProvisionedTenant(tenant.ProviderTiDBCloudNative)
	logProvisionStage(ctx, "admin_tenant_pool_metadata_resume_done", updated.TenantID, tenant.ProviderTiDBCloudNative, started, "cluster_id", updated.ClusterID, "organization_id", updated.OrganizationID)
	s.startProvisionedTenantSchemaInit(ctx, &provisionTenantResult{
		TenantID:       updated.TenantID,
		Status:         meta.TenantProvisioning,
		Provider:       tenant.ProviderTiDBCloudNative,
		TenantDSN:      tenantDSN(updated.Username, updated.Password, updated.Host, updated.Port, updated.DBName, dbtls, tenant.ProviderTiDBCloudNative),
		CloudProvider:  cloudProvider,
		Region:         region,
		OrganizationID: strings.TrimSpace(updated.OrganizationID),
	})
}

func (s *Server) tenantPoolMetadataResumePersistContext(ctx context.Context) (context.Context, context.CancelFunc) {
	parent := backgroundWithTrace(ctx)
	if s.forkWorkerCtx != nil {
		parent = contextWithTrace(s.forkWorkerCtx, ctx)
	}
	return context.WithTimeout(parent, tenantPoolMetadataResumePersistTimeout)
}

func (s *Server) cleanupPoolProvisionedClusters(ctx context.Context, clusters []*tenant.ClusterInfo, cred tenant.CredentialProvisionRequest, tenantIDs []string, reason string) {
	cleanupCtx := backgroundWithTrace(ctx)
	seenTenants := make(map[string]struct{}, len(tenantIDs)+len(clusters))
	deprovisionFailed := make(map[string]struct{}, len(clusters))
	for _, tenantID := range tenantIDs {
		tenantID = strings.TrimSpace(tenantID)
		if tenantID == "" {
			continue
		}
		seenTenants[tenantID] = struct{}{}
	}
	for _, cluster := range clusters {
		if cluster == nil {
			continue
		}
		tenantID := strings.TrimSpace(cluster.TenantID)
		if tenantID != "" {
			if _, ok := seenTenants[tenantID]; !ok {
				if err := s.meta.UpdateTenantStatus(context.Background(), tenantID, meta.TenantFailed); err != nil {
					logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_mark_failed", zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(err))
				}
			}
		}
		if strings.TrimSpace(cluster.ClusterID) == "" {
			continue
		}
		t := &meta.Tenant{
			ID:        tenantID,
			Provider:  tenant.ProviderTiDBCloudNative,
			ClusterID: strings.TrimSpace(cluster.ClusterID),
			DBHost:    cluster.Host,
			DBPort:    cluster.Port,
			DBUser:    cluster.Username,
			DBName:    cluster.DBName,
		}
		if err := s.deprovisionTenantCluster(cleanupCtx, t, cred); err != nil {
			if tenantID != "" {
				deprovisionFailed[tenantID] = struct{}{}
			}
			logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_cluster_failed", zap.String("tenant_id", tenantID), zap.String("cluster_id", cluster.ClusterID), zap.String("reason", reason), zap.Error(err))
		}
	}
	for tenantID := range seenTenants {
		if _, failed := deprovisionFailed[tenantID]; failed {
			if err := s.meta.UpdateTenantStatus(context.Background(), tenantID, meta.TenantFailed); err != nil {
				logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_mark_failed", zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(err))
			}
			continue
		}
		_ = s.meta.RevokeTenantAPIKeys(cleanupCtx, tenantID)
		if err := s.meta.MarkTenantDeleted(cleanupCtx, tenantID); err != nil {
			logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_mark_deleted_failed", zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(err))
			_ = s.meta.UpdateTenantStatus(context.Background(), tenantID, meta.TenantFailed)
		} else {
			s.clearLocalTenantMetrics(tenantID)
		}
	}
}

func (s *Server) insertPendingPoolTenant(ctx context.Context, tenantID, provider string, now time.Time) error {
	autoProfile, err := s.defaultAutoEmbeddingProfileForTenant(ctx, tenantID, provider, now)
	if err != nil {
		return fmt.Errorf("build tenant auto-embedding profile: %w", err)
	}
	if err := s.meta.InsertTenant(ctx, &meta.Tenant{
		ID:               tenantID,
		Status:           meta.TenantPending,
		DBHost:           "",
		DBPort:           0,
		DBUser:           "",
		DBPasswordCipher: []byte{},
		DBName:           "",
		DBTLS:            true,
		Provider:         provider,
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		return fmt.Errorf("persist tenant: %w", err)
	}
	if autoProfile != nil {
		if err := s.meta.UpsertTenantAutoEmbeddingProfile(ctx, autoProfile); err != nil {
			_ = s.meta.UpdateTenantStatus(context.Background(), tenantID, meta.TenantFailed)
			return fmt.Errorf("persist tenant auto-embedding profile: %w", err)
		}
	}
	return nil
}

func (s *Server) cleanupCreatedPoolTenants(ctx context.Context, results []*provisionTenantResult, cred tenant.CredentialProvisionRequest, reason string) {
	cleanupCtx := backgroundWithTrace(ctx)
	for _, res := range results {
		if res == nil || strings.TrimSpace(res.TenantID) == "" {
			continue
		}
		tenantID := strings.TrimSpace(res.TenantID)
		t, err := s.meta.GetTenant(cleanupCtx, tenantID)
		if err != nil {
			logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_get_tenant_failed", zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(err))
			continue
		}
		if t.Provider == tenant.ProviderTiDBCloudNativeShared {
			if t.Status == meta.TenantDeleted {
				logger.Info(cleanupCtx, "admin_tenant_pool_cleanup_shared_member_skipped",
					zap.String("tenant_id", tenantID), zap.String("reason", reason),
					zap.String("skip_reason", "already_deleted"))
				continue
			}
			if err := s.discardUnclaimedSharedPoolTenant(cleanupCtx, t); err != nil {
				if errors.Is(err, meta.ErrTenantPoolMembershipNotFree) {
					logger.Info(cleanupCtx, "admin_tenant_pool_cleanup_shared_member_skipped",
						zap.String("tenant_id", tenantID), zap.String("reason", reason),
						zap.String("skip_reason", "free_membership_no_longer_owned"))
					continue
				}
				logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_shared_member_failed",
					zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(err))
				if updated, markErr := s.meta.MarkUnclaimedSharedTenantFailed(cleanupCtx, tenantID); markErr != nil {
					logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_shared_member_mark_failed_failed",
						zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(markErr))
				} else if !updated {
					logger.Info(cleanupCtx, "admin_tenant_pool_cleanup_shared_member_mark_failed_skipped",
						zap.String("tenant_id", tenantID), zap.String("reason", reason),
						zap.String("skip_reason", "free_membership_no_longer_owned"))
				}
			}
			continue
		}
		if err := s.deprovisionTenantCluster(cleanupCtx, t, cred); err != nil {
			logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_created_cluster_failed", zap.String("tenant_id", tenantID), zap.String("cluster_id", t.ClusterID), zap.String("reason", reason), zap.Error(err))
			_ = s.meta.UpdateTenantStatus(context.Background(), tenantID, meta.TenantFailed)
			continue
		}
		_ = s.meta.RevokeTenantAPIKeys(cleanupCtx, tenantID)
		if err := s.meta.MarkTenantDeleted(cleanupCtx, tenantID); err != nil {
			logger.Warn(cleanupCtx, "admin_tenant_pool_cleanup_mark_deleted_failed", zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(err))
		} else {
			s.clearLocalTenantMetrics(tenantID)
		}
	}
}

func (s *Server) discardUnclaimedSharedPoolTenant(ctx context.Context, t *meta.Tenant) error {
	if t == nil {
		return meta.ErrNotFound
	}
	fsID, err := s.meta.ResolveFsID(ctx, t.ID)
	if err != nil {
		return err
	}
	placement, err := s.meta.GetTenantPlacement(ctx, fsID)
	if err != nil {
		return err
	}
	if err := s.meta.DiscardUnclaimedSharedTenant(ctx, t.ID, fsID, placement.DbID, s.sharedDBReopenRatio); err != nil {
		return err
	}
	_ = s.meta.RevokeTenantAPIKeys(ctx, t.ID)
	s.clearLocalTenantMetrics(t.ID)
	return nil
}

func (s *Server) deleteTenantPoolMetadata(ctx context.Context, poolID, reason string) {
	if strings.TrimSpace(poolID) == "" {
		return
	}
	cleanupCtx := backgroundWithTrace(ctx)
	if err := s.meta.DeleteTenantPool(cleanupCtx, poolID); err != nil && !errors.Is(err, meta.ErrNotFound) {
		logger.Warn(cleanupCtx, "admin_tenant_pool_delete_metadata_failed", zap.String("pool_id", poolID), zap.String("reason", reason), zap.Error(err))
	}
}

func firstResultOrganizationID(results []*provisionTenantResult) string {
	for _, res := range results {
		if res != nil && strings.TrimSpace(res.OrganizationID) != "" {
			return strings.TrimSpace(res.OrganizationID)
		}
	}
	return ""
}

func (s *Server) replenishTenantPoolAsync(ctx context.Context, pool *meta.TenantPool, cred tenant.CredentialProvisionRequest) {
	s.replenishTenantPoolAsyncWithStarters(ctx, pool, cred, s.startServerWorker, s.startServerWorker)
}

func (s *Server) enqueueTenantPoolLeaderReplenishment(ctx context.Context, pool *meta.TenantPool, cred tenant.CredentialProvisionRequest) bool {
	if pool == nil || pool.PoolID == "" || pool.OrganizationID == "" || pool.Size <= 0 {
		return false
	}
	if s.leader != nil && !s.leader.IsLeader() {
		return false
	}
	return s.enqueueTenantPoolLeaderReconcile(ctx, pool.PoolID, func(workerCtx context.Context) {
		s.replenishTenantPool(workerCtx, pool, cred)
	})
}

func (s *Server) replenishTenantPoolAsyncWithStarters(
	ctx context.Context,
	pool *meta.TenantPool,
	cred tenant.CredentialProvisionRequest,
	workStarter tenantPoolWorkerStarter,
	timerStarter tenantPoolWorkerStarter,
) {
	if pool == nil || pool.PoolID == "" || pool.OrganizationID == "" || pool.Size <= 0 {
		return
	}
	if workStarter == nil || timerStarter == nil {
		return
	}
	decision := s.requestTenantPoolReplenishmentAt(pool.PoolID, time.Now())
	if !decision.start {
		if decision.scheduleAfter > 0 {
			s.scheduleTenantPoolReplenishment(ctx, pool, cred, decision.scheduleAfter, workStarter, timerStarter)
		}
		return
	}
	workerCtx := backgroundWithTrace(ctx)
	if !workStarter(workerCtx, func(ctx context.Context) {
		defer func() {
			finished := s.finishTenantPoolReplenishmentAt(pool.PoolID, time.Now())
			if finished.scheduleAfter > 0 {
				s.scheduleTenantPoolReplenishment(ctx, pool, cred, finished.scheduleAfter, workStarter, timerStarter)
			}
		}()
		s.replenishTenantPool(ctx, pool, cred)
	}) {
		finished := s.finishTenantPoolReplenishmentAt(pool.PoolID, time.Now())
		if finished.scheduleAfter > 0 {
			s.scheduleTenantPoolReplenishment(ctx, pool, cred, finished.scheduleAfter, workStarter, timerStarter)
		}
	}
}

func (s *Server) replenishTenantPool(ctx context.Context, pool *meta.TenantPool, cred tenant.CredentialProvisionRequest) {
	replenishStarted := time.Now()
	metricResult := "ok"
	defer func() {
		metrics.RecordOperation(adminTenantPoolMetricsComponent, "replenish", metricResult, time.Since(replenishStarted))
	}()
	sharedProvider := s.defaultTenantProvider == tenant.ProviderTiDBCloudNativeShared
	createdAny := false
	sharedWaveRemaining := 0
	// A started shared wave accounts for in-flight pending/provisioning slots;
	// sharedWaveRemaining is the hard cap that guarantees termination.
	for {
		if err := ctx.Err(); err != nil {
			metricResult = adminTenantPoolMetricResult(err)
			return
		}
		current, err := s.meta.GetTenantPoolByID(ctx, pool.PoolID)
		if err != nil {
			if !errors.Is(err, meta.ErrNotFound) {
				logger.Warn(ctx, "admin_tenant_pool_replenish_get_pool_failed", zap.String("pool_id", pool.PoolID), zap.Error(err))
				metricResult = "error"
			} else {
				metricResult = "not_found"
			}
			return
		}
		if current.Status != meta.TenantPoolActive || current.OrganizationID == "" || current.Size <= 0 {
			metricResult = "skipped"
			return
		}
		freeSize, err := s.meta.CountFreeTenantPoolBindings(ctx, current.OrganizationID)
		if err != nil {
			logger.Warn(ctx, "admin_tenant_pool_replenish_free_count_failed", zap.String("pool_id", current.PoolID), zap.Error(err))
			metricResult = "error"
			return
		}
		s.recordTenantPoolCapacity(current.PoolID, current.OrganizationID, current.Size, freeSize)
		if !createdAny && !s.tenantPoolBelowRefillWatermark(freeSize, current.Size) {
			logger.Info(ctx, "admin_tenant_pool_replenish_skipped",
				zap.String("pool_id", current.PoolID),
				zap.String("organization_id", current.OrganizationID),
				zap.Int("pool_size", current.Size),
				zap.Int("free_size", freeSize),
				zap.Float64("refill_free_ratio", s.effectiveTenantPoolRefillFreeRatio()))
			metricResult = "noop"
			return
		}

		// Active inventory is immediately usable. Shared pending/provisioning
		// inventory offsets missing capacity until the stuck-pool watchdog makes
		// the physical attempt terminal. This avoids a refill/watchdog race at an
		// independent time-based lease boundary.
		slotSize := 0
		if sharedProvider {
			plannedSize, countErr := s.meta.CountTenantPoolPlannedSlots(ctx, current.OrganizationID)
			if countErr != nil {
				logger.Warn(ctx, "admin_tenant_pool_replenish_count_failed", zap.String("pool_id", current.PoolID), zap.Error(countErr))
				metricResult = "error"
				return
			}
			slotSize = freeSize + plannedSize
		} else {
			var countErr error
			slotSize, countErr = s.meta.CountTenantPoolFreeSlots(ctx, current.OrganizationID)
			if countErr != nil {
				logger.Warn(ctx, "admin_tenant_pool_replenish_count_failed", zap.String("pool_id", current.PoolID), zap.Error(countErr))
				metricResult = "error"
				return
			}
		}
		missing := current.Size - slotSize
		if missing <= 0 {
			if !createdAny {
				metricResult = "noop"
			}
			return
		}
		if sharedProvider {
			maxTenants, _ := s.managedSharedDBPolicy()
			if !createdAny {
				sharedWaveRemaining = managedSharedDBRefillTenantCount(missing, maxTenants, s.managedSharedDBCloudBatchSize)
			}
			if sharedWaveRemaining <= 0 {
				return
			}
			missing = min(missing, sharedWaveRemaining)
		}
		results, err := s.createFreePoolTenants(ctx, current.PoolID, missing, cred, nil)
		if err != nil {
			logger.Warn(ctx, "admin_tenant_pool_replenish_failed", zap.String("pool_id", current.PoolID), zap.Error(err))
			metricResult = "cluster_error"
			return
		}
		for _, res := range results {
			s.startProvisionedTenantSchemaInit(ctx, res)
		}
		if len(results) > 0 {
			createdAny = true
			metricResult = "ok"
			if sharedProvider {
				sharedWaveRemaining -= len(results)
			}
		}
		if !sharedProvider || len(results) == 0 || sharedWaveRemaining <= 0 {
			return
		}
	}
}

func (s *Server) scheduleTenantPoolReplenishment(
	ctx context.Context,
	pool *meta.TenantPool,
	cred tenant.CredentialProvisionRequest,
	delay time.Duration,
	workStarter tenantPoolWorkerStarter,
	timerStarter tenantPoolWorkerStarter,
) {
	workerCtx := backgroundWithTrace(ctx)
	if timerStarter(workerCtx, func(ctx context.Context) {
		timer := time.NewTimer(delay)
		defer timer.Stop()
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
			s.replenishTenantPoolAsyncWithStarters(ctx, pool, cred, workStarter, timerStarter)
		}
	}) {
		return
	}
	s.clearTenantPoolReplenishmentScheduled(pool.PoolID)
}

func (s *Server) tenantPoolBelowRefillWatermark(freeSize, poolSize int) bool {
	if poolSize <= 0 {
		return false
	}
	if freeSize < 0 {
		freeSize = 0
	}
	return float64(freeSize)/float64(poolSize) < s.effectiveTenantPoolRefillFreeRatio()
}

func (s *Server) effectiveTenantPoolRefillFreeRatio() float64 {
	if s == nil {
		return DefaultTenantPoolRefillFreeRatio
	}
	return normalizeTenantPoolRefillFreeRatio(s.tenantPoolRefillFreeRatio)
}

func (s *Server) resumePendingTenantPoolAsync(ctx context.Context, pool *meta.TenantPool, cred tenant.CredentialProvisionRequest) {
	if pool == nil || pool.OrganizationID == "" || pool.PoolID == "" {
		return
	}
	if actual, ok := s.tenantPoolResumeJobs.Load(pool.PoolID); ok {
		if existing, typeOK := actual.(*tenantPoolResumeJob); typeOK {
			existing.rerun.Store(true)
		}
		return
	}
	if !s.beginTenantPoolResumeScan(pool.PoolID) {
		return
	}
	workerCtx := backgroundWithTrace(ctx)
	if !s.startServerWorker(workerCtx, func(ctx context.Context) {
		empty := false
		defer func() { s.finishTenantPoolResumeScan(pool.PoolID, empty) }()
		clusters, err := s.pendingTenantPoolResumeClusters(ctx, pool.PoolID, pool.Size)
		if err != nil {
			logger.Warn(ctx, "admin_tenant_pool_pending_resume_list_failed", zap.String("pool_id", pool.PoolID), zap.Error(err))
			return
		}
		empty = len(clusters) == 0
		s.startPoolClustersMetadataResume(ctx, pool.PoolID, clusters, cred)
	}) {
		s.finishTenantPoolResumeScan(pool.PoolID, false)
	}
}

func (s *Server) pendingTenantPoolResumeClusters(ctx context.Context, poolID string, limit int) ([]*tenant.ClusterInfo, error) {
	pool, err := s.meta.GetTenantPoolByID(ctx, poolID)
	if err != nil {
		return nil, err
	}
	if pool.OrganizationID == "" || pool.Status != meta.TenantPoolActive {
		return []*tenant.ClusterInfo{}, nil
	}
	if limit <= 0 || limit < pool.Size {
		limit = pool.Size
	}
	rows, err := s.meta.ListPendingTenantPoolBindingsForResume(ctx, pool.OrganizationID, limit)
	if err != nil {
		return nil, err
	}
	clusters := make([]*tenant.ClusterInfo, 0, len(rows))
	for _, row := range rows {
		plainPass, err := s.pool.Decrypt(ctx, row.Tenant.DBPasswordCipher)
		if err != nil || strings.TrimSpace(string(plainPass)) == "" {
			logger.Warn(ctx, "admin_tenant_pool_pending_resume_password_failed",
				zap.String("tenant_id", row.Tenant.ID),
				zap.String("pool_id", pool.PoolID),
				zap.Error(err))
			continue
		}
		cluster := clusterInfoFromTenant(&row.Tenant)
		cluster.OrganizationID = row.Binding.OrganizationID
		cluster.Password = string(plainPass)
		clusters = append(clusters, cluster)
	}
	return clusters, nil
}

func (s *Server) tenantPoolLock(poolID string) *sync.Mutex {
	if strings.TrimSpace(poolID) == "" {
		return &sync.Mutex{}
	}
	v, _ := s.tenantPoolLocks.LoadOrStore(poolID, &sync.Mutex{})
	return v.(*sync.Mutex)
}

func (s *Server) beginTenantPoolReplenishmentAt(poolID string, now time.Time) bool {
	return s.requestTenantPoolReplenishmentAt(poolID, now).start
}

func (s *Server) requestTenantPoolReplenishmentAt(poolID string, now time.Time) tenantPoolReplenishDecision {
	if strings.TrimSpace(poolID) == "" {
		return tenantPoolReplenishDecision{}
	}
	value, _ := s.tenantPoolReplenishJobs.LoadOrStore(poolID, &tenantPoolWorkGate{})
	gate := value.(*tenantPoolWorkGate)
	gate.mu.Lock()
	defer gate.mu.Unlock()
	if gate.running {
		gate.rerun = true
		return tenantPoolReplenishDecision{}
	}
	if now.Before(gate.nextAllowed) {
		gate.rerun = true
		if gate.scheduled {
			return tenantPoolReplenishDecision{}
		}
		gate.scheduled = true
		return tenantPoolReplenishDecision{scheduleAfter: gate.nextAllowed.Sub(now)}
	}
	gate.running = true
	gate.rerun = false
	gate.scheduled = false
	return tenantPoolReplenishDecision{start: true}
}

func (s *Server) finishTenantPoolReplenishmentAt(poolID string, now time.Time) tenantPoolReplenishDecision {
	value, ok := s.tenantPoolReplenishJobs.Load(poolID)
	if !ok {
		return tenantPoolReplenishDecision{}
	}
	gate := value.(*tenantPoolWorkGate)
	gate.mu.Lock()
	defer gate.mu.Unlock()
	gate.running = false
	gate.nextAllowed = now.Add(tenantPoolReplenishMinInterval)
	if !gate.rerun || gate.scheduled {
		return tenantPoolReplenishDecision{}
	}
	gate.scheduled = true
	return tenantPoolReplenishDecision{scheduleAfter: tenantPoolReplenishMinInterval}
}

func (s *Server) clearTenantPoolReplenishmentScheduled(poolID string) {
	value, ok := s.tenantPoolReplenishJobs.Load(poolID)
	if !ok {
		return
	}
	gate := value.(*tenantPoolWorkGate)
	gate.mu.Lock()
	gate.scheduled = false
	gate.mu.Unlock()
}

func (s *Server) beginTenantPoolResumeScan(poolID string) bool {
	return s.beginTenantPoolResumeScanAt(poolID, time.Now())
}

func (s *Server) beginTenantPoolResumeScanAt(poolID string, now time.Time) bool {
	if strings.TrimSpace(poolID) == "" {
		return false
	}
	value, _ := s.tenantPoolResumeScans.LoadOrStore(poolID, &tenantPoolWorkGate{})
	gate := value.(*tenantPoolWorkGate)
	gate.mu.Lock()
	defer gate.mu.Unlock()
	if gate.running || now.Before(gate.nextAllowed) {
		return false
	}
	gate.running = true
	return true
}

func (s *Server) finishTenantPoolResumeScan(poolID string, empty bool) {
	s.finishTenantPoolResumeScanAt(poolID, time.Now(), empty)
}

func (s *Server) finishTenantPoolResumeScanAt(poolID string, now time.Time, empty bool) {
	value, ok := s.tenantPoolResumeScans.Load(poolID)
	if !ok {
		return
	}
	gate := value.(*tenantPoolWorkGate)
	gate.mu.Lock()
	gate.running = false
	if empty {
		gate.nextAllowed = now.Add(tenantPoolPendingResumeEmptyInterval)
	} else {
		gate.nextAllowed = time.Time{}
	}
	gate.mu.Unlock()
}

func (s *Server) tenantPoolCreateLock(cred tenant.CredentialProvisionRequest) *sync.Mutex {
	key := strings.TrimSpace(cred.PublicKey)
	if key == "" {
		return &sync.Mutex{}
	}
	// A TiDB Cloud public key belongs to a single org, so this serializes the
	// first-create path before the org id is discoverable from managed clusters.
	v, _ := s.tenantPoolCreateLocks.LoadOrStore(key, &sync.Mutex{})
	return v.(*sync.Mutex)
}

func tenantPoolCreateDatabaseLockKey(cred tenant.CredentialProvisionRequest) string {
	key := strings.TrimSpace(cred.PublicKey)
	if key == "" {
		return ""
	}
	return "create:" + key
}

// claimAdminTenantFromPool tries to hand out a pre-warmed tenant from the
// caller org's tenant pool. sharedPoolMatched reports that the org has a
// registered shared-schema pool instead, so callers can route provisioning
// through the shared provider.
func (s *Server) claimAdminTenantFromPool(ctx context.Context, cred tenant.CredentialProvisionRequest, quotaOpt *quotaRequest) (*provisionTenantResult, *meta.TenantPool, bool, bool, error) {
	return s.claimAdminTenantFromPoolWithAccess(ctx, cred, quotaOpt, nil, meta.TenantMetadata{})
}

func (s *Server) claimAdminTenantFromPoolWithAccess(ctx context.Context, cred tenant.CredentialProvisionRequest, quotaOpt *quotaRequest, access *tiDBCloudAccessProfile, metadata meta.TenantMetadata) (*provisionTenantResult, *meta.TenantPool, bool, bool, error) {
	claimStarted := time.Now()
	manager, ok := s.provisioner.(tenant.TenantPoolClusterManager)
	if !ok {
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_skipped", "provider", tenant.ProviderTiDBCloudNative, "reason", "pool_manager_unavailable", "duration_ms", durationMillis(claimStarted))...)
		return nil, nil, false, false, nil
	}
	if _, ok := s.provisioner.(tenant.ManagedClusterLister); !ok {
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_skipped", "provider", tenant.ProviderTiDBCloudNative, "reason", "managed_cluster_lister_unavailable", "duration_ms", durationMillis(claimStarted))...)
		return nil, nil, false, false, nil
	}
	if access == nil {
		var err error
		access, err = s.resolveTiDBCloudAccessProfile(ctx, cred, "tenant_pool_claim")
		if err != nil {
			return nil, nil, false, false, err
		}
	}
	if quotaOpt != nil {
		if err := s.validateQuotaSetRequestWithoutSpendingLimit(*quotaOpt); err != nil {
			return nil, nil, false, false, err
		}
	}
	stageStarted := time.Now()
	orgID := strings.TrimSpace(access.OrganizationID)
	if orgID == "" {
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_org_lookup_done", "provider", tenant.ProviderTiDBCloudNative, "organization_id", orgID, "duration_ms", durationMillis(stageStarted), "has_error", true)...)
		return nil, nil, false, false, fmt.Errorf("TiDB Cloud organization is required")
	}
	logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_org_lookup_done", "provider", tenant.ProviderTiDBCloudNative, "organization_id", orgID, "duration_ms", durationMillis(stageStarted))...)
	cleanupCred := cred
	defer s.startTenantFailedCleanupAsync(ctx, orgID, cleanupCred)
	stageStarted = time.Now()
	pool, err := s.meta.GetTenantPoolByOrganization(ctx, orgID)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			// A registered external/manual shared DB remains a provision target,
			// but it must not bypass an existing logical pool's mixed inventory.
			// This check therefore runs only after the logical pool lookup misses.
			if sharedDB, sharedErr := s.meta.FindSharedDBForOrg(ctx, orgID); sharedErr == nil && sharedDB != nil {
				logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_skipped",
					"provider", tenant.ProviderTiDBCloudNative, "organization_id", orgID,
					"reason", "external_shared_pool_registered", "duration_ms", durationMillis(claimStarted))...)
				return nil, nil, false, true, nil
			} else if sharedErr != nil && !errors.Is(sharedErr, meta.ErrNotFound) {
				return nil, nil, false, false, sharedErr
			}
			logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_pool_lookup_missed", "provider", tenant.ProviderTiDBCloudNative, "organization_id", orgID, "duration_ms", durationMillis(stageStarted))...)
			return nil, nil, false, false, nil
		}
		return nil, nil, false, false, err
	}
	logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_pool_lookup_done", "provider", tenant.ProviderTiDBCloudNative, "pool_id", pool.PoolID, "organization_id", orgID, "pool_size", pool.Size, "pool_status", pool.Status, "duration_ms", durationMillis(stageStarted))...)
	if pool.Status != meta.TenantPoolActive {
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_skipped", "provider", tenant.ProviderTiDBCloudNative, "pool_id", pool.PoolID, "organization_id", orgID, "reason", "pool_inactive", "pool_status", pool.Status, "duration_ms", durationMillis(claimStarted))...)
		return nil, nil, false, false, nil
	}
	// Native replenishment needs the request-scoped customer credential. Shared
	// bulk replenishment uses server-owned credentials and is exclusively driven
	// by the leader reconciler; requests retain only their direct miss fallback.
	if s.defaultTenantProvider != tenant.ProviderTiDBCloudNativeShared {
		defer s.replenishTenantPoolAsync(ctx, pool, cred)
	}
	s.resumePendingTenantPoolAsync(ctx, pool, cred)
	stageStarted = time.Now()
	claimCtx := ctx
	var claimedQuotaOpt *quotaRequest
	claimOnce := func() (tenantPoolClaimSelection, error) {
		var nativeCandidate *meta.TenantWithTiDBCloudOrgBinding
		if rows, listErr := s.meta.ListFreeTenantPoolBindings(claimCtx, orgID, false, 1); listErr != nil {
			return tenantPoolClaimSelection{}, listErr
		} else if len(rows) > 0 {
			nativeCandidate = &rows[0]
		}
		sharedCandidate, sharedErr := s.meta.GetOldestFreeTenantPoolMembership(claimCtx, pool.PoolID)
		if sharedErr != nil && !errors.Is(sharedErr, meta.ErrNotFound) {
			return tenantPoolClaimSelection{}, sharedErr
		}
		if sharedCandidate != nil && (nativeCandidate == nil || sharedCandidate.Membership.CreatedAt.Before(nativeCandidate.Binding.CreatedAt) ||
			(sharedCandidate.Membership.CreatedAt.Equal(nativeCandidate.Binding.CreatedAt) && sharedCandidate.Tenant.ID < nativeCandidate.Tenant.ID)) {
			sharedQuotaOpt := quotaRequestForShared(quotaOpt)
			if access.IsFree {
				sharedQuotaOpt, sharedErr = s.normalizeTiDBCloudFreeProvisionQuota(sharedQuotaOpt)
				if sharedErr != nil {
					return tenantPoolClaimSelection{}, sharedErr
				}
			} else if sharedQuotaOpt != nil && sharedQuotaOpt.anySet() {
				if err := s.validateQuotaSetRequest(*sharedQuotaOpt); err != nil {
					return tenantPoolClaimSelection{}, err
				}
			}
			if access.IsFree {
				if headroomErr := s.checkFreePoolClaimHeadroom(claimCtx, orgID); headroomErr != nil {
					return tenantPoolClaimSelection{}, headroomErr
				}
			}
			result, claimErr := s.claimSharedTenantPoolMember(claimCtx, pool, sharedCandidate, sharedQuotaOpt, metadata, access.IsFree)
			if claimErr != nil {
				return tenantPoolClaimSelection{}, claimErr
			}
			return tenantPoolClaimSelection{sharedResult: result}, nil
		}
		if nativeCandidate == nil {
			return tenantPoolClaimSelection{}, nil
		}
		nativeQuotaOpt := quotaOpt
		if access.IsFree {
			var normalizeErr error
			nativeQuotaOpt, normalizeErr = s.normalizeTiDBCloudFreeProvisionQuota(nativeQuotaOpt)
			if normalizeErr != nil {
				return tenantPoolClaimSelection{}, normalizeErr
			}
		} else if nativeQuotaOpt != nil {
			if err := s.validateQuotaSetRequest(*nativeQuotaOpt); err != nil {
				return tenantPoolClaimSelection{}, err
			}
		}
		var claimQuotaPatch meta.QuotaConfigPatch
		if nativeQuotaOpt != nil {
			var patchErr error
			claimQuotaPatch, patchErr = quotaConfigPatchFromRequest(*nativeQuotaOpt)
			if patchErr != nil {
				return tenantPoolClaimSelection{}, patchErr
			}
		}
		if access.IsFree {
			if headroomErr := s.checkFreePoolClaimHeadroom(claimCtx, orgID); headroomErr != nil {
				return tenantPoolClaimSelection{}, headroomErr
			}
		}
		var row *meta.TenantWithTiDBCloudOrgBinding
		var claimErr error
		if access.IsFree {
			row, claimErr = s.meta.ClaimFreeTenantPoolBindingWithMetadataForFreeOrganization(claimCtx, orgID, nativeCandidate.Tenant.ID, claimQuotaPatch, metadata)
		} else {
			row, claimErr = s.meta.ClaimFreeTenantPoolBindingWithMetadata(claimCtx, orgID, nativeCandidate.Tenant.ID, claimQuotaPatch, metadata)
		}
		if claimErr != nil {
			return tenantPoolClaimSelection{}, claimErr
		}
		claimedQuotaOpt = nativeQuotaOpt
		return tenantPoolClaimSelection{native: row}, nil
	}
	var selection tenantPoolClaimSelection
	if access.IsFree {
		err = s.meta.WithTiDBCloudFreeQuotaLock(ctx, orgID, func(lockCtx context.Context) error {
			return s.meta.WithTenantPoolLock(lockCtx, pool.PoolID, func(poolLockCtx context.Context) error {
				claimCtx = poolLockCtx
				var claimErr error
				selection, claimErr = retryTenantPoolClaimCAS(claimOnce)
				return claimErr
			})
		})
	} else {
		selection, err = retryTenantPoolClaimCAS(claimOnce)
	}
	if err != nil {
		return nil, nil, false, false, err
	}
	if selection.sharedResult != nil {
		return selection.sharedResult, pool, true, false, nil
	}
	quotaOpt = claimedQuotaOpt
	row := selection.native
	if row == nil {
		logger.Info(ctx, "server_event", eventFields(ctx, "admin_tenant_pool_claim_free_tenant_missed", "provider", tenant.ProviderTiDBCloudNative, "pool_id", pool.PoolID, "organization_id", orgID, "duration_ms", durationMillis(stageStarted))...)
		return nil, pool, false, false, nil
	}
	logProvisionStage(ctx, "admin_tenant_pool_claim_free_tenant_claimed", row.Tenant.ID, row.Tenant.Provider, stageStarted, "pool_id", pool.PoolID, "organization_id", orgID, "cluster_id", row.Binding.ClusterID, "status", row.Tenant.Status)
	usedAt := time.Now().UTC()
	var opts tenant.QuotaUpdateOptions
	if access.IsFree {
		zero := int64(0)
		opts.TiDBCloudSpendingLimitMonthly = &zero
	} else if quotaOpt != nil {
		opts.TiDBCloudSpendingLimitMonthly = quotaOpt.TiDBCloudSpendingLimit
	}
	cluster := clusterInfoFromTenant(&row.Tenant)
	stageStarted = time.Now()
	cloudCfg, err := manager.MarkClusterPoolUsed(ctx, cluster, cred, usedAt, opts)
	if err != nil {
		s.releaseClaimedPoolTenant(ctx, manager, cluster, cred, row.Tenant.ID, "mark_used_error")
		return nil, nil, false, false, err
	}
	logProvisionStage(ctx, "admin_tenant_pool_claim_cluster_marked_used", row.Tenant.ID, row.Tenant.Provider, stageStarted, "pool_id", pool.PoolID, "organization_id", orgID, "cluster_id", cluster.ClusterID, "quota_requested", quotaOpt != nil)
	success := false
	defer func() {
		if !success {
			s.releaseClaimedPoolTenant(ctx, manager, cluster, cred, row.Tenant.ID, "claim_error")
		}
	}()
	if !access.IsFree && (quotaOpt == nil || quotaOpt.TiDBCloudSpendingLimit == nil) &&
		cloudCfg != nil && cloudCfg.TiDBCloudSpendingLimitMonthly != nil {
		if err := s.meta.SetQuotaConfigPatch(ctx, row.Tenant.ID, meta.QuotaConfigPatch{
			TiDBCloudSpendingLimit: cloudCfg.TiDBCloudSpendingLimitMonthly,
		}); err != nil {
			return nil, nil, false, false, fmt.Errorf("persist default spending limit after pool claim: %w", err)
		}
	}
	stageStarted = time.Now()
	plainPass, err := s.pool.Decrypt(ctx, row.Tenant.DBPasswordCipher)
	if err != nil {
		return nil, nil, false, false, err
	}
	logProvisionStage(ctx, "admin_tenant_pool_claim_db_password_decrypted", row.Tenant.ID, row.Tenant.Provider, stageStarted, "pool_id", pool.PoolID, "organization_id", orgID)
	stageStarted = time.Now()
	apiToken, apiKeyID, err := s.issueOwnerAPIKey(ctx, row.Tenant.ID, "default", 1, time.Time{}, apiKeyIssueSource{})
	if err != nil {
		return nil, nil, false, false, err
	}
	logProvisionStage(ctx, "admin_tenant_pool_claim_api_key_issued", row.Tenant.ID, row.Tenant.Provider, stageStarted, "pool_id", pool.PoolID, "organization_id", orgID, "api_key_id", apiKeyID)
	cloudProvider, region := provisioningCloudRegion(s.provisioner)
	success = true
	logProvisionStage(ctx, "admin_tenant_pool_claim_done", row.Tenant.ID, row.Tenant.Provider, claimStarted, "pool_id", pool.PoolID, "organization_id", orgID, "cluster_id", cluster.ClusterID, "api_key_id", apiKeyID, "status", row.Tenant.Status)
	return &provisionTenantResult{
		TenantID:       row.Tenant.ID,
		APIKey:         apiToken,
		APIKeyID:       apiKeyID,
		Status:         row.Tenant.Status,
		Provider:       tenant.ProviderTiDBCloudNative,
		TenantDSN:      tenantDSN(row.Tenant.DBUser, string(plainPass), row.Tenant.DBHost, row.Tenant.DBPort, row.Tenant.DBName, row.Tenant.DBTLS, row.Tenant.Provider),
		CloudProvider:  cloudProvider,
		Region:         region,
		OrganizationID: row.Binding.OrganizationID,
	}, pool, true, false, nil
}

func (s *Server) checkFreePoolClaimHeadroom(ctx context.Context, organizationID string) error {
	count, err := s.meta.CountTiDBCloudOrganizationTenants(ctx, organizationID)
	if err != nil {
		return err
	}
	if count >= s.tidbCloudFreePlanLimits.TenantCount {
		return tenant.ErrTiDBCloudFreeTenantLimitReached
	}
	return nil
}

func (s *Server) claimSharedTenantPoolMember(ctx context.Context, pool *meta.TenantPool, row *meta.TenantWithPoolMembership, quotaOpt *quotaRequest, metadata meta.TenantMetadata, freeOrganization bool) (*provisionTenantResult, error) {
	if pool == nil || row == nil {
		return nil, meta.ErrNotFound
	}
	// Resolve immutable free-member routing metadata before the claim commits.
	// A transient read failure must not strand a used member whose owner token
	// was committed but never returned to the caller.
	fsID, err := s.meta.ResolveFsID(ctx, row.Tenant.ID)
	if err != nil {
		return nil, err
	}
	placement, err := s.meta.GetTenantPlacement(ctx, fsID)
	if err != nil {
		return nil, err
	}
	dbPool, err := s.meta.GetSharedDB(ctx, placement.DbID)
	if err != nil {
		return nil, err
	}
	var patch meta.QuotaConfigPatch
	if quotaOpt = quotaRequestForShared(quotaOpt); quotaOpt != nil && quotaOpt.anySet() {
		var err error
		patch, err = quotaConfigPatchFromRequest(*quotaOpt)
		if err != nil {
			return nil, err
		}
	}
	rawToken, apiKeyID, key, err := s.buildOwnerAPIKey(ctx, row.Tenant.ID, "default", 1, time.Time{}, apiKeyIssueSource{})
	if err != nil {
		return nil, err
	}
	var claimErr error
	if freeOrganization {
		claimErr = s.meta.ClaimSharedTenantPoolMembershipWithMetadataForFreeOrganization(ctx, row.Tenant.ID, pool.PoolID, patch, key, metadata)
	} else {
		claimErr = s.meta.ClaimSharedTenantPoolMembershipWithMetadata(ctx, row.Tenant.ID, pool.PoolID, patch, key, metadata)
	}
	if claimErr != nil {
		return nil, claimErr
	}
	return &provisionTenantResult{TenantID: row.Tenant.ID, APIKey: rawToken, APIKeyID: apiKeyID,
		Status: meta.TenantActive, Provider: tenant.ProviderTiDBCloudNativeShared,
		CloudProvider: dbPool.CloudProvider, Region: dbPool.Region,
		OrganizationID: row.Membership.TiDBCloudOrganizationID}, nil
}

func (s *Server) releaseClaimedPoolTenant(ctx context.Context, manager tenant.TenantPoolClusterManager, cluster *tenant.ClusterInfo, cred tenant.CredentialProvisionRequest, tenantID, reason string) {
	releaseCtx := backgroundWithTrace(ctx)
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" && cluster != nil {
		tenantID = strings.TrimSpace(cluster.TenantID)
	}
	if tenantID != "" {
		if err := s.meta.ReleaseTenantPoolClaim(releaseCtx, tenantID); err != nil {
			logger.Warn(releaseCtx, "admin_tenant_pool_release_binding_failed", zap.String("tenant_id", tenantID), zap.String("reason", reason), zap.Error(err))
		}
	}
	if err := manager.MarkClusterPoolFree(releaseCtx, cluster, cred); err != nil {
		clusterID := ""
		if cluster != nil {
			clusterID = cluster.ClusterID
		}
		logger.Warn(releaseCtx, "admin_tenant_pool_release_cluster_failed", zap.String("tenant_id", tenantID), zap.String("cluster_id", clusterID), zap.String("reason", reason), zap.Error(err))
	}
}