package meta

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"math"
	"strings"
	"time"

	"github.com/google/uuid"
)

// MaxTiDBCloudSpendingLimit is the upper bound used by physical TiDB Cloud
// spending targets and the dedicated tenant compatibility API. In the TiDB
// Cloud unit, 1,000,000 represents $10,000.
const MaxTiDBCloudSpendingLimit = int64(1_000_000)

// SharedDBRoleShared marks a db_pool row as a multi-tenant shared-schema
// database. A "dedicated" role is reserved for future use and rejected for
// now.
const SharedDBRoleShared = "shared"

// Placement values for TenantPlacement.Placement.
const (
	PlacementShared    = "shared"
	PlacementDedicated = "dedicated"
)

// Schema shape values for TenantPlacement.SchemaShape.
const (
	SchemaShapeStandalone = "standalone"
	SchemaShapeShared     = "shared"
)

// PlacementStatusDeleting retains the physical resource identity while a
// durable tenant cleanup job is still running. Capacity has already been
// released, so allocators must not treat this row as active inventory.
const PlacementStatusDeleting = "deleting"

const (
	SharedDBStatusPending      = "pending"
	SharedDBStatusProvisioning = "provisioning"
	SharedDBStatusActive       = "active"
	SharedDBStatusFailed       = "failed"
	SharedDBStatusDraining     = "draining"
)

const sharedDBStatusActive = SharedDBStatusActive

// SharedDBCapacityMode selects the capacity bound used by a reservation.
type SharedDBCapacityMode string

const (
	SharedDBCapacityNormal    SharedDBCapacityMode = "normal"
	SharedDBCapacityEmergency SharedDBCapacityMode = "emergency"
)

// SharedDBHardCap computes runtime emergency capacity from a persisted soft
// capacity. The ratio is deliberately not stored in db_pool.
func SharedDBHardCap(softCap int, ratio float64) (int, error) {
	if softCap <= 0 {
		return 0, fmt.Errorf("soft capacity must be positive")
	}
	if math.IsNaN(ratio) || math.IsInf(ratio, 0) || ratio < 1 {
		return 0, fmt.Errorf("hard-cap ratio must be finite and at least 1")
	}
	value := float64(softCap) * ratio
	if value > float64(int(^uint(0)>>1)) {
		return 0, fmt.Errorf("hard capacity overflows int")
	}
	return int(math.Ceil(value)), nil
}

// SharedDBReopenThresholdForRatio computes the runtime hysteresis threshold.
// The ratio is intentionally not persisted in db_pool; it is deployment
// policy and must be in the open interval (0,1).
func SharedDBReopenThresholdForRatio(softCap int, ratio float64) (int, error) {
	if softCap <= 0 {
		return 0, nil
	}
	if math.IsNaN(ratio) || math.IsInf(ratio, 0) || ratio <= 0 || ratio >= 1 {
		return 0, fmt.Errorf("reopen ratio must be finite and in (0,1)")
	}
	return int(math.Floor(float64(softCap) * ratio)), nil
}

func sharedDBUUID(raw string) (string, error) {
	if strings.TrimSpace(raw) == "" {
		return uuid.NewString(), nil
	}
	parsed, err := uuid.Parse(strings.TrimSpace(raw))
	if err != nil {
		return "", fmt.Errorf("invalid shared db pool uuid %q: %w", raw, err)
	}
	return parsed.String(), nil
}

func validateSharedDBOrganizationID(raw string) (string, error) {
	organizationID := strings.TrimSpace(raw)
	if organizationID == "" {
		return "", fmt.Errorf("tidbcloud organization id is required")
	}
	if organizationID != raw || strings.ContainsAny(organizationID, "*%?") {
		return "", fmt.Errorf("tidbcloud organization id must be an exact value")
	}
	return organizationID, nil
}

// SharedDB is one physical database registered in db_pool. PasswordCipher
// holds the same encrypted envelope as tenants.db_password; the plaintext
// never crosses this layer. TLSMode is the go-sql-driver tls DSN parameter
// verbatim ("true", "skip-verify", a custom registered config name, or ""
// for plaintext) so the runtime handle reopens the DB with exactly the TLS
// mode it was registered with. MaxTenants of 0 makes the pool drain-only:
// existing tenants may leave, but no new tenant may enter.
type SharedDB struct {
	ID                      int64
	UUID                    string
	TiDBCloudOrganizationID string
	ClusterID               string
	ProvisioningKey         []byte
	CloudProvider           string
	Region                  string
	Role                    string
	Host                    string
	Port                    int
	User                    string
	PasswordCipher          []byte
	Name                    string
	TLSMode                 string
	MaxTenants              int
	TenantCount             int
	SoftCapReached          bool
	SpendingLimit           *int64
	SchemaVersion           int
	Status                  string
	CreatedAt               time.Time
	UpdatedAt               time.Time
}

// StuckSharedDBPoolFailure identifies the durable work released when a
// pending or provisioning physical shared DB makes no progress before a
// caller-provided cutoff.
type StuckSharedDBPoolFailure struct {
	DBID      int64
	TenantIDs []string
}

// TenantPlacement maps one filesystem (fs_id) to the physical database that
// hosts it. TargetDbID is reserved for future migrations; Epoch is reserved
// for optimistic concurrency during migrations and stays 1 unless a migration
// explicitly bumps it.
type TenantPlacement struct {
	FsID        int64
	DbID        int64
	Placement   string
	SchemaShape string
	Status      string
	TargetDbID  *int64
	Epoch       int64
}

// SharedDBPoolMetricSnapshot is the central-meta view used by the existing
// tenant metrics pass. TenantCount is the capacity counter maintained with
// placement writes; TenantStates is recomputed from placements so counter
// drift remains visible in exported metrics.
type SharedDBPoolMetricSnapshot struct {
	ID                      int64
	UUID                    string
	TiDBCloudOrganizationID string
	Status                  string
	MaxTenants              int
	TenantCount             int
	SoftCapReached          bool
	SpendingLimit           *int64
	UpdatedAt               time.Time
	StatusUpdatedAt         time.Time
	TenantStates            []SharedDBPoolTenantStateCount
}

// SharedDBPoolTenantStateCount is one persisted tenant-status count within a
// physical shared DB pool.
type SharedDBPoolTenantStateCount struct {
	State TenantStatus
	Count int64
}

// sharedDBSelectColumns lists the db_pool columns read into SharedDB. The
// `role` column is backtick-quoted because ROLE is reserved in MySQL 8.0.
const sharedDBSelectColumns = "db_id, uuid, org_id, cluster_id, provisioning_key, cloud_provider, region, " +
	"`role`, db_host, db_port, db_user, db_password, db_name, db_tls, max_tenants, tenant_count, " +
	"soft_cap_reached, spending_limit, schema_version, status, created_at, updated_at"

// CreateManagedSharedDBPool commits the durable identity and fixed business
// policy for one Drive9-managed physical pool before any TiDB Cloud API call.
// Connection and cluster fields intentionally remain NULL while provisioning.
func (s *Store) CreateManagedSharedDBPool(ctx context.Context, in *SharedDB) (id int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "create_managed_shared_db_pool", start, &err)
	values, err := managedSharedDBInsertValuesFor(in)
	if err != nil {
		return 0, err
	}
	return insertManagedSharedDBPool(ctx, s.db, values)
}

type managedSharedDBInsertValues struct {
	poolUUID        string
	organizationID  string
	provisioningKey []byte
	cloudProvider   string
	region          string
	passwordCipher  any
	databaseName    any
	maxTenants      int
	spendingLimit   int64
}

func managedSharedDBInsertValuesFor(in *SharedDB) (managedSharedDBInsertValues, error) {
	if in == nil {
		return managedSharedDBInsertValues{}, fmt.Errorf("shared db pool is required")
	}
	organizationID, err := validateSharedDBOrganizationID(in.TiDBCloudOrganizationID)
	if err != nil {
		return managedSharedDBInsertValues{}, err
	}
	poolUUID, err := sharedDBUUID(in.UUID)
	if err != nil {
		return managedSharedDBInsertValues{}, err
	}
	if len(in.ProvisioningKey) != 32 {
		return managedSharedDBInsertValues{}, fmt.Errorf("provisioning key must be a 32-byte SHA-256 fingerprint")
	}
	if in.CloudProvider == "" {
		return managedSharedDBInsertValues{}, fmt.Errorf("cloud provider is required")
	}
	if in.Region == "" {
		return managedSharedDBInsertValues{}, fmt.Errorf("region is required")
	}
	if in.MaxTenants <= 0 {
		return managedSharedDBInsertValues{}, fmt.Errorf("managed max tenants must be positive")
	}
	if in.SpendingLimit == nil || *in.SpendingLimit <= 0 || *in.SpendingLimit > int64(math.MaxInt32) {
		return managedSharedDBInsertValues{}, fmt.Errorf("managed spending limit must be in (0,%d]", int64(math.MaxInt32))
	}
	var passwordCipher any
	if len(in.PasswordCipher) != 0 {
		passwordCipher = in.PasswordCipher
	}
	var databaseName any
	if in.Name != "" {
		databaseName = in.Name
	}
	return managedSharedDBInsertValues{poolUUID: poolUUID, organizationID: organizationID,
		provisioningKey: append([]byte(nil), in.ProvisioningKey...), cloudProvider: in.CloudProvider,
		region: in.Region, passwordCipher: passwordCipher, databaseName: databaseName,
		maxTenants: in.MaxTenants, spendingLimit: *in.SpendingLimit}, nil
}

type managedSharedDBExecer interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
}

func insertManagedSharedDBPool(ctx context.Context, execer managedSharedDBExecer, values managedSharedDBInsertValues) (int64, error) {
	res, err := execer.ExecContext(ctx, `INSERT INTO db_pool
		(uuid, org_id, provisioning_key, cloud_provider, region, `+"`role`"+`, db_tls,
		 db_password, db_name, max_tenants, tenant_count, soft_cap_reached, spending_limit, schema_version, status)
		VALUES (?, ?, ?, ?, ?, ?, '', ?, ?, ?, 0, 0, ?, 0, ?)`, values.poolUUID, values.organizationID, values.provisioningKey,
		values.cloudProvider, values.region, SharedDBRoleShared, values.passwordCipher, values.databaseName,
		values.maxTenants, values.spendingLimit,
		SharedDBStatusPending)
	if err != nil {
		return 0, fmt.Errorf("insert managed db_pool row: %w", err)
	}
	id, err := res.LastInsertId()
	if err != nil {
		return 0, fmt.Errorf("resolve managed db_pool id: %w", err)
	}
	return id, nil
}

// UpdateManagedSharedDBPoolCloudResult persists the cloud identity and any
// connection metadata currently available for a managed pool. A pool stays
// pending while connection metadata is incomplete, transitions to
// provisioning once the metadata is ready, and remains provisioning until its
// shared schema is ready.
func (s *Store) UpdateManagedSharedDBPoolCloudResult(ctx context.Context, in *SharedDB) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_managed_shared_db_pool_cloud_result", start, &err)
	if in == nil || in.ID <= 0 {
		return fmt.Errorf("managed shared db pool id must be positive")
	}
	if in.TiDBCloudOrganizationID == "" {
		return fmt.Errorf("tidbcloud organization id is required")
	}
	clusterID := strings.TrimSpace(in.ClusterID)
	if clusterID == "" {
		return fmt.Errorf("cluster id is required")
	}
	if clusterID != in.ClusterID {
		return fmt.Errorf("cluster id must be an exact value")
	}
	if in.Port < 0 {
		return fmt.Errorf("db port must not be negative")
	}
	if len(in.TLSMode) > 32 {
		return fmt.Errorf("tls mode %q is too long", in.TLSMode)
	}
	nullString := func(value string) any {
		if value == "" {
			return nil
		}
		return value
	}
	nullInt := func(value int) any {
		if value == 0 {
			return nil
		}
		return value
	}
	nullBytes := func(value []byte) any {
		if len(value) == 0 {
			return nil
		}
		return value
	}
	var advancedTenants []string
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var currentOrganizationID string
		var currentClusterID, currentHost, currentUser, currentName sql.NullString
		var currentPort sql.NullInt64
		var currentPassword []byte
		var currentStatus string
		if err := tx.QueryRowContext(ctx, `SELECT org_id, cluster_id, db_host, db_port, db_user, db_password, db_name, status
			FROM db_pool WHERE db_id = ? FOR UPDATE`, in.ID).
			Scan(&currentOrganizationID, &currentClusterID, &currentHost, &currentPort, &currentUser, &currentPassword, &currentName, &currentStatus); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return fmt.Errorf("lock cloud result for db pool %d: %w", in.ID, err)
		}
		if currentStatus != SharedDBStatusPending && currentStatus != SharedDBStatusProvisioning {
			return ErrNotFound
		}
		// Once complete metadata advances the pool to provisioning, finishers may
		// safely run without retaining this MetaDB row lock only if the Cloud
		// allocation identity cannot be reassigned underneath them. Connection
		// metadata remains refreshable for endpoint and credential rotation.
		if currentStatus == SharedDBStatusProvisioning &&
			((strings.TrimSpace(currentOrganizationID) != "" && currentOrganizationID != in.TiDBCloudOrganizationID) ||
				(strings.TrimSpace(currentClusterID.String) != "" && currentClusterID.String != clusterID)) {
			return fmt.Errorf("%w: db pool %d", ErrSharedDBPoolIdentityConflict, in.ID)
		}
		host := currentHost.String
		if in.Host != "" {
			host = in.Host
		}
		port := int(currentPort.Int64)
		if in.Port > 0 {
			port = in.Port
		}
		user := currentUser.String
		if in.User != "" {
			user = in.User
		}
		password := currentPassword
		if len(in.PasswordCipher) > 0 {
			password = in.PasswordCipher
		}
		name := currentName.String
		if in.Name != "" {
			name = in.Name
		}
		nextStatus := currentStatus
		metadataReady := clusterID != "" && host != "" && port > 0 && user != "" && len(password) > 0 && name != ""
		if currentStatus == SharedDBStatusPending && metadataReady {
			nextStatus = SharedDBStatusProvisioning
		}
		if _, err := tx.ExecContext(ctx, `UPDATE db_pool
			SET org_id = ?, cluster_id = ?, provisioning_key = NULL,
				db_host = COALESCE(?, db_host), db_port = COALESCE(?, db_port),
				db_user = COALESCE(?, db_user), db_password = COALESCE(?, db_password),
				db_name = COALESCE(?, db_name), db_tls = ?,
				status_updated_at = CASE WHEN status <> ? THEN CURRENT_TIMESTAMP(3) ELSE status_updated_at END,
				status = ?
			WHERE db_id = ?`, in.TiDBCloudOrganizationID, clusterID,
			nullString(in.Host), nullInt(in.Port), nullString(in.User), nullBytes(in.PasswordCipher),
			nullString(in.Name), in.TLSMode, nextStatus, nextStatus, in.ID); err != nil {
			return fmt.Errorf("persist cloud result for db pool %d: %w", in.ID, err)
		}
		if nextStatus == SharedDBStatusProvisioning && metadataReady {
			// Collect the tenant ids before the bulk UPDATE so the API-key
			// resolve cache can be evicted precisely after commit.
			rows, err := tx.QueryContext(ctx, `SELECT t.id
				FROM tenants t
				JOIN fs_registry f ON f.tenant_id = t.id
				JOIN tenant_placements p ON p.fs_id = f.fs_id
				WHERE p.db_id = ? AND t.provider = ? AND t.status = ?`,
				in.ID, tidbCloudNativeSharedProvider, TenantPending)
			if err != nil {
				return fmt.Errorf("list advancing tenants for db pool %d: %w", in.ID, err)
			}
			for rows.Next() {
				var tenantID string
				if err := rows.Scan(&tenantID); err != nil {
					_ = rows.Close()
					return fmt.Errorf("scan advancing tenant for db pool %d: %w", in.ID, err)
				}
				advancedTenants = append(advancedTenants, tenantID)
			}
			if err := rows.Err(); err != nil {
				_ = rows.Close()
				return fmt.Errorf("list advancing tenants for db pool %d: %w", in.ID, err)
			}
			if err := rows.Close(); err != nil {
				return fmt.Errorf("list advancing tenants for db pool %d: %w", in.ID, err)
			}
			if _, err := tx.ExecContext(ctx, `UPDATE tenants t
				JOIN fs_registry f ON f.tenant_id = t.id
				JOIN tenant_placements p ON p.fs_id = f.fs_id
				SET t.status = ?, t.updated_at = ?
				WHERE p.db_id = ? AND t.provider = ? AND t.status = ?`, TenantProvisioning, time.Now().UTC(),
				in.ID, tidbCloudNativeSharedProvider, TenantPending); err != nil {
				return fmt.Errorf("advance tenants for db pool %d: %w", in.ID, err)
			}
		}
		return nil
	})
	if err != nil {
		return err
	}
	for _, tenantID := range advancedTenants {
		s.apiKeys.evictTenant(tenantID)
	}
	return nil
}

// PrepareManagedSharedDBPoolRoot durably stores the root credential and
// database name before the first Cloud create call. It is idempotent for a
// pending row that has not yet acquired a cluster identity.
func (s *Store) PrepareManagedSharedDBPoolRoot(ctx context.Context, dbID int64, passwordCipher []byte, databaseName string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "prepare_managed_shared_db_pool_root", start, &err)
	if dbID <= 0 {
		return fmt.Errorf("db pool id must be positive")
	}
	if len(passwordCipher) == 0 {
		return fmt.Errorf("managed root password cipher is required")
	}
	if databaseName == "" {
		return fmt.Errorf("managed database name is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE db_pool
		SET db_password = COALESCE(db_password, ?), db_name = COALESCE(db_name, ?)
		WHERE db_id = ? AND status = ? AND cluster_id IS NULL`,
		passwordCipher, databaseName, dbID, SharedDBStatusPending)
	if err != nil {
		return fmt.Errorf("prepare managed db pool %d root credential: %w", dbID, err)
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("prepare managed db pool %d rows affected: %w", dbID, err)
	}
	if affected > 0 {
		return nil
	}
	var exists int
	if err := s.db.QueryRowContext(ctx, `SELECT 1 FROM db_pool WHERE db_id = ? AND status = ? AND cluster_id IS NULL
		AND db_password IS NOT NULL AND db_name IS NOT NULL`, dbID, SharedDBStatusPending).Scan(&exists); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return err
	}
	return nil
}

// UpdateSharedDBSchemaVersion records the checked-in shared schema version
// after the idempotent ensure succeeds.
func (s *Store) UpdateSharedDBSchemaVersion(ctx context.Context, dbID int64, version int) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_shared_db_schema_version", start, &err)
	if dbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	if version <= 0 {
		return fmt.Errorf("schema version must be positive")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE db_pool SET schema_version = ? WHERE db_id = ?`, version, dbID)
	if err != nil {
		return fmt.Errorf("update schema version for db pool %d: %w", dbID, err)
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("schema version rows affected for db pool %d: %w", dbID, err)
	}
	if affected == 0 {
		var exists int
		if scanErr := s.db.QueryRowContext(ctx, `SELECT 1 FROM db_pool WHERE db_id = ?`, dbID).Scan(&exists); scanErr != nil {
			if errors.Is(scanErr, sql.ErrNoRows) {
				return ErrNotFound
			}
			return scanErr
		}
	}
	return nil
}

// RefreshManagedSharedDBProvisioningProgress keeps a long-running remote
// schema operation visible to the stuck-pool watchdog without retaining a
// MetaDB connection or advisory lock for the duration of the DDL.
func (s *Store) RefreshManagedSharedDBProvisioningProgress(ctx context.Context, dbID int64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "refresh_managed_shared_db_provisioning_progress", start, &err)
	if dbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE db_pool SET updated_at = CURRENT_TIMESTAMP(3)
		WHERE db_id = ? AND role = ? AND status = ?`, dbID, SharedDBRoleShared, SharedDBStatusProvisioning)
	if err != nil {
		return fmt.Errorf("refresh provisioning progress for db pool %d: %w", dbID, err)
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("refresh provisioning progress rows affected for db pool %d: %w", dbID, err)
	}
	if affected == 0 {
		var status string
		if err := s.db.QueryRowContext(ctx, `SELECT status FROM db_pool WHERE db_id = ? AND role = ?`, dbID, SharedDBRoleShared).Scan(&status); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return err
		}
		if status != SharedDBStatusProvisioning {
			return fmt.Errorf("%w: db pool %d has status %q", ErrSharedDBPoolNotProvisioning, dbID, status)
		}
	}
	return nil
}

// ActivateSharedDBPool flips a managed pool to active only after the cloud
// identity, connection fields, and shared schema version are complete.
func (s *Store) ActivateSharedDBPool(ctx context.Context, dbID int64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "activate_shared_db_pool", start, &err)
	if dbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE db_pool SET status = ?, status_updated_at = CURRENT_TIMESTAMP(3)
		WHERE db_id = ? AND status = ? AND org_id IS NOT NULL AND cluster_id IS NOT NULL
			AND db_host IS NOT NULL AND db_port IS NOT NULL AND db_user IS NOT NULL
			AND db_password IS NOT NULL AND db_name IS NOT NULL AND schema_version > 0`,
		SharedDBStatusActive, dbID, SharedDBStatusProvisioning)
	if err != nil {
		return fmt.Errorf("activate db pool %d: %w", dbID, err)
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("activate db pool rows affected %d: %w", dbID, err)
	}
	if affected > 0 {
		return nil
	}
	var status string
	if scanErr := s.db.QueryRowContext(ctx, `SELECT status FROM db_pool WHERE db_id = ?`, dbID).Scan(&status); scanErr != nil {
		if errors.Is(scanErr, sql.ErrNoRows) {
			return ErrNotFound
		}
		return scanErr
	}
	if status == SharedDBStatusActive {
		return nil
	}
	return fmt.Errorf("db pool %d is not ready for activation", dbID)
}

func (s *Store) MarkSharedDBPoolFailed(ctx context.Context, dbID int64) error {
	res, err := s.db.ExecContext(ctx, `UPDATE db_pool SET status = ?, status_updated_at = CURRENT_TIMESTAMP(3)
		WHERE db_id = ? AND status IN (?, ?) AND tenant_count = 0`,
		SharedDBStatusFailed, dbID, SharedDBStatusPending, SharedDBStatusProvisioning)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		var status string
		var tenantCount int
		if err := s.db.QueryRowContext(ctx, `SELECT status, tenant_count FROM db_pool WHERE db_id = ?`, dbID).
			Scan(&status, &tenantCount); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return err
		}
		if (status == SharedDBStatusPending || status == SharedDBStatusProvisioning) && tenantCount > 0 {
			return fmt.Errorf("db pool %d has %d placed tenants and cannot be marked failed", dbID, tenantCount)
		}
		return ErrNotFound
	}
	return nil
}

// testHookAfterStuckSharedDBPoolLocked, if set, runs after the watchdog has
// locked the db_pool row and before it locks matching tenant rows. Tests stall
// here to interleave a tenant status change: unistore does not honor
// FOR UPDATE SKIP LOCKED as a lock probe.
var testHookAfterStuckSharedDBPoolLocked func(dbID int64)

// MarkStuckSharedDBPoolFailed atomically fails a stale physical pool and its
// still-pending/provisioning shared tenants. The expected status and
// updatedBefore cutoff form a CAS guard so concurrent metadata/schema progress
// wins over the watchdog.
func (s *Store) MarkStuckSharedDBPoolFailed(ctx context.Context, dbID int64, expectedStatus string, updatedBefore time.Time) (result *StuckSharedDBPoolFailure, changed bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_stuck_shared_db_pool_failed", start, &err)
	if dbID <= 0 {
		return nil, false, fmt.Errorf("db_id must be positive")
	}
	if expectedStatus != SharedDBStatusPending && expectedStatus != SharedDBStatusProvisioning {
		return nil, false, fmt.Errorf("unsupported stuck shared db status %q", expectedStatus)
	}
	cutoff := updatedBefore.UTC()
	var tenantIDs []string
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var status string
		var updatedAt time.Time
		if err := tx.QueryRowContext(ctx, `SELECT status, updated_at
			FROM db_pool WHERE db_id = ? AND role = ? FOR UPDATE`, dbID, SharedDBRoleShared).
			Scan(&status, &updatedAt); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return err
		}
		if status != expectedStatus || updatedAt.After(cutoff) {
			return nil
		}
		if hook := testHookAfterStuckSharedDBPoolLocked; hook != nil {
			hook(dbID)
		}
		// Lock the matching tenant rows so the IDs returned to cache eviction and
		// logs are exactly the rows the conditional failure update can transition.
		rows, err := tx.QueryContext(ctx, `SELECT t.id
			FROM tenant_placements p
			JOIN fs_registry f ON f.fs_id = p.fs_id
			JOIN tenants t ON t.id = f.tenant_id
			WHERE p.db_id = ? AND t.provider = ? AND t.status IN (?, ?)
			ORDER BY t.id FOR UPDATE`, dbID, tidbCloudNativeSharedProvider, TenantPending, TenantProvisioning)
		if err != nil {
			return fmt.Errorf("list stuck shared tenants for db pool %d: %w", dbID, err)
		}
		for rows.Next() {
			var tenantID string
			if err := rows.Scan(&tenantID); err != nil {
				_ = rows.Close()
				return err
			}
			tenantIDs = append(tenantIDs, tenantID)
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return err
		}
		if err := rows.Close(); err != nil {
			return err
		}
		now := time.Now().UTC()
		if _, err := tx.ExecContext(ctx, `UPDATE tenants t
			JOIN fs_registry f ON f.tenant_id = t.id
			JOIN tenant_placements p ON p.fs_id = f.fs_id
			SET t.status = ?, t.updated_at = ?
			WHERE p.db_id = ? AND t.provider = ? AND t.status IN (?, ?)`,
			TenantFailed, now, dbID, tidbCloudNativeSharedProvider, TenantPending, TenantProvisioning); err != nil {
			return fmt.Errorf("fail stuck shared tenants for db pool %d: %w", dbID, err)
		}
		res, err := tx.ExecContext(ctx, `UPDATE db_pool SET status = ?, status_updated_at = ?, updated_at = ?
			WHERE db_id = ? AND role = ? AND status = ? AND updated_at <= ?`,
			SharedDBStatusFailed, now, now, dbID, SharedDBRoleShared, expectedStatus, cutoff)
		if err != nil {
			return fmt.Errorf("fail stuck shared db pool %d: %w", dbID, err)
		}
		affected, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if affected != 1 {
			return fmt.Errorf("fail stuck shared db pool %d affected %d rows, want 1", dbID, affected)
		}
		changed = true
		result = &StuckSharedDBPoolFailure{
			DBID: dbID, TenantIDs: append([]string(nil), tenantIDs...),
		}
		return nil
	})
	if err != nil {
		return nil, false, err
	}
	if changed {
		for _, tenantID := range tenantIDs {
			s.apiKeys.evictTenant(tenantID)
		}
	}
	return result, changed, nil
}

// ConfirmFailedSharedDBPoolCloudDeleted atomically clears the expected Cloud
// identity and marks the pool's placements deleting. A deleting placement is
// durable proof that physical shared rows no longer need to be purged because
// the entire Cloud cluster has already been removed.
func (s *Store) ConfirmFailedSharedDBPoolCloudDeleted(ctx context.Context, dbID int64, clusterID string) (confirmed bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "confirm_failed_shared_db_pool_cloud_deleted", start, &err)
	if dbID <= 0 {
		return false, fmt.Errorf("db_id must be positive")
	}
	clusterID = strings.TrimSpace(clusterID)
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var currentClusterID sql.NullString
		var role, status string
		if scanErr := tx.QueryRowContext(ctx, `SELECT cluster_id, role, status
			FROM db_pool WHERE db_id = ? FOR UPDATE`, dbID).Scan(&currentClusterID, &role, &status); scanErr != nil {
			if errors.Is(scanErr, sql.ErrNoRows) {
				return nil
			}
			return scanErr
		}
		if role != SharedDBRoleShared || status != SharedDBStatusFailed || strings.TrimSpace(currentClusterID.String) != clusterID {
			return nil
		}
		if clusterID != "" {
			if _, execErr := tx.ExecContext(ctx, `UPDATE db_pool SET cluster_id = NULL, updated_at = ? WHERE db_id = ?`,
				time.Now().UTC(), dbID); execErr != nil {
				return execErr
			}
		}
		if _, execErr := tx.ExecContext(ctx, `UPDATE tenant_placements SET status = ? WHERE db_id = ?`,
			PlacementStatusDeleting, dbID); execErr != nil {
			return execErr
		}
		confirmed = true
		return nil
	})
	return confirmed, err
}

// RepairFailedSharedDBPoolTenantCountIfEmpty repairs denormalized positive
// drift only when the authoritative placement set is empty.
func (s *Store) RepairFailedSharedDBPoolTenantCountIfEmpty(ctx context.Context, dbID int64) (repaired bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "repair_failed_shared_db_pool_tenant_count", start, &err)
	if dbID <= 0 {
		return false, fmt.Errorf("db_id must be positive")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE db_pool
		SET tenant_count = 0, soft_cap_reached = 0, updated_at = ?
		WHERE db_id = ? AND role = ? AND status = ? AND tenant_count <> 0
			AND NOT EXISTS (SELECT 1 FROM tenant_placements p WHERE p.db_id = db_pool.db_id)`,
		time.Now().UTC(), dbID, SharedDBRoleShared, SharedDBStatusFailed)
	if err != nil {
		return false, err
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return affected == 1, nil
}

// DeleteFailedSharedDBPoolIfEmpty removes only a locally empty failed pool
// whose Cloud identity has already been cleared (or never existed).
func (s *Store) DeleteFailedSharedDBPoolIfEmpty(ctx context.Context, dbID int64) (deleted bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_failed_shared_db_pool_if_empty", start, &err)
	if dbID <= 0 {
		return false, fmt.Errorf("db_id must be positive")
	}
	res, err := s.db.ExecContext(ctx, `DELETE FROM db_pool
		WHERE db_id = ? AND role = ? AND status = ? AND tenant_count = 0 AND cluster_id IS NULL
			AND NOT EXISTS (SELECT 1 FROM tenant_placements p WHERE p.db_id = db_pool.db_id)`,
		dbID, SharedDBRoleShared, SharedDBStatusFailed)
	if err != nil {
		return false, err
	}
	affected, err := res.RowsAffected()
	return affected == 1, err
}

// ListSharedDBsByStatus returns shared DB-pool rows in stable ID order.
func (s *Store) ListSharedDBsByStatus(ctx context.Context, status string, limit int) (out []*SharedDB, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_shared_dbs_by_status", start, &err)
	return s.listSharedDBsByStatusAfter(ctx, status, 0, limit)
}

// ListSharedDBsByStatusAfter returns one stable keyset page ordered by db_id.
func (s *Store) ListSharedDBsByStatusAfter(ctx context.Context, status string, afterID int64, limit int) (out []*SharedDB, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_shared_dbs_by_status_after", start, &err)
	return s.listSharedDBsByStatusAfter(ctx, status, afterID, limit)
}

func (s *Store) listSharedDBsByStatusAfter(ctx context.Context, status string, afterID int64, limit int) (out []*SharedDB, err error) {
	switch status {
	case SharedDBStatusPending, SharedDBStatusProvisioning, SharedDBStatusActive, SharedDBStatusFailed, SharedDBStatusDraining:
	default:
		return nil, fmt.Errorf("unsupported shared db status %q", status)
	}
	if limit <= 0 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx, "SELECT "+sharedDBSelectColumns+" FROM db_pool "+
		"WHERE `role` = ? AND status = ? AND db_id > ? ORDER BY db_id LIMIT ?", SharedDBRoleShared, status, afterID, limit)
	if err != nil {
		return nil, fmt.Errorf("list shared dbs in status %q: %w", status, err)
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		rec, scanErr := scanSharedDBScanner(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		out = append(out, rec)
	}
	return out, rows.Err()
}

// ListActiveSharedDBsWithProvisioningTenantsAfter returns active physical
// pools that still have an eligible shared tenant awaiting final activation.
// This is the durable recovery path for a crash after physical activation but
// before ActivateSharedTenantsBatch finishes.
func (s *Store) ListActiveSharedDBsWithProvisioningTenantsAfter(ctx context.Context, afterID int64, limit int) (out []*SharedDB, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_active_shared_dbs_with_provisioning_tenants_after", start, &err)
	if afterID < 0 {
		return nil, fmt.Errorf("after db id must not be negative")
	}
	if limit <= 0 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx, "SELECT "+sharedDBSelectColumns+` FROM db_pool d
		WHERE d.`+"`role`"+` = ? AND d.status = ? AND d.db_id > ?
			AND EXISTS (
				SELECT 1 FROM tenant_placements p
				JOIN fs_registry f ON f.fs_id = p.fs_id
				JOIN tenants t ON t.id = f.tenant_id
				WHERE p.db_id = d.db_id AND p.status = ? AND t.provider = ? AND t.status = ?
					AND (EXISTS (SELECT 1 FROM tenant_api_keys k
						WHERE k.tenant_id = t.id AND k.scope_kind = ? AND k.status = ?)
						OR EXISTS (SELECT 1 FROM tenant_pool_memberships m
							WHERE m.tenant_id = t.id AND m.pool_status = ?)))
		ORDER BY d.db_id LIMIT ?`, SharedDBRoleShared, SharedDBStatusActive, afterID,
		SharedDBStatusActive, tidbCloudNativeSharedProvider, TenantProvisioning,
		APIKeyScopeKindOwner, APIKeyActive, TenantPoolBindingFree, limit)
	if err != nil {
		return nil, fmt.Errorf("list active shared dbs with provisioning tenants: %w", err)
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		rec, scanErr := scanSharedDBScanner(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		out = append(out, rec)
	}
	return out, rows.Err()
}

// FindSharedDBForAllocation selects the oldest eligible exact-organization
// pool, preferring active over provisioning over pending. Managed rows use one fixed
// physical spending limit, so tenant virtual values do not participate in
// allocation.
func (s *Store) FindSharedDBForAllocation(ctx context.Context, organizationID string) (db *SharedDB, err error) {
	start := time.Now()
	defer observeMeta(ctx, "find_shared_db_for_allocation", start, &err)
	organizationID, err = validateSharedDBOrganizationID(organizationID)
	if err != nil {
		return nil, err
	}
	query := "SELECT " + sharedDBSelectColumns + " FROM db_pool d " +
		"WHERE d.org_id = ? AND d.`role` = ? " +
		"AND d.status IN (?, ?, ?) " +
		"AND d.max_tenants > 0 AND d.soft_cap_reached = 0 AND d.tenant_count < d.max_tenants " +
		"ORDER BY CASE d.status WHEN 'active' THEN 0 WHEN 'provisioning' THEN 1 ELSE 2 END, d.db_id LIMIT 1"
	db, err = scanSharedDBRow(s.db.QueryRowContext(ctx, query, organizationID, SharedDBRoleShared,
		SharedDBStatusActive, SharedDBStatusProvisioning, SharedDBStatusPending))
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("find shared db for allocation: %w", err)
	}
	return db, nil
}

// FindSharedDBForEmergency selects the least-loaded active managed pool below
// its own runtime-derived hard cap. It is used only after a physical create
// failure on a direct request; refill and prewarm must never call it.
func (s *Store) FindSharedDBForEmergency(ctx context.Context, organizationID string, hardCapRatio float64) (*SharedDB, error) {
	if organizationID == "" {
		return nil, fmt.Errorf("organization id is required")
	}
	if hardCapRatio <= 1 {
		return nil, fmt.Errorf("invalid emergency capacity arguments")
	}
	query := "SELECT " + sharedDBSelectColumns + " FROM db_pool d " +
		"WHERE d.org_id = ? AND d.`role` = ? AND d.status = ? " +
		"AND d.spending_limit IS NOT NULL " +
		"AND d.max_tenants > 0 AND d.tenant_count < CEIL(d.max_tenants * ?) " +
		"ORDER BY d.tenant_count ASC, d.db_id ASC LIMIT 1"
	db, err := scanSharedDBRow(s.db.QueryRowContext(ctx, query, organizationID, SharedDBRoleShared,
		SharedDBStatusActive, hardCapRatio))
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("find emergency shared db: %w", err)
	}
	return db, nil
}

// RegisterSharedDB registers a manually configured physical database in
// db_pool. Its stable UUID is derived from the full manual connection identity
// when omitted; managed cloud pools use CreateManagedSharedDBPool and global
// cluster_id uniqueness instead.
func (s *Store) RegisterSharedDB(ctx context.Context, in *SharedDB) (dbID int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "register_shared_db", start, &err)
	if in == nil {
		return 0, fmt.Errorf("shared db is required")
	}
	organizationID, err := validateSharedDBOrganizationID(in.TiDBCloudOrganizationID)
	if err != nil {
		return 0, err
	}
	if in.Host == "" {
		return 0, fmt.Errorf("db host is required")
	}
	if in.Port <= 0 {
		return 0, fmt.Errorf("db port must be positive")
	}
	if in.User == "" {
		return 0, fmt.Errorf("db user is required")
	}
	poolUUID := strings.TrimSpace(in.UUID)
	if poolUUID == "" {
		poolUUID = uuid.NewSHA1(uuid.NameSpaceOID, []byte(strings.Join(
			[]string{organizationID, in.Host, in.Name, in.User}, "\x00"))).String()
	} else {
		poolUUID, err = sharedDBUUID(poolUUID)
		if err != nil {
			return 0, err
		}
	}
	if len(in.PasswordCipher) == 0 {
		return 0, fmt.Errorf("db password cipher is required")
	}
	if in.Name == "" {
		return 0, fmt.Errorf("db name is required")
	}
	if in.MaxTenants < 0 {
		return 0, fmt.Errorf("max tenants must not be negative")
	}
	if len(in.TLSMode) > 32 {
		return 0, fmt.Errorf("tls mode %q is too long", in.TLSMode)
	}
	role := in.Role
	if role == "" {
		role = SharedDBRoleShared
	}
	if role != SharedDBRoleShared {
		return 0, fmt.Errorf("unsupported db role %q", role)
	}
	status := in.Status
	if status == "" {
		status = sharedDBStatusActive
	}
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		err := tx.QueryRowContext(ctx, `SELECT db_id FROM db_pool
			WHERE org_id = ? AND db_host = ? AND db_name = ? AND db_user = ?
			ORDER BY db_id LIMIT 1 FOR UPDATE`, organizationID, in.Host, in.Name, in.User).Scan(&dbID)
		if err == nil {
			if _, err := tx.ExecContext(ctx, `UPDATE db_pool
				SET db_port = ?, db_password = ?, db_tls = ?, max_tenants = ?,
					status_updated_at = CASE WHEN status <> ? THEN CURRENT_TIMESTAMP(3) ELSE status_updated_at END,
					status = ?
				WHERE db_id = ?`, in.Port, in.PasswordCipher, in.TLSMode, in.MaxTenants, status, status, dbID); err != nil {
				return fmt.Errorf("update existing manual db_pool row: %w", err)
			}
			return nil
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return fmt.Errorf("find existing manual db_pool row: %w", err)
		}
		if _, err := tx.ExecContext(ctx,
			"INSERT INTO db_pool (uuid, org_id, `role`, db_host, db_port, db_user, db_password, db_name, "+
				"db_tls, max_tenants, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "+
				"ON DUPLICATE KEY UPDATE org_id = VALUES(org_id), db_host = VALUES(db_host), "+
				"db_port = VALUES(db_port), db_user = VALUES(db_user), db_name = VALUES(db_name), "+
				"db_password = VALUES(db_password), db_tls = VALUES(db_tls), "+
				"max_tenants = VALUES(max_tenants), "+
				"status_updated_at = CASE WHEN status <> VALUES(status) THEN CURRENT_TIMESTAMP(3) ELSE status_updated_at END, "+
				"status = VALUES(status)",
			poolUUID, organizationID, role, in.Host, in.Port, in.User, in.PasswordCipher, in.Name,
			in.TLSMode, in.MaxTenants, status); err != nil {
			return fmt.Errorf("upsert manual db_pool row: %w", err)
		}
		if err := tx.QueryRowContext(ctx, `SELECT db_id FROM db_pool WHERE uuid = ?`, poolUUID).Scan(&dbID); err != nil {
			return fmt.Errorf("resolve db_id after upsert: %w", err)
		}
		return nil
	})
	if err != nil {
		return 0, err
	}
	return dbID, nil
}

// GetSharedDB returns the db_pool row for dbID, or ErrNotFound.
func (s *Store) GetSharedDB(ctx context.Context, dbID int64) (db *SharedDB, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_shared_db", start, &err)
	if dbID <= 0 {
		return nil, fmt.Errorf("db_id must be positive")
	}
	db, err = scanSharedDBRow(s.db.QueryRowContext(ctx,
		"SELECT "+sharedDBSelectColumns+" FROM db_pool WHERE db_id = ?", dbID))
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("get shared db %d: %w", dbID, err)
	}
	return db, nil
}

// GetSharedDBForTenant resolves the physical shared resource identity used by
// credential authorization and metrics. tenants.cluster_id is intentionally
// not involved for shared-schema tenants.
func (s *Store) GetSharedDBForTenant(ctx context.Context, tenantID string) (*SharedDB, error) {
	db, err := scanSharedDBRow(s.db.QueryRowContext(ctx, `SELECT `+sharedDBSelectColumns+`
		FROM db_pool WHERE db_id = (SELECT p.db_id FROM tenant_placements p
		JOIN fs_registry f ON f.fs_id = p.fs_id WHERE f.tenant_id = ?)`, tenantID))
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	return db, nil
}

// FindSharedDBForOrg returns the oldest open active shared database serving
// exactly orgID. Shared DB pools never match across organizations.
func (s *Store) FindSharedDBForOrg(ctx context.Context, orgID string) (db *SharedDB, err error) {
	start := time.Now()
	defer observeMeta(ctx, "find_shared_db_for_org", start, &err)
	orgID, err = validateSharedDBOrganizationID(orgID)
	if err != nil {
		return nil, err
	}
	return s.findActiveSharedDB(ctx, orgID)
}

func (s *Store) findActiveSharedDB(ctx context.Context, orgID string) (*SharedDB, error) {
	// Capacity is enforced at selection time: normal allocation only sees open
	// positive-capacity pools; the transactional reserve remains the final
	// guard against races.
	db, err := scanSharedDBRow(s.db.QueryRowContext(ctx,
		"SELECT "+sharedDBSelectColumns+" FROM db_pool "+
			"WHERE org_id = ? AND `role` = ? AND status = ? "+
			"AND max_tenants > 0 AND soft_cap_reached = 0 AND tenant_count < max_tenants ORDER BY db_id LIMIT 1",
		orgID, SharedDBRoleShared, sharedDBStatusActive))
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("find shared db for org %q: %w", orgID, err)
	}
	return db, nil
}

// ListSharedDBs returns all active db_pool rows ordered by db_id.
func (s *Store) ListSharedDBs(ctx context.Context) (out []*SharedDB, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_shared_dbs", start, &err)
	rows, err := s.db.QueryContext(ctx,
		"SELECT "+sharedDBSelectColumns+" FROM db_pool WHERE `role` = ? AND status = ? ORDER BY db_id",
		SharedDBRoleShared, sharedDBStatusActive)
	if err != nil {
		return nil, fmt.Errorf("list shared dbs: %w", err)
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		rec, scanErr := scanSharedDBScanner(rows)
		if scanErr != nil {
			err = scanErr
			return nil, err
		}
		out = append(out, rec)
	}
	return out, rows.Err()
}

// ListSharedDBPoolMetricSnapshots returns one snapshot per physical shared
// database. It is intentionally a read-only aggregate used by the existing
// tenant metrics pass; it does not reconcile or mutate capacity counters.
// The query counts durable placements as an active baseline, then reclassifies
// only non-active tenants. This keeps the common path index-only and avoids two
// point lookups for every active placement.
const listSharedDBPoolMetricSnapshotsSQL = `WITH non_active_tenant_states AS (
	SELECT p.db_id, t.status AS tenant_status, COUNT(*) AS tenant_count
		FROM tenants t FORCE INDEX (idx_tenant_status)
		STRAIGHT_JOIN fs_registry f ON f.tenant_id = t.id
		STRAIGHT_JOIN tenant_placements p ON p.fs_id = f.fs_id
		WHERE t.status <> ?
		GROUP BY p.db_id, t.status
), tenant_state_deltas AS (
	SELECT db_id, ? AS tenant_status, COUNT(*) AS tenant_count
		FROM tenant_placements
		GROUP BY db_id
	UNION ALL
	SELECT db_id, tenant_status, tenant_count
		FROM non_active_tenant_states
	UNION ALL
	SELECT db_id, ? AS tenant_status, -SUM(tenant_count) AS tenant_count
		FROM non_active_tenant_states
		GROUP BY db_id
), tenant_states AS (
	SELECT db_id, tenant_status, SUM(tenant_count) AS tenant_count
		FROM tenant_state_deltas
		GROUP BY db_id, tenant_status
		HAVING SUM(tenant_count) > 0
)
SELECT
	d.db_id, d.uuid, COALESCE(d.org_id, ''), d.status, d.max_tenants, d.tenant_count, d.soft_cap_reached,
	d.spending_limit, d.updated_at, d.status_updated_at,
	COALESCE(tenant_states.tenant_status, ''), COALESCE(tenant_states.tenant_count, 0)
	FROM db_pool d
	LEFT JOIN tenant_states ON tenant_states.db_id = d.db_id
	WHERE d.` + "`role`" + ` = ?
	ORDER BY d.db_id, tenant_states.tenant_status`

func (s *Store) ListSharedDBPoolMetricSnapshots(ctx context.Context) (out []SharedDBPoolMetricSnapshot, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_shared_db_pool_metric_snapshots", start, &err)
	rows, err := s.db.QueryContext(ctx, listSharedDBPoolMetricSnapshotsSQL,
		TenantActive, TenantActive, TenantActive, SharedDBRoleShared)
	if err != nil {
		return nil, fmt.Errorf("list shared db pool metric snapshots: %w", err)
	}
	defer func() { _ = rows.Close() }()
	byID := make(map[int64]int)
	for rows.Next() {
		var id int64
		var dbPoolUUID, organizationID, status, tenantStatus string
		var maxTenants, tenantCount int
		var softCapReached bool
		var spendingLimit sql.NullInt64
		var updatedAt, statusUpdatedAt time.Time
		var stateCount int64
		if err := rows.Scan(&id, &dbPoolUUID, &organizationID, &status, &maxTenants, &tenantCount, &softCapReached,
			&spendingLimit, &updatedAt, &statusUpdatedAt, &tenantStatus, &stateCount); err != nil {
			return nil, fmt.Errorf("scan shared db pool metric snapshot: %w", err)
		}
		index, ok := byID[id]
		if !ok {
			snapshot := SharedDBPoolMetricSnapshot{
				ID: id, UUID: dbPoolUUID, TiDBCloudOrganizationID: organizationID, Status: status,
				MaxTenants: maxTenants, TenantCount: tenantCount, SoftCapReached: softCapReached, UpdatedAt: updatedAt, StatusUpdatedAt: statusUpdatedAt,
			}
			if spendingLimit.Valid {
				value := spendingLimit.Int64
				snapshot.SpendingLimit = &value
			}
			out = append(out, snapshot)
			index = len(out) - 1
			byID[id] = index
		}
		if tenantStatus != "" && stateCount > 0 {
			out[index].TenantStates = append(out[index].TenantStates, SharedDBPoolTenantStateCount{
				State: TenantStatus(tenantStatus), Count: stateCount,
			})
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list shared db pool metric snapshot rows: %w", err)
	}
	return out, nil
}

// IncrSharedDBTenantCount adjusts the denormalized tenant_count of a db_pool
// row by delta (negative to decrement), clamped at >= 0. Returns ErrNotFound
// when dbID is unknown.
func (s *Store) IncrSharedDBTenantCount(ctx context.Context, dbID int64, delta int) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "incr_shared_db_tenant_count", start, &err)
	if dbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	res, err := s.db.ExecContext(ctx,
		`UPDATE db_pool SET tenant_count = GREATEST(tenant_count + ?, 0) WHERE db_id = ?`,
		delta, dbID)
	if err != nil {
		return fmt.Errorf("adjust tenant_count for db %d: %w", dbID, err)
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("tenant_count rows affected for db %d: %w", dbID, err)
	}
	if affected > 0 {
		return nil
	}
	// MySQL reports 0 affected rows when the value did not change (delta 0 or
	// a clamp no-op); confirm the row exists before reporting success.
	var one int
	err = s.db.QueryRowContext(ctx, `SELECT 1 FROM db_pool WHERE db_id = ?`, dbID).Scan(&one)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return fmt.Errorf("check shared db %d: %w", dbID, err)
	}
	return nil
}

// UpsertTenantPlacement records or replaces the placement of a filesystem.
// The epoch is left untouched by upserts: it stays at its default of 1 unless
// a migration explicitly bumps it.
func (s *Store) UpsertTenantPlacement(ctx context.Context, p *TenantPlacement) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_tenant_placement", start, &err)
	if p == nil {
		return fmt.Errorf("tenant placement is required")
	}
	if p.FsID <= 0 {
		return fmt.Errorf("fs_id must be positive")
	}
	if p.DbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	switch p.Placement {
	case PlacementShared, PlacementDedicated:
	default:
		return fmt.Errorf("unsupported placement %q", p.Placement)
	}
	switch p.SchemaShape {
	case SchemaShapeStandalone, SchemaShapeShared:
	default:
		return fmt.Errorf("unsupported schema shape %q", p.SchemaShape)
	}
	status := p.Status
	if status == "" {
		status = sharedDBStatusActive
	}
	switch status {
	case sharedDBStatusActive, "migrating":
	default:
		return fmt.Errorf("unsupported placement status %q", status)
	}
	var target any
	if p.TargetDbID != nil {
		target = *p.TargetDbID
	}
	if _, err := s.db.ExecContext(ctx,
		`INSERT INTO tenant_placements (fs_id, db_id, placement, schema_shape, status, target_db_id)
			VALUES (?, ?, ?, ?, ?, ?)
			ON DUPLICATE KEY UPDATE db_id = VALUES(db_id), placement = VALUES(placement),
				schema_shape = VALUES(schema_shape), status = VALUES(status),
				target_db_id = VALUES(target_db_id)`,
		p.FsID, p.DbID, p.Placement, p.SchemaShape, status, target); err != nil {
		return fmt.Errorf("upsert tenant placement for fs_id %d: %w", p.FsID, err)
	}
	return nil
}

// GetTenantPlacement returns the placement for fsID, or ErrNotFound. Callers
// treat a missing row as "standalone, legacy path".
func (s *Store) GetTenantPlacement(ctx context.Context, fsID int64) (p *TenantPlacement, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tenant_placement", start, &err)
	if fsID <= 0 {
		return nil, fmt.Errorf("fs_id must be positive")
	}
	var target sql.NullInt64
	p = &TenantPlacement{}
	err = s.db.QueryRowContext(ctx,
		`SELECT fs_id, db_id, placement, schema_shape, status, target_db_id, epoch
			FROM tenant_placements WHERE fs_id = ?`, fsID).
		Scan(&p.FsID, &p.DbID, &p.Placement, &p.SchemaShape, &p.Status, &target, &p.Epoch)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("get tenant placement for fs_id %d: %w", fsID, err)
	}
	if target.Valid {
		p.TargetDbID = &target.Int64
	}
	return p, nil
}

// DeleteTenantPlacement removes the placement for fsID. It is idempotent:
// deleting an unknown fsID succeeds.
func (s *Store) DeleteTenantPlacement(ctx context.Context, fsID int64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_tenant_placement", start, &err)
	if fsID <= 0 {
		return fmt.Errorf("fs_id must be positive")
	}
	if _, err := s.db.ExecContext(ctx,
		`DELETE FROM tenant_placements WHERE fs_id = ?`, fsID); err != nil {
		return fmt.Errorf("delete tenant placement for fs_id %d: %w", fsID, err)
	}
	return nil
}

// ErrSharedDBCapacityExhausted is returned when a shared-pool reservation
// cannot be made because the target pool has no free capacity
// (max_tenants reached).
var ErrSharedDBCapacityExhausted = errors.New("shared db capacity exhausted")

// ErrSharedDBQuotaNotMaterialized is returned when a shared tenant quota
// mutation is attempted before the compatibility row has been persisted.
var ErrSharedDBQuotaNotMaterialized = errors.New("shared tenant quota is not materialized")

// CompleteSharedTenantProvision atomically performs every meta write that
// places a pending tenant into a shared pool, in one transaction:
//
//  1. the conditional capacity reservation (two concurrent provisions
//     cannot take the same last slot — the loser gets
//     ErrSharedDBCapacityExhausted),
//  2. the placement insert,
//  3. the provider re-label and readiness transition on the tenant row
//     (pending while connection metadata is incomplete, provisioning while
//     schema init is running, and active only when the DB pool is active), and
//  4. the owner API key insert (ErrDuplicate on key id collision).
//
// Either all of it commits or none does: a shared tenant can never become
// active without its placement row, its reserved capacity, and its owner
// key, and a failed attempt leaves no partial state behind.
func (s *Store) CompleteSharedTenantProvision(ctx context.Context, tenantID, provider string, p *TenantPlacement, k *APIKey) (err error) {
	return s.completeSharedTenantProvision(ctx, tenantID, provider, p, k, nil, SharedDBCapacityNormal, 0)
}

// CompleteSharedTenantProvisionEmergency uses a caller-computed runtime hard
// cap. It is reserved for direct-request fallback after physical DB creation
// fails; tenant-pool refill must use the normal wrapper.
func (s *Store) CompleteSharedTenantProvisionEmergency(ctx context.Context, tenantID, provider string, p *TenantPlacement, k *APIKey, hardCap int) error {
	if hardCap <= 0 {
		return fmt.Errorf("emergency hard cap must be positive")
	}
	return s.completeSharedTenantProvision(ctx, tenantID, provider, p, k, nil, SharedDBCapacityEmergency, hardCap)
}

// CompleteSharedTenantPoolMember reserves a shared placement and records a
// free logical-pool membership in the same transaction. Prewarmed members do
// not receive an owner key until claim.
func (s *Store) CompleteSharedTenantPoolMember(ctx context.Context, tenantID, provider string, p *TenantPlacement, membership *TenantPoolMembership) error {
	return s.completeSharedTenantProvision(ctx, tenantID, provider, p, nil, membership, SharedDBCapacityNormal, 0)
}

func (s *Store) completeSharedTenantProvision(ctx context.Context, tenantID, provider string, p *TenantPlacement, k *APIKey, membership *TenantPoolMembership, capacityMode SharedDBCapacityMode, hardCap int) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "complete_shared_tenant_provision", start, &err)
	if tenantID == "" {
		return fmt.Errorf("tenant id is required")
	}
	if provider != tidbCloudNativeSharedProvider {
		return fmt.Errorf("shared tenant provider %q is required", tidbCloudNativeSharedProvider)
	}
	if p == nil {
		return fmt.Errorf("tenant placement is required")
	}
	if p.FsID <= 0 {
		return fmt.Errorf("fs_id must be positive")
	}
	if p.DbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	if p.Placement != PlacementShared || p.SchemaShape != SchemaShapeShared {
		return fmt.Errorf("provision requires shared placement and schema shape")
	}
	if k == nil && membership == nil {
		return fmt.Errorf("api key or tenant pool membership is required")
	}
	if capacityMode != SharedDBCapacityNormal && capacityMode != SharedDBCapacityEmergency {
		return fmt.Errorf("unsupported shared db capacity mode %q", capacityMode)
	}
	var scopeKind APIKeyScopeKind
	if k != nil {
		if k.ExpiresAt != nil {
			return fmt.Errorf("shared pool activation does not support api keys with expires_at")
		}
		scopeKind, err = apiKeyScopeKindForInsert(k)
		if err != nil {
			return err
		}
	}
	if membership != nil {
		if membership.TenantID != tenantID || strings.TrimSpace(membership.PoolID) == "" {
			return fmt.Errorf("valid tenant pool membership is required")
		}
	}
	status := p.Status
	if status == "" {
		status = sharedDBStatusActive
	}
	now := time.Now().UTC()
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var dedicatedBindingCount int
		if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM tenant_tidbcloud_org_bindings
			WHERE tenant_id = ?`, tenantID).Scan(&dedicatedBindingCount); err != nil {
			return err
		}
		if dedicatedBindingCount != 0 {
			return fmt.Errorf("tenant %q already has a dedicated tidbcloud org binding", tenantID)
		}
		var dbPoolStatus string
		var clusterID sql.NullString
		var maxTenants, tenantCount int
		var softCapReached bool
		if err := tx.QueryRowContext(ctx, `SELECT status, cluster_id, max_tenants, tenant_count, soft_cap_reached
			FROM db_pool WHERE db_id = ? FOR UPDATE`, p.DbID).
			Scan(&dbPoolStatus, &clusterID, &maxTenants, &tenantCount, &softCapReached); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return fmt.Errorf("lock db pool %d: %w", p.DbID, err)
		}
		switch capacityMode {
		case SharedDBCapacityNormal:
			if dbPoolStatus != SharedDBStatusActive &&
				((dbPoolStatus != SharedDBStatusProvisioning && dbPoolStatus != SharedDBStatusPending) ||
					(membership == nil && (!clusterID.Valid || clusterID.String == ""))) {
				return fmt.Errorf("db pool %d is not physically ready: status=%s cluster=%q membership=%t: %w", p.DbID, dbPoolStatus, clusterID.String, membership != nil, ErrSharedDBCapacityExhausted)
			}
			if maxTenants <= 0 || softCapReached || tenantCount >= maxTenants {
				return fmt.Errorf("db pool %d soft capacity exhausted: count=%d max=%d latch=%t: %w", p.DbID, tenantCount, maxTenants, softCapReached, ErrSharedDBCapacityExhausted)
			}
		case SharedDBCapacityEmergency:
			if dbPoolStatus != SharedDBStatusActive || maxTenants <= 0 || hardCap < maxTenants || tenantCount >= hardCap {
				return ErrSharedDBCapacityExhausted
			}
		}
		var res sql.Result
		if capacityMode == SharedDBCapacityEmergency {
			res, err = tx.ExecContext(ctx, `UPDATE db_pool
				SET tenant_count = tenant_count + 1, soft_cap_reached = 1
				WHERE db_id = ? AND status = ?
					AND max_tenants > 0 AND tenant_count + 1 <= ?`,
				p.DbID, SharedDBStatusActive, hardCap)
		} else {
			normalQuery := `UPDATE db_pool
				SET soft_cap_reached = CASE
					WHEN max_tenants > 0 AND tenant_count + 1 >= max_tenants THEN 1
					ELSE soft_cap_reached END,
					tenant_count = tenant_count + 1
				WHERE db_id = ? AND (%s)
					AND max_tenants > 0 AND soft_cap_reached = 0 AND tenant_count + 1 <= max_tenants`
			if membership != nil {
				normalQuery = fmt.Sprintf(normalQuery, "status IN (?, ?, ?)")
				res, err = tx.ExecContext(ctx, normalQuery, p.DbID, SharedDBStatusActive, SharedDBStatusProvisioning, SharedDBStatusPending)
			} else {
				normalQuery = fmt.Sprintf(normalQuery, "status = ? OR (status IN (?, ?) AND cluster_id IS NOT NULL)")
				res, err = tx.ExecContext(ctx, normalQuery, p.DbID, SharedDBStatusActive, SharedDBStatusProvisioning, SharedDBStatusPending)
			}
		}
		if err != nil {
			return fmt.Errorf("reserve capacity on db %d: %w", p.DbID, err)
		}
		affected, err := res.RowsAffected()
		if err != nil {
			return fmt.Errorf("reserve capacity rows affected for db %d: %w", p.DbID, err)
		}
		if affected == 0 {
			return ErrSharedDBCapacityExhausted
		}
		var tenantStatus TenantStatus
		switch dbPoolStatus {
		case SharedDBStatusActive:
			tenantStatus = TenantActive
		case SharedDBStatusProvisioning:
			tenantStatus = TenantProvisioning
		case SharedDBStatusPending:
			tenantStatus = TenantPending
		default:
			return fmt.Errorf("db pool %d is not allocatable in status %q", p.DbID, dbPoolStatus)
		}
		var target any
		if p.TargetDbID != nil {
			target = *p.TargetDbID
		}
		if _, err := tx.ExecContext(ctx,
			`INSERT INTO tenant_placements (fs_id, db_id, placement, schema_shape, status, target_db_id)
				VALUES (?, ?, ?, ?, ?, ?)`,
			p.FsID, p.DbID, p.Placement, p.SchemaShape, status, target); err != nil {
			return fmt.Errorf("insert placement for fs_id %d: %w", p.FsID, err)
		}
		res, err = tx.ExecContext(ctx,
			`UPDATE tenants SET provider = ?, status = ?, updated_at = ? WHERE id = ?`,
			provider, tenantStatus, now, tenantID)
		if err != nil {
			return fmt.Errorf("activate tenant %s: %w", tenantID, err)
		}
		affected, err = res.RowsAffected()
		if err != nil {
			return fmt.Errorf("activate tenant rows affected for %s: %w", tenantID, err)
		}
		if affected == 0 {
			return ErrNotFound
		}
		if k != nil {
			if _, err := tx.ExecContext(ctx, `INSERT INTO tenant_api_keys
			(id, tenant_id, key_name, jwt_ciphertext, jwt_hash, token_version, status, scope_kind,
			 issued_by_provider, issued_by_subject_key, issued_by_metadata_json,
			 issued_at, revoked_at, created_at, updated_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
				k.ID, k.TenantID, k.KeyName, k.JWTCiphertext, k.JWTHash, k.TokenVersion, k.Status, scopeKind,
				k.IssuedByProvider, k.IssuedBySubjectKey, nullableBytes(k.IssuedByMetadataJSON),
				k.IssuedAt.UTC(), k.RevokedAt, k.CreatedAt.UTC(), k.UpdatedAt.UTC()); err != nil {
				if isDuplicateEntry(err) {
					return ErrDuplicate
				}
				return fmt.Errorf("insert owner api key: %w", err)
			}
		}
		if membership != nil {
			createdAt := membership.CreatedAt
			if createdAt.IsZero() {
				createdAt = now
			}
			var organizationID any
			if strings.TrimSpace(membership.TiDBCloudOrganizationID) != "" {
				organizationID = strings.TrimSpace(membership.TiDBCloudOrganizationID)
			}
			if _, err := tx.ExecContext(ctx, `INSERT INTO tenant_pool_memberships
				(tenant_id, tidbcloud_organization_id, pool_id, pool_status, created_at, updated_at)
				VALUES (?, ?, ?, ?, ?, ?)`, tenantID, organizationID, membership.PoolID,
				TenantPoolBindingFree, createdAt, now); err != nil {
				if isDuplicateEntry(err) {
					return ErrDuplicate
				}
				return fmt.Errorf("insert tenant pool membership: %w", err)
			}
		}
		return nil
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	return nil
}

// ActivateSharedTenantsBatch activates ready shared tenants after their
// physical DB pool becomes active. Direct-create tenants are eligible through
// an owner key; prewarmed tenants are eligible through a free pool membership.
func (s *Store) ActivateSharedTenantsBatch(ctx context.Context, dbID int64, limit int) (activated int, err error) {
	start := time.Now()
	defer observeMeta(ctx, "activate_shared_tenants_batch", start, &err)
	if dbID <= 0 {
		return 0, fmt.Errorf("db_id must be positive")
	}
	if limit <= 0 {
		return 0, fmt.Errorf("activation limit must be positive")
	}
	var tenantIDs []string
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var poolStatus string
		if scanErr := tx.QueryRowContext(ctx, `SELECT status FROM db_pool WHERE db_id = ? FOR UPDATE`, dbID).Scan(&poolStatus); scanErr != nil {
			if errors.Is(scanErr, sql.ErrNoRows) {
				return ErrNotFound
			}
			return scanErr
		}
		if poolStatus != SharedDBStatusActive {
			return fmt.Errorf("db pool %d is not active", dbID)
		}
		rows, queryErr := tx.QueryContext(ctx, `SELECT t.id
			FROM tenants t
			JOIN fs_registry f ON f.tenant_id = t.id
			JOIN tenant_placements p ON p.fs_id = f.fs_id
			WHERE p.db_id = ? AND p.status = ? AND t.provider = ? AND t.status = ?
				AND (EXISTS (SELECT 1 FROM tenant_api_keys k
					WHERE k.tenant_id = t.id AND k.scope_kind = ? AND k.status = ?)
					OR EXISTS (SELECT 1 FROM tenant_pool_memberships m
						WHERE m.tenant_id = t.id AND m.pool_status = ?))
			ORDER BY t.created_at, t.id LIMIT ? FOR UPDATE`, dbID, SharedDBStatusActive,
			tidbCloudNativeSharedProvider, TenantProvisioning, APIKeyScopeKindOwner, APIKeyActive,
			TenantPoolBindingFree, limit)
		if queryErr != nil {
			return queryErr
		}
		for rows.Next() {
			var tenantID string
			if scanErr := rows.Scan(&tenantID); scanErr != nil {
				_ = rows.Close()
				return scanErr
			}
			tenantIDs = append(tenantIDs, tenantID)
		}
		if rowsErr := rows.Err(); rowsErr != nil {
			_ = rows.Close()
			return rowsErr
		}
		if closeErr := rows.Close(); closeErr != nil {
			return closeErr
		}
		now := time.Now().UTC()
		for _, tenantID := range tenantIDs {
			res, updateErr := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ?
				WHERE id = ? AND status = ? AND provider = ?`, TenantActive, now, tenantID,
				TenantProvisioning, tidbCloudNativeSharedProvider)
			if updateErr != nil {
				return updateErr
			}
			n, rowsErr := res.RowsAffected()
			if rowsErr != nil {
				return rowsErr
			}
			activated += int(n)
		}
		return nil
	})
	if err != nil {
		return activated, err
	}
	for _, tenantID := range tenantIDs {
		s.apiKeys.evictTenant(tenantID)
	}
	return activated, nil
}

// DeleteTenantPlacementAndDecrCount atomically removes a tenant's placement
// and releases its capacity slot in a single transaction. Making the two
// writes atomic keeps the shared delete path retry-safe: a failure rolls both
// back, so a retried delete re-enters with the placement row still present
// and cannot orphan the row or double-decrement the counter. The reopen ratio
// is runtime policy supplied by the server and is never persisted. Returns
// ErrNotFound when the placement row does not exist.
func (s *Store) DeleteTenantPlacementAndDecrCount(ctx context.Context, fsID, dbID int64, reopenRatio float64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_tenant_placement_and_decr_count", start, &err)
	if fsID <= 0 {
		return fmt.Errorf("fs_id must be positive")
	}
	if dbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	if _, err := SharedDBReopenThresholdForRatio(1, reopenRatio); err != nil {
		return err
	}
	return s.InTx(ctx, func(tx *sql.Tx) error {
		return releaseTenantPlacementAndDecrCountTx(ctx, tx, fsID, dbID, reopenRatio, true)
	})
}

// FinalizeSharedTenantDeleteMetadata atomically releases shared DB capacity
// and logical-pool membership. Tenants requiring external cleanup retain a
// deleting placement as their authorization anchor until job finalization;
// tenants without external cleanup remove placement and become deleted in the
// same transaction. A failed transition always retains the active placement.
func (s *Store) FinalizeSharedTenantDeleteMetadata(ctx context.Context, tenantID string, fsID, dbID int64, reopenRatio float64, markDeleted bool) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "finalize_shared_tenant_delete_metadata", start, &err)
	if strings.TrimSpace(tenantID) == "" {
		return fmt.Errorf("tenant id is required")
	}
	if fsID <= 0 {
		return fmt.Errorf("fs_id must be positive")
	}
	if dbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	if _, err := SharedDBReopenThresholdForRatio(1, reopenRatio); err != nil {
		return err
	}
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var status TenantStatus
		if scanErr := tx.QueryRowContext(ctx, `SELECT status FROM tenants WHERE id = ? FOR UPDATE`, tenantID).Scan(&status); scanErr != nil {
			if errors.Is(scanErr, sql.ErrNoRows) {
				return ErrNotFound
			}
			return scanErr
		}
		if status != TenantDeleting {
			return ErrNotFound
		}
		if err := releaseTenantPlacementAndDecrCountTx(ctx, tx, fsID, dbID, reopenRatio, markDeleted); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM tenant_pool_memberships WHERE tenant_id = ?`, tenantID); err != nil {
			return fmt.Errorf("delete tenant pool membership for %s: %w", tenantID, err)
		}
		if !markDeleted {
			return nil
		}
		now := time.Now().UTC()
		res, err := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ?`, TenantDeleted, now, tenantID)
		if err != nil {
			return err
		}
		if err := requireAffected(res); err != nil {
			return err
		}
		if err := deleteTenantExtractConfigsTx(ctx, tx, tenantID); err != nil {
			return err
		}
		if err := deleteTenantEmbeddingConfigTx(ctx, tx, tenantID); err != nil {
			return err
		}
		if err := deleteTenantObjectNamespaceTx(ctx, tx, tenantID); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM tenant_tidbcloud_org_bindings WHERE tenant_id = ?`, tenantID); err != nil {
			return err
		}
		return insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkMetricsCleanup)
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	return nil
}

// DiscardUnclaimedSharedTenant atomically removes a shared tenant staged by a
// failed provisioning wave. Eligibility is anchored by a free pool membership,
// which serializes this compensation with claim and proves the tenant has not
// been handed to a user. Active tenants without that anchor must go through the
// purge-confirmed delete path instead.
func (s *Store) DiscardUnclaimedSharedTenant(ctx context.Context, tenantID string, fsID, dbID int64, reopenRatio float64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "discard_unclaimed_shared_tenant", start, &err)
	if strings.TrimSpace(tenantID) == "" {
		return fmt.Errorf("tenant id is required")
	}
	if fsID <= 0 {
		return fmt.Errorf("fs_id must be positive")
	}
	if dbID <= 0 {
		return fmt.Errorf("db_id must be positive")
	}
	if _, err := SharedDBReopenThresholdForRatio(1, reopenRatio); err != nil {
		return err
	}
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var status TenantStatus
		var provider string
		if err := tx.QueryRowContext(ctx, `SELECT t.status, t.provider
			FROM tenants t
			JOIN tenant_pool_memberships m ON m.tenant_id = t.id
			JOIN fs_registry f ON f.tenant_id = t.id
			JOIN tenant_placements p ON p.fs_id = f.fs_id
			WHERE t.id = ? AND m.pool_status = ? AND f.fs_id = ? AND p.db_id = ?
			FOR UPDATE`, tenantID, TenantPoolBindingFree, fsID, dbID).Scan(&status, &provider); err != nil {
			if !errors.Is(err, sql.ErrNoRows) {
				return err
			}
			// A joined miss is ambiguous: losing the free membership is a benign
			// claim race, while missing tenant/placement metadata is corruption.
			var freeMembership int
			if scanErr := tx.QueryRowContext(ctx, `SELECT 1 FROM tenant_pool_memberships
				WHERE tenant_id = ? AND pool_status = ?`, tenantID, TenantPoolBindingFree).Scan(&freeMembership); scanErr != nil {
				if errors.Is(scanErr, sql.ErrNoRows) {
					return ErrTenantPoolMembershipNotFree
				}
				return scanErr
			}
			return ErrNotFound
		}
		if provider != tidbCloudNativeSharedProvider ||
			(status != TenantPending && status != TenantProvisioning && status != TenantActive) {
			return fmt.Errorf("tenant %s is not an unclaimed shared tenant", tenantID)
		}
		if err := releaseTenantPlacementAndDecrCountTx(ctx, tx, fsID, dbID, reopenRatio, true); err != nil {
			return err
		}
		res, err := tx.ExecContext(ctx, `DELETE FROM tenant_pool_memberships
			WHERE tenant_id = ? AND pool_status = ?`, tenantID, TenantPoolBindingFree)
		if err != nil {
			return fmt.Errorf("delete tenant pool membership for %s: %w", tenantID, err)
		}
		if err := requireAffected(res); err != nil {
			return err
		}
		now := time.Now().UTC()
		res, err = tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ?
			WHERE id = ? AND status = ?`, TenantDeleted, now, tenantID, status)
		if err != nil {
			return err
		}
		if err := requireAffected(res); err != nil {
			return err
		}
		if err := deleteTenantExtractConfigsTx(ctx, tx, tenantID); err != nil {
			return err
		}
		if err := deleteTenantEmbeddingConfigTx(ctx, tx, tenantID); err != nil {
			return err
		}
		if err := deleteTenantObjectNamespaceTx(ctx, tx, tenantID); err != nil {
			return err
		}
		return insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkMetricsCleanup)
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	return nil
}

// MarkUnclaimedSharedTenantFailed marks a staged shared tenant failed only
// while its free membership still proves the tenant has not been claimed.
func (s *Store) MarkUnclaimedSharedTenantFailed(ctx context.Context, tenantID string) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_unclaimed_shared_tenant_failed", start, &err)
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return false, fmt.Errorf("tenant id is required")
	}
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var tenantStatus TenantStatus
		if err := tx.QueryRowContext(ctx, `SELECT t.status FROM tenants t
			JOIN tenant_pool_memberships m ON m.tenant_id = t.id
			WHERE t.id = ? AND t.provider = ? AND t.status IN (?, ?, ?)
				AND m.pool_status = ? FOR UPDATE`, tenantID, tidbCloudNativeSharedProvider,
			TenantPending, TenantProvisioning, TenantActive, TenantPoolBindingFree).Scan(&tenantStatus); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return nil
			}
			return err
		}
		res, err := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ?
			WHERE id = ? AND provider = ? AND status = ?`, TenantFailed, time.Now().UTC(), tenantID,
			tidbCloudNativeSharedProvider, tenantStatus)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		updated = n == 1
		return nil
	})
	if err != nil {
		return false, err
	}
	if updated {
		s.apiKeys.evictTenant(tenantID)
	}
	return updated, nil
}

func releaseTenantPlacementAndDecrCountTx(ctx context.Context, tx *sql.Tx, fsID, dbID int64, reopenRatio float64, deletePlacement bool) error {
	var (
		res sql.Result
		err error
	)
	if deletePlacement {
		res, err = tx.ExecContext(ctx, `DELETE FROM tenant_placements WHERE fs_id = ? AND db_id = ?`, fsID, dbID)
	} else {
		res, err = tx.ExecContext(ctx, `UPDATE tenant_placements SET status = ?
			WHERE fs_id = ? AND db_id = ? AND status <> ?`, PlacementStatusDeleting, fsID, dbID, PlacementStatusDeleting)
	}
	if err != nil {
		return fmt.Errorf("release tenant placement for fs_id %d: %w", fsID, err)
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("delete tenant placement rows affected for fs_id %d: %w", fsID, err)
	}
	if affected == 0 {
		return ErrNotFound
	}
	var maxTenants, tenantCount int
	var softCapReached bool
	if err := tx.QueryRowContext(ctx, `SELECT max_tenants, tenant_count, soft_cap_reached FROM db_pool WHERE db_id = ? FOR UPDATE`, dbID).
		Scan(&maxTenants, &tenantCount, &softCapReached); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return fmt.Errorf("lock capacity for db %d: %w", dbID, err)
	}
	reopenThreshold, err := SharedDBReopenThresholdForRatio(maxTenants, reopenRatio)
	if err != nil {
		return err
	}
	newTenantCount := tenantCount - 1
	if newTenantCount < 0 {
		newTenantCount = 0
	}
	nextSoftCapReached := softCapReached
	if maxTenants == 0 {
		nextSoftCapReached = true
	} else if newTenantCount <= reopenThreshold {
		nextSoftCapReached = false
	}
	_, err = tx.ExecContext(ctx, `UPDATE db_pool
			SET soft_cap_reached = ?, tenant_count = ?
			WHERE db_id = ?`, nextSoftCapReached, newTenantCount, dbID)
	if err != nil {
		return fmt.Errorf("release capacity on db %d: %w", dbID, err)
	}
	// The SELECT ... FOR UPDATE above already established that the pool
	// exists. RowsAffected may be zero when the delete leaves both values
	// unchanged (for example an already-zero counter), which is still a
	// valid atomic placement release.
	return nil
}

// scanSharedDBRow scans one db_pool row selected with sharedDBSelectColumns.
func scanSharedDBRow(row *sql.Row) (*SharedDB, error) {
	return scanSharedDBScanner(row)
}

type sharedDBScanner interface {
	Scan(dest ...any) error
}

func scanSharedDBScanner(row sharedDBScanner) (*SharedDB, error) {
	var rec SharedDB
	var organizationID, clusterID, cloudProvider, region sql.NullString
	var host, user, name sql.NullString
	var port, spendingLimit sql.NullInt64
	if err := row.Scan(&rec.ID, &rec.UUID, &organizationID, &clusterID, &rec.ProvisioningKey, &cloudProvider, &region,
		&rec.Role, &host, &port, &user, &rec.PasswordCipher, &name, &rec.TLSMode,
		&rec.MaxTenants, &rec.TenantCount, &rec.SoftCapReached, &spendingLimit, &rec.SchemaVersion, &rec.Status,
		&rec.CreatedAt, &rec.UpdatedAt); err != nil {
		return nil, err
	}
	rec.TiDBCloudOrganizationID = organizationID.String
	rec.ClusterID = clusterID.String
	rec.CloudProvider = cloudProvider.String
	rec.Region = region.String
	rec.Host = host.String
	rec.Port = int(port.Int64)
	rec.User = user.String
	rec.Name = name.String
	if spendingLimit.Valid {
		value := spendingLimit.Int64
		rec.SpendingLimit = &value
	}
	return &rec, nil
}