// Package meta provides control-plane metadata storage for multi-tenant auth.
package meta

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/mysqlutil"
	"github.com/mem9-ai/drive9/pkg/pathutil"
	"github.com/mem9-ai/drive9/pkg/schemaspec"
	"go.uber.org/zap"
)

var (
	ErrNotFound                        = errors.New("not found")
	ErrDuplicate                       = errors.New("duplicate entry")
	ErrStorageQuotaExceeded            = errors.New("tenant storage quota exceeded")
	ErrFileCountQuotaExceeded          = errors.New("tenant file count quota exceeded")
	ErrReservationAlreadyExists        = errors.New("upload reservation already exists")
	ErrQuotaReservationBusy            = errors.New("quota reservation busy")
	ErrTiDBCloudFreeQuotaBusy          = errors.New("free tenant quota check is busy; retry later")
	ErrTiDBCloudFreeTenantLimitReached = errors.New("free TiDB Cloud tenant limit reached")
	ErrTenantPoolHasFreeMembers        = errors.New("tenant pool still has live free members")
	ErrTenantPoolMembershipNotFree     = errors.New("tenant pool membership is absent or no longer free")
	ErrSharedDBPoolIdentityConflict    = errors.New("shared db pool physical identity changed after provisioning started")
	ErrSharedDBPoolNotProvisioning     = errors.New("shared db pool is not provisioning")
	ErrAppendLogBindingMismatch        = errors.New("append-log binding does not match configured target")
	ErrAPIKeyTerminal                  = errors.New("api key is terminal")
	ErrAPIKeyLimitExceeded             = errors.New("api key limit exceeded")
	// ErrAPIKeyVersionUnchanged is returned by RotateAPIKey when next.TokenVersion
	// does not strictly advance past the optimistic-lock oldVersion. Rotation must
	// never keep or regress the version so a residual-window stale Bearer cannot
	// re-pass the caller-version conflict check.
	ErrAPIKeyVersionUnchanged = errors.New("api key token_version must change on rotate")
	// ErrAPIKeyExpired is returned when activating a key whose expires_at is past.
	ErrAPIKeyExpired = errors.New("api key is expired")
	// ErrTenantNotActive is returned when a key-issuance path locks a tenant that
	// is not in an allowed status for that path (e.g. deleting concurrent with
	// generate, or non-provisioning/active for issuer rotate).
	ErrTenantNotActive = errors.New("tenant is not active")
	// ErrAPIKeyTenantMismatch is returned when RotateIssuerAPIKey is given a next
	// record whose TenantID does not match the locked tenantID parameter.
	ErrAPIKeyTenantMismatch = errors.New("api key tenant_id does not match")
	errAppendLogBindingRace = errors.New("append-log binding compare-and-set lost race")
)

type TenantStatus string

const (
	TenantPending      TenantStatus = "pending"
	TenantProvisioning TenantStatus = "provisioning"
	TenantActive       TenantStatus = "active"
	TenantFailed       TenantStatus = "failed"
	TenantSuspended    TenantStatus = "suspended"
	TenantDeleting     TenantStatus = "deleting"
	TenantDeleted      TenantStatus = "deleted"
)

// Tenant notify work-mask bits are defined in the meta package because the
// outbox is persisted here. Server and backend aliases reference these
// constants instead of allocating bits independently.
const (
	TenantNotifyWorkSSE = 1 << iota
	TenantNotifyWorkSemantic
	TenantNotifyWorkFileGC
	TenantNotifyWorkMetricsCleanup
	TenantNotifyWorkAPIKeyCacheCleanup
	TenantNotifyWorkDeleteReconcile
	TenantNotifyWorkStorageTierScan
	TenantNotifyWorkEmbeddingConfigCleanup
)

var allTenantStatuses = []TenantStatus{
	TenantPending,
	TenantProvisioning,
	TenantActive,
	TenantFailed,
	TenantSuspended,
	TenantDeleting,
	TenantDeleted,
}

// purgeVisibleTenantStatuses is the set of tenant statuses the free-member
// purge can see and act on. The pool-delete orphan cleanup in
// DeleteTenantPoolAndDetachUsedMembers intentionally matches terminal
// statuses positively instead of complementing this list, so a live status
// added here later cannot be mistaken for garbage there.
var purgeVisibleTenantStatuses = []TenantStatus{
	TenantPending,
	TenantProvisioning,
	TenantActive,
	TenantFailed,
}

const tidbCloudNativeProvider = "tidb_cloud_native"
const tidbCloudNativeSharedProvider = "tidb_cloud_native_shared"
const maxTiDBCloudOrgBindingDuplicateTuples = 20

type TenantKind string

const (
	TenantKindLive TenantKind = "live"
	TenantKindFork TenantKind = "fork"
)

type APIKeyStatus string

const (
	APIKeyActive   APIKeyStatus = "active"
	APIKeyDisabled APIKeyStatus = "disabled"
	APIKeyRevoked  APIKeyStatus = "revoked"
)

// DefaultMaxActiveAPIKeys is the soft per-tenant cap on non-revoked, non-expired keys.
const DefaultMaxActiveAPIKeys = 100

type APIKeyScopeKind string

const (
	APIKeyScopeKindOwner APIKeyScopeKind = "owner"
	APIKeyScopeKindFS    APIKeyScopeKind = "fs_scoped"
)

type Tenant struct {
	ID                 string
	DisplayName        string
	Label              map[string]string
	Status             TenantStatus
	Kind               TenantKind
	ParentTenantID     string
	StorageNamespaceID string
	DBHost             string
	DBPort             int
	DBUser             string
	DBPasswordCipher   []byte
	DBName             string
	DBTLS              bool
	Provider           string
	TiDBCloudOrgID     string
	ClusterID          string
	BranchID           string
	ClaimURL           string
	ClaimExpiresAt     *time.Time
	SchemaVersion      int
	S3EncryptionMode   S3EncryptionMode
	S3KMSKeyID         string
	S3BucketKeyEnabled *bool
	CreatedAt          time.Time
	UpdatedAt          time.Time
}

type TenantCounts struct {
	Statuses []TenantStatusCount
}

type TenantStatusCount struct {
	Status TenantStatus
	Count  int64
}

func TenantStatuses() []TenantStatus {
	return append([]TenantStatus(nil), allTenantStatuses...)
}

func (c TenantCounts) Count(status TenantStatus) int64 {
	for _, rec := range c.Statuses {
		if rec.Status == status {
			return rec.Count
		}
	}
	return 0
}

type TenantAutoEmbeddingProfile struct {
	TenantID      string
	EmbeddingMode string
	Model         string
	Dimensions    int
	OptionsJSON   string
	APIBase       string
	APIKeyCipher  []byte
	CreatedAt     time.Time
	UpdatedAt     time.Time
}

const (
	TenantEmbeddingModeAuto    = "auto"
	TenantEmbeddingModeFTSOnly = "fts_only"
)

func ResolveTenantEmbeddingMode(persisted string, defaultFTSOnly bool) (mode string, wasNull bool, err error) {
	switch persisted {
	case "":
		if defaultFTSOnly {
			return TenantEmbeddingModeFTSOnly, true, nil
		}
		return TenantEmbeddingModeAuto, true, nil
	case TenantEmbeddingModeAuto, TenantEmbeddingModeFTSOnly:
		return persisted, false, nil
	default:
		return "", false, fmt.Errorf("unsupported tenant embedding mode %q", persisted)
	}
}

type StorageNamespaceState string

const (
	StorageNamespaceActive   StorageNamespaceState = "active"
	StorageNamespaceDeleting StorageNamespaceState = "deleting"
	StorageNamespaceDeleted  StorageNamespaceState = "deleted"
)

// AppendLogBindingState records whether a storage namespace has an immutable
// S3 Express target for append-log objects.
type AppendLogBindingState string

const (
	AppendLogBindingUnbound AppendLogBindingState = "unbound"
	AppendLogBindingBound   AppendLogBindingState = "bound"
)

type StorageNamespace struct {
	ID                    string
	OwnerTenantID         string
	Backend               string
	Bucket                string
	Prefix                string
	AppendLogBindingState AppendLogBindingState
	AppendLogBucket       string
	AppendLogPrefix       string
	AppendLogRegion       string
	State                 StorageNamespaceState
	CreatedAt             time.Time
	UpdatedAt             time.Time
}

type ObjectGCCandidateState string

const staleObjectGCPreparingAge = 24 * time.Hour

const (
	ObjectGCCandidatePending ObjectGCCandidateState = "pending"
	// Preparing fences a possibly in-flight S3 mutation. The worker may
	// recover it only after the mutation's maximum lifetime has elapsed.
	ObjectGCCandidatePreparing ObjectGCCandidateState = "preparing"
	// Publishing is fail-closed: only producer cleanup or a committed
	// overwrite/delete may replace it. The GC worker never claims this state
	// because an earlier metadata COMMIT may have an unknown outcome.
	ObjectGCCandidatePublishing ObjectGCCandidateState = "publishing"
	ObjectGCCandidateDeleting   ObjectGCCandidateState = "deleting"
	ObjectGCCandidateDeleted    ObjectGCCandidateState = "deleted"
	ObjectGCCandidateFailed     ObjectGCCandidateState = "failed"
)

type ObjectGCCandidateReason string

const (
	ObjectGCReasonOverwrite   ObjectGCCandidateReason = "overwrite"
	ObjectGCReasonFileDelete  ObjectGCCandidateReason = "file_delete"
	ObjectGCReasonFailedWrite ObjectGCCandidateReason = "failed_write"
	ObjectGCReasonForkDelete  ObjectGCCandidateReason = "fork_delete"
)

type TenantDeleteJobState string

const (
	TenantDeleteJobPending TenantDeleteJobState = "pending"
	TenantDeleteJobRunning TenantDeleteJobState = "running"
	TenantDeleteJobDeleted TenantDeleteJobState = "deleted"
)

type TenantDeleteJob struct {
	TenantID                string
	NamespaceID             string
	Backend                 string
	Bucket                  string
	Prefix                  string
	State                   TenantDeleteJobState
	Attempts                int
	LastError               string
	NotBefore               time.Time
	DeletedObjects          int64
	AbortedMultipartUploads int64
	CreatedAt               time.Time
	UpdatedAt               time.Time
	CompletedAt             *time.Time
}

func (t Tenant) S3EncryptionPolicy() S3EncryptionPolicy {
	return S3EncryptionPolicy{
		Mode:             t.S3EncryptionMode,
		KMSKeyID:         t.S3KMSKeyID,
		BucketKeyEnabled: t.S3BucketKeyEnabledValue(),
	}
}

func (t Tenant) S3BucketKeyEnabledValue() bool {
	if t.S3BucketKeyEnabled == nil {
		return true
	}
	return *t.S3BucketKeyEnabled
}

type APIKey struct {
	ID                   string
	TenantID             string
	KeyName              string
	JWTCiphertext        []byte
	JWTHash              string
	TokenVersion         int
	Status               APIKeyStatus
	ScopeKind            APIKeyScopeKind
	IssuedByProvider     string
	IssuedBySubjectKey   string
	IssuedByMetadataJSON []byte
	IssuedAt             time.Time
	RevokedAt            *time.Time
	ExpiresAt            *time.Time
	CreatedAt            time.Time
	UpdatedAt            time.Time
}

type APIKeyFSScope struct {
	TenantID   string
	APIKeyID   string
	Prefix     string
	PrefixHash string
	Ops        string
	CreatedAt  time.Time
	UpdatedAt  time.Time
}

type TenantWithAPIKey struct {
	Tenant         Tenant
	APIKey         APIKey
	TiDBCloudOrgID string
}

type ExternalBinding struct {
	Provider     string
	SubjectKey   string
	TenantID     string
	MetadataJSON []byte
	CreatedAt    time.Time
	UpdatedAt    time.Time
}

type TenantTiDBCloudOrgBinding struct {
	TenantID       string
	OrganizationID string
	ClusterID      string
	BranchID       string
	PoolID         string
	PoolStatus     TenantPoolBindingStatus
	UsedAt         *time.Time
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

type TenantWithTiDBCloudOrgBinding struct {
	Tenant  Tenant
	Binding TenantTiDBCloudOrgBinding
}

type metaQueryExecer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

type TenantPoolBindingStatus string

const (
	TenantPoolBindingUsed TenantPoolBindingStatus = "used"
	TenantPoolBindingFree TenantPoolBindingStatus = "free"
)

type TenantPoolStatus string

const (
	TenantPoolActive   TenantPoolStatus = "active"
	TenantPoolDeleting TenantPoolStatus = "deleting"
)

type TenantPool struct {
	PoolID         string
	OrganizationID string
	Size           int
	Status         TenantPoolStatus
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

type TenantPoolBindingStatusCount struct {
	PoolID         string
	OrganizationID string
	Status         TenantPoolBindingStatus
	Count          int64
}

type Store struct {
	db *sql.DB
	// apiKeys caches ResolveByAPIKeyHash results (see api_key_cache.go).
	apiKeys *apiKeyResolveCache
}

func Open(dsn string) (*Store, error) {
	return OpenContext(context.Background(), dsn)
}

func OpenContext(ctx context.Context, dsn string) (*Store, error) {
	if strings.Contains(dsn, "multiStatements=true") {
		return nil, fmt.Errorf("multiStatements=true is not allowed in production DSN")
	}
	db, err := mysqlutil.OpenInstrumented(ctx, dsn, mysqlutil.RoleMeta)
	if err != nil {
		return nil, fmt.Errorf("open db: %w", err)
	}
	s := &Store{db: db, apiKeys: newAPIKeyResolveCache()}
	if err := s.migrate(); err != nil {
		_ = mysqlutil.CloseInstrumented(db)
		return nil, fmt.Errorf("migrate: %w", err)
	}
	return s, nil
}

func (s *Store) Close() error { return mysqlutil.CloseInstrumented(s.db) }
func (s *Store) DB() *sql.DB  { return s.db }

const metaSchemaMigrateLockNamePrefix = "dat9_meta_schema_migrate:"
const metaSchemaMigrateLockTimeoutSeconds = 30
const externalBindingLockTimeoutSeconds = 90
const externalBindingReleaseLockTimeout = 5 * time.Second
const tidbCloudOrgBindingLockTimeoutSeconds = 90
const tidbCloudOrgBindingReleaseLockTimeout = 5 * time.Second
const tenantPoolLockTimeoutSeconds = 300
const tenantPoolReleaseLockTimeout = 5 * time.Second

var tidbCloudFreeQuotaLockTimeoutSeconds = 5

const tidbCloudFreeQuotaReleaseLockTimeout = 5 * time.Second

func (s *Store) migrate() (err error) {
	ctx := context.Background()
	releaseLock, err := acquireMetaSchemaMigrationLock(ctx, s.db)
	if err != nil {
		return fmt.Errorf("acquire meta schema migration lock: %w", err)
	}
	defer func() {
		if releaseErr := releaseLock(); releaseErr != nil {
			err = errors.Join(err, fmt.Errorf("release meta schema migration lock: %w", releaseErr))
		}
	}()

	stmts := metaInitSchemaStatements()
	spec, err := metaSchemaSpecFromStatements(stmts)
	if err != nil {
		return fmt.Errorf("parse meta schema statements: %w", err)
	}
	if err := migrateLegacyDBPoolClusterIDIndex(ctx, s.db); err != nil {
		return err
	}
	if err := dropObsoleteMetaIndexes(ctx, s.db); err != nil {
		return err
	}
	if err := ensureDBPoolUUID(ctx, s.db); err != nil {
		return err
	}
	diffs, err := diffMetaSchema(ctx, s.db, spec)
	if err != nil {
		return fmt.Errorf("diff meta schema: %w", err)
	}
	if err := applyMetaSchemaRepairs(ctx, s.db, plannedMetaSchemaRepairs(diffs)); err != nil {
		return err
	}
	if err := ensureTenantQuotaConfigLLMDefaults(ctx, s.db); err != nil {
		return err
	}
	if err := dropObsoleteTenantQuotaConfigColumns(ctx, s.db); err != nil {
		return err
	}
	if err := expandManagedDBPoolSchema(ctx, s.db); err != nil {
		return err
	}
	if err := backfillTiDBCloudOrgBindingBranchIDs(ctx, s.db); err != nil {
		return err
	}
	if err := deleteStaleTiDBCloudOrgBindings(ctx, s.db); err != nil {
		return err
	}
	if err := ensureTiDBCloudOrgBindingUniqueIndex(ctx, s.db); err != nil {
		return err
	}
	if err := backfillOrgObjectBackendIdentityHashes(ctx, s.db); err != nil {
		return err
	}
	if err := ensureOrgObjectBackendIdentityIndex(ctx, s.db); err != nil {
		return err
	}
	diffs, err = diffMetaSchema(ctx, s.db, spec)
	if err != nil {
		return fmt.Errorf("re-diff meta schema: %w", err)
	}
	if len(diffs) > 0 {
		return &metaSchemaDiffError{diffs: diffs}
	}
	return nil
}

func ensureTenantQuotaConfigLLMDefaults(ctx context.Context, db *sql.DB) error {
	defaults := []struct {
		column string
		value  int64
	}{
		{column: "max_media_llm_files", value: defaultMaxMediaLLMFilesValue},
		{column: "max_video_llm_files", value: defaultMaxVideoLLMFilesValue},
	}
	for _, quotaDefault := range defaults {
		var observed sql.NullInt64
		if err := db.QueryRowContext(ctx, `SELECT column_default
			FROM information_schema.columns
			WHERE table_schema = DATABASE() AND table_name = 'tenant_quota_config' AND column_name = ?`,
			quotaDefault.column).Scan(&observed); err != nil {
			return fmt.Errorf("inspect tenant_quota_config.%s default: %w", quotaDefault.column, err)
		}
		if observed.Valid && observed.Int64 == quotaDefault.value {
			continue
		}
		query := fmt.Sprintf("ALTER TABLE tenant_quota_config MODIFY COLUMN %s BIGINT NOT NULL DEFAULT %d",
			quotaDefault.column, quotaDefault.value)
		if _, err := db.ExecContext(ctx, query); err != nil {
			return fmt.Errorf("update tenant_quota_config.%s default: %w", quotaDefault.column, err)
		}
	}
	return nil
}

func dropObsoleteTenantQuotaConfigColumns(ctx context.Context, db *sql.DB) error {
	for _, column := range []string{"quota_limits_overridden"} {
		var exists int
		if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
			FROM information_schema.columns
			WHERE table_schema = DATABASE() AND table_name = 'tenant_quota_config' AND column_name = ?`, column).Scan(&exists); err != nil {
			return fmt.Errorf("inspect tenant_quota_config.%s: %w", column, err)
		}
		if exists == 0 {
			continue
		}
		if _, err := db.ExecContext(ctx, fmt.Sprintf("ALTER TABLE tenant_quota_config DROP COLUMN %s", column)); err != nil {
			return fmt.Errorf("drop tenant_quota_config.%s: %w", column, err)
		}
	}
	return nil
}

func expandManagedDBPoolSchema(ctx context.Context, db *sql.DB) error {
	nullableColumns := []struct {
		name       string
		definition string
	}{
		{name: "org_id", definition: "VARCHAR(64) NULL"},
		{name: "db_host", definition: "VARCHAR(255) NULL"},
		{name: "db_port", definition: "INT NULL"},
		{name: "db_user", definition: "VARCHAR(255) NULL"},
		{name: "db_password", definition: "VARBINARY(2048) NULL"},
		{name: "db_name", definition: "VARCHAR(255) NULL"},
	}
	for _, column := range nullableColumns {
		var nullable string
		if err := db.QueryRowContext(ctx, `SELECT is_nullable
			FROM information_schema.columns
			WHERE table_schema = DATABASE() AND table_name = 'db_pool' AND column_name = ?`, column.name).Scan(&nullable); err != nil {
			return fmt.Errorf("inspect db_pool.%s nullability: %w", column.name, err)
		}
		if nullable == "YES" {
			continue
		}
		if _, err := db.ExecContext(ctx, fmt.Sprintf("ALTER TABLE db_pool MODIFY COLUMN %s %s", column.name, column.definition)); err != nil {
			return fmt.Errorf("make db_pool.%s nullable: %w", column.name, err)
		}
	}

	var softCapColumnExists int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM information_schema.columns
		WHERE table_schema = DATABASE() AND table_name = 'db_pool' AND column_name = 'soft_cap_reached'`).Scan(&softCapColumnExists); err != nil {
		return fmt.Errorf("inspect db_pool.soft_cap_reached: %w", err)
	}
	if softCapColumnExists == 0 {
		if _, err := db.ExecContext(ctx, `ALTER TABLE db_pool ADD COLUMN soft_cap_reached TINYINT(1) NOT NULL DEFAULT 0 AFTER tenant_count`); err != nil {
			return fmt.Errorf("add db_pool.soft_cap_reached: %w", err)
		}
	}
	var statusUpdatedAtColumnExists int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM information_schema.columns
		WHERE table_schema = DATABASE() AND table_name = 'db_pool' AND column_name = 'status_updated_at'`).Scan(&statusUpdatedAtColumnExists); err != nil {
		return fmt.Errorf("inspect db_pool.status_updated_at: %w", err)
	}
	if statusUpdatedAtColumnExists == 0 {
		if _, err := db.ExecContext(ctx, `ALTER TABLE db_pool ADD COLUMN status_updated_at DATETIME(3) NULL AFTER status`); err != nil {
			return fmt.Errorf("add db_pool.status_updated_at: %w", err)
		}
		if _, err := db.ExecContext(ctx, `UPDATE db_pool SET status_updated_at = updated_at WHERE status_updated_at IS NULL`); err != nil {
			return fmt.Errorf("backfill db_pool.status_updated_at: %w", err)
		}
		if _, err := db.ExecContext(ctx, `ALTER TABLE db_pool MODIFY COLUMN status_updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3)`); err != nil {
			return fmt.Errorf("make db_pool.status_updated_at non-null: %w", err)
		}
	}
	// Only ever open the latch during migration. Do not clear a latched pool
	// below the soft cap here; deletion owns the hysteresis transition.
	if _, err := db.ExecContext(ctx, `UPDATE db_pool
		SET soft_cap_reached = 1
		WHERE max_tenants > 0 AND tenant_count >= max_tenants AND soft_cap_reached = 0`); err != nil {
		return fmt.Errorf("backfill db_pool.soft_cap_reached: %w", err)
	}
	return ensureDBPoolClusterIDUniqueIndex(ctx, db)
}

func ensureDBPoolClusterIDUniqueIndex(ctx context.Context, db *sql.DB) error {
	exists, err := metaIndexExists(ctx, db, "db_pool", "uk_db_pool_cluster_id")
	if err != nil {
		return fmt.Errorf("check db_pool cluster id unique index: %w", err)
	}
	if !exists {
		var duplicateCount int
		if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM (
			SELECT cluster_id
			FROM db_pool
			WHERE cluster_id IS NOT NULL
			GROUP BY cluster_id
			HAVING COUNT(*) > 1
		) duplicates`).Scan(&duplicateCount); err != nil {
			return fmt.Errorf("preflight db_pool cluster id unique index: %w", err)
		}
		if duplicateCount > 0 {
			return fmt.Errorf("preflight db_pool cluster id unique index: found %d duplicate cluster ids", duplicateCount)
		}
		if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX uk_db_pool_cluster_id ON db_pool(cluster_id)`); err != nil && !isIgnorableMetaSchemaError(err) {
			return fmt.Errorf("create db_pool cluster id unique index: %w", err)
		}
	}
	return nil
}

func migrateLegacyDBPoolClusterIDIndex(ctx context.Context, db *sql.DB) error {
	exists, err := metaIndexExists(ctx, db, "db_pool", "uk_db_pool_cloud_resource")
	if err != nil {
		return fmt.Errorf("check legacy db_pool cluster id unique index: %w", err)
	}
	if !exists {
		return nil
	}
	if err := ensureDBPoolClusterIDUniqueIndex(ctx, db); err != nil {
		return err
	}
	if err := dropMetaIndexIfExists(ctx, db, "db_pool", "uk_db_pool_cloud_resource"); err != nil {
		return fmt.Errorf("drop legacy db_pool cluster id unique index: %w", err)
	}
	return nil
}

func ensureDBPoolUUID(ctx context.Context, db *sql.DB) error {
	var tableExists int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM information_schema.tables
		WHERE table_schema = DATABASE() AND table_name = 'db_pool'`).Scan(&tableExists); err != nil {
		return fmt.Errorf("inspect db_pool table for uuid migration: %w", err)
	}
	if tableExists == 0 {
		return nil
	}

	var columnExists int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM information_schema.columns
		WHERE table_schema = DATABASE() AND table_name = 'db_pool' AND column_name = 'uuid'`).Scan(&columnExists); err != nil {
		return fmt.Errorf("inspect db_pool.uuid: %w", err)
	}
	if columnExists == 0 {
		if _, err := db.ExecContext(ctx, `ALTER TABLE db_pool ADD COLUMN uuid CHAR(36) NULL AFTER db_id`); err != nil {
			return fmt.Errorf("add db_pool.uuid: %w", err)
		}
	}

	type dbPoolUUIDRow struct {
		id   int64
		uuid sql.NullString
	}
	rows, err := db.QueryContext(ctx, `SELECT db_id, uuid FROM db_pool ORDER BY db_id`)
	if err != nil {
		return fmt.Errorf("list db_pool rows for uuid migration: %w", err)
	}
	var existing []dbPoolUUIDRow
	for rows.Next() {
		var row dbPoolUUIDRow
		if err := rows.Scan(&row.id, &row.uuid); err != nil {
			_ = rows.Close()
			return fmt.Errorf("scan db_pool uuid migration row: %w", err)
		}
		existing = append(existing, row)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return fmt.Errorf("iterate db_pool uuid migration rows: %w", err)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf("close db_pool uuid migration rows: %w", err)
	}
	for _, row := range existing {
		raw := strings.TrimSpace(row.uuid.String)
		if raw != "" {
			value, valueErr := sharedDBUUID(raw)
			if valueErr != nil {
				return fmt.Errorf("db_pool %d uuid migration: %w", row.id, valueErr)
			}
			if value != raw {
				if _, err := db.ExecContext(ctx, `UPDATE db_pool SET uuid = ? WHERE db_id = ?`, value, row.id); err != nil {
					return fmt.Errorf("canonicalize db_pool %d uuid: %w", row.id, err)
				}
			}
			continue
		}
		value, valueErr := sharedDBUUID("")
		if valueErr != nil {
			return fmt.Errorf("db_pool %d uuid migration: %w", row.id, valueErr)
		}
		if _, err := db.ExecContext(ctx, `UPDATE db_pool SET uuid = ?
			WHERE db_id = ? AND (uuid IS NULL OR uuid = '')`, value, row.id); err != nil {
			return fmt.Errorf("backfill db_pool %d uuid: %w", row.id, err)
		}
	}
	var isNullable string
	if err := db.QueryRowContext(ctx, `SELECT is_nullable
		FROM information_schema.columns
		WHERE table_schema = DATABASE() AND table_name = 'db_pool' AND column_name = 'uuid'`).Scan(&isNullable); err != nil {
		return fmt.Errorf("inspect db_pool.uuid nullability: %w", err)
	}
	if isNullable != "NO" {
		if _, err := db.ExecContext(ctx, `ALTER TABLE db_pool MODIFY COLUMN uuid CHAR(36) NOT NULL`); err != nil {
			return fmt.Errorf("make db_pool.uuid non-null: %w", err)
		}
	}
	exists, err := metaIndexExists(ctx, db, "db_pool", "uk_db_pool_uuid")
	if err != nil {
		return fmt.Errorf("check db_pool uuid unique index: %w", err)
	}
	if !exists {
		if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX uk_db_pool_uuid ON db_pool(uuid)`); err != nil && !isIgnorableMetaSchemaError(err) {
			return fmt.Errorf("create db_pool uuid unique index: %w", err)
		}
	}
	return nil
}

func acquireMetaSchemaMigrationLock(ctx context.Context, db *sql.DB) (func() error, error) {
	conn, err := db.Conn(ctx)
	if err != nil {
		return nil, err
	}

	var got sql.NullInt64
	if err := conn.QueryRowContext(ctx,
		"SELECT GET_LOCK(CONCAT(?, DATABASE()), ?)",
		metaSchemaMigrateLockNamePrefix,
		metaSchemaMigrateLockTimeoutSeconds,
	).Scan(&got); err != nil {
		_ = conn.Close()
		return nil, err
	}
	if !got.Valid {
		_ = conn.Close()
		return nil, fmt.Errorf("named lock returned NULL")
	}
	if got.Int64 != 1 {
		_ = conn.Close()
		return nil, fmt.Errorf("timed out waiting for named lock")
	}

	return func() error {
		defer func() { _ = conn.Close() }()

		var released sql.NullInt64
		if err := conn.QueryRowContext(ctx,
			"SELECT RELEASE_LOCK(CONCAT(?, DATABASE()))",
			metaSchemaMigrateLockNamePrefix,
		).Scan(&released); err != nil {
			return err
		}
		if !released.Valid {
			return fmt.Errorf("named lock release returned NULL")
		}
		if released.Int64 != 1 {
			return fmt.Errorf("named lock was not held by current connection")
		}
		return nil
	}, nil
}

type metaColumnMeta struct {
	columnType string
}

type metaTableMeta struct {
	tableName string
	columns   map[string]metaColumnMeta
}

type metaSchemaDiffKind string

const (
	metaSchemaDiffMissingTable  metaSchemaDiffKind = "missing_table"
	metaSchemaDiffMissingColumn metaSchemaDiffKind = "missing_column"
	metaSchemaDiffMissingIndex  metaSchemaDiffKind = "missing_index"
	metaSchemaDiffColumnType    metaSchemaDiffKind = "column_type_mismatch"
)

type metaSchemaDiff struct {
	kind       metaSchemaDiffKind
	tableName  string
	columnName string
	indexName  string
	detail     string
	repairSQL  string
}

type metaSchemaDiffError struct {
	diffs []metaSchemaDiff
}

func (e *metaSchemaDiffError) Error() string {
	if e == nil || len(e.diffs) == 0 {
		return ""
	}
	parts := make([]string, 0, len(e.diffs))
	for _, d := range e.diffs {
		parts = append(parts, d.detail)
	}
	return "meta schema contract mismatch: " + strings.Join(parts, "; ")
}

type metaSchemaSpec struct {
	tables []metaTableSpec
}

type metaTableSpec struct {
	name            string
	createStatement string
	columns         map[string]metaColumnSpec
	indexes         map[string]metaIndexSpec
}

type metaColumnSpec struct {
	columnType string
	addSQL     string
	modifySQL  string
}

type metaIndexSpec struct {
	createSQL string
	isPrimary bool
}

func metaInitSchemaStatements() []string {
	stmts := []string{
		`CREATE TABLE IF NOT EXISTS tenants (
			id               VARCHAR(64) PRIMARY KEY,
			display_name     VARCHAR(255) NOT NULL DEFAULT '',
			label            JSON NULL,
			status           VARCHAR(20) NOT NULL DEFAULT 'provisioning',
			kind             VARCHAR(16) NOT NULL DEFAULT 'live',
			parent_tenant_id VARCHAR(64) NOT NULL DEFAULT '',
			storage_namespace_id VARCHAR(64) NOT NULL DEFAULT '',
			db_host          VARCHAR(255) NOT NULL,
			db_port          INT NOT NULL,
			db_user          VARCHAR(255) NOT NULL,
			db_password      VARBINARY(2048) NOT NULL,
			db_name          VARCHAR(255) NOT NULL,
			db_tls           TINYINT(1) NOT NULL DEFAULT 1,
			provider         VARCHAR(50) NOT NULL,
			cluster_id       VARCHAR(255) NULL,
			branch_id        VARCHAR(255) NOT NULL DEFAULT '',
			claim_url        TEXT NULL,
			claim_expires_at DATETIME(3) NULL,
			schema_version   INT UNSIGNED NOT NULL DEFAULT 1,
			s3_encryption_mode VARCHAR(16) NOT NULL DEFAULT 'inherit',
			s3_kms_key_id VARCHAR(256) NOT NULL DEFAULT '',
			s3_bucket_key_enabled TINYINT(1) NOT NULL DEFAULT 1,
			created_at       DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at       DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			deleted_at       DATETIME(3) NULL,
			INDEX idx_tenant_status (status),
			INDEX idx_tenant_status_created_id (status, created_at, id),
			INDEX idx_tenant_provider (provider),
			INDEX idx_tenant_namespace (storage_namespace_id, kind, status),
			INDEX idx_tenant_parent (parent_tenant_id)
		)`,
		`CREATE TABLE IF NOT EXISTS fs_registry (
			fs_id      BIGINT AUTO_INCREMENT PRIMARY KEY,
			tenant_id  VARCHAR(64) NOT NULL UNIQUE,
			created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3)
		)`,
		// db_pool is the fleet-wide registry of physical databases available
		// for tenant placement under the shared-schema layout. org_id scopes a
		// database to one exact TiDB Cloud organization. db_password holds the same
		// encrypted envelope as tenants.db_password. tenant_count is maintained
		// atomically with placement writes via CompleteSharedTenantProvision and
		// DeleteTenantPlacementAndDecrCount.
		// The `role` column name is backtick-quoted because ROLE is a reserved
		// word in MySQL 8.0.
		`CREATE TABLE IF NOT EXISTS db_pool (
			db_id        BIGINT AUTO_INCREMENT PRIMARY KEY,
			uuid         CHAR(36) NOT NULL,
			org_id       VARCHAR(64) NULL,
			cluster_id   VARCHAR(255) NULL,
			provisioning_key VARBINARY(32) NULL,
			cloud_provider VARCHAR(32) NULL,
			region       VARCHAR(64) NULL,
			` + "`role`" + `       VARCHAR(20) NOT NULL,
			db_host      VARCHAR(255) NULL,
			db_port      INT NULL,
			db_user      VARCHAR(255) NULL,
			db_password  VARBINARY(2048) NULL,
			db_name      VARCHAR(255) NULL,
			db_tls       VARCHAR(32) NOT NULL DEFAULT '',
			max_tenants  INT NOT NULL DEFAULT 0,
			tenant_count INT NOT NULL DEFAULT 0,
			soft_cap_reached TINYINT(1) NOT NULL DEFAULT 0,
			spending_limit BIGINT NULL,
			schema_version INT UNSIGNED NOT NULL DEFAULT 0,
			status       VARCHAR(20) NOT NULL DEFAULT 'active',
			status_updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			created_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			UNIQUE INDEX uk_db_pool_uuid (uuid),
			UNIQUE INDEX uk_db_pool_cluster_id (cluster_id),
			INDEX idx_db_pool_allocate (org_id, status, db_id),
			INDEX idx_db_pool_provisioning_key (provisioning_key, status, db_id),
			INDEX idx_db_pool_role_status_id (` + "`role`" + `, status, db_id)
		)`,
		// tenant_placements records which physical database (db_pool row) hosts
		// each filesystem (fs_registry row). A missing row means the tenant
		// still lives on its legacy standalone database. epoch is reserved for
		// optimistic concurrency during future migrations; it stays 1 unless a
		// migration explicitly bumps it.
		`CREATE TABLE IF NOT EXISTS tenant_placements (
			fs_id        BIGINT PRIMARY KEY,
			db_id        BIGINT NOT NULL,
			placement    VARCHAR(20) NOT NULL,
			schema_shape VARCHAR(20) NOT NULL,
			status       VARCHAR(20) NOT NULL DEFAULT 'active',
			target_db_id BIGINT NULL,
			epoch        BIGINT NOT NULL DEFAULT 1,
			created_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			INDEX idx_placement_db (db_id, status)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_auto_embedding_profiles (
			tenant_id      VARCHAR(64) PRIMARY KEY,
			embedding_mode VARCHAR(32) NULL,
			model          VARCHAR(255) NOT NULL DEFAULT 'tidbcloud_free/amazon/titan-embed-text-v2',
			dimensions     INT UNSIGNED NOT NULL DEFAULT 1024,
			options_json   VARCHAR(2048) NOT NULL DEFAULT '{"dimensions":1024}',
			api_base       TEXT NULL,
			api_key_cipher VARBINARY(2048) NULL,
			created_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_extract_configs (
			tenant_id      VARCHAR(64)     NOT NULL,
			media_type     VARCHAR(16)     NOT NULL,
			enabled        TINYINT(1)      NOT NULL DEFAULT 0,
			api_base       VARCHAR(512)    NULL,
			api_key_cipher VARBINARY(2048) NULL,
			model          VARCHAR(255)    NULL,
			prompt         MEDIUMTEXT      NULL,
			protocol       VARCHAR(32)     NULL,
			created_at     DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at     DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id, media_type)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_embedding_configs (
			tenant_id       VARCHAR(64)     NOT NULL,
			enabled         TINYINT(1)      NOT NULL DEFAULT 0,
			api_base        VARCHAR(512)    NULL,
			api_key_cipher  VARBINARY(2048) NULL,
			model           VARCHAR(255)    NULL,
			generation      BIGINT UNSIGNED NOT NULL DEFAULT 1,
			config_revision BIGINT UNSIGNED NOT NULL DEFAULT 1,
			created_at      DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at      DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id)
		)`,
		`CREATE TABLE IF NOT EXISTS org_object_backends (
			id                   VARCHAR(64) NOT NULL,
			organization_id      VARCHAR(64) NOT NULL,
			name                 VARCHAR(128) NOT NULL DEFAULT '',
			scheme               VARCHAR(16) NOT NULL,
			endpoint             VARCHAR(512) NOT NULL DEFAULT '',
			sts_endpoint         VARCHAR(512) NOT NULL DEFAULT '',
			region               VARCHAR(64) NOT NULL DEFAULT '',
			account_id           VARCHAR(128) NOT NULL DEFAULT '',
			force_path_style     TINYINT(1) NOT NULL DEFAULT 0,
			bucket               VARCHAR(255) NOT NULL,
			prefix               VARCHAR(2048) NOT NULL DEFAULT '',
			credential_kind      VARCHAR(16) NOT NULL,
			role_arn             VARCHAR(2048) NOT NULL DEFAULT '',
			access_key_id        VARCHAR(128) NOT NULL DEFAULT '',
			secret_cipher        VARBINARY(8192) NULL,
			external_id_cipher   VARBINARY(2048) NULL,
			max_session_ttl_sec  INT UNSIGNED NOT NULL DEFAULT 3600,
			identity_hash        CHAR(64) NOT NULL DEFAULT '',
			created_at           DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at           DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (id),
			UNIQUE INDEX uk_org_object_backend_identity (identity_hash),
			INDEX idx_org_object_backends_org (organization_id)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_object_namespaces (
			tenant_id    VARCHAR(64) NOT NULL,
			namespace_id VARCHAR(255) NOT NULL DEFAULT '',
			created_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id)
		)`,
		`CREATE TABLE IF NOT EXISTS storage_namespaces (
			namespace_id               VARCHAR(64) PRIMARY KEY,
			owner_tenant_id            VARCHAR(64) NOT NULL,
			backend                    VARCHAR(16) NOT NULL,
			bucket                     VARCHAR(255) NOT NULL DEFAULT '',
			prefix                     VARCHAR(2048) NOT NULL,
			append_log_binding_state   VARCHAR(16) NOT NULL DEFAULT 'unbound',
			append_log_bucket          VARCHAR(255) NOT NULL DEFAULT '',
			append_log_prefix          VARCHAR(2048) NOT NULL DEFAULT '',
			append_log_region          VARCHAR(64) NOT NULL DEFAULT '',
			state                      VARCHAR(20) NOT NULL DEFAULT 'active',
			created_at                 DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at                 DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			INDEX idx_storage_namespace_owner (owner_tenant_id),
			INDEX idx_storage_namespace_state (state)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_delete_jobs (
			tenant_id                  VARCHAR(64) PRIMARY KEY,
			namespace_id               VARCHAR(64) NOT NULL,
			backend                    VARCHAR(16) NOT NULL,
			bucket                     VARCHAR(255) NOT NULL DEFAULT '',
			prefix                     VARCHAR(2048) NOT NULL,
			state                      VARCHAR(20) NOT NULL DEFAULT 'pending',
			attempts                   INT NOT NULL DEFAULT 0,
			last_error                 TEXT NULL,
			not_before                 DATETIME(3) NOT NULL,
			deleted_objects            BIGINT NOT NULL DEFAULT 0,
			aborted_multipart_uploads BIGINT NOT NULL DEFAULT 0,
			created_at                 DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at                 DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			completed_at               DATETIME(3) NULL,
			INDEX idx_tenant_delete_jobs_due (state, not_before),
			INDEX idx_tenant_delete_jobs_namespace (namespace_id, state)
		)`,
		`CREATE TABLE IF NOT EXISTS object_gc_candidates (
			namespace_id     VARCHAR(64) NOT NULL,
			storage_ref      TEXT NOT NULL,
			storage_ref_hash VARCHAR(64) NOT NULL,
			content_layout   VARCHAR(32) NOT NULL DEFAULT 'single',
			reason           VARCHAR(32) NOT NULL,
			source_tenant_id VARCHAR(64) NOT NULL DEFAULT '',
			source_file_id   VARCHAR(64) NOT NULL DEFAULT '',
			not_before       DATETIME(3) NOT NULL,
			state            VARCHAR(20) NOT NULL DEFAULT 'pending',
			attempts         INT NOT NULL DEFAULT 0,
			last_error       TEXT NULL,
			created_at       DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at       DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			deleted_at       DATETIME(3) NULL,
			PRIMARY KEY (namespace_id, storage_ref_hash),
			INDEX idx_object_gc_due (state, not_before),
			INDEX idx_object_gc_namespace_due (namespace_id, state, not_before)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_api_keys (
			id             VARCHAR(64) PRIMARY KEY,
			tenant_id      VARCHAR(64) NOT NULL,
			key_name       VARCHAR(64) NOT NULL DEFAULT 'default',
			jwt_ciphertext VARBINARY(4096) NOT NULL,
			jwt_hash       VARCHAR(128) NOT NULL,
			token_version  INT NOT NULL DEFAULT 1,
			status         VARCHAR(20) NOT NULL DEFAULT 'active',
			scope_kind     VARCHAR(32) NOT NULL DEFAULT 'owner',
			issued_by_provider VARCHAR(64) NOT NULL DEFAULT '',
			issued_by_subject_key VARCHAR(512) NOT NULL DEFAULT '',
			issued_by_metadata_json JSON NULL,
			issued_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			revoked_at     DATETIME(3) NULL,
			expires_at     DATETIME(3) NULL,
			created_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			UNIQUE INDEX idx_api_keys_hash (jwt_hash),
			INDEX idx_api_keys_tenant (tenant_id, status),
			INDEX idx_api_keys_issuer (tenant_id, issued_by_provider, issued_by_subject_key, status, created_at)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_external_bindings (
			provider      VARCHAR(64) NOT NULL,
			subject_key   VARCHAR(512) NOT NULL,
			tenant_id     VARCHAR(64) NOT NULL,
			metadata_json JSON NULL,
			created_at    DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at    DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			UNIQUE INDEX uk_external_binding_subject (provider, subject_key),
			INDEX idx_external_binding_tenant (tenant_id)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_tidbcloud_org_bindings (
				tenant_id       VARCHAR(64) PRIMARY KEY,
				organization_id VARCHAR(64) NOT NULL,
				cluster_id      VARCHAR(255) NOT NULL,
				branch_id       VARCHAR(255) NOT NULL DEFAULT '',
				pool_id         VARCHAR(64) NOT NULL DEFAULT '',
				pool_status     VARCHAR(20) NOT NULL DEFAULT 'used',
				used_at         DATETIME(3) NULL,
				created_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
				updated_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
				UNIQUE INDEX uk_tidbcloud_org_cluster_branch (organization_id, cluster_id, branch_id),
				INDEX idx_tidbcloud_org_cluster (organization_id, cluster_id, branch_id, created_at, tenant_id),
				INDEX idx_tidbcloud_org_created (organization_id, created_at, tenant_id),
				INDEX idx_tidbcloud_pool_free (organization_id, pool_status, created_at, tenant_id)
			)`,
		`CREATE TABLE IF NOT EXISTS tenant_tidbcloud_pools (
			pool_id         VARCHAR(64) PRIMARY KEY,
			organization_id VARCHAR(64) NULL,
			size            INT NOT NULL,
			status          VARCHAR(20) NOT NULL DEFAULT 'active',
			created_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			UNIQUE INDEX uk_tidbcloud_pool_org (organization_id),
			INDEX idx_tidbcloud_pool_status (status, created_at),
			INDEX idx_tidbcloud_pool_status_id (status, pool_id)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_pool_memberships (
			tenant_id                    VARCHAR(64) PRIMARY KEY,
			tidbcloud_organization_id    VARCHAR(64) NULL,
			pool_id                      VARCHAR(64) NOT NULL,
			pool_status                  VARCHAR(20) NOT NULL DEFAULT 'free',
			used_at                      DATETIME(3) NULL,
			created_at                   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at                   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			INDEX idx_tenant_pool_claim (pool_id, pool_status, created_at, tenant_id),
			INDEX idx_tenant_pool_org_status (tidbcloud_organization_id, pool_status, pool_id, created_at, tenant_id)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_api_key_fs_scopes (
			tenant_id   VARCHAR(64) NOT NULL,
			api_key_id  VARCHAR(64) NOT NULL,
			prefix      TEXT NOT NULL,
			prefix_hash VARCHAR(64) NOT NULL,
			ops         VARCHAR(255) NOT NULL,
			created_at  DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at  DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id, api_key_id, prefix_hash),
			INDEX idx_fs_scopes_api_key (api_key_id),
			INDEX idx_fs_scopes_tenant_key (tenant_id, api_key_id)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_quota_config (
			tenant_id             VARCHAR(64) PRIMARY KEY,
			max_storage_bytes     BIGINT NOT NULL DEFAULT 53687091200,
			max_file_size_bytes   BIGINT NOT NULL DEFAULT 0,
			max_file_count        BIGINT NOT NULL DEFAULT 0,
			max_media_llm_files   BIGINT NOT NULL DEFAULT 100,
			max_video_llm_files   BIGINT NOT NULL DEFAULT 30,
			max_monthly_cost_mc   BIGINT NOT NULL DEFAULT 0,
			tidbcloud_spending_limit BIGINT NULL,
			tidbcloud_spending_limit_checked_at DATETIME(3) NULL,
			created_at            DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at            DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_quota_usage (
			tenant_id          VARCHAR(64) PRIMARY KEY,
			storage_bytes      BIGINT NOT NULL DEFAULT 0,
			db_storage_bytes   BIGINT NULL,
			s3_storage_bytes   BIGINT NULL,
			reserved_bytes     BIGINT NOT NULL DEFAULT 0,
			file_count         BIGINT NOT NULL DEFAULT 0,
			media_file_count   BIGINT NOT NULL DEFAULT 0,
			video_file_count   BIGINT NOT NULL DEFAULT 0,
			updated_at         DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_file_meta (
			tenant_id    VARCHAR(64) NOT NULL,
			file_id      VARCHAR(64) NOT NULL,
			size_bytes   BIGINT NOT NULL DEFAULT 0,
			is_media     TINYINT(1) NOT NULL DEFAULT 0,
			storage_type VARCHAR(32) NULL,
			layer_owned  TINYINT(1) NOT NULL DEFAULT 0,
			created_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id, file_id),
			INDEX idx_file_meta_tenant_tier (tenant_id, storage_type)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_layer_storage_meta (
			tenant_id    VARCHAR(64) NOT NULL,
			layer_id     VARCHAR(64) NOT NULL,
			path_hash    VARCHAR(64) NOT NULL,
			entry_seq    BIGINT NOT NULL,
			size_bytes   BIGINT NOT NULL DEFAULT 0,
			storage_type VARCHAR(32) NULL,
			created_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id, layer_id, path_hash, entry_seq),
			INDEX idx_layer_storage_tenant_tier (tenant_id, storage_type)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_storage_tier_scan (
			tenant_id              VARCHAR(64) PRIMARY KEY,
			status                 VARCHAR(20) NOT NULL DEFAULT 'pending',
			scan_phase             VARCHAR(16) NOT NULL DEFAULT 'core',
			scan_cursor            VARCHAR(64) NOT NULL DEFAULT '',
			layer_path_hash        VARCHAR(64) NOT NULL DEFAULT '',
			layer_entry_seq        BIGINT NOT NULL DEFAULT 0,
			watermark              BIGINT NOT NULL DEFAULT 0,
			lease_owner            VARCHAR(128) NOT NULL DEFAULT '',
			lease_until            DATETIME(3) NULL,
			attempt_count          INT NOT NULL DEFAULT 0,
			last_error             TEXT NULL,
			full_scan_completed_at DATETIME(3) NULL,
			created_at             DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at             DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			INDEX idx_tier_scan_claim (status, lease_until, updated_at)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_upload_reservations (
			tenant_id      VARCHAR(64) NOT NULL,
			upload_id      VARCHAR(64) NOT NULL,
			reserved_bytes BIGINT NOT NULL,
			file_count_delta BIGINT NOT NULL DEFAULT 0,
			target_path    VARCHAR(4096) NOT NULL,
			status         VARCHAR(20) NOT NULL DEFAULT 'active',
			expires_at     DATETIME(3) NOT NULL,
			created_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id, upload_id),
			INDEX idx_active_reservations (tenant_id, status, expires_at)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_llm_usage (
			id              BIGINT AUTO_INCREMENT PRIMARY KEY,
			tenant_id       VARCHAR(64) NOT NULL,
			task_type       VARCHAR(32) NOT NULL,
			task_id         VARCHAR(64) NOT NULL,
			cost_millicents BIGINT NOT NULL DEFAULT 0,
			raw_units       BIGINT NOT NULL DEFAULT 0,
			raw_unit_type   VARCHAR(16) NOT NULL,
			created_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			INDEX idx_tenant_month (tenant_id, created_at)
		)`,
		`CREATE TABLE IF NOT EXISTS tenant_monthly_llm_cost (
			tenant_id    VARCHAR(64) NOT NULL,
			month_start  DATE NOT NULL,
			total_mc     BIGINT NOT NULL DEFAULT 0,
			PRIMARY KEY (tenant_id, month_start)
		)`,
		`CREATE TABLE IF NOT EXISTS quota_mutation_log (
			id             BIGINT AUTO_INCREMENT PRIMARY KEY,
			tenant_id      VARCHAR(64) NOT NULL,
			mutation_type  VARCHAR(32) NOT NULL,
			mutation_data  JSON NOT NULL,
			status         VARCHAR(20) NOT NULL DEFAULT 'pending',
			retry_count    INT NOT NULL DEFAULT 0,
			created_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			applied_at     DATETIME(3) NULL,
			INDEX idx_pending (status, created_at),
			INDEX idx_pending_tenant_age (status, tenant_id, created_at),
			INDEX idx_tenant_order (tenant_id, id)
		)`,
		// pod_registry tracks all drive9-server pods for cross-pod SSE push
		// notification routing. Each pod upserts its row on startup and heartbeats
		// every ~10s. The leader marks pods with stale heartbeats as inactive so
		// writers don't push to dead pods. This table lives in the central meta DB
		// (always provisioned).
		`CREATE TABLE IF NOT EXISTS pod_registry (
			pod_id         VARCHAR(64) PRIMARY KEY,
			addr           VARCHAR(255) NOT NULL,
			last_heartbeat DATETIME(3) NOT NULL,
			status         VARCHAR(20) NOT NULL DEFAULT 'active',
			created_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			INDEX idx_pod_heartbeat (last_heartbeat)
		)`,
		// pod_subscriptions maps each pod to the tenant IDs for which it has active
		// SSE subscribers. Writers consult this (via a locally cached reverse index)
		// to push notifications only to pods that care about a given tenant, avoiding
		// wasteful fan-out to pods with no subscribers for that tenant. Each pod
		// periodically upserts its current subscriber set and prunes stale entries.
		`CREATE TABLE IF NOT EXISTS pod_subscriptions (
			pod_id     VARCHAR(64) NOT NULL,
			tenant_id VARCHAR(64) NOT NULL,
			updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (pod_id, tenant_id)
		)`,
		// tenant_notify_outbox is the unified outbox that replaces per-tenant TiDB
		// polling for SSE, semantic, file_gc, and quota work. Each write path inserts
		// a lightweight row (tenant_id + work_mask) into this always-provisioned meta
		// DB table. Every pod polls it at 200ms and dispatches by work_mask: SSE bits
		// wake the local bus (broadcast to all pods with subscribers); semantic/file_gc/
		// quota bits kick the unified worker only on the shard owner. This eliminates
		// all periodic per-tenant TiDB scans, enabling serverless scale-to-zero.
		`CREATE TABLE IF NOT EXISTS tenant_notify_outbox (
			id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
			tenant_id   VARCHAR(64) NOT NULL,
			work_mask   TINYINT UNSIGNED NOT NULL DEFAULT 0,
			created_at  DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			INDEX idx_tenant_notify_created (created_at)
		)`,
		// tenant_outbox_cursor stores each pod's last-processed outbox id so a pod
		// can resume after restart without skipping work and without re-reading all
		// history. Each pod owns its own row (keyed by pod_id).
		`CREATE TABLE IF NOT EXISTS tenant_outbox_cursor (
			pod_id      VARCHAR(64) NOT NULL,
			last_id     BIGINT UNSIGNED NOT NULL DEFAULT 0,
			updated_at  DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (pod_id)
		)`,
		// delete_reconcile_pending durably records that a tenant's
		// delete-orphan reconciliation sweep is still owed (due_at). It is the
		// restart-safe retry path for WorkDeleteReconcile: the outbox row is
		// consumed once, but the pending row survives pod restarts and is
		// re-dispatched by the outbox poller's pending scan until the worker
		// completes a successful sweep and deletes it.
		`CREATE TABLE IF NOT EXISTS delete_reconcile_pending (
			tenant_id  VARCHAR(64) NOT NULL,
			due_at     DATETIME(3) NOT NULL,
			created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			PRIMARY KEY (tenant_id),
			INDEX idx_delete_reconcile_due (due_at)
		)`,
		// shared_maintenance_state holds one row per cluster-wide maintenance
		// task (e.g. the shared-pool fs_events sweep) so all pods can throttle
		// against each other via an atomic claim instead of each running the
		// task on its own interval.
		`CREATE TABLE IF NOT EXISTS shared_maintenance_state (
			name        VARCHAR(64) NOT NULL,
			last_run_at DATETIME(3) NOT NULL,
			PRIMARY KEY (name)
		)`,
		// billing_ops_delta stores per-pod flushes of the in-process file ops
		// counter for the billing worker (billing_ops.go).
		billingOpsDeltaSchemaStatement,
	}
	// Billing worker tables (billing_storage_snapshot, billing_usage_ledger,
	// billing_sqs_outbox, billing_collect_checkpoint) are defined in
	// billing.go and appended here so every process opening the meta store
	// converges on the same schema.
	return append(stmts, billingSchemaStatements()...)
}

func dropObsoleteMetaIndexes(ctx context.Context, db *sql.DB) error {
	if err := dropMetaIndexIfExists(ctx, db, "tenant_api_keys", "idx_api_keys_tenant_name"); err != nil {
		return fmt.Errorf("drop obsolete meta index idx_api_keys_tenant_name: %w", err)
	}
	if err := dropMetaIndexIfColumns(ctx, db, "tenant_tidbcloud_org_bindings", "idx_tidbcloud_org_cluster",
		[]string{"organization_id", "cluster_id", "created_at", "tenant_id"}); err != nil {
		return fmt.Errorf("drop obsolete meta index idx_tidbcloud_org_cluster: %w", err)
	}
	if err := dropMetaIndexIfExists(ctx, db, "db_pool", "uk_db_pool_endpoint"); err != nil {
		return fmt.Errorf("drop obsolete meta index uk_db_pool_endpoint: %w", err)
	}
	if err := dropMetaIndexIfExists(ctx, db, "org_object_backends", "uk_org_object_backend"); err != nil {
		return fmt.Errorf("drop obsolete meta index uk_org_object_backend: %w", err)
	}
	return nil
}

func dropMetaIndexIfExists(ctx context.Context, db *sql.DB, tableName, indexName string) error {
	if err := validateMetaIdentifier(tableName); err != nil {
		return fmt.Errorf("invalid table name %q: %w", tableName, err)
	}
	if err := validateMetaIdentifier(indexName); err != nil {
		return fmt.Errorf("invalid index name %q: %w", indexName, err)
	}
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM information_schema.statistics
		WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?`,
		tableName, indexName).Scan(&count); err != nil {
		return err
	}
	if count == 0 {
		return nil
	}
	_, err := db.ExecContext(ctx, fmt.Sprintf("DROP INDEX %s ON %s", indexName, tableName))
	return err
}

func dropMetaIndexIfColumns(ctx context.Context, db *sql.DB, tableName, indexName string, columns []string) error {
	if err := validateMetaIdentifier(tableName); err != nil {
		return fmt.Errorf("invalid table name %q: %w", tableName, err)
	}
	if err := validateMetaIdentifier(indexName); err != nil {
		return fmt.Errorf("invalid index name %q: %w", indexName, err)
	}
	got, err := loadMetaIndexColumns(ctx, db, tableName, indexName)
	if err != nil {
		return err
	}
	if !sameStringSlice(got, columns) {
		return nil
	}
	_, err = db.ExecContext(ctx, fmt.Sprintf("DROP INDEX %s ON %s", indexName, tableName))
	return err
}

func validateMetaIdentifier(s string) error {
	if s == "" {
		return fmt.Errorf("identifier is empty")
	}
	for _, r := range s {
		if r == '_' || ('a' <= r && r <= 'z') || ('A' <= r && r <= 'Z') || ('0' <= r && r <= '9') {
			continue
		}
		return fmt.Errorf("identifier contains unsupported character %q", r)
	}
	return nil
}

func loadMetaIndexColumns(ctx context.Context, db *sql.DB, tableName, indexName string) ([]string, error) {
	rows, err := db.QueryContext(ctx, `SELECT column_name
		FROM information_schema.statistics
		WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?
		ORDER BY seq_in_index`, tableName, indexName)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	var out []string
	for rows.Next() {
		var col string
		if err := rows.Scan(&col); err != nil {
			return nil, err
		}
		out = append(out, strings.ToLower(col))
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func sameStringSlice(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func metaSchemaSpecFromStatements(stmts []string) (metaSchemaSpec, error) {
	tables := make([]metaTableSpec, 0)
	for _, stmt := range stmts {
		table, ok, err := parseCreateMetaTableSpec(stmt)
		if err != nil {
			return metaSchemaSpec{}, err
		}
		if ok {
			tables = append(tables, table)
		}
	}
	return metaSchemaSpec{tables: tables}, nil
}

func parseCreateMetaTableSpec(stmt string) (metaTableSpec, bool, error) {
	tableName, defs, ok, err := parseMetaCreateTableStatement(stmt)
	if err != nil {
		return metaTableSpec{}, false, err
	}
	if !ok {
		return metaTableSpec{}, false, nil
	}
	columns := make(map[string]metaColumnSpec)
	indexes := make(map[string]metaIndexSpec)
	for _, def := range splitMetaTopLevelComma(defs) {
		def = strings.TrimSpace(def)
		if def == "" {
			continue
		}
		if idxName, idxSpec, ok := parseMetaConstraintIndexDefinition(tableName, def); ok {
			indexes[idxName] = idxSpec
			continue
		}
		if idxName, idxSQL, ok := parseMetaInlineIndexDefinition(tableName, def); ok {
			indexes[idxName] = metaIndexSpec{createSQL: idxSQL}
			continue
		}
		if isMetaConstraintDefinition(def) {
			continue
		}
		colName, colSpec, ok := parseMetaColumnDefinition(tableName, def)
		if ok {
			columns[colName] = colSpec
			normalizedDef := normalizeMetaSQLFragment(def)
			if strings.Contains(normalizedDef, " primary key") {
				indexes["primary"] = metaIndexSpec{isPrimary: true}
			}
		}
	}
	return metaTableSpec{
		name:            tableName,
		createStatement: strings.TrimSpace(stmt),
		columns:         columns,
		indexes:         indexes,
	}, true, nil
}

func diffMetaSchema(ctx context.Context, db *sql.DB, spec metaSchemaSpec) ([]metaSchemaDiff, error) {
	var diffs []metaSchemaDiff
	for _, table := range spec.tables {
		tableDiffs, err := diffMetaTable(ctx, db, table)
		if err != nil {
			return nil, err
		}
		diffs = append(diffs, tableDiffs...)
	}
	return diffs, nil
}

func diffMetaTable(ctx context.Context, db *sql.DB, table metaTableSpec) ([]metaSchemaDiff, error) {
	meta, err := loadMetaTableMeta(ctx, db, table.name)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return []metaSchemaDiff{{
				kind:      metaSchemaDiffMissingTable,
				tableName: table.name,
				detail:    fmt.Sprintf("%s schema contract: missing table", table.name),
				repairSQL: table.createStatement,
			}}, nil
		}
		return nil, fmt.Errorf("load %s table metadata: %w", table.name, err)
	}
	indexNames, err := loadMetaIndexNames(ctx, db, table.name)
	if err != nil {
		return nil, fmt.Errorf("load %s index metadata: %w", table.name, err)
	}
	return diffMetaTableMetaWithObservedIndexes(table, meta, "", indexNames), nil
}

func diffMetaTableMeta(table metaTableSpec, meta metaTableMeta, createStmt string) []metaSchemaDiff {
	return diffMetaTableMetaWithObservedIndexes(table, meta, createStmt, parseObservedMetaIndexes(createStmt))
}

func diffMetaTableMetaWithObservedIndexes(table metaTableSpec, meta metaTableMeta, createStmt string, observedIndexes map[string]struct{}) []metaSchemaDiff {
	var diffs []metaSchemaDiff
	for _, name := range sortedMetaColumnNames(table.columns) {
		spec := table.columns[name]
		col, ok := meta.columns[name]
		if !ok {
			diffs = append(diffs, metaSchemaDiff{
				kind:       metaSchemaDiffMissingColumn,
				tableName:  table.name,
				columnName: name,
				detail:     fmt.Sprintf("%s schema contract: missing %s column", table.name, name),
				repairSQL:  spec.addSQL,
			})
			continue
		}
		observedType := normalizeMetaSQLFragment(col.columnType)
		desiredType := normalizeMetaSQLFragment(spec.columnType)
		if observedType != desiredType {
			// Only generate repair SQL for the one known-safe widening:
			// tenants.schema_version INT → INT UNSIGNED.
			var repairSQL string
			if table.name == "tenants" && name == "schema_version" &&
				observedType == "int" && desiredType == "int unsigned" {
				repairSQL = spec.modifySQL
			}
			if table.name == "org_object_backends" && name == "secret_cipher" &&
				observedType == "varbinary(2048)" && desiredType == "varbinary(8192)" {
				repairSQL = spec.modifySQL
			}
			diffs = append(diffs, metaSchemaDiff{
				kind:       metaSchemaDiffColumnType,
				tableName:  table.name,
				columnName: name,
				detail:     fmt.Sprintf("%s schema contract: %s column type = %q, want %s", table.name, name, col.columnType, spec.columnType),
				repairSQL:  repairSQL,
			})
		}
	}
	for _, name := range sortedMetaIndexNames(table.indexes) {
		spec := table.indexes[name]
		if !hasObservedMetaIndex(observedIndexes, name, spec) {
			detail := fmt.Sprintf("%s schema contract: missing %s index", table.name, name)
			if spec.isPrimary {
				detail = fmt.Sprintf("%s schema contract: missing primary key constraint", table.name)
			}
			diffs = append(diffs, metaSchemaDiff{
				kind:      metaSchemaDiffMissingIndex,
				tableName: table.name,
				indexName: name,
				detail:    detail,
				repairSQL: spec.createSQL,
			})
		}
	}
	return diffs
}

func hasObservedMetaIndex(observedIndexes map[string]struct{}, indexName string, spec metaIndexSpec) bool {
	if len(observedIndexes) == 0 {
		return false
	}
	name := strings.ToLower(indexName)
	if spec.isPrimary {
		name = "primary"
	}
	_, ok := observedIndexes[name]
	return ok
}

func parseObservedMetaIndexes(createStmt string) map[string]struct{} {
	_, defs, ok, err := parseMetaCreateTableStatement(createStmt)
	if err != nil || !ok {
		return nil
	}
	observed := make(map[string]struct{})
	for _, def := range splitMetaTopLevelComma(defs) {
		def = strings.TrimSpace(def)
		if def == "" {
			continue
		}
		if idxName, _, ok := parseMetaConstraintIndexDefinition("", def); ok {
			observed[strings.ToLower(idxName)] = struct{}{}
			continue
		}
		if idxName, _, ok := parseMetaInlineIndexDefinition("", def); ok {
			observed[strings.ToLower(idxName)] = struct{}{}
			continue
		}
		if strings.Contains(normalizeMetaSQLFragment(def), " primary key") {
			observed["primary"] = struct{}{}
		}
	}
	return observed
}

func plannedMetaSchemaRepairs(diffs []metaSchemaDiff) []string {
	tableMissing := make(map[string]bool)
	for _, diff := range diffs {
		if diff.kind == metaSchemaDiffMissingTable {
			tableMissing[diff.tableName] = true
		}
	}

	seen := make(map[string]struct{})
	plans := make([]string, 0, len(diffs))
	for _, diff := range diffs {
		if diff.repairSQL == "" {
			continue
		}
		if !isSafeMetaRepairDiff(diff, tableMissing) {
			continue
		}
		if _, ok := seen[diff.repairSQL]; ok {
			continue
		}
		seen[diff.repairSQL] = struct{}{}
		plans = append(plans, diff.repairSQL)
	}
	return plans
}

func applyMetaSchemaRepairs(ctx context.Context, db *sql.DB, stmts []string) error {
	for _, stmt := range stmts {
		if _, err := db.ExecContext(ctx, stmt); err != nil {
			if isIgnorableMetaSchemaError(err) {
				continue
			}
			return fmt.Errorf("apply meta schema repair %q: %w", schemaSnippet(stmt), err)
		}
	}
	return nil
}

func backfillTiDBCloudOrgBindingBranchIDs(ctx context.Context, db *sql.DB) error {
	_, err := db.ExecContext(ctx, `UPDATE tenant_tidbcloud_org_bindings b
		JOIN tenants t ON t.id = b.tenant_id
		SET b.branch_id = TRIM(COALESCE(t.branch_id, ''))
		WHERE b.branch_id <> TRIM(COALESCE(t.branch_id, ''))`)
	if err != nil {
		return fmt.Errorf("backfill tidbcloud org binding branch ids: %w", err)
	}
	return nil
}

func deleteStaleTiDBCloudOrgBindings(ctx context.Context, db metaQueryExecer) error {
	_, err := db.ExecContext(ctx, `DELETE b
		FROM tenant_tidbcloud_org_bindings b
		LEFT JOIN tenants t ON t.id = b.tenant_id
		WHERE t.id IS NULL OR t.status = ?`, TenantDeleted)
	if err != nil {
		return fmt.Errorf("delete stale tidbcloud org bindings: %w", err)
	}
	return nil
}

func ensureTiDBCloudOrgBindingUniqueIndex(ctx context.Context, db *sql.DB) error {
	exists, err := metaIndexExists(ctx, db, "tenant_tidbcloud_org_bindings", "uk_tidbcloud_org_cluster_branch")
	if err != nil {
		return fmt.Errorf("check tidbcloud org binding unique index: %w", err)
	}
	if exists {
		return nil
	}
	if err := ensureNoDuplicateTiDBCloudOrgBindingTuples(ctx, db); err != nil {
		return fmt.Errorf("preflight tidbcloud org binding unique index: %w", err)
	}
	_, err = db.ExecContext(ctx, `CREATE UNIQUE INDEX uk_tidbcloud_org_cluster_branch
		ON tenant_tidbcloud_org_bindings(organization_id, cluster_id, branch_id)`)
	if err != nil {
		if isIgnorableMetaSchemaError(err) {
			return nil
		}
		return fmt.Errorf("create tidbcloud org binding unique index: %w", err)
	}
	return nil
}

func metaIndexExists(ctx context.Context, db *sql.DB, tableName, indexName string) (bool, error) {
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM information_schema.statistics
		WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?`,
		tableName, indexName).Scan(&count); err != nil {
		return false, fmt.Errorf("check meta index %s.%s exists: %w", tableName, indexName, err)
	}
	return count > 0, nil
}

func metaTableExists(ctx context.Context, db *sql.DB, tableName string) (bool, error) {
	if err := validateMetaIdentifier(tableName); err != nil {
		return false, err
	}
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM information_schema.tables
		WHERE table_schema = DATABASE() AND table_name = ?`, tableName).Scan(&count); err != nil {
		return false, err
	}
	return count > 0, nil
}

func backfillOrgObjectBackendIdentityHashes(ctx context.Context, db *sql.DB) error {
	exists, err := metaTableExists(ctx, db, "org_object_backends")
	if err != nil {
		return fmt.Errorf("check org_object_backends exists: %w", err)
	}
	if !exists {
		return nil
	}
	rows, err := db.QueryContext(ctx, `SELECT id, organization_id, scheme, bucket, prefix, endpoint, identity_hash FROM org_object_backends`)
	if err != nil {
		return fmt.Errorf("list org object backends for identity hash backfill: %w", err)
	}
	defer func() { _ = rows.Close() }()
	type row struct {
		id, org, scheme, bucket, prefix, endpoint, hash string
	}
	var pending []row
	for rows.Next() {
		var r row
		if err := rows.Scan(&r.id, &r.org, &r.scheme, &r.bucket, &r.prefix, &r.endpoint, &r.hash); err != nil {
			return fmt.Errorf("scan org object backend identity hash: %w", err)
		}
		want := orgObjectBackendIdentityHash(&OrgObjectBackend{
			OrganizationID: r.org,
			Scheme:         r.scheme,
			Bucket:         r.bucket,
			Prefix:         r.prefix,
			Endpoint:       r.endpoint,
		})
		if r.hash != want {
			pending = append(pending, row{id: r.id, hash: want})
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate org object backend identity hashes: %w", err)
	}
	for _, r := range pending {
		if _, err := db.ExecContext(ctx, `UPDATE org_object_backends SET identity_hash = ? WHERE id = ?`, r.hash, r.id); err != nil {
			return fmt.Errorf("backfill org object backend identity hash %s: %w", r.id, err)
		}
	}
	return nil
}

func ensureOrgObjectBackendIdentityIndex(ctx context.Context, db *sql.DB) error {
	exists, err := metaTableExists(ctx, db, "org_object_backends")
	if err != nil {
		return fmt.Errorf("check org_object_backends exists: %w", err)
	}
	if !exists {
		return nil
	}
	hasIndex, err := metaIndexExists(ctx, db, "org_object_backends", "uk_org_object_backend_identity")
	if err != nil {
		return fmt.Errorf("check org object backend identity index: %w", err)
	}
	if hasIndex {
		return nil
	}
	var dupes int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM (
		SELECT identity_hash FROM org_object_backends GROUP BY identity_hash HAVING COUNT(*) > 1
	) d`).Scan(&dupes); err != nil {
		return fmt.Errorf("preflight org object backend identity index: %w", err)
	}
	if dupes > 0 {
		return fmt.Errorf("cannot create org object backend identity index: found %d duplicate identity_hash group(s)", dupes)
	}
	_, err = db.ExecContext(ctx, `CREATE UNIQUE INDEX uk_org_object_backend_identity ON org_object_backends(identity_hash)`)
	if err != nil {
		if isIgnorableMetaSchemaError(err) {
			return nil
		}
		return fmt.Errorf("create org object backend identity index: %w", err)
	}
	return nil
}

func ensureNoDuplicateTiDBCloudOrgBindingTuples(ctx context.Context, db *sql.DB) error {
	var total int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM (
		SELECT organization_id, cluster_id, branch_id
		FROM tenant_tidbcloud_org_bindings
		GROUP BY organization_id, cluster_id, branch_id
		HAVING COUNT(*) > 1
	) duplicates`).Scan(&total); err != nil {
		return fmt.Errorf("count duplicate tidbcloud org binding tuples: %w", err)
	}
	if total == 0 {
		return nil
	}

	rows, err := db.QueryContext(ctx, `SELECT organization_id, cluster_id, branch_id, COUNT(*) AS n, GROUP_CONCAT(tenant_id ORDER BY tenant_id)
		FROM tenant_tidbcloud_org_bindings
		GROUP BY organization_id, cluster_id, branch_id
		HAVING COUNT(*) > 1
		ORDER BY n DESC, organization_id, cluster_id, branch_id
		LIMIT ?`, maxTiDBCloudOrgBindingDuplicateTuples)
	if err != nil {
		return fmt.Errorf("list duplicate tidbcloud org binding tuples: %w", err)
	}
	defer func() { _ = rows.Close() }()

	conflicts := make([]string, 0, min(total, maxTiDBCloudOrgBindingDuplicateTuples))
	for rows.Next() {
		var organizationID, clusterID, branchID, tenantIDs string
		var count int
		if err := rows.Scan(&organizationID, &clusterID, &branchID, &count, &tenantIDs); err != nil {
			return fmt.Errorf("scan duplicate tidbcloud org binding tuple: %w", err)
		}
		conflicts = append(conflicts, fmt.Sprintf("organization_id=%q cluster_id=%q branch_id=%q count=%d tenant_ids=%s", organizationID, clusterID, branchID, count, tenantIDs))
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate duplicate tidbcloud org binding tuples: %w", err)
	}
	if total > len(conflicts) {
		conflicts = append(conflicts, fmt.Sprintf("... %d more duplicate tuple(s) omitted", total-len(conflicts)))
	}
	return fmt.Errorf("cannot create tidbcloud org binding unique index: found %d duplicate tuple(s): %s", total, strings.Join(conflicts, "; "))
}

func parseMetaCreateTableStatement(stmt string) (tableName string, definitions string, ok bool, err error) {
	return schemaspec.ParseCreateTableStatement(stmt)
}

func splitMetaTopLevelComma(definitions string) []string {
	return schemaspec.SplitTopLevelComma(definitions)
}

func parseMetaInlineIndexDefinition(tableName, def string) (indexName, createSQL string, ok bool) {
	n := normalizeMetaSQLFragment(def)
	if strings.HasPrefix(n, "unique index ") {
		name, cols := parseMetaIndexNameAndColumns(def, "UNIQUE INDEX")
		if name == "" || cols == "" {
			return "", "", false
		}
		return name, fmt.Sprintf("CREATE UNIQUE INDEX %s ON %s%s", name, tableName, cols), true
	}
	if strings.HasPrefix(n, "index ") {
		name, cols := parseMetaIndexNameAndColumns(def, "INDEX")
		if name == "" || cols == "" {
			return "", "", false
		}
		return name, fmt.Sprintf("CREATE INDEX %s ON %s%s", name, tableName, cols), true
	}
	return "", "", false
}

func parseMetaConstraintIndexDefinition(tableName, def string) (indexName string, spec metaIndexSpec, ok bool) {
	n := normalizeMetaSQLFragment(def)
	if strings.HasPrefix(n, "primary key") {
		return "primary", metaIndexSpec{isPrimary: true}, true
	}
	if strings.HasPrefix(n, "unique key ") {
		name, cols := parseMetaIndexNameAndColumns(def, "UNIQUE KEY")
		if name == "" || cols == "" {
			return "", metaIndexSpec{}, false
		}
		return name, metaIndexSpec{
			createSQL: fmt.Sprintf("CREATE UNIQUE INDEX %s ON %s%s", name, tableName, cols),
		}, true
	}
	if strings.HasPrefix(n, "constraint ") && strings.Contains(n, " unique key ") {
		// CONSTRAINT name UNIQUE KEY idx_name (...) -> parse using UNIQUE KEY token.
		name, cols := parseMetaIndexNameAndColumns(def, "UNIQUE KEY")
		if name == "" || cols == "" {
			return "", metaIndexSpec{}, false
		}
		return name, metaIndexSpec{
			createSQL: fmt.Sprintf("CREATE UNIQUE INDEX %s ON %s%s", name, tableName, cols),
		}, true
	}
	return "", metaIndexSpec{}, false
}

func parseMetaIndexNameAndColumns(def, prefix string) (indexName, columns string) {
	trimmed := strings.TrimSpace(def)
	upper := strings.ToUpper(trimmed)
	p := strings.Index(upper, prefix)
	if p < 0 {
		return "", ""
	}
	rest := strings.TrimSpace(trimmed[p+len(prefix):])
	name, remainder := splitMetaIdentifierAndRest(rest)
	if name == "" {
		return "", ""
	}
	open := strings.Index(remainder, "(")
	if open < 0 {
		return "", ""
	}
	return strings.ToLower(name), strings.TrimSpace(remainder[open:])
}

func parseMetaColumnDefinition(tableName, def string) (string, metaColumnSpec, bool) {
	name, rest := splitMetaIdentifierAndRest(def)
	if name == "" || rest == "" {
		return "", metaColumnSpec{}, false
	}
	colType := parseMetaColumnType(rest)
	if colType == "" {
		return "", metaColumnSpec{}, false
	}
	return strings.ToLower(name), metaColumnSpec{
		columnType: strings.ToLower(strings.TrimSpace(colType)),
		addSQL:     fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", tableName, name, strings.TrimSpace(rest)),
		modifySQL:  fmt.Sprintf("ALTER TABLE %s MODIFY COLUMN %s %s", tableName, name, strings.TrimSpace(rest)),
	}, true
}

func parseMetaColumnType(rest string) string {
	return schemaspec.ParseColumnType(rest)
}

func splitMetaIdentifierAndRest(s string) (identifier string, rest string) {
	return schemaspec.SplitIdentifierAndRest(s)
}

func isMetaConstraintDefinition(def string) bool {
	n := normalizeMetaSQLFragment(def)
	return strings.HasPrefix(n, "primary key") || strings.HasPrefix(n, "constraint ") || strings.HasPrefix(n, "unique key ")
}

func isSafeMetaRepairDiff(diff metaSchemaDiff, tableMissing map[string]bool) bool {
	switch diff.kind {
	case metaSchemaDiffMissingTable:
		return true
	case metaSchemaDiffMissingColumn:
		return isSafeAddColumnRepairSQL(diff.repairSQL)
	case metaSchemaDiffMissingIndex:
		normalized := normalizeMetaSQLFragment(diff.repairSQL)
		if strings.HasPrefix(normalized, "create index ") {
			return true
		}
		if strings.HasPrefix(normalized, "create unique index ") {
			return tableMissing[diff.tableName]
		}
		return false
	case metaSchemaDiffColumnType:
		return isSafeModifyColumnRepairSQL(diff)
	default:
		return false
	}
}

func isSafeAddColumnRepairSQL(sqlText string) bool {
	return schemaspec.IsSafeAddColumnRepairSQL(sqlText)
}

// isSafeModifyColumnRepairSQL returns true only for the single known-safe
// column widening: tenants.schema_version INT → INT UNSIGNED.
// schema_version values come from CRC32Version which produces uint32 (non-negative);
// widening from INT to INT UNSIGNED is data-preserving for this column.
// General MODIFY COLUMN is not considered safe without an explicit entry here.
func isSafeModifyColumnRepairSQL(diff metaSchemaDiff) bool {
	n := strings.TrimSuffix(normalizeMetaSQLFragment(diff.repairSQL), ";")
	if diff.tableName == "tenants" && diff.columnName == "schema_version" {
		return n == "alter table tenants modify column schema_version int unsigned not null default 1"
	}
	if diff.tableName == "org_object_backends" && diff.columnName == "secret_cipher" {
		return n == "alter table org_object_backends modify column secret_cipher varbinary(8192) null"
	}
	return false
}

func isIgnorableMetaSchemaError(err error) bool {
	return schemaspec.IsIgnorableMySQLError(err)
}

func loadMetaTableMeta(ctx context.Context, db *sql.DB, tableName string) (metaTableMeta, error) {
	rows, err := db.QueryContext(ctx, `SELECT column_name, column_type
		FROM information_schema.columns
		WHERE table_schema = DATABASE() AND table_name = ?`, tableName)
	if err != nil {
		return metaTableMeta{}, err
	}
	defer func() { _ = rows.Close() }()

	columns := make(map[string]metaColumnMeta)
	for rows.Next() {
		var name, columnType string
		if err := rows.Scan(&name, &columnType); err != nil {
			return metaTableMeta{}, err
		}
		columns[strings.ToLower(name)] = metaColumnMeta{columnType: columnType}
	}
	if err := rows.Err(); err != nil {
		return metaTableMeta{}, err
	}
	if len(columns) == 0 {
		return metaTableMeta{}, sql.ErrNoRows
	}
	return metaTableMeta{tableName: tableName, columns: columns}, nil
}

func loadMetaIndexNames(ctx context.Context, db *sql.DB, tableName string) (map[string]struct{}, error) {
	rows, err := db.QueryContext(ctx, `SELECT DISTINCT index_name
		FROM information_schema.statistics
		WHERE table_schema = DATABASE() AND table_name = ?`, tableName)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	indexNames := make(map[string]struct{})
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		indexNames[strings.ToLower(name)] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return indexNames, nil
}

func normalizeMetaSQLFragment(s string) string {
	return schemaspec.NormalizeSQLFragment(s)
}

func sortedMetaColumnNames(columns map[string]metaColumnSpec) []string {
	names := make([]string, 0, len(columns))
	for name := range columns {
		names = append(names, name)
	}
	sort.Strings(names)
	return names
}

func sortedMetaIndexNames(indexes map[string]metaIndexSpec) []string {
	names := make([]string, 0, len(indexes))
	for name := range indexes {
		names = append(names, name)
	}
	sort.Strings(names)
	return names
}

func schemaSnippet(stmt string) string {
	return schemaspec.SQLSnippet(stmt)
}

func (s *Store) InsertTenant(ctx context.Context, t *Tenant) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_tenant", start, &err)
	// Retry the whole tx on lock conflicts: concurrent cold Acquires allocate
	// fs_ids through EnsureFsID, and the two-table write below (tenants +
	// fs_registry) can deadlock with them on the unique index of
	// fs_registry.tenant_id. The deadlock victim retries cleanly.
	return withMetaLockConflictRetry(ctx, "insert_tenant", func() error {
		return s.InTx(ctx, func(tx *sql.Tx) error {
			return insertTenantTx(ctx, tx, t)
		})
	})
}

func insertTenantTx(ctx context.Context, tx *sql.Tx, t *Tenant) error {
	labelJSON, err := tenantLabelJSON(t.Label)
	if err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO tenants
		(id, display_name, label, status, kind, parent_tenant_id, storage_namespace_id, db_host, db_port, db_user, db_password, db_name, db_tls,
		 provider, cluster_id, branch_id, claim_url, claim_expires_at, schema_version,
		 s3_encryption_mode, s3_kms_key_id, s3_bucket_key_enabled, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		t.ID, strings.TrimSpace(t.DisplayName), labelJSON, t.Status, tenantKindForInsert(t), t.ParentTenantID, t.StorageNamespaceID,
		t.DBHost, t.DBPort, t.DBUser, t.DBPasswordCipher, t.DBName, boolToInt(t.DBTLS),
		t.Provider, nullStr(t.ClusterID), t.BranchID, nullStr(t.ClaimURL), t.ClaimExpiresAt, t.SchemaVersion,
		tenantS3EncryptionModeForInsert(t), t.S3KMSKeyID, boolToInt(tenantS3BucketKeyEnabledForInsert(t)),
		t.CreatedAt.UTC(), t.UpdatedAt.UTC())
	if isDuplicateEntry(err) {
		return ErrDuplicate
	}
	if err != nil {
		return err
	}
	// Every tenant gets a stable internal fs_id at creation time (see
	// fs_registry). INSERT IGNORE keeps this idempotent for tenants that were
	// pre-registered by BackfillFsRegistry.
	_, err = tx.ExecContext(ctx, `INSERT IGNORE INTO fs_registry (tenant_id) VALUES (?)`, t.ID)
	return err
}

func tenantLabelJSON(label map[string]string) ([]byte, error) {
	if label == nil {
		label = map[string]string{}
	}
	raw, err := json.Marshal(label)
	if err != nil {
		return nil, fmt.Errorf("marshal tenant label: %w", err)
	}
	return raw, nil
}

func scanTenantLabel(raw []byte) (map[string]string, error) {
	if len(raw) == 0 || string(raw) == "null" {
		return map[string]string{}, nil
	}
	var label map[string]string
	if err := json.Unmarshal(raw, &label); err != nil {
		return nil, fmt.Errorf("decode tenant label: %w", err)
	}
	if label == nil {
		label = map[string]string{}
	}
	return label, nil
}

type TenantListFilter struct {
	DisplayNameContains string
	LabelKey            string
	LabelValue          string
}

type TenantMetadata struct {
	DisplayName string
	Label       map[string]string
}

const tenantTiDBCloudOrganizationVisibilityPredicateSQL = `(
	(t.provider = ? AND EXISTS (
		SELECT 1 FROM tenant_tidbcloud_org_bindings b
		WHERE b.tenant_id = t.id
			AND b.organization_id = ?
			AND b.pool_status <> ?
	)) OR
	(t.provider = ? AND EXISTS (
		SELECT 1 FROM fs_registry f
		JOIN tenant_placements p ON p.fs_id = f.fs_id
		JOIN db_pool d ON d.db_id = p.db_id
		WHERE f.tenant_id = t.id
			AND d.org_id = ?
			AND NOT EXISTS (
				SELECT 1 FROM tenant_pool_memberships m
				WHERE m.tenant_id = t.id AND m.pool_status = ?
			)
	))
)`

func (s *Store) UpsertTenantAutoEmbeddingProfile(ctx context.Context, p *TenantAutoEmbeddingProfile) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_tenant_auto_embedding_profile", start, &err)
	if p == nil {
		return fmt.Errorf("nil tenant auto-embedding profile")
	}
	_, err = s.db.ExecContext(ctx, `INSERT INTO tenant_auto_embedding_profiles
		(tenant_id, embedding_mode, model, dimensions, options_json, api_base, api_key_cipher, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE
			embedding_mode = VALUES(embedding_mode),
			model = VALUES(model),
			dimensions = VALUES(dimensions),
			options_json = VALUES(options_json),
			api_base = VALUES(api_base),
			api_key_cipher = VALUES(api_key_cipher),
			updated_at = VALUES(updated_at)`,
		p.TenantID, nullStr(p.EmbeddingMode), p.Model, p.Dimensions, p.OptionsJSON, nullStr(p.APIBase), nullableBytes(p.APIKeyCipher),
		p.CreatedAt.UTC(), p.UpdatedAt.UTC())
	return err
}

func (s *Store) GetTenantAutoEmbeddingProfile(ctx context.Context, tenantID string) (out *TenantAutoEmbeddingProfile, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tenant_auto_embedding_profile", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT tenant_id, embedding_mode, model, dimensions, options_json,
			api_base, api_key_cipher, created_at, updated_at
		FROM tenant_auto_embedding_profiles WHERE tenant_id = ?`, tenantID)
	var rec TenantAutoEmbeddingProfile
	var embeddingMode sql.NullString
	var apiBase sql.NullString
	var apiKeyCipher []byte
	if err = row.Scan(&rec.TenantID, &embeddingMode, &rec.Model, &rec.Dimensions, &rec.OptionsJSON,
		&apiBase, &apiKeyCipher, &rec.CreatedAt, &rec.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			err = ErrNotFound
			return nil, err
		}
		return nil, err
	}
	if embeddingMode.Valid {
		rec.EmbeddingMode = embeddingMode.String
	}
	if apiBase.Valid {
		rec.APIBase = apiBase.String
	}
	rec.APIKeyCipher = apiKeyCipher
	return &rec, nil
}

func (s *Store) EnsureTenantAutoEmbeddingProfile(ctx context.Context, tenantID string) (out *TenantAutoEmbeddingProfile, err error) {
	start := time.Now()
	defer observeMeta(ctx, "ensure_tenant_auto_embedding_profile", start, &err)
	profile, err := s.GetTenantAutoEmbeddingProfile(ctx, tenantID)
	if err == nil {
		return profile, nil
	}
	if !errors.Is(err, ErrNotFound) {
		return nil, err
	}
	_, err = s.db.ExecContext(ctx, `INSERT INTO tenant_auto_embedding_profiles (tenant_id) VALUES (?)`, tenantID)
	if err != nil && !isDuplicateEntry(err) {
		return nil, err
	}
	return s.GetTenantAutoEmbeddingProfile(ctx, tenantID)
}

func (s *Store) SetTenantAutoEmbeddingProfileModeIfNull(ctx context.Context, tenantID, mode string) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "set_tenant_auto_embedding_profile_mode_if_null", start, &err)
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_auto_embedding_profiles
		SET embedding_mode = ?, updated_at = CURRENT_TIMESTAMP(3)
		WHERE tenant_id = ? AND embedding_mode IS NULL`, mode, tenantID)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

func (s *Store) InsertAPIKey(ctx context.Context, k *APIKey) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_api_key", start, &err)
	scopeKind, err := apiKeyScopeKindForInsert(k)
	if err != nil {
		return err
	}
	if err := guardAPIKeyExpiresAt(scopeKind, k); err != nil {
		return err
	}
	_, err = s.db.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, expires_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, nullTimeFromPtr(k.ExpiresAt), k.CreatedAt.UTC(), k.UpdatedAt.UTC())
	if isDuplicateEntry(err) {
		return ErrDuplicate
	}
	return err
}

func guardAPIKeyExpiresAt(scopeKind APIKeyScopeKind, k *APIKey) error {
	if k == nil {
		return fmt.Errorf("api key is required")
	}
	if scopeKind == APIKeyScopeKindFS && k.ExpiresAt == nil {
		return fmt.Errorf("fs_scoped api key requires expires_at")
	}
	return nil
}

func (s *Store) GetExternalBinding(ctx context.Context, provider, subjectKey string) (out *ExternalBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_external_binding", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT provider, subject_key, tenant_id, metadata_json, created_at, updated_at
		FROM tenant_external_bindings
		WHERE provider = ? AND subject_key = ?`, provider, subjectKey)
	var rec ExternalBinding
	var metadata []byte
	if err = row.Scan(&rec.Provider, &rec.SubjectKey, &rec.TenantID, &metadata, &rec.CreatedAt, &rec.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			err = ErrNotFound
			return nil, err
		}
		return nil, err
	}
	rec.MetadataJSON = metadata
	return &rec, nil
}

func (s *Store) InsertExternalBinding(ctx context.Context, b *ExternalBinding) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_external_binding", start, &err)
	if b == nil {
		return fmt.Errorf("external binding is required")
	}
	_, err = s.db.ExecContext(ctx, `INSERT INTO tenant_external_bindings
		(provider, subject_key, tenant_id, metadata_json, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?)`,
		b.Provider, b.SubjectKey, b.TenantID, nullableBytes(b.MetadataJSON), b.CreatedAt.UTC(), b.UpdatedAt.UTC())
	if isDuplicateEntry(err) {
		return ErrDuplicate
	}
	return err
}

func (s *Store) DeleteExternalBinding(ctx context.Context, provider, subjectKey string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_external_binding", start, &err)
	res, err := s.db.ExecContext(ctx, `DELETE FROM tenant_external_bindings WHERE provider = ? AND subject_key = ?`, provider, subjectKey)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Store) UpsertTenantTiDBCloudOrgBinding(ctx context.Context, b *TenantTiDBCloudOrgBinding) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_tidbcloud_org_binding", start, &err)
	if b == nil {
		return fmt.Errorf("tidbcloud org binding is required")
	}
	tenantID := strings.TrimSpace(b.TenantID)
	organizationID := strings.TrimSpace(b.OrganizationID)
	clusterID := strings.TrimSpace(b.ClusterID)
	if tenantID == "" {
		return fmt.Errorf("tenant_id is required")
	}
	if organizationID == "" {
		return fmt.Errorf("organization_id is required")
	}
	if clusterID == "" {
		return fmt.Errorf("cluster_id is required")
	}
	branchID := strings.TrimSpace(b.BranchID)
	// tenants.branch_id is the source of truth; callers cannot override the
	// branch dimension used for duplicate-ownership checks.
	if tenantBranchID, tenantProvider, ok, lookupErr := s.lookupTenantBindingIdentity(ctx, tenantID); lookupErr != nil {
		return lookupErr
	} else if ok {
		if tenantProvider == tidbCloudNativeSharedProvider {
			return fmt.Errorf("shared tenant %q cannot own a dedicated tidbcloud org binding", tenantID)
		}
		branchID = tenantBranchID
	}
	now := time.Now().UTC()
	createdAt := b.CreatedAt
	if createdAt.IsZero() {
		createdAt = now
	}
	updatedAt := b.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = now
	}
	return s.withTiDBCloudOrgBindingLock(ctx, organizationID, clusterID, branchID, func(ctx context.Context, conn metaQueryExecer) error {
		if err := deleteStaleTiDBCloudOrgBindingConflicts(ctx, conn, tenantID, organizationID, clusterID, branchID); err != nil {
			return err
		}
		if err := ensureTiDBCloudOrgBindingAvailable(ctx, conn, tenantID, organizationID, clusterID, branchID); err != nil {
			return err
		}
		_, execErr := conn.ExecContext(ctx, `INSERT INTO tenant_tidbcloud_org_bindings
			(tenant_id, organization_id, cluster_id, branch_id, pool_id, pool_status, used_at, created_at, updated_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
			ON DUPLICATE KEY UPDATE
				organization_id = VALUES(organization_id),
				cluster_id = VALUES(cluster_id),
				branch_id = VALUES(branch_id),
				pool_id = VALUES(pool_id),
				pool_status = VALUES(pool_status),
				used_at = VALUES(used_at),
			updated_at = VALUES(updated_at)`,
			tenantID, organizationID, clusterID, branchID, strings.TrimSpace(b.PoolID), tenantPoolBindingStatusForInsert(b.PoolStatus), b.UsedAt,
			createdAt.UTC(), updatedAt.UTC())
		return execErr
	})
}

func deleteStaleTiDBCloudOrgBindingConflicts(ctx context.Context, q metaQueryExecer, tenantID, organizationID, clusterID, branchID string) error {
	_, err := q.ExecContext(ctx, `DELETE b
		FROM tenant_tidbcloud_org_bindings b
		LEFT JOIN tenants t ON t.id = b.tenant_id
		WHERE b.organization_id = ? AND b.cluster_id = ? AND b.branch_id = ?
			AND b.tenant_id <> ? AND (t.id IS NULL OR t.status = ?)`,
		organizationID, clusterID, branchID, tenantID, TenantDeleted)
	if err != nil {
		return fmt.Errorf("delete stale tidbcloud org binding conflicts: %w", err)
	}
	return nil
}

func (s *Store) lookupTenantBindingIdentity(ctx context.Context, tenantID string) (branchID, provider string, ok bool, err error) {
	var nullableBranchID sql.NullString
	err = s.db.QueryRowContext(ctx, `SELECT branch_id, provider FROM tenants WHERE id = ?`, tenantID).
		Scan(&nullableBranchID, &provider)
	if errors.Is(err, sql.ErrNoRows) {
		return "", "", false, nil
	}
	if err != nil {
		return "", "", false, err
	}
	if nullableBranchID.Valid {
		return strings.TrimSpace(nullableBranchID.String), provider, true, nil
	}
	return "", provider, true, nil
}

func (s *Store) withTiDBCloudOrgBindingLock(ctx context.Context, organizationID, clusterID, branchID string, fn func(context.Context, metaQueryExecer) error) (err error) {
	if fn == nil {
		return fmt.Errorf("tidbcloud org binding lock callback is required")
	}
	lockName := tidbCloudOrgBindingLockName(organizationID, clusterID, branchID)
	conn, err := s.db.Conn(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = conn.Close() }()

	var databaseName sql.NullString
	if err := conn.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&databaseName); err != nil {
		return err
	}
	lockName = tenantPoolDatabaseLockName(lockName, databaseName.String)

	var got sql.NullInt64
	if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", lockName, tidbCloudOrgBindingLockTimeoutSeconds).Scan(&got); err != nil {
		return err
	}
	if !got.Valid {
		return fmt.Errorf("tidbcloud org binding named lock returned NULL")
	}
	if got.Int64 != 1 {
		return fmt.Errorf("timed out waiting for tidbcloud org binding named lock")
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), tidbCloudOrgBindingReleaseLockTimeout)
		defer cancel()
		var released sql.NullInt64
		releaseErr := conn.QueryRowContext(releaseCtx, "SELECT RELEASE_LOCK(?)", lockName).Scan(&released)
		if releaseErr != nil {
			err = errors.Join(err, releaseErr)
			return
		}
		if !released.Valid {
			err = errors.Join(err, fmt.Errorf("tidbcloud org binding named lock release returned NULL"))
			return
		}
		if released.Int64 != 1 {
			err = errors.Join(err, fmt.Errorf("tidbcloud org binding named lock was not held by current connection"))
		}
	}()

	return fn(ctx, conn)
}

func ensureTiDBCloudOrgBindingAvailable(ctx context.Context, q metaQueryExecer, tenantID, organizationID, clusterID, branchID string) error {
	var existingTenantID string
	err := q.QueryRowContext(ctx, `SELECT b.tenant_id
		FROM tenant_tidbcloud_org_bindings b
		JOIN tenants t ON t.id = b.tenant_id
		WHERE b.organization_id = ? AND b.cluster_id = ? AND b.branch_id = ?
			AND b.tenant_id <> ? AND t.status <> ?
		LIMIT 1`,
		organizationID, clusterID, branchID, tenantID, TenantDeleted).Scan(&existingTenantID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	return fmt.Errorf("%w: tidbcloud organization cluster branch already bound to tenant %s", ErrDuplicate, existingTenantID)
}

func (s *Store) GetTenantTiDBCloudOrgBinding(ctx context.Context, tenantID string) (out *TenantTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tidbcloud_org_binding", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT tenant_id, organization_id, cluster_id, branch_id, pool_id, pool_status, used_at, created_at, updated_at
		FROM tenant_tidbcloud_org_bindings WHERE tenant_id = ?`, tenantID)
	var rec TenantTiDBCloudOrgBinding
	var usedAt sql.NullTime
	if err = row.Scan(&rec.TenantID, &rec.OrganizationID, &rec.ClusterID, &rec.BranchID, &rec.PoolID, &rec.PoolStatus, &usedAt, &rec.CreatedAt, &rec.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			err = ErrNotFound
			return nil, err
		}
		return nil, err
	}
	if usedAt.Valid {
		t := usedAt.Time.UTC()
		rec.UsedAt = &t
	}
	return &rec, nil
}

func (s *Store) CreateTenantPool(ctx context.Context, p *TenantPool) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "create_tidbcloud_pool", start, &err)
	if p == nil {
		return fmt.Errorf("tenant pool is required")
	}
	poolID := strings.TrimSpace(p.PoolID)
	if poolID == "" {
		return fmt.Errorf("pool_id is required")
	}
	if p.Size < 0 {
		return fmt.Errorf("pool size must be non-negative")
	}
	now := time.Now().UTC()
	createdAt := p.CreatedAt
	if createdAt.IsZero() {
		createdAt = now
	}
	updatedAt := p.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = now
	}
	_, err = s.db.ExecContext(ctx, `INSERT INTO tenant_tidbcloud_pools
		(pool_id, organization_id, size, status, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?)`,
		poolID, nullStr(strings.TrimSpace(p.OrganizationID)), p.Size, tenantPoolStatusForInsert(p.Status),
		createdAt.UTC(), updatedAt.UTC())
	if isDuplicateEntry(err) {
		return ErrDuplicate
	}
	return err
}

func (s *Store) GetTenantPoolByOrganization(ctx context.Context, organizationID string) (out *TenantPool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tidbcloud_pool_by_org", start, &err)
	organizationID = strings.TrimSpace(organizationID)
	if organizationID == "" {
		return nil, fmt.Errorf("organization_id is required")
	}
	row := s.db.QueryRowContext(ctx, `SELECT pool_id, organization_id, size, status, created_at, updated_at
		FROM tenant_tidbcloud_pools WHERE organization_id = ?`, organizationID)
	return scanTenantPoolRow(row)
}

func (s *Store) GetTenantPoolByID(ctx context.Context, poolID string) (out *TenantPool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tidbcloud_pool_by_id", start, &err)
	poolID = strings.TrimSpace(poolID)
	if poolID == "" {
		return nil, fmt.Errorf("pool_id is required")
	}
	row := s.db.QueryRowContext(ctx, `SELECT pool_id, organization_id, size, status, created_at, updated_at
		FROM tenant_tidbcloud_pools WHERE pool_id = ?`, poolID)
	return scanTenantPoolRow(row)
}

// ListTenantPoolsByStatusAfter returns one stable keyset page ordered by pool ID.
func (s *Store) ListTenantPoolsByStatusAfter(ctx context.Context, status TenantPoolStatus, afterPoolID string, limit int) (out []*TenantPool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tidbcloud_pools_by_status_after", start, &err)
	switch status {
	case TenantPoolActive, TenantPoolDeleting:
	default:
		return nil, fmt.Errorf("unsupported tenant pool status %q", status)
	}
	if limit <= 0 {
		limit = 100
	}
	rows, err := s.db.QueryContext(ctx, `SELECT pool_id, organization_id, size, status, created_at, updated_at
		FROM tenant_tidbcloud_pools
		WHERE status = ? AND pool_id > ?
		ORDER BY pool_id
		LIMIT ?`, status, strings.TrimSpace(afterPoolID), limit)
	if err != nil {
		return nil, fmt.Errorf("list tenant pools in status %q: %w", status, err)
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		rec, scanErr := scanTenantPoolRow(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		out = append(out, rec)
	}
	return out, rows.Err()
}

func (s *Store) UpdateTenantPoolOrganization(ctx context.Context, poolID, organizationID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tidbcloud_pool_org", start, &err)
	poolID = strings.TrimSpace(poolID)
	organizationID = strings.TrimSpace(organizationID)
	if poolID == "" || organizationID == "" {
		return fmt.Errorf("pool_id and organization_id are required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_tidbcloud_pools
		SET organization_id = ?, updated_at = ? WHERE pool_id = ?`, organizationID, time.Now().UTC(), poolID)
	if err != nil {
		if isDuplicateEntry(err) {
			return ErrDuplicate
		}
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Store) CountFreeTenantPoolBindings(ctx context.Context, organizationID string) (out int, err error) {
	return s.countFreeTenantPoolBindingsByStatus(ctx, organizationID, []TenantStatus{TenantActive})
}

func (s *Store) CountTenantPoolFreeSlots(ctx context.Context, organizationID string) (out int, err error) {
	return s.countFreeTenantPoolBindingsByStatus(ctx, organizationID, []TenantStatus{TenantPending, TenantProvisioning, TenantActive})
}

// CountTenantPoolPlannedSlots counts shared free pending/provisioning inventory
// whose durable physical attempt remains non-terminal. The stuck-pool watchdog
// is the sole authority that expires a pending/provisioning physical attempt;
// active physical pools are finalized by the activation reconciler. Native
// pending tenants are intentionally excluded because leader refill has no
// request-scoped customer credential with which to advance them.
func (s *Store) CountTenantPoolPlannedSlots(ctx context.Context, organizationID string) (out int, err error) {
	start := time.Now()
	defer observeMeta(ctx, "count_tidbcloud_pool_planned_slots", start, &err)
	organizationID = strings.TrimSpace(organizationID)
	if organizationID == "" {
		return 0, fmt.Errorf("organization_id is required")
	}
	row := s.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(slot_count), 0) FROM (
		SELECT COUNT(*) AS slot_count
			FROM tenant_pool_memberships m
			STRAIGHT_JOIN tenants t ON t.id = m.tenant_id
			STRAIGHT_JOIN fs_registry f ON f.tenant_id = t.id
			STRAIGHT_JOIN tenant_placements p ON p.fs_id = f.fs_id
			STRAIGHT_JOIN db_pool d ON d.db_id = p.db_id
			WHERE m.tidbcloud_organization_id = ? AND m.pool_status = ? AND t.provider = ?
				AND t.status IN (?, ?) AND d.status IN (?, ?, ?)
		UNION ALL
		SELECT COALESCE(SUM(GREATEST(d.max_tenants - d.tenant_count, 0)), 0) AS slot_count
			FROM db_pool d
			WHERE d.org_id = ? AND d.`+"`role`"+` = ? AND d.provisioning_key IS NOT NULL
				AND d.status IN (?, ?) AND d.tenant_count = 0 AND d.max_tenants > 0
	) inventory`,
		organizationID, TenantPoolBindingFree, tidbCloudNativeSharedProvider,
		TenantPending, TenantProvisioning, SharedDBStatusPending, SharedDBStatusProvisioning, SharedDBStatusActive,
		organizationID, SharedDBRoleShared, SharedDBStatusPending, SharedDBStatusProvisioning)
	if err = row.Scan(&out); err != nil {
		return 0, err
	}
	return out, nil
}

const countFreeTenantPoolBindingsSQL = `SELECT COALESCE(SUM(slot_count), 0) FROM (
	SELECT COUNT(*) AS slot_count
		FROM tenant_tidbcloud_org_bindings b
		STRAIGHT_JOIN tenants t ON t.id = b.tenant_id
		WHERE b.organization_id = ? AND b.pool_status = ? AND t.provider = ?
			AND t.status IN (%s)
	UNION ALL
	SELECT COUNT(*) AS slot_count
		FROM tenant_pool_memberships m
		STRAIGHT_JOIN tenants t ON t.id = m.tenant_id
		WHERE m.tidbcloud_organization_id = ? AND m.pool_status = ? AND t.provider = ?
			AND t.status IN (%s)
) inventory`

func (s *Store) countFreeTenantPoolBindingsByStatus(ctx context.Context, organizationID string, statuses []TenantStatus) (out int, err error) {
	start := time.Now()
	defer observeMeta(ctx, "count_free_tidbcloud_pool_bindings", start, &err)
	organizationID = strings.TrimSpace(organizationID)
	if organizationID == "" {
		return 0, fmt.Errorf("organization_id is required")
	}
	if len(statuses) == 0 {
		return 0, fmt.Errorf("tenant statuses are required")
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(statuses)), ",")
	args := make([]any, 0, 6+2*len(statuses))
	args = append(args, organizationID, TenantPoolBindingFree, tidbCloudNativeProvider)
	for _, status := range statuses {
		args = append(args, status)
	}
	args = append(args, organizationID, TenantPoolBindingFree, tidbCloudNativeSharedProvider)
	for _, status := range statuses {
		args = append(args, status)
	}
	row := s.db.QueryRowContext(ctx, fmt.Sprintf(countFreeTenantPoolBindingsSQL, placeholders, placeholders), args...)
	if err = row.Scan(&out); err != nil {
		return 0, fmt.Errorf("count free tidbcloud pool bindings for organization %s: %w", organizationID, err)
	}
	return out, nil
}

// Count raw durable bindings first, then subtract deleted-tenant cleanup debt by
// driving from the selective tenant status index. Tenant rows are durable control-
// plane records; production deletion marks them deleted instead of hard-deleting
// them, so this avoids a tenant lookup for every live binding without counting debt.
const countTenantPoolBindingsByStatusSQL = `WITH binding_deltas AS (
	SELECT organization_id, pool_id, pool_status, COUNT(*) AS delta
		FROM tenant_tidbcloud_org_bindings
		WHERE pool_id <> ''
		GROUP BY organization_id, pool_id, pool_status
	UNION ALL
	SELECT tidbcloud_organization_id AS organization_id, pool_id, pool_status, COUNT(*) AS delta
		FROM tenant_pool_memberships
		WHERE pool_id <> ''
		GROUP BY tidbcloud_organization_id, pool_id, pool_status
	UNION ALL
	SELECT b.organization_id, b.pool_id, b.pool_status, -COUNT(*) AS delta
		FROM tenants t
		STRAIGHT_JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = t.id
		WHERE t.status = ? AND b.pool_id <> ''
		GROUP BY b.organization_id, b.pool_id, b.pool_status
	UNION ALL
	SELECT m.tidbcloud_organization_id AS organization_id, m.pool_id, m.pool_status, -COUNT(*) AS delta
		FROM tenants t
		STRAIGHT_JOIN tenant_pool_memberships m ON m.tenant_id = t.id
		WHERE t.status = ? AND m.pool_id <> ''
		GROUP BY m.tidbcloud_organization_id, m.pool_id, m.pool_status
), binding_counts AS (
	SELECT organization_id, pool_id, pool_status, SUM(delta) AS binding_count
		FROM binding_deltas
		GROUP BY organization_id, pool_id, pool_status
)
SELECT
	p.pool_id,
	p.organization_id,
	statuses.pool_status,
	COALESCE(binding_counts.binding_count, 0)
	FROM tenant_tidbcloud_pools p
	CROSS JOIN (
		SELECT ? AS pool_status
		UNION ALL
		SELECT ? AS pool_status
	) statuses
	LEFT JOIN binding_counts
		ON binding_counts.pool_id = p.pool_id
		AND binding_counts.organization_id = p.organization_id
		AND binding_counts.pool_status = statuses.pool_status
	WHERE p.pool_id <> ''
		AND p.organization_id IS NOT NULL
		AND p.organization_id <> ''
	ORDER BY p.pool_id, p.organization_id, statuses.pool_status`

func (s *Store) CountTenantPoolBindingsByStatus(ctx context.Context) (out []TenantPoolBindingStatusCount, err error) {
	start := time.Now()
	defer observeMeta(ctx, "count_tidbcloud_pool_bindings_by_status", start, &err)
	rows, err := s.db.QueryContext(ctx, countTenantPoolBindingsByStatusSQL,
		TenantDeleted, TenantDeleted, TenantPoolBindingFree, TenantPoolBindingUsed)
	if err != nil {
		return nil, fmt.Errorf("count tenant pool bindings by status query: %w", err)
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var rec TenantPoolBindingStatusCount
		if err := rows.Scan(&rec.PoolID, &rec.OrganizationID, &rec.Status, &rec.Count); err != nil {
			return nil, fmt.Errorf("count tenant pool bindings by status scan: %w", err)
		}
		out = append(out, rec)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("count tenant pool bindings by status rows: %w", err)
	}
	return out, nil
}

func (s *Store) ListFreeTenantPoolBindings(ctx context.Context, organizationID string, newestFirst bool, limit int) (out []TenantWithTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_free_tidbcloud_pool_bindings", start, &err)
	return s.listFreeTenantPoolBindings(ctx, organizationID, newestFirst, limit, []TenantStatus{TenantActive})
}

func (s *Store) ListFreeTenantPoolBindingsForDelete(ctx context.Context, organizationID string, newestFirst bool, limit int) (out []TenantWithTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_free_tidbcloud_pool_bindings_for_delete", start, &err)
	return s.listFreeTenantPoolBindings(ctx, organizationID, newestFirst, limit, purgeVisibleTenantStatuses)
}

func (s *Store) ListTenantPoolFreeSlotsForDelete(ctx context.Context, organizationID string, newestFirst bool, limit int) (out []TenantWithTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tidbcloud_pool_free_slots_for_delete", start, &err)
	return s.listFreeTenantPoolBindings(ctx, organizationID, newestFirst, limit, []TenantStatus{TenantPending, TenantProvisioning, TenantActive})
}

// ListUnconfirmedFreeTenantPoolDeleteIDs returns free native tenants whose
// cluster delete still needs to be re-driven with customer credentials.
func (s *Store) ListUnconfirmedFreeTenantPoolDeleteIDs(ctx context.Context, organizationID, poolID string, limit int) (out []string, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_unconfirmed_free_tenant_pool_delete_ids", start, &err)
	organizationID = strings.TrimSpace(organizationID)
	if organizationID == "" {
		return nil, fmt.Errorf("organization_id is required")
	}
	poolID = strings.TrimSpace(poolID)
	if poolID == "" {
		return nil, fmt.Errorf("pool_id is required")
	}
	if limit <= 0 {
		limit = 100
	}
	rows, err := s.db.QueryContext(ctx, `SELECT t.id
		FROM tenant_tidbcloud_org_bindings b
		JOIN tenants t ON t.id = b.tenant_id
		LEFT JOIN tenant_delete_jobs j ON j.tenant_id = t.id
		WHERE b.organization_id = ? AND b.pool_id = ? AND b.pool_status = ? AND t.provider = ? AND t.status = ?
			AND (COALESCE(t.storage_namespace_id, '') = '' OR j.tenant_id IS NULL)
		ORDER BY b.created_at DESC, b.tenant_id DESC
		LIMIT ?`, organizationID, poolID, TenantPoolBindingFree, tidbCloudNativeProvider, TenantDeleting, limit)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var tenantID string
		if err := rows.Scan(&tenantID); err != nil {
			return nil, err
		}
		out = append(out, tenantID)
	}
	return out, rows.Err()
}

func (s *Store) ListPendingTenantPoolBindingsForResume(ctx context.Context, organizationID string, limit int) (out []TenantWithTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_pending_tidbcloud_pool_bindings_for_resume", start, &err)
	return s.listFreeTenantPoolBindings(ctx, organizationID, false, limit, []TenantStatus{TenantPending})
}

func (s *Store) listFreeTenantPoolBindings(ctx context.Context, organizationID string, newestFirst bool, limit int, statuses []TenantStatus) (out []TenantWithTiDBCloudOrgBinding, err error) {
	organizationID = strings.TrimSpace(organizationID)
	if organizationID == "" {
		return nil, fmt.Errorf("organization_id is required")
	}
	if len(statuses) == 0 {
		return nil, fmt.Errorf("tenant statuses are required")
	}
	if limit <= 0 {
		limit = 100
	}
	order := "ASC"
	if newestFirst {
		order = "DESC"
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(statuses)), ",")
	query := `SELECT
			t.id, t.display_name, t.label, t.status, t.kind, t.parent_tenant_id, t.storage_namespace_id,
			t.db_host, t.db_port, t.db_user, t.db_password, t.db_name,
			t.db_tls, t.provider, t.cluster_id, t.branch_id, t.claim_url, t.claim_expires_at, t.schema_version,
			t.s3_encryption_mode, t.s3_kms_key_id, t.s3_bucket_key_enabled, t.created_at, t.updated_at,
			b.tenant_id, b.organization_id, b.cluster_id, b.branch_id, b.pool_id, b.pool_status, b.used_at, b.created_at, b.updated_at
		FROM tenant_tidbcloud_org_bindings b
		JOIN tenants t ON t.id = b.tenant_id
			WHERE b.organization_id = ? AND b.pool_status = ? AND t.provider = ?
				AND t.status IN (` + placeholders + `)
			ORDER BY b.created_at ` + order + `, b.tenant_id ` + order + `
			LIMIT ?`
	args := make([]any, 0, 4+len(statuses))
	args = append(args, organizationID, TenantPoolBindingFree, tidbCloudNativeProvider)
	for _, status := range statuses {
		args = append(args, status)
	}
	args = append(args, limit)
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	return scanTenantBindingRows(rows)
}

func (s *Store) ClaimOldestFreeTenantPoolBinding(ctx context.Context, organizationID string) (out *TenantWithTiDBCloudOrgBinding, err error) {
	return s.claimFreeTenantPoolBinding(ctx, organizationID, "", QuotaConfigPatch{}, TenantMetadata{}, false)
}

func (s *Store) ClaimFreeTenantPoolBinding(ctx context.Context, organizationID, tenantID string, quota QuotaConfigPatch) (out *TenantWithTiDBCloudOrgBinding, err error) {
	return s.ClaimFreeTenantPoolBindingWithMetadata(ctx, organizationID, tenantID, quota, TenantMetadata{})
}

func (s *Store) ClaimFreeTenantPoolBindingWithMetadata(ctx context.Context, organizationID, tenantID string, quota QuotaConfigPatch, metadata TenantMetadata) (out *TenantWithTiDBCloudOrgBinding, err error) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return nil, fmt.Errorf("tenant_id is required")
	}
	return s.claimFreeTenantPoolBinding(ctx, organizationID, tenantID, quota, metadata, false)
}

// ClaimFreeTenantPoolBindingWithMetadataForFreeOrganization claims a native
// pool tenant and shadow-writes its legacy classification for rollback.
func (s *Store) ClaimFreeTenantPoolBindingWithMetadataForFreeOrganization(ctx context.Context, organizationID, tenantID string, quota QuotaConfigPatch, metadata TenantMetadata) (out *TenantWithTiDBCloudOrgBinding, err error) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return nil, fmt.Errorf("tenant_id is required")
	}
	return s.claimFreeTenantPoolBinding(ctx, organizationID, tenantID, quota, metadata, true)
}

func (s *Store) claimFreeTenantPoolBinding(ctx context.Context, organizationID, tenantID string, quota QuotaConfigPatch, metadata TenantMetadata, freeOrganization bool) (out *TenantWithTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "claim_free_tidbcloud_pool_binding", start, &err)
	organizationID = strings.TrimSpace(organizationID)
	if organizationID == "" {
		return nil, fmt.Errorf("organization_id is required")
	}
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		query := `SELECT
					t.id, t.display_name, t.label, t.status, t.kind, t.parent_tenant_id, t.storage_namespace_id,
				t.db_host, t.db_port, t.db_user, t.db_password, t.db_name,
				t.db_tls, t.provider, t.cluster_id, t.branch_id, t.claim_url, t.claim_expires_at, t.schema_version,
				t.s3_encryption_mode, t.s3_kms_key_id, t.s3_bucket_key_enabled, t.created_at, t.updated_at,
				b.tenant_id, b.organization_id, b.cluster_id, b.branch_id, b.pool_id, b.pool_status, b.used_at, b.created_at, b.updated_at
			FROM tenant_tidbcloud_org_bindings b
			JOIN tenants t ON t.id = b.tenant_id
				WHERE b.organization_id = ? AND b.pool_status = ? AND t.provider = ?
					AND t.status = ?`
		args := []any{organizationID, TenantPoolBindingFree, tidbCloudNativeProvider, TenantActive}
		if tenantID != "" {
			query += ` AND t.id = ?`
			args = append(args, tenantID)
		}
		query += `
				ORDER BY b.created_at ASC, b.tenant_id ASC
				LIMIT 1 FOR UPDATE`
		row := tx.QueryRowContext(ctx, query, args...)
		rec, scanErr := scanTenantBindingRow(row)
		if scanErr != nil {
			return scanErr
		}
		if quota.MaxMediaLLMFiles != nil || quota.MaxVideoLLMFiles != nil {
			if err := s.validateQuotaExtractConfigsTx(ctx, tx, rec.Tenant.ID, quota); err != nil {
				return err
			}
		}
		if err := upsertTenantQuotaPatchTx(ctx, tx, rec.Tenant.ID, quota); err != nil {
			return err
		}
		labelJSON, err := tenantLabelJSON(metadata.Label)
		if err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `UPDATE tenants SET display_name = ?, label = ?, updated_at = ? WHERE id = ?`,
			strings.TrimSpace(metadata.DisplayName), labelJSON, time.Now().UTC(), rec.Tenant.ID); err != nil {
			return err
		}
		rec.Tenant.DisplayName = strings.TrimSpace(metadata.DisplayName)
		rec.Tenant.Label = cloneStringMap(metadata.Label)
		if rec.Tenant.Label == nil {
			rec.Tenant.Label = map[string]string{}
		}
		now := time.Now().UTC()
		res, execErr := tx.ExecContext(ctx, `UPDATE tenant_tidbcloud_org_bindings
			SET pool_status = ?, used_at = ?, updated_at = ?
			WHERE tenant_id = ? AND pool_status = ?`,
			TenantPoolBindingUsed, now, now, rec.Binding.TenantID, TenantPoolBindingFree)
		if execErr != nil {
			return execErr
		}
		n, _ := res.RowsAffected()
		if n == 0 {
			return ErrNotFound
		}
		rec.Binding.PoolStatus = TenantPoolBindingUsed
		rec.Binding.UsedAt = &now
		rec.Binding.UpdatedAt = now
		if freeOrganization || quota.TiDBCloudSpendingLimit == nil {
			if err := setLegacyTiDBCloudFreeClassificationTx(ctx, tx, rec.Tenant.ID, freeOrganization); err != nil {
				return err
			}
		}
		out = rec
		return nil
	})
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, err
	}
	return out, nil
}

func (s *Store) UpdateTenantPoolSize(ctx context.Context, poolID string, size int) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tidbcloud_pool_size", start, &err)
	poolID = strings.TrimSpace(poolID)
	if poolID == "" {
		return fmt.Errorf("pool_id is required")
	}
	if size < 0 {
		return fmt.Errorf("pool size must be non-negative")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_tidbcloud_pools SET size = ?, updated_at = ? WHERE pool_id = ?`, size, time.Now().UTC(), poolID)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Store) UpdateTenantPoolStatus(ctx context.Context, poolID string, status TenantPoolStatus) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tidbcloud_pool_status", start, &err)
	poolID = strings.TrimSpace(poolID)
	if poolID == "" {
		return fmt.Errorf("pool_id is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_tidbcloud_pools SET status = ?, updated_at = ? WHERE pool_id = ?`,
		tenantPoolStatusForInsert(status), time.Now().UTC(), poolID)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Store) DeleteTenantPool(ctx context.Context, poolID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_tidbcloud_pool", start, &err)
	poolID = strings.TrimSpace(poolID)
	if poolID == "" {
		return fmt.Errorf("pool_id is required")
	}
	res, err := s.db.ExecContext(ctx, `DELETE FROM tenant_tidbcloud_pools WHERE pool_id = ?`, poolID)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Store) DetachUsedTenantPoolBindings(ctx context.Context, poolID string) error {
	_, err := s.db.ExecContext(ctx, `UPDATE tenant_tidbcloud_org_bindings
		SET pool_id = '', updated_at = ? WHERE pool_id = ? AND pool_status = ?`,
		time.Now().UTC(), strings.TrimSpace(poolID), TenantPoolBindingUsed)
	return err
}

// DeleteTenantPoolAndDetachUsedMembers atomically removes logical-pool
// ownership from used native/shared members and deletes the logical pool row.
func (s *Store) DeleteTenantPoolAndDetachUsedMembers(ctx context.Context, poolID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_tenant_pool_and_detach_used_members", start, &err)
	poolID = strings.TrimSpace(poolID)
	if poolID == "" {
		return fmt.Errorf("pool_id is required")
	}
	// Native deleting members cannot be resumed without caller-held cloud
	// credentials, so their free bindings remain the recovery handle until
	// cluster deletion is confirmed. Shared deleting members are different:
	// the leader reconciler can re-run their idempotent purge, so their
	// memberships remain reachable until purge completion. Drop only terminal
	// or handed-off native bindings and missing/deleted shared memberships here.
	// Both predicates are positive so every other live status fails closed.
	if _, err := s.db.ExecContext(ctx, `DELETE b FROM tenant_tidbcloud_org_bindings b
		LEFT JOIN tenants t ON t.id = b.tenant_id
		LEFT JOIN tenant_delete_jobs j ON j.tenant_id = t.id
		WHERE b.pool_id = ? AND b.pool_status = ?
			AND (t.id IS NULL OR t.status = ?
				OR (t.status = ? AND j.tenant_id IS NOT NULL))`,
		poolID, TenantPoolBindingFree, TenantDeleted, TenantDeleting); err != nil {
		return err
	}
	if _, err := s.db.ExecContext(ctx, `DELETE m FROM tenant_pool_memberships m
		LEFT JOIN tenants t ON t.id = m.tenant_id
		WHERE m.pool_id = ? AND m.pool_status = ?
			AND (t.id IS NULL OR t.status = ?)`,
		poolID, TenantPoolBindingFree, TenantDeleted); err != nil {
		return err
	}
	return s.InTx(ctx, func(tx *sql.Tx) error {
		var lockedPoolID string
		if err := tx.QueryRowContext(ctx, `SELECT pool_id FROM tenant_tidbcloud_pools
			WHERE pool_id = ? FOR UPDATE`, poolID).Scan(&lockedPoolID); errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		} else if err != nil {
			return err
		}
		var freeTenantID string
		if err := tx.QueryRowContext(ctx, `SELECT tenant_id FROM tenant_tidbcloud_org_bindings
			WHERE pool_id = ? AND pool_status = ? LIMIT 1 FOR UPDATE`,
			poolID, TenantPoolBindingFree).Scan(&freeTenantID); err == nil {
			return fmt.Errorf("tenant pool %q still has free native member %q: %w", poolID, freeTenantID, ErrTenantPoolHasFreeMembers)
		} else if !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		if err := tx.QueryRowContext(ctx, `SELECT tenant_id FROM tenant_pool_memberships
			WHERE pool_id = ? AND pool_status = ? LIMIT 1 FOR UPDATE`,
			poolID, TenantPoolBindingFree).Scan(&freeTenantID); err == nil {
			return fmt.Errorf("tenant pool %q still has free shared member %q: %w", poolID, freeTenantID, ErrTenantPoolHasFreeMembers)
		} else if !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		now := time.Now().UTC()
		if _, err := tx.ExecContext(ctx, `UPDATE tenant_tidbcloud_org_bindings
			SET pool_id = '', updated_at = ? WHERE pool_id = ? AND pool_status = ?`,
			now, poolID, TenantPoolBindingUsed); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM tenant_pool_memberships
			WHERE pool_id = ? AND pool_status = ?`, poolID, TenantPoolBindingUsed); err != nil {
			return err
		}
		res, err := tx.ExecContext(ctx, `DELETE FROM tenant_tidbcloud_pools WHERE pool_id = ?`, poolID)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n == 0 {
			return ErrNotFound
		}
		return nil
	})
}

func (s *Store) UpdateTenantPoolBindingStatus(ctx context.Context, tenantID string, status TenantPoolBindingStatus, usedAt *time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tidbcloud_pool_binding_status", start, &err)
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return fmt.Errorf("tenant_id is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_tidbcloud_org_bindings
		SET pool_status = ?, used_at = ?, updated_at = ? WHERE tenant_id = ?`,
		tenantPoolBindingStatusForInsert(status), usedAt, time.Now().UTC(), tenantID)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Store) ReleaseTenantPoolClaim(ctx context.Context, tenantID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "release_tidbcloud_pool_claim", start, &err)
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return fmt.Errorf("tenant_id is required")
	}
	return s.InTx(ctx, func(tx *sql.Tx) error {
		res, err := tx.ExecContext(ctx, `UPDATE tenant_tidbcloud_org_bindings
			SET pool_status = ?, used_at = NULL, updated_at = ?
			WHERE tenant_id = ? AND pool_status = ?`,
			TenantPoolBindingFree, time.Now().UTC(), tenantID, TenantPoolBindingUsed)
		if err != nil {
			return err
		}
		if err := requireAffected(res); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM tenant_quota_config WHERE tenant_id = ?`, tenantID); err != nil {
			return err
		}
		return nil
	})
}

func (s *Store) MarkFreeTenantPoolTenantDeleting(ctx context.Context, tenantID string, from TenantStatus) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_free_tidbcloud_pool_tenant_deleting", 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 id string
		row := tx.QueryRowContext(ctx, `SELECT t.id
			FROM tenants t
			JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = t.id
			WHERE t.id = ? AND t.status = ? AND b.pool_status = ?
			LIMIT 1 FOR UPDATE`, tenantID, from, TenantPoolBindingFree)
		if scanErr := row.Scan(&id); scanErr != nil {
			return scanErr
		}
		res, execErr := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
			TenantDeleting, time.Now().UTC(), tenantID, from)
		if execErr != nil {
			return execErr
		}
		n, _ := res.RowsAffected()
		updated = n > 0
		if updated {
			return insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkMetricsCleanup)
		}
		return nil
	})
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	if updated {
		s.apiKeys.evictTenant(tenantID)
	}
	return updated, nil
}

func (s *Store) ListTenantsByTiDBCloudOrganizations(ctx context.Context, organizationIDs []string, offset, limit int) (out []TenantWithTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenants_by_tidbcloud_orgs", start, &err)
	organizationIDs = compactStrings(organizationIDs)
	if len(organizationIDs) == 0 {
		return []TenantWithTiDBCloudOrgBinding{}, nil
	}
	if offset < 0 {
		offset = 0
	}
	if limit <= 0 {
		limit = 10
	}
	placeholders := make([]string, len(organizationIDs))
	args := make([]any, 0, len(organizationIDs)+5)
	for i, id := range organizationIDs {
		placeholders[i] = "?"
		args = append(args, id)
	}
	args = append(args, tidbCloudNativeProvider, TenantPoolBindingFree, TenantDeleted, limit, offset)
	query := `SELECT
			t.id, t.display_name, t.label, t.status, t.kind, t.parent_tenant_id, t.storage_namespace_id,
			t.db_host, t.db_port, t.db_user, t.db_password, t.db_name,
			t.db_tls, t.provider, t.cluster_id, t.branch_id, t.claim_url, t.claim_expires_at, t.schema_version,
			t.s3_encryption_mode, t.s3_kms_key_id, t.s3_bucket_key_enabled, t.created_at, t.updated_at,
			b.tenant_id, b.organization_id, b.cluster_id, b.branch_id, b.pool_id, b.pool_status, b.used_at, b.created_at, b.updated_at
			FROM tenant_tidbcloud_org_bindings b
			JOIN tenants t ON t.id = b.tenant_id
			WHERE b.organization_id IN (` + strings.Join(placeholders, ",") + `)
				AND t.provider = ?
				AND b.pool_status <> ?
				AND t.status <> ?
			ORDER BY t.created_at DESC, t.id DESC
			LIMIT ? OFFSET ?`
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	return scanTenantBindingRows(rows)
}

func (s *Store) ListTenantsByTiDBCloudOrgClusterBindings(ctx context.Context, bindings []TenantTiDBCloudOrgBinding, offset, limit int) (out []TenantWithTiDBCloudOrgBinding, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenants_by_tidbcloud_org_clusters", start, &err)
	bindings = compactTiDBCloudOrgClusterBindings(bindings)
	if len(bindings) == 0 {
		return []TenantWithTiDBCloudOrgBinding{}, nil
	}
	if offset < 0 {
		offset = 0
	}
	if limit <= 0 {
		limit = 10
	}
	placeholders := make([]string, 0, len(bindings))
	args := make([]any, 0, len(bindings)*2+5)
	for _, binding := range bindings {
		placeholders = append(placeholders, "(?, ?)")
		args = append(args, binding.OrganizationID, binding.ClusterID)
	}
	args = append(args, tidbCloudNativeProvider, TenantPoolBindingFree, TenantDeleted, limit, offset)
	query := `SELECT
				t.id, t.display_name, t.label, t.status, t.kind, t.parent_tenant_id, t.storage_namespace_id,
			t.db_host, t.db_port, t.db_user, t.db_password, t.db_name,
			t.db_tls, t.provider, t.cluster_id, t.branch_id, t.claim_url, t.claim_expires_at, t.schema_version,
			t.s3_encryption_mode, t.s3_kms_key_id, t.s3_bucket_key_enabled, t.created_at, t.updated_at,
			b.tenant_id, b.organization_id, b.cluster_id, b.branch_id, b.pool_id, b.pool_status, b.used_at, b.created_at, b.updated_at
			FROM tenant_tidbcloud_org_bindings b
			JOIN tenants t ON t.id = b.tenant_id
			WHERE (b.organization_id, b.cluster_id) IN (` + strings.Join(placeholders, ",") + `)
				AND t.provider = ?
				AND b.pool_status <> ?
				AND t.status <> ?
			ORDER BY t.created_at DESC, t.id DESC
			LIMIT ? OFFSET ?`
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	return scanTenantBindingRows(rows)
}

// ListTenantsByTiDBCloudOrganization lists dedicated and shared tenants owned
// by one IAM-authenticated TiDB Cloud organization. Pagination is applied
// after both provider sources are combined.
func (s *Store) ListTenantsByTiDBCloudOrganization(ctx context.Context, organizationID string, offset, limit int) (out []Tenant, err error) {
	return s.ListTenantsByTiDBCloudOrganizationFiltered(ctx, organizationID, TenantListFilter{}, offset, limit)
}

// ListTenantsByTiDBCloudOrganizationFiltered lists dedicated and shared
// tenants owned by one organization, applying metadata filters before
// pagination. DisplayNameContains searches the effective display name, which
// falls back to the tenant ID when display_name is empty.
func (s *Store) ListTenantsByTiDBCloudOrganizationFiltered(ctx context.Context, organizationID string, filter TenantListFilter, offset, limit int) (out []Tenant, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenants_by_tidbcloud_organization", start, &err)
	organizationID = strings.TrimSpace(organizationID)
	if organizationID == "" {
		return []Tenant{}, nil
	}
	if offset < 0 {
		offset = 0
	}
	if limit <= 0 {
		limit = 10
	}
	args := []any{TenantDeleted, tidbCloudNativeProvider, organizationID, TenantPoolBindingFree,
		tidbCloudNativeSharedProvider, organizationID, TenantPoolBindingFree}
	query := `SELECT
			t.id, t.display_name, t.label, t.status, t.kind, t.parent_tenant_id, t.storage_namespace_id,
			t.db_host, t.db_port, t.db_user, t.db_password, t.db_name,
			t.db_tls, t.provider, t.cluster_id, t.branch_id, t.claim_url, t.claim_expires_at, t.schema_version,
			t.s3_encryption_mode, t.s3_kms_key_id, t.s3_bucket_key_enabled, t.created_at, t.updated_at
		FROM tenants t
		WHERE t.status <> ? AND ` + tenantTiDBCloudOrganizationVisibilityPredicateSQL
	if filter.DisplayNameContains != "" {
		query += ` AND COALESCE(NULLIF(t.display_name, ''), t.id) LIKE ?`
		args = append(args, "%"+filter.DisplayNameContains+"%")
	}
	if filter.LabelKey != "" {
		query += ` AND JSON_CONTAINS(COALESCE(t.label, JSON_OBJECT()), JSON_OBJECT(?, ?))`
		args = append(args, filter.LabelKey, filter.LabelValue)
	}
	query += ` ORDER BY t.created_at DESC, t.id DESC LIMIT ? OFFSET ?`
	args = append(args, limit, offset)
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	return scanTenantRows(rows)
}

func (s *Store) TenantDisplayNameExistsInTiDBCloudOrganization(ctx context.Context, organizationID, displayName string) (bool, error) {
	organizationID = strings.TrimSpace(organizationID)
	displayName = strings.TrimSpace(displayName)
	if organizationID == "" || displayName == "" {
		return false, nil
	}
	var exists bool
	err := s.db.QueryRowContext(ctx, `SELECT EXISTS(
		SELECT 1 FROM tenants t
		WHERE t.status <> ? AND COALESCE(NULLIF(t.display_name, ''), t.id) = ? AND `+tenantTiDBCloudOrganizationVisibilityPredicateSQL+`
	)`, TenantDeleted, displayName, tidbCloudNativeProvider, organizationID, TenantPoolBindingFree,
		tidbCloudNativeSharedProvider, organizationID, TenantPoolBindingFree).Scan(&exists)
	return exists, err
}

func compactStrings(values []string) []string {
	out := make([]string, 0, len(values))
	seen := make(map[string]bool, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" || seen[value] {
			continue
		}
		seen[value] = true
		out = append(out, value)
	}
	sort.Strings(out)
	return out
}

func compactTiDBCloudOrgClusterBindings(bindings []TenantTiDBCloudOrgBinding) []TenantTiDBCloudOrgBinding {
	out := make([]TenantTiDBCloudOrgBinding, 0, len(bindings))
	seen := make(map[string]bool, len(bindings))
	for _, binding := range bindings {
		orgID := strings.TrimSpace(binding.OrganizationID)
		clusterID := strings.TrimSpace(binding.ClusterID)
		if orgID == "" || clusterID == "" {
			continue
		}
		key := orgID + "\x00" + clusterID
		if seen[key] {
			continue
		}
		seen[key] = true
		out = append(out, TenantTiDBCloudOrgBinding{OrganizationID: orgID, ClusterID: clusterID})
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].OrganizationID != out[j].OrganizationID {
			return out[i].OrganizationID < out[j].OrganizationID
		}
		return out[i].ClusterID < out[j].ClusterID
	})
	return out
}

func (s *Store) WithExternalBindingLock(ctx context.Context, provider, subjectKey string, fn func(context.Context) error) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "external_binding_lock", start, &err)
	if fn == nil {
		return fmt.Errorf("external binding lock callback is required")
	}
	lockName := externalBindingLockName(provider, subjectKey)
	conn, err := s.db.Conn(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = conn.Close() }()

	var got sql.NullInt64
	if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(CONCAT(?, DATABASE()), ?)", lockName, externalBindingLockTimeoutSeconds).Scan(&got); err != nil {
		return err
	}
	if !got.Valid {
		return fmt.Errorf("external binding named lock returned NULL")
	}
	if got.Int64 != 1 {
		return fmt.Errorf("timed out waiting for external binding named lock")
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), externalBindingReleaseLockTimeout)
		defer cancel()
		var released sql.NullInt64
		releaseErr := conn.QueryRowContext(releaseCtx, "SELECT RELEASE_LOCK(CONCAT(?, DATABASE()))", lockName).Scan(&released)
		if releaseErr != nil {
			err = errors.Join(err, releaseErr)
			return
		}
		if !released.Valid {
			err = errors.Join(err, fmt.Errorf("external binding named lock release returned NULL"))
			return
		}
		if released.Int64 != 1 {
			err = errors.Join(err, fmt.Errorf("external binding named lock was not held by current connection"))
		}
	}()

	return fn(ctx)
}

func (s *Store) WithTenantPoolLock(ctx context.Context, poolID string, fn func(context.Context) error) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "tenant_pool_lock", start, &err)
	if fn == nil {
		return fmt.Errorf("tenant pool lock callback is required")
	}
	lockName := tenantPoolLockName(poolID)
	if lockName == "" {
		return fmt.Errorf("pool_id is required")
	}
	conn, err := s.db.Conn(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = conn.Close() }()

	var databaseName sql.NullString
	if err := conn.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&databaseName); err != nil {
		return err
	}
	lockName = tenantPoolDatabaseLockName(lockName, databaseName.String)

	var got sql.NullInt64
	if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", lockName, tenantPoolLockTimeoutSeconds).Scan(&got); err != nil {
		return err
	}
	if !got.Valid {
		return fmt.Errorf("tenant pool named lock returned NULL")
	}
	if got.Int64 != 1 {
		return fmt.Errorf("timed out waiting for tenant pool named lock")
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), tenantPoolReleaseLockTimeout)
		defer cancel()
		var released sql.NullInt64
		releaseErr := conn.QueryRowContext(releaseCtx, "SELECT RELEASE_LOCK(?)", lockName).Scan(&released)
		if releaseErr != nil {
			err = errors.Join(err, releaseErr)
			return
		}
		if !released.Valid {
			err = errors.Join(err, fmt.Errorf("tenant pool named lock release returned NULL"))
			return
		}
		if released.Int64 != 1 {
			err = errors.Join(err, fmt.Errorf("tenant pool named lock was not held by current connection"))
		}
	}()

	return fn(ctx)
}

// WithSharedDBAllocationLock serializes compact DB-pool selection and durable
// pool planning for one TiDB Cloud organization or unresolved credential
// fingerprint. Physical Cloud work uses the per-db_pool work locks below so
// already-planned pools in one organization do not serialize each other.
func (s *Store) WithSharedDBAllocationLock(ctx context.Context, identity string, fn func(context.Context) error) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "shared_db_allocation_lock", start, &err)
	if strings.TrimSpace(identity) == "" {
		return fmt.Errorf("shared db allocation identity is required")
	}
	if fn == nil {
		return fmt.Errorf("shared db allocation lock callback is required")
	}
	sum := sha256.Sum256([]byte(identity))
	lockName := "d9_dbpool:" + hex.EncodeToString(sum[:20])
	conn, err := s.db.Conn(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = conn.Close() }()
	var databaseName sql.NullString
	if err := conn.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&databaseName); err != nil {
		return err
	}
	lockName = tenantPoolDatabaseLockName(lockName, databaseName.String)
	var got sql.NullInt64
	if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", lockName, tenantPoolLockTimeoutSeconds).Scan(&got); err != nil {
		return err
	}
	if !got.Valid {
		return fmt.Errorf("shared db allocation named lock returned NULL")
	}
	if got.Int64 != 1 {
		return fmt.Errorf("timed out waiting for shared db allocation named lock")
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), tenantPoolReleaseLockTimeout)
		defer cancel()
		var released sql.NullInt64
		releaseErr := conn.QueryRowContext(releaseCtx, "SELECT RELEASE_LOCK(?)", lockName).Scan(&released)
		if releaseErr != nil {
			err = errors.Join(err, releaseErr)
			return
		}
		if !released.Valid || released.Int64 != 1 {
			err = errors.Join(err, fmt.Errorf("shared db allocation named lock was not held by current connection"))
		}
	}()
	return fn(ctx)
}

// WithSharedDBPoolWorkLock serializes one lifecycle attempt for a durable
// db_pool row. Unlike WithSharedDBAllocationLock, this lock is scoped to one
// physical pool, so unrelated pools in the same organization can enter Cloud
// create and continuation concurrently. The session-owned advisory lock is
// released automatically if its process or database connection disappears.
func (s *Store) WithSharedDBPoolWorkLock(ctx context.Context, dbID int64, fn func(context.Context) error) error {
	if fn == nil {
		return fmt.Errorf("shared db pool work lock callback is required")
	}
	return s.withSharedDBPoolWorkLocks(ctx, []int64{dbID}, tenantPoolLockTimeoutSeconds, true,
		func(lockCtx context.Context, _ []int64) error { return fn(lockCtx) })
}

// WithSharedDBPoolWorkClaims tries to own each supplied physical pool
// without waiting. The callback receives the subset this connection owns, so
// one busy pool never delays unrelated members of a Cloud batch. All claims
// remain held through the callback and are released together afterward.
func (s *Store) WithSharedDBPoolWorkClaims(ctx context.Context, dbIDs []int64, fn func(context.Context, []int64) error) error {
	if fn == nil {
		return fmt.Errorf("shared db pool work claims callback is required")
	}
	return s.withSharedDBPoolWorkLocks(ctx, dbIDs, 0, false, fn)
}

func (s *Store) withSharedDBPoolWorkLocks(ctx context.Context, dbIDs []int64, waitSeconds int, requireAll bool, fn func(context.Context, []int64) error) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "shared_db_pool_work_lock", start, &err)
	if len(dbIDs) == 0 {
		return nil
	}
	ids := append([]int64(nil), dbIDs...)
	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
	unique := ids[:0]
	for _, dbID := range ids {
		if dbID <= 0 {
			return fmt.Errorf("db_id must be positive")
		}
		if len(unique) == 0 || unique[len(unique)-1] != dbID {
			unique = append(unique, dbID)
		}
	}
	conn, err := s.db.Conn(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = conn.Close() }()
	var databaseName sql.NullString
	if err := conn.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&databaseName); err != nil {
		return err
	}
	lockNames := make([]string, 0, len(unique))
	ownedIDs := make([]int64, 0, len(unique))
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), tenantPoolReleaseLockTimeout)
		defer cancel()
		for i := len(lockNames) - 1; i >= 0; i-- {
			var released sql.NullInt64
			releaseErr := conn.QueryRowContext(releaseCtx, "SELECT RELEASE_LOCK(?)", lockNames[i]).Scan(&released)
			if releaseErr != nil {
				err = errors.Join(err, releaseErr)
				continue
			}
			if !released.Valid || released.Int64 != 1 {
				err = errors.Join(err, fmt.Errorf("shared db pool work named lock was not held by current connection"))
			}
		}
	}()
	for _, dbID := range unique {
		lockName := tenantPoolDatabaseLockName(fmt.Sprintf("d9_dbwork:%d", dbID), databaseName.String)
		var got sql.NullInt64
		if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", lockName, waitSeconds).Scan(&got); err != nil {
			return err
		}
		if !got.Valid {
			return fmt.Errorf("shared db pool work named lock returned NULL")
		}
		if got.Int64 != 1 {
			if requireAll {
				return fmt.Errorf("timed out waiting for shared db pool work named lock")
			}
			continue
		}
		lockNames = append(lockNames, lockName)
		ownedIDs = append(ownedIDs, dbID)
	}
	if len(ownedIDs) == 0 {
		return nil
	}
	return fn(ctx, ownedIDs)
}

func externalBindingLockName(provider, subjectKey string) string {
	sum := sha256.Sum256([]byte(provider + "\x00" + subjectKey))
	return "d9_extbind:" + hex.EncodeToString(sum[:20])
}

func tidbCloudOrgBindingLockName(organizationID, clusterID, branchID string) string {
	sum := sha256.Sum256([]byte(strings.TrimSpace(organizationID) + "\x00" + strings.TrimSpace(clusterID) + "\x00" + strings.TrimSpace(branchID)))
	return "d9_tidb_orgbind:" + hex.EncodeToString(sum[:16])
}

func tenantPoolLockName(poolID string) string {
	poolID = strings.TrimSpace(poolID)
	if poolID == "" {
		return ""
	}
	sum := sha256.Sum256([]byte(poolID))
	return "d9_tenant_pool:" + hex.EncodeToString(sum[:16])
}

func tenantPoolDatabaseLockName(baseLockName, databaseName string) string {
	sum := sha256.Sum256([]byte(strings.TrimSpace(databaseName)))
	return baseLockName + ":" + hex.EncodeToString(sum[:4])
}

// apiKeyResolveDBTimeout bounds the detached context used for the shared DB
// query inside the resolve singleflight (see ResolveByAPIKeyHash), so a slow
// metadb cannot park flight waiters forever.
const apiKeyResolveDBTimeout = 10 * time.Second

// ResolveByAPIKeyHash resolves a tenant and its API key by jwt_hash. Results
// are cached in-process for a short TTL (see api_key_cache.go) because the
// auth middleware calls this on every authenticated request; ErrNotFound is
// never cached so newly created keys are visible immediately.
func (s *Store) ResolveByAPIKeyHash(ctx context.Context, hash string) (out *TenantWithAPIKey, err error) {
	start := time.Now()
	defer observeMeta(ctx, "resolve_api_key_hash", start, &err)
	if cached, ok := s.apiKeys.get(hash); ok {
		return cached, nil
	}
	v, err, _ := s.apiKeys.flight.Do(hash, func() (any, error) {
		// Double-check under the flight window: a concurrent caller may have
		// filled the entry while this one waited.
		if cached, ok := s.apiKeys.lookup(hash); ok {
			return cached, nil
		}
		// Detach the shared query from the flight leader's request context:
		// every waiter shares this one query, so a leader's client disconnect
		// must not cancel it for all of them. It is bounded by its own
		// timeout instead.
		dbCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), apiKeyResolveDBTimeout)
		defer cancel()
		rec, dberr := s.resolveByAPIKeyHashDB(dbCtx, hash)
		if dberr != nil {
			return nil, dberr
		}
		s.apiKeys.fill(hash, cloneTenantWithAPIKey(*rec))
		return rec, nil
	})
	if err != nil {
		return nil, err
	}
	// The detached query protects the OTHER flight waiters from one caller's
	// disconnect; this caller itself keeps the pre-fix prompt-cancellation
	// behavior (the auth middleware maps it to 499 client-canceled).
	if ctx.Err() != nil {
		err = ctx.Err()
		return nil, err
	}
	// Clone per caller: flight waiters share the fetched record, but callers
	// may mutate it (tenant.Pool writes Tenant.StorageNamespaceID on a cold
	// open), so every caller must get a private copy — same as the hit path.
	rec := cloneTenantWithAPIKey(*v.(*TenantWithAPIKey))
	return &rec, nil
}

func (s *Store) resolveByAPIKeyHashDB(ctx context.Context, hash string) (out *TenantWithAPIKey, err error) {
	row := s.db.QueryRowContext(ctx, `SELECT
			t.id, t.display_name, t.label, t.status, t.kind, t.parent_tenant_id, t.storage_namespace_id,
			t.db_host, t.db_port, t.db_user, t.db_password, t.db_name, t.db_tls,
			t.provider, t.cluster_id, t.branch_id, t.claim_url, t.claim_expires_at, t.schema_version,
			t.s3_encryption_mode, t.s3_kms_key_id, t.s3_bucket_key_enabled, t.created_at, t.updated_at,
			k.id, k.tenant_id, k.key_name, k.jwt_ciphertext, k.jwt_hash, k.token_version, k.status, k.scope_kind,
			k.issued_by_provider, k.issued_by_subject_key, k.issued_by_metadata_json, k.issued_at,
			k.revoked_at, k.expires_at, k.created_at, k.updated_at,
			COALESCE(CASE WHEN t.provider = ? THEN d.org_id ELSE b.organization_id END, '')
		FROM tenant_api_keys k
		JOIN tenants t ON t.id = k.tenant_id
		LEFT JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = t.id
		LEFT JOIN fs_registry f ON f.tenant_id = t.id
		LEFT JOIN tenant_placements p ON p.fs_id = f.fs_id
		LEFT JOIN db_pool d ON d.db_id = p.db_id
		WHERE k.jwt_hash = ?`, tidbCloudNativeSharedProvider, hash)

	var rec TenantWithAPIKey
	var dbTLS int
	var claimURL sql.NullString
	var claimExp sql.NullTime
	var clusterID sql.NullString
	var parentTenantID sql.NullString
	var storageNamespaceID sql.NullString
	var revokedAt sql.NullTime
	var expiresAt sql.NullTime
	var s3BucketKeyEnabled int
	var issuedByProvider sql.NullString
	var issuedBySubjectKey sql.NullString
	var issuedByMetadataJSON []byte
	var labelJSON []byte
	if err = row.Scan(
		&rec.Tenant.ID, &rec.Tenant.DisplayName, &labelJSON, &rec.Tenant.Status, &rec.Tenant.Kind, &parentTenantID, &storageNamespaceID,
		&rec.Tenant.DBHost, &rec.Tenant.DBPort, &rec.Tenant.DBUser,
		&rec.Tenant.DBPasswordCipher, &rec.Tenant.DBName, &dbTLS, &rec.Tenant.Provider, &clusterID,
		&rec.Tenant.BranchID, &claimURL, &claimExp, &rec.Tenant.SchemaVersion, &rec.Tenant.S3EncryptionMode,
		&rec.Tenant.S3KMSKeyID, &s3BucketKeyEnabled, &rec.Tenant.CreatedAt, &rec.Tenant.UpdatedAt,
		&rec.APIKey.ID, &rec.APIKey.TenantID, &rec.APIKey.KeyName, &rec.APIKey.JWTCiphertext,
		&rec.APIKey.JWTHash, &rec.APIKey.TokenVersion, &rec.APIKey.Status, &rec.APIKey.ScopeKind,
		&issuedByProvider, &issuedBySubjectKey, &issuedByMetadataJSON, &rec.APIKey.IssuedAt,
		&revokedAt, &expiresAt, &rec.APIKey.CreatedAt, &rec.APIKey.UpdatedAt, &rec.TiDBCloudOrgID,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			err = ErrNotFound
			return nil, err
		}
		return nil, err
	}
	rec.Tenant.Label, err = scanTenantLabel(labelJSON)
	if err != nil {
		return nil, err
	}
	rec.Tenant.DBTLS = dbTLS == 1
	rec.Tenant.S3BucketKeyEnabled = boolPtr(s3BucketKeyEnabled == 1)
	if clusterID.Valid {
		rec.Tenant.ClusterID = clusterID.String
	}
	if parentTenantID.Valid {
		rec.Tenant.ParentTenantID = parentTenantID.String
	}
	if storageNamespaceID.Valid {
		rec.Tenant.StorageNamespaceID = storageNamespaceID.String
	}
	if claimURL.Valid {
		rec.Tenant.ClaimURL = claimURL.String
	}
	if claimExp.Valid {
		t := claimExp.Time.UTC()
		rec.Tenant.ClaimExpiresAt = &t
	}
	if revokedAt.Valid {
		t := revokedAt.Time.UTC()
		rec.APIKey.RevokedAt = &t
	}
	if expiresAt.Valid {
		t := expiresAt.Time.UTC()
		rec.APIKey.ExpiresAt = &t
	}
	rec.Tenant.TiDBCloudOrgID = rec.TiDBCloudOrgID
	if issuedByProvider.Valid {
		rec.APIKey.IssuedByProvider = issuedByProvider.String
	}
	if issuedBySubjectKey.Valid {
		rec.APIKey.IssuedBySubjectKey = issuedBySubjectKey.String
	}
	rec.APIKey.IssuedByMetadataJSON = issuedByMetadataJSON
	out = &rec
	return out, nil
}

func (s *Store) InsertAPIKeyFSScope(ctx context.Context, scope *APIKeyFSScope) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_api_key_fs_scope", start, &err)
	if scope == nil {
		return fmt.Errorf("fs scope is required")
	}
	prefix, err := canonicalFSScopePrefix(scope.Prefix)
	if err != nil {
		return err
	}
	if err := validateFSScopeOps(scope.Ops); err != nil {
		return err
	}
	if strings.TrimSpace(prefix) == "" {
		return fmt.Errorf("fs scope prefix is required")
	}
	prefixHash := fsScopePrefixHash(prefix)
	createdAt := scope.CreatedAt
	if createdAt.IsZero() {
		createdAt = time.Now().UTC()
	}
	updatedAt := scope.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = createdAt
	}
	_, err = s.db.ExecContext(ctx, `INSERT INTO tenant_api_key_fs_scopes
		(tenant_id, api_key_id, prefix, prefix_hash, ops, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?)`,
		scope.TenantID, scope.APIKeyID, prefix, prefixHash, scope.Ops,
		createdAt.UTC(), updatedAt.UTC())
	if isDuplicateEntry(err) {
		return ErrDuplicate
	}
	return err
}

func (s *Store) ListAPIKeyFSScopes(ctx context.Context, tenantID, apiKeyID string) (out []APIKeyFSScope, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_api_key_fs_scopes", start, &err)
	rows, err := s.db.QueryContext(ctx, `SELECT tenant_id, api_key_id, prefix, prefix_hash, ops, created_at, updated_at
		FROM tenant_api_key_fs_scopes
		WHERE tenant_id = ? AND api_key_id = ?
		ORDER BY prefix`, tenantID, apiKeyID)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var scope APIKeyFSScope
		if err := rows.Scan(&scope.TenantID, &scope.APIKeyID, &scope.Prefix, &scope.PrefixHash, &scope.Ops, &scope.CreatedAt, &scope.UpdatedAt); err != nil {
			return nil, err
		}
		out = append(out, scope)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func apiKeyScopeKindForInsert(k *APIKey) (APIKeyScopeKind, error) {
	if k == nil || k.ScopeKind == "" {
		return APIKeyScopeKindOwner, nil
	}
	if !isValidAPIKeyScopeKind(k.ScopeKind) {
		return "", fmt.Errorf("unsupported api key scope kind %q", k.ScopeKind)
	}
	return k.ScopeKind, nil
}

func isValidAPIKeyScopeKind(kind APIKeyScopeKind) bool {
	switch kind {
	case APIKeyScopeKindOwner, APIKeyScopeKindFS:
		return true
	default:
		return false
	}
}

func canonicalFSScopePrefix(raw string) (string, error) {
	if strings.TrimSpace(raw) == "" {
		return "", fmt.Errorf("fs scope prefix is required")
	}
	if strings.TrimSpace(raw) == ":" {
		return "", fmt.Errorf("fs scope prefix is required")
	}
	raw = strings.TrimPrefix(raw, ":")
	prefix, err := pathutil.Canonicalize(raw)
	if err != nil {
		return "", fmt.Errorf("invalid fs scope prefix: %w", err)
	}
	return prefix, nil
}

func validateFSScopeOps(raw string) error {
	ops := make(map[string]bool)
	hasPseudoRoot := false
	for _, part := range strings.Split(raw, ",") {
		op := strings.TrimSpace(part)
		if op == "" {
			continue
		}
		switch op {
		case "pseudoroot":
			hasPseudoRoot = true
		case "read", "list", "search", "write", "delete":
			ops[op] = true
		default:
			return fmt.Errorf("unknown fs scope op %q", op)
		}
	}
	if len(ops) == 0 && !hasPseudoRoot {
		return fmt.Errorf("empty fs scope ops")
	}
	if ops["search"] && !ops["read"] {
		return fmt.Errorf("search fs scope requires read")
	}
	return nil
}

func fsScopePrefixHash(prefix string) string {
	sum := sha256.Sum256([]byte(prefix))
	return hex.EncodeToString(sum[:])
}

func (s *Store) GetTenant(ctx context.Context, id string) (out *Tenant, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tenant", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT id, display_name, label, status, kind, parent_tenant_id, storage_namespace_id,
		db_host, db_port, db_user, db_password, db_name,
		db_tls, provider, cluster_id, branch_id, claim_url, claim_expires_at, schema_version,
		s3_encryption_mode, s3_kms_key_id, s3_bucket_key_enabled, created_at, updated_at
		FROM tenants WHERE id = ?`, id)
	var dbTLS int
	var clusterID sql.NullString
	var parentTenantID sql.NullString
	var storageNamespaceID sql.NullString
	var claimURL sql.NullString
	var claimExp sql.NullTime
	var s3BucketKeyEnabled int
	var labelJSON []byte
	var rec Tenant
	if err = row.Scan(&rec.ID, &rec.DisplayName, &labelJSON, &rec.Status, &rec.Kind, &parentTenantID, &storageNamespaceID,
		&rec.DBHost, &rec.DBPort, &rec.DBUser, &rec.DBPasswordCipher,
		&rec.DBName, &dbTLS, &rec.Provider, &clusterID, &rec.BranchID, &claimURL, &claimExp, &rec.SchemaVersion,
		&rec.S3EncryptionMode, &rec.S3KMSKeyID, &s3BucketKeyEnabled, &rec.CreatedAt, &rec.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			err = ErrNotFound
			return nil, err
		}
		return nil, err
	}
	rec.Label, err = scanTenantLabel(labelJSON)
	if err != nil {
		return nil, err
	}
	rec.DBTLS = dbTLS == 1
	rec.S3BucketKeyEnabled = boolPtr(s3BucketKeyEnabled == 1)
	if clusterID.Valid {
		rec.ClusterID = clusterID.String
	}
	if parentTenantID.Valid {
		rec.ParentTenantID = parentTenantID.String
	}
	if storageNamespaceID.Valid {
		rec.StorageNamespaceID = storageNamespaceID.String
	}
	if claimURL.Valid {
		rec.ClaimURL = claimURL.String
	}
	if claimExp.Valid {
		t := claimExp.Time.UTC()
		rec.ClaimExpiresAt = &t
	}
	out = &rec
	return out, nil
}

func (s *Store) ListTenantsByStatus(ctx context.Context, status TenantStatus, limit int) (out []Tenant, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenants_by_status", start, &err)
	if limit <= 0 {
		limit = 100
	}
	rows, err := s.db.QueryContext(ctx, `SELECT id, display_name, label, status, kind, parent_tenant_id, storage_namespace_id,
		db_host, db_port, db_user, db_password, db_name,
		db_tls, provider, cluster_id, branch_id, claim_url, claim_expires_at, schema_version,
		s3_encryption_mode, s3_kms_key_id, s3_bucket_key_enabled, created_at, updated_at
		FROM tenants WHERE status = ? ORDER BY created_at ASC, id ASC LIMIT ?`, status, limit)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	return scanTenantRows(rows)
}

func (s *Store) CountTenants(ctx context.Context) (out TenantCounts, err error) {
	start := time.Now()
	defer observeMeta(ctx, "count_tenants", start, &err)
	counts := make(map[TenantStatus]int64, len(allTenantStatuses))
	for _, status := range allTenantStatuses {
		counts[status] = 0
	}
	rows, err := s.db.QueryContext(ctx, `SELECT status, COUNT(*) FROM tenants GROUP BY status`)
	if err != nil {
		return out, fmt.Errorf("count tenants: %w", err)
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var status TenantStatus
		var count int64
		if err := rows.Scan(&status, &count); err != nil {
			return out, fmt.Errorf("scan tenant count: %w", err)
		}
		if _, ok := counts[status]; ok {
			counts[status] = count
		}
	}
	if err := rows.Err(); err != nil {
		return out, fmt.Errorf("count tenants rows: %w", err)
	}
	for _, status := range allTenantStatuses {
		out.Statuses = append(out.Statuses, TenantStatusCount{Status: status, Count: counts[status]})
	}
	return out, nil
}

// ListTenantsByStatusAfter returns one keyset page of tenants after
// (afterCreatedAt, afterID), ordered by (created_at, id). Pass a zero
// afterCreatedAt and empty afterID to scan from the beginning.
func (s *Store) ListTenantsByStatusAfter(ctx context.Context, status TenantStatus, afterCreatedAt time.Time, afterID string, limit int) (out []Tenant, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenants_by_status_after", start, &err)
	if limit <= 0 {
		limit = 100
	}

	var rows *sql.Rows
	if afterCreatedAt.IsZero() && afterID == "" {
		rows, err = s.db.QueryContext(ctx, `SELECT id, display_name, label, status, kind, parent_tenant_id, storage_namespace_id,
			db_host, db_port, db_user, db_password, db_name,
			db_tls, provider, cluster_id, branch_id, claim_url, claim_expires_at, schema_version,
			s3_encryption_mode, s3_kms_key_id, s3_bucket_key_enabled, created_at, updated_at
			FROM tenants WHERE status = ? ORDER BY created_at ASC, id ASC LIMIT ?`, status, limit)
	} else {
		rows, err = s.db.QueryContext(ctx, `SELECT id, display_name, label, status, kind, parent_tenant_id, storage_namespace_id,
			db_host, db_port, db_user, db_password, db_name,
			db_tls, provider, cluster_id, branch_id, claim_url, claim_expires_at, schema_version,
			s3_encryption_mode, s3_kms_key_id, s3_bucket_key_enabled, created_at, updated_at
			FROM tenants
			WHERE status = ? AND (created_at > ? OR (created_at = ? AND id > ?))
			ORDER BY created_at ASC, id ASC LIMIT ?`, status, afterCreatedAt.UTC(), afterCreatedAt.UTC(), afterID, limit)
	}
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	return scanTenantRows(rows)
}

func scanTenantRows(rows *sql.Rows) ([]Tenant, error) {
	out := make([]Tenant, 0)
	for rows.Next() {
		var t Tenant
		var dbTLS int
		var clusterID sql.NullString
		var parentTenantID sql.NullString
		var storageNamespaceID sql.NullString
		var claimURL sql.NullString
		var claimExp sql.NullTime
		var s3BucketKeyEnabled int
		var labelJSON []byte
		if err := rows.Scan(&t.ID, &t.DisplayName, &labelJSON, &t.Status, &t.Kind, &parentTenantID, &storageNamespaceID,
			&t.DBHost, &t.DBPort, &t.DBUser, &t.DBPasswordCipher,
			&t.DBName, &dbTLS, &t.Provider, &clusterID, &t.BranchID, &claimURL, &claimExp, &t.SchemaVersion,
			&t.S3EncryptionMode, &t.S3KMSKeyID, &s3BucketKeyEnabled, &t.CreatedAt, &t.UpdatedAt); err != nil {
			return nil, err
		}
		label, err := scanTenantLabel(labelJSON)
		if err != nil {
			return nil, err
		}
		t.Label = label
		t.DBTLS = dbTLS == 1
		t.S3BucketKeyEnabled = boolPtr(s3BucketKeyEnabled == 1)
		if clusterID.Valid {
			t.ClusterID = clusterID.String
		}
		if parentTenantID.Valid {
			t.ParentTenantID = parentTenantID.String
		}
		if storageNamespaceID.Valid {
			t.StorageNamespaceID = storageNamespaceID.String
		}
		if claimURL.Valid {
			t.ClaimURL = claimURL.String
		}
		if claimExp.Valid {
			ts := claimExp.Time.UTC()
			t.ClaimExpiresAt = &ts
		}
		out = append(out, t)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func scanTenantBindingRows(rows *sql.Rows) ([]TenantWithTiDBCloudOrgBinding, error) {
	out := make([]TenantWithTiDBCloudOrgBinding, 0)
	for rows.Next() {
		rec, err := scanTenantBindingScanner(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, *rec)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

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

func scanTenantBindingRow(row tenantBindingScanner) (*TenantWithTiDBCloudOrgBinding, error) {
	return scanTenantBindingScanner(row)
}

func scanTenantBindingScanner(row tenantBindingScanner) (*TenantWithTiDBCloudOrgBinding, error) {
	var rec TenantWithTiDBCloudOrgBinding
	var dbTLS int
	var clusterID sql.NullString
	var parentTenantID sql.NullString
	var storageNamespaceID sql.NullString
	var claimURL sql.NullString
	var claimExp sql.NullTime
	var s3BucketKeyEnabled int
	var usedAt sql.NullTime
	var labelJSON []byte
	if err := row.Scan(&rec.Tenant.ID, &rec.Tenant.DisplayName, &labelJSON, &rec.Tenant.Status, &rec.Tenant.Kind, &parentTenantID, &storageNamespaceID,
		&rec.Tenant.DBHost, &rec.Tenant.DBPort, &rec.Tenant.DBUser, &rec.Tenant.DBPasswordCipher,
		&rec.Tenant.DBName, &dbTLS, &rec.Tenant.Provider, &clusterID, &rec.Tenant.BranchID, &claimURL, &claimExp, &rec.Tenant.SchemaVersion,
		&rec.Tenant.S3EncryptionMode, &rec.Tenant.S3KMSKeyID, &s3BucketKeyEnabled, &rec.Tenant.CreatedAt, &rec.Tenant.UpdatedAt,
		&rec.Binding.TenantID, &rec.Binding.OrganizationID, &rec.Binding.ClusterID, &rec.Binding.BranchID, &rec.Binding.PoolID, &rec.Binding.PoolStatus, &usedAt,
		&rec.Binding.CreatedAt, &rec.Binding.UpdatedAt); err != nil {
		return nil, err
	}
	label, err := scanTenantLabel(labelJSON)
	if err != nil {
		return nil, err
	}
	rec.Tenant.Label = label
	rec.Tenant.DBTLS = dbTLS == 1
	rec.Tenant.S3BucketKeyEnabled = boolPtr(s3BucketKeyEnabled == 1)
	if clusterID.Valid {
		rec.Tenant.ClusterID = clusterID.String
	}
	if parentTenantID.Valid {
		rec.Tenant.ParentTenantID = parentTenantID.String
	}
	if storageNamespaceID.Valid {
		rec.Tenant.StorageNamespaceID = storageNamespaceID.String
	}
	if claimURL.Valid {
		rec.Tenant.ClaimURL = claimURL.String
	}
	if claimExp.Valid {
		ts := claimExp.Time.UTC()
		rec.Tenant.ClaimExpiresAt = &ts
	}
	if usedAt.Valid {
		ts := usedAt.Time.UTC()
		rec.Binding.UsedAt = &ts
	}
	return &rec, nil
}

// UpdateTenantStatus uses a per-tenant transaction for deleting/deleted
// transitions so the status change and metrics-cleanup outbox row commit
// atomically. Batch cleanup callers intentionally keep one transaction per
// tenant: their loops include external cleanup work, and one large transaction
// would extend lock lifetimes and couple otherwise independent tenant failures.
func (s *Store) UpdateTenantStatus(ctx context.Context, id string, status TenantStatus) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_status", start, &err)
	if tenantStatusNeedsMetricsCleanup(status) {
		err = s.InTx(ctx, func(tx *sql.Tx) error {
			res, err := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ?`, status, time.Now().UTC(), id)
			if err != nil {
				return err
			}
			if err := requireAffected(res); err != nil {
				return err
			}
			return insertTenantNotifyTx(ctx, tx, id, TenantNotifyWorkMetricsCleanup)
		})
		if err != nil {
			return err
		}
		s.apiKeys.evictTenant(id)
		return nil
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ?`, status, time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	s.apiKeys.evictTenant(id)
	return nil
}

func tenantStatusNeedsMetricsCleanup(status TenantStatus) bool {
	return status == TenantDeleting || status == TenantDeleted
}

// UpdateTenantProvider rewrites a tenant's persisted provider. It exists for
// the shared-pool placement path, where the tenant row is inserted with the
// request provider (e.g. tidb_cloud_native) and must be re-labeled as
// tidb_cloud_native_shared once placement on a shared-schema DB is decided,
// so provider-driven capability checks classify the tenant correctly.
func (s *Store) UpdateTenantProvider(ctx context.Context, id, provider string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_provider", start, &err)
	res, err := s.db.ExecContext(ctx, `UPDATE tenants SET provider = ?, updated_at = ? WHERE id = ?`, provider, time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	s.apiKeys.evictTenant(id)
	return nil
}

// UpdateTenantStatusIf has the same per-tenant transaction boundary as
// UpdateTenantStatus when moving to deleting/deleted, while preserving the
// conditional update and per-tenant failure isolation expected by cleanup
// loops.
func (s *Store) UpdateTenantStatusIf(ctx context.Context, id string, from, to TenantStatus) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_status_if", start, &err)
	if tenantStatusNeedsMetricsCleanup(to) {
		err = s.InTx(ctx, func(tx *sql.Tx) error {
			res, err := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
				to, time.Now().UTC(), id, from)
			if err != nil {
				return err
			}
			n, _ := res.RowsAffected()
			updated = n > 0
			if !updated {
				return nil
			}
			return insertTenantNotifyTx(ctx, tx, id, TenantNotifyWorkMetricsCleanup)
		})
		if err != nil {
			return false, err
		}
		if updated {
			s.apiKeys.evictTenant(id)
		}
		return updated, nil
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
		to, time.Now().UTC(), id, from)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	updated = n > 0
	if updated {
		s.apiKeys.evictTenant(id)
	}
	return updated, nil
}

func (s *Store) FinalizeTenantConnection(ctx context.Context, id string, from, to TenantStatus, cluster *Tenant) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "finalize_tenant_connection", start, &err)
	if cluster == nil {
		return false, fmt.Errorf("tenant connection is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenants
		SET status = ?, db_host = ?, db_port = ?, db_user = ?, db_password = ?, db_name = ?, db_tls = ?,
			provider = ?, cluster_id = ?, branch_id = ?, claim_url = ?, claim_expires_at = ?, updated_at = ?
		WHERE id = ? AND status = ?`,
		to, cluster.DBHost, cluster.DBPort, cluster.DBUser, cluster.DBPasswordCipher, cluster.DBName, boolToInt(cluster.DBTLS),
		cluster.Provider, nullStr(cluster.ClusterID), cluster.BranchID, nullStr(cluster.ClaimURL), cluster.ClaimExpiresAt,
		time.Now().UTC(), id, from)
	if err != nil {
		return false, fmt.Errorf("finalize tenant connection: %w", err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return false, fmt.Errorf("read finalized tenant rows affected: %w", err)
	}
	return n > 0, nil
}

func (s *Store) UpdateTenantConnection(ctx context.Context, id string, cluster *Tenant) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_connection", start, &err)
	if cluster == nil {
		return fmt.Errorf("tenant connection is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenants
		SET db_host = ?, db_port = ?, db_user = ?, db_password = ?, db_name = ?, db_tls = ?,
			provider = ?, cluster_id = ?, branch_id = ?, claim_url = ?, claim_expires_at = ?, updated_at = ?
		WHERE id = ?`,
		cluster.DBHost, cluster.DBPort, cluster.DBUser, cluster.DBPasswordCipher, cluster.DBName, boolToInt(cluster.DBTLS),
		cluster.Provider, nullStr(cluster.ClusterID), cluster.BranchID, nullStr(cluster.ClaimURL), cluster.ClaimExpiresAt,
		time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	s.apiKeys.evictTenant(id)
	return nil
}

func (s *Store) PersistTiDBCloudTenantClusterReference(ctx context.Context, id, organizationID string, cluster *Tenant) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "persist_tidbcloud_tenant_cluster_reference", start, &err)
	id = strings.TrimSpace(id)
	organizationID = strings.TrimSpace(organizationID)
	if id == "" || organizationID == "" {
		return fmt.Errorf("tenant id and tidbcloud organization id are required")
	}
	if cluster == nil || strings.TrimSpace(cluster.ClusterID) == "" {
		return fmt.Errorf("tenant cluster reference is required")
	}
	clusterID := strings.TrimSpace(cluster.ClusterID)
	branchID := strings.TrimSpace(cluster.BranchID)
	return s.withTiDBCloudOrgBindingLock(ctx, organizationID, clusterID, branchID, func(ctx context.Context, q metaQueryExecer) error {
		conn, ok := q.(*sql.Conn)
		if !ok {
			return fmt.Errorf("tidbcloud org binding lock connection is unavailable")
		}
		tx, err := conn.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		defer func() { _ = tx.Rollback() }()
		if err := deleteStaleTiDBCloudOrgBindingConflicts(ctx, tx, id, organizationID, clusterID, branchID); err != nil {
			return err
		}
		if err := ensureTiDBCloudOrgBindingAvailable(ctx, tx, id, organizationID, clusterID, branchID); err != nil {
			return err
		}
		now := time.Now().UTC()
		res, err := tx.ExecContext(ctx, `UPDATE tenants
			SET provider = ?, cluster_id = ?, branch_id = ?, claim_url = ?, claim_expires_at = ?, updated_at = ?
			WHERE id = ? AND status = ?`,
			cluster.Provider, clusterID, branchID, nullStr(cluster.ClaimURL), cluster.ClaimExpiresAt, now, id, TenantPending)
		if err != nil {
			return err
		}
		if err := requireAffected(res); err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx, `INSERT INTO tenant_tidbcloud_org_bindings
			(tenant_id, organization_id, cluster_id, branch_id, pool_id, pool_status, used_at, created_at, updated_at)
			VALUES (?, ?, ?, ?, '', ?, NULL, ?, ?)
			ON DUPLICATE KEY UPDATE
				organization_id = VALUES(organization_id), cluster_id = VALUES(cluster_id), branch_id = VALUES(branch_id),
				pool_id = '', pool_status = VALUES(pool_status), used_at = NULL, updated_at = VALUES(updated_at)`,
			id, organizationID, clusterID, branchID, TenantPoolBindingUsed, now, now)
		if err != nil {
			return err
		}
		return tx.Commit()
	})
}

func (s *Store) UpdateTenantClusterReference(ctx context.Context, id string, cluster *Tenant) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_cluster_reference", start, &err)
	if cluster == nil {
		return fmt.Errorf("tenant cluster reference is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenants
		SET db_host = ?, db_port = ?, db_user = ?, db_name = ?, db_tls = ?,
			provider = ?, cluster_id = ?, branch_id = ?, claim_url = ?, claim_expires_at = ?, updated_at = ?
		WHERE id = ?`,
		cluster.DBHost, cluster.DBPort, cluster.DBUser, cluster.DBName, boolToInt(cluster.DBTLS),
		cluster.Provider, nullStr(cluster.ClusterID), cluster.BranchID, nullStr(cluster.ClaimURL), cluster.ClaimExpiresAt,
		time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	s.apiKeys.evictTenant(id)
	return nil
}

func (s *Store) ClearTenantProvisionMetadata(ctx context.Context, tenantID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "clear_tenant_provision_metadata", start, &err)
	now := time.Now().UTC()
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		res, err := tx.ExecContext(ctx, `UPDATE tenants
			SET db_host = '', db_port = 0, db_user = '', db_password = ?, db_name = '', db_tls = 1,
				cluster_id = NULL, branch_id = '', claim_url = NULL, claim_expires_at = NULL, updated_at = ?
			WHERE id = ?`,
			[]byte{}, now, tenantID)
		if err != nil {
			return err
		}
		if err := requireAffected(res); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM tenant_tidbcloud_org_bindings WHERE tenant_id = ?`, tenantID); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	return nil
}

func (s *Store) UpdateTenantDBCredentialIf(ctx context.Context, id, fromDBUser, dbUser string, dbPasswordCipher []byte) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_db_credential_if", start, &err)
	res, err := s.db.ExecContext(ctx, `UPDATE tenants SET db_user = ?, db_password = ?, updated_at = ? WHERE id = ? AND db_user = ?`,
		dbUser, dbPasswordCipher, time.Now().UTC(), id, fromDBUser)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	updated = n > 0
	if updated {
		s.apiKeys.evictTenant(id)
	}
	return updated, nil
}

func (s *Store) UpdateTenantBranch(ctx context.Context, id string, cluster *Tenant) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_branch", start, &err)
	if cluster == nil {
		return fmt.Errorf("tenant branch is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE tenants
		SET provider = ?, cluster_id = ?, branch_id = ?, claim_url = ?, claim_expires_at = ?, updated_at = ?
		WHERE id = ?`,
		cluster.Provider, nullStr(cluster.ClusterID), cluster.BranchID, nullStr(cluster.ClaimURL), cluster.ClaimExpiresAt,
		time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	s.apiKeys.evictTenant(id)
	return nil
}

func (s *Store) RevokeTenantAPIKeys(ctx context.Context, tenantID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "revoke_tenant_api_keys", start, &err)
	now := time.Now().UTC()
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		res, err := tx.ExecContext(ctx, `UPDATE tenant_api_keys
			SET status = ?, revoked_at = COALESCE(revoked_at, ?), updated_at = ?
			WHERE tenant_id = ? AND status IN (?, ?)`,
			APIKeyRevoked, now, now, tenantID, APIKeyActive, APIKeyDisabled)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n == 0 {
			return nil
		}
		return insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkAPIKeyCacheCleanup)
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	return nil
}

func (s *Store) RevokeAPIKey(ctx context.Context, tenantID, apiKeyID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "revoke_api_key", start, &err)
	now := time.Now().UTC()
	updated := false
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		res, err := tx.ExecContext(ctx, `UPDATE tenant_api_keys
			SET status = ?, revoked_at = COALESCE(revoked_at, ?), updated_at = ?
			WHERE tenant_id = ? AND id = ? AND status IN (?, ?)`,
			APIKeyRevoked, now, now, tenantID, apiKeyID, APIKeyActive, APIKeyDisabled)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		updated = n > 0
		if !updated {
			return nil
		}
		return insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkAPIKeyCacheCleanup)
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	if !updated {
		return ErrNotFound
	}
	return nil
}

func (s *Store) RevokeAPIKeysByIssuer(ctx context.Context, tenantID, provider, subjectKey, exceptAPIKeyID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "revoke_api_keys_by_issuer", start, &err)
	now := time.Now().UTC()
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		res, err := tx.ExecContext(ctx, `UPDATE tenant_api_keys
			SET status = ?, revoked_at = COALESCE(revoked_at, ?), updated_at = ?
			WHERE tenant_id = ? AND issued_by_provider = ? AND issued_by_subject_key = ? AND status IN (?, ?)
				AND (? = '' OR id <> ?)`,
			APIKeyRevoked, now, now, tenantID, provider, subjectKey, APIKeyActive, APIKeyDisabled, exceptAPIKeyID, exceptAPIKeyID)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n == 0 {
			return nil
		}
		return insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkAPIKeyCacheCleanup)
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	return nil
}

func (s *Store) GetActiveAPIKeyByIssuer(ctx context.Context, tenantID, provider, subjectKey string) (out *APIKey, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_active_api_key_by_issuer", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT
			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, expires_at, created_at, updated_at
		FROM tenant_api_keys
		WHERE tenant_id = ? AND issued_by_provider = ? AND issued_by_subject_key = ? AND status = ?
			AND ((expires_at IS NULL AND scope_kind = ?) OR expires_at > UTC_TIMESTAMP(3))
		ORDER BY created_at DESC
		LIMIT 1`, tenantID, provider, subjectKey, APIKeyActive, APIKeyScopeKindOwner)

	var rec APIKey
	var revokedAt sql.NullTime
	var expiresAt sql.NullTime
	var issuedByProvider sql.NullString
	var issuedBySubjectKey sql.NullString
	var issuedByMetadataJSON []byte
	if err = row.Scan(
		&rec.ID, &rec.TenantID, &rec.KeyName, &rec.JWTCiphertext,
		&rec.JWTHash, &rec.TokenVersion, &rec.Status, &rec.ScopeKind,
		&issuedByProvider, &issuedBySubjectKey, &issuedByMetadataJSON, &rec.IssuedAt,
		&revokedAt, &expiresAt, &rec.CreatedAt, &rec.UpdatedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			err = ErrNotFound
			return nil, err
		}
		return nil, err
	}
	if revokedAt.Valid {
		t := revokedAt.Time.UTC()
		rec.RevokedAt = &t
	}
	if expiresAt.Valid {
		t := expiresAt.Time.UTC()
		rec.ExpiresAt = &t
	}
	if issuedByProvider.Valid {
		rec.IssuedByProvider = issuedByProvider.String
	}
	if issuedBySubjectKey.Valid {
		rec.IssuedBySubjectKey = issuedBySubjectKey.String
	}
	rec.IssuedByMetadataJSON = issuedByMetadataJSON
	out = &rec
	return out, nil
}

func (s *Store) UpsertStorageNamespace(ctx context.Context, ns *StorageNamespace) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_storage_namespace", start, &err)
	if ns == nil {
		return fmt.Errorf("storage namespace is required")
	}
	appendLogState, appendLogBucket, appendLogPrefix, appendLogRegion, err := normalizeAppendLogBinding(ns)
	if err != nil {
		return err
	}
	state := ns.State
	if state == "" {
		state = StorageNamespaceActive
	}
	now := time.Now().UTC()
	_, err = s.db.ExecContext(ctx, `INSERT INTO storage_namespaces
		(namespace_id, owner_tenant_id, backend, bucket, prefix,
		 append_log_binding_state, append_log_bucket, append_log_prefix, append_log_region,
		 state, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE
			owner_tenant_id = VALUES(owner_tenant_id),
			backend = VALUES(backend),
			bucket = VALUES(bucket),
			prefix = VALUES(prefix),
			state = VALUES(state),
			updated_at = VALUES(updated_at)`,
		ns.ID, ns.OwnerTenantID, ns.Backend, ns.Bucket, ns.Prefix,
		appendLogState, appendLogBucket, appendLogPrefix, appendLogRegion,
		state, now, now)
	return err
}

func (s *Store) GetStorageNamespace(ctx context.Context, id string) (out *StorageNamespace, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_storage_namespace", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT namespace_id, owner_tenant_id, backend, bucket, prefix,
		append_log_binding_state, append_log_bucket, append_log_prefix, append_log_region,
		state, created_at, updated_at
		FROM storage_namespaces WHERE namespace_id = ?`, id)
	var ns StorageNamespace
	if err = row.Scan(&ns.ID, &ns.OwnerTenantID, &ns.Backend, &ns.Bucket, &ns.Prefix,
		&ns.AppendLogBindingState, &ns.AppendLogBucket, &ns.AppendLogPrefix, &ns.AppendLogRegion,
		&ns.State, &ns.CreatedAt, &ns.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			err = ErrNotFound
			return nil, err
		}
		return nil, err
	}
	return &ns, nil
}

func (s *Store) UpdateStorageNamespaceState(ctx context.Context, id string, state StorageNamespaceState) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_storage_namespace_state", start, &err)
	if id == "" {
		return fmt.Errorf("namespace id is required")
	}
	if state == "" {
		return fmt.Errorf("namespace state is required")
	}
	res, err := s.db.ExecContext(ctx, `UPDATE storage_namespaces SET state = ?, updated_at = ? WHERE namespace_id = ?`,
		state, time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func normalizeAppendLogBinding(ns *StorageNamespace) (AppendLogBindingState, string, string, string, error) {
	state := ns.AppendLogBindingState
	if state == "" {
		state = AppendLogBindingUnbound
	}
	bucket := strings.TrimSpace(ns.AppendLogBucket)
	prefix := NormalizeAppendLogBasePrefix(ns.AppendLogPrefix)
	region := strings.TrimSpace(ns.AppendLogRegion)
	switch state {
	case AppendLogBindingUnbound:
		if bucket != "" || prefix != "" || region != "" {
			return "", "", "", "", fmt.Errorf("unbound append-log namespace must not have a target")
		}
	case AppendLogBindingBound:
		if ns.Backend != "s3" {
			return "", "", "", "", fmt.Errorf("append-log binding requires s3 storage backend")
		}
		if bucket == "" || region == "" {
			return "", "", "", "", fmt.Errorf("bound append-log namespace requires bucket and region")
		}
	default:
		return "", "", "", "", fmt.Errorf("unsupported append-log binding state %q", state)
	}
	return state, bucket, prefix, region, nil
}

// NormalizeAppendLogBasePrefix returns the canonical optional base prefix used
// before append-log/<storage-namespace-id>/. It matches the existing writer
// behavior: leading slashes are stripped and a single missing trailing slash
// is added, while repeated trailing slashes are preserved.
func NormalizeAppendLogBasePrefix(prefix string) string {
	prefix = strings.TrimLeft(prefix, "/")
	if prefix != "" && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	return prefix
}

// EnsureAppendLogStorageNamespaceBinding atomically binds an S3 storage
// namespace to one immutable S3 Express target. Repeating the same binding is
// idempotent; attempting to replace a bound target fails closed.
func (s *Store) EnsureAppendLogStorageNamespaceBinding(ctx context.Context, namespaceID, bucket, prefix, region string) (out *StorageNamespace, err error) {
	start := time.Now()
	defer observeMeta(ctx, "ensure_append_log_storage_namespace_binding", start, &err)
	namespaceID = strings.TrimSpace(namespaceID)
	if namespaceID == "" {
		return nil, fmt.Errorf("storage namespace id is required")
	}
	_, bucket, prefix, region, err = normalizeAppendLogBinding(&StorageNamespace{
		Backend:               "s3",
		AppendLogBindingState: AppendLogBindingBound,
		AppendLogBucket:       bucket,
		AppendLogPrefix:       prefix,
		AppendLogRegion:       region,
	})
	if err != nil {
		return nil, err
	}
	for attempt := 0; attempt < 2; attempt++ {
		out = nil
		err = s.InTx(ctx, func(tx *sql.Tx) error {
			ns, getErr := getStorageNamespaceTx(ctx, tx, namespaceID)
			if getErr != nil {
				return getErr
			}
			// Serialize with the deleting mark using the same tenant-first lock order.
			var tenantStatus TenantStatus
			if getErr := tx.QueryRowContext(ctx, `SELECT status FROM tenants WHERE id = ? FOR UPDATE`, ns.OwnerTenantID).Scan(&tenantStatus); getErr != nil {
				if errors.Is(getErr, sql.ErrNoRows) {
					return ErrNotFound
				}
				return getErr
			}
			if tenantStatus != TenantActive {
				return ErrNotFound
			}
			var state StorageNamespaceState
			if getErr := tx.QueryRowContext(ctx, `SELECT state FROM storage_namespaces WHERE namespace_id = ? FOR UPDATE`, ns.ID).Scan(&state); getErr != nil {
				if errors.Is(getErr, sql.ErrNoRows) {
					return ErrNotFound
				}
				return getErr
			}
			if state != StorageNamespaceActive {
				return ErrNotFound
			}
			ns.State = state
			if _, _, _, _, getErr = normalizeAppendLogBinding(ns); getErr != nil {
				return fmt.Errorf("namespace %q: %w", namespaceID, getErr)
			}
			switch ns.AppendLogBindingState {
			case "", AppendLogBindingUnbound:
				if ns.Backend != "s3" {
					return fmt.Errorf("append-log binding requires s3 storage backend")
				}
				now := time.Now().UTC()
				res, updateErr := tx.ExecContext(ctx, `UPDATE storage_namespaces
					SET append_log_binding_state = ?, append_log_bucket = ?, append_log_prefix = ?, append_log_region = ?, updated_at = ?
					WHERE namespace_id = ? AND append_log_binding_state = ? AND state = ?`,
					AppendLogBindingBound, bucket, prefix, region, now, namespaceID, AppendLogBindingUnbound, StorageNamespaceActive)
				if updateErr != nil {
					return updateErr
				}
				if n, rowsErr := res.RowsAffected(); rowsErr != nil {
					return rowsErr
				} else if n == 0 {
					return errAppendLogBindingRace
				}
				ns.AppendLogBindingState = AppendLogBindingBound
				ns.AppendLogBucket = bucket
				ns.AppendLogPrefix = prefix
				ns.AppendLogRegion = region
				ns.UpdatedAt = now
				out = ns
				return nil
			case AppendLogBindingBound:
				if ns.AppendLogBucket != bucket || ns.AppendLogPrefix != prefix || ns.AppendLogRegion != region {
					return fmt.Errorf("namespace %q: %w", namespaceID, ErrAppendLogBindingMismatch)
				}
				out = ns
				return nil
			default:
				return fmt.Errorf("namespace %q: unsupported append-log binding state %q", namespaceID, ns.AppendLogBindingState)
			}
		})
		if !errors.Is(err, errAppendLogBindingRace) {
			break
		}
	}
	if err != nil {
		return nil, err
	}
	return out, nil
}

func (s *Store) EnsureTenantStorageNamespace(ctx context.Context, tenantID, backendName, bucket, prefix string) (out *StorageNamespace, err error) {
	start := time.Now()
	defer observeMeta(ctx, "ensure_tenant_storage_namespace", start, &err)
	if tenantID == "" {
		return nil, fmt.Errorf("tenant id is required")
	}
	if backendName == "" {
		return nil, fmt.Errorf("storage backend is required")
	}
	if prefix == "" {
		return nil, fmt.Errorf("storage prefix is required")
	}
	ns := &StorageNamespace{
		ID:                    tenantID,
		OwnerTenantID:         tenantID,
		Backend:               backendName,
		Bucket:                bucket,
		Prefix:                prefix,
		AppendLogBindingState: AppendLogBindingUnbound,
		State:                 StorageNamespaceActive,
	}
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		row := tx.QueryRowContext(ctx, `SELECT storage_namespace_id FROM tenants WHERE id = ? FOR UPDATE`, tenantID)
		var existing sql.NullString
		if err := row.Scan(&existing); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return err
		}
		if existing.Valid && existing.String != "" {
			existingNS, err := getStorageNamespaceTx(ctx, tx, existing.String)
			if err != nil {
				return err
			}
			ns = existingNS
			return nil
		}
		now := time.Now().UTC()
		if _, err := tx.ExecContext(ctx, `INSERT INTO storage_namespaces
			(namespace_id, owner_tenant_id, backend, bucket, prefix,
			 append_log_binding_state, append_log_bucket, append_log_prefix, append_log_region,
			 state, created_at, updated_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
			ON DUPLICATE KEY UPDATE updated_at = updated_at`,
			ns.ID, ns.OwnerTenantID, ns.Backend, ns.Bucket, ns.Prefix,
			ns.AppendLogBindingState, ns.AppendLogBucket, ns.AppendLogPrefix, ns.AppendLogRegion,
			ns.State, now, now); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `UPDATE tenants SET storage_namespace_id = ?, updated_at = ? WHERE id = ?`,
			ns.ID, now, tenantID); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	s.apiKeys.evictTenant(tenantID)
	return ns, nil
}

func (s *Store) EnqueueTenantDeleteJob(ctx context.Context, job *TenantDeleteJob) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "enqueue_tenant_delete_job", start, &err)
	tenantID := ""
	if job != nil {
		tenantID = job.TenantID
	}
	if err := validateTenantDeleteJob(tenantID, job); err != nil {
		return err
	}
	now := time.Now().UTC()
	return upsertTenantDeleteJob(ctx, s.db, job, now)
}

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

func upsertTenantDeleteJob(ctx context.Context, execer tenantDeleteJobExecer, job *TenantDeleteJob, now time.Time) error {
	notBefore := job.NotBefore.UTC()
	if notBefore.IsZero() {
		notBefore = now
	}
	// Deliberately do not update state on duplicate. The existing row may be
	// running or terminal; only its owning transition may change that state.
	// Enqueue refreshes payload/timing without stealing it.
	_, err := execer.ExecContext(ctx, `INSERT INTO tenant_delete_jobs
		(tenant_id, namespace_id, backend, bucket, prefix, state, not_before, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE
			namespace_id = VALUES(namespace_id),
			backend = VALUES(backend),
			bucket = VALUES(bucket),
			prefix = VALUES(prefix),
			not_before = LEAST(not_before, VALUES(not_before)),
			updated_at = VALUES(updated_at)`,
		job.TenantID, job.NamespaceID, job.Backend, job.Bucket, job.Prefix,
		TenantDeleteJobPending, notBefore, now, now)
	return err
}

// TenantDeleteJobArmed reports whether the tenant's cleanup job is due or
// already owned by a worker.
func (s *Store) TenantDeleteJobArmed(ctx context.Context, tenantID string) (out bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "tenant_delete_job_armed", start, &err)
	if tenantID == "" {
		return false, fmt.Errorf("tenant id is required")
	}
	row := s.db.QueryRowContext(ctx, `SELECT 1 FROM tenant_delete_jobs
		WHERE tenant_id = ? AND state IN (?, ?) AND not_before <= ? LIMIT 1`,
		tenantID, TenantDeleteJobPending, TenantDeleteJobRunning, time.Now().UTC())
	var one int
	if err := row.Scan(&one); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func validateTenantDeleteJob(tenantID string, job *TenantDeleteJob) error {
	if job == nil {
		return fmt.Errorf("tenant delete job is required")
	}
	if tenantID == "" || job.TenantID != tenantID {
		return fmt.Errorf("tenant id is required")
	}
	if job.NamespaceID == "" {
		return fmt.Errorf("namespace id is required")
	}
	if job.Backend == "" {
		return fmt.Errorf("storage backend is required")
	}
	if job.Prefix == "" {
		return fmt.Errorf("storage prefix is required")
	}
	if job.State != "" && job.State != TenantDeleteJobPending {
		return fmt.Errorf("invalid initial tenant delete job state %q", job.State)
	}
	return nil
}

// markTenantDeletingWithDeleteJobTx marks the tenant deleting, marks its
// storage namespace deleting, and enqueues the cleanup job inside tx.
// The caller is responsible for any additional guard (for example a pool
// free-membership check) before invoking it. updated_at is always bumped so
// RowsAffected stays meaningful when from == TenantDeleting.
func markTenantDeletingWithDeleteJobTx(ctx context.Context, tx *sql.Tx, tenantID string, from TenantStatus, job *TenantDeleteJob) (bool, error) {
	now := time.Now().UTC()
	res, err := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
		TenantDeleting, now, tenantID, from)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return false, nil
	}
	res, err = tx.ExecContext(ctx, `UPDATE storage_namespaces SET state = ?, updated_at = ? WHERE namespace_id = ?`,
		StorageNamespaceDeleting, now, job.NamespaceID)
	if err != nil {
		return false, err
	}
	if err := requireAffected(res); err != nil {
		return false, err
	}
	if err := upsertTenantDeleteJob(ctx, tx, job, now); err != nil {
		return false, err
	}
	return true, insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkMetricsCleanup)
}

// MarkTenantDeletingWithDeleteJob atomically marks a tenant deleting, marks
// its storage namespace deleting, and enqueues the tenant's S3 cleanup job in
// a single transaction, so a deleting tenant always has a durable cleanup
// path. Resumable callers may use NotBefore as a fallback.
func (s *Store) MarkTenantDeletingWithDeleteJob(ctx context.Context, tenantID string, from TenantStatus, job *TenantDeleteJob) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_tenant_deleting_with_delete_job", start, &err)
	if err := validateTenantDeleteJob(tenantID, job); err != nil {
		return false, err
	}
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var txErr error
		updated, txErr = markTenantDeletingWithDeleteJobTx(ctx, tx, tenantID, from, job)
		return txErr
	})
	if err != nil {
		return false, err
	}
	if updated {
		s.apiKeys.evictTenant(tenantID)
	}
	return updated, nil
}

// RollbackTenantDeletingWithDeleteJob undoes MarkTenantDeletingWithDeleteJob
// when the in-request external steps fail: it restores the tenant status and
// namespace state and drops the not-yet-running cleanup job, all in one
// transaction. It reports rolledBack=false when another actor owns or has
// progressed the delete.
func (s *Store) RollbackTenantDeletingWithDeleteJob(ctx context.Context, tenantID, namespaceID string, to TenantStatus) (rolledBack bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "rollback_tenant_deleting_with_delete_job", 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 jobState TenantDeleteJobState
		var provider string
		if err := tx.QueryRowContext(ctx, `SELECT j.state, t.provider
			FROM tenant_delete_jobs j
			JOIN tenants t ON t.id = j.tenant_id
			WHERE j.tenant_id = ? FOR UPDATE`, tenantID).Scan(&jobState, &provider); errors.Is(err, sql.ErrNoRows) {
			return nil
		} else if err != nil {
			return err
		}
		// Cluster-backed jobs are created only after Cloud deprovision succeeds,
		// so their presence is already past the rollback point. Only an
		// idempotent shared-schema purge may roll back a pending cleanup job.
		if provider != tidbCloudNativeSharedProvider || jobState != TenantDeleteJobPending {
			return nil
		}
		now := time.Now().UTC()
		res, err := tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
			to, now, tenantID, TenantDeleting)
		if err != nil {
			return err
		}
		n, _ := res.RowsAffected()
		if n == 0 {
			return nil
		}
		rolledBack = true
		if _, err := tx.ExecContext(ctx, `DELETE FROM tenant_delete_jobs WHERE tenant_id = ? AND state = ?`,
			tenantID, TenantDeleteJobPending); err != nil {
			return err
		}
		if namespaceID != "" {
			if _, err := tx.ExecContext(ctx, `UPDATE storage_namespaces SET state = ?, updated_at = ?
				WHERE namespace_id = ? AND state = ?`,
				StorageNamespaceActive, now, namespaceID, StorageNamespaceDeleting); err != nil {
				return err
			}
		}
		return nil
	})
	return rolledBack, err
}

// CancelTenantDeleteJob removes a claimed cleanup job only while the tenant is
// still absent or not deleting. The tenant-status predicate serializes cancel
// with a concurrent re-delete: either cancel removes the old row before the
// new mark inserts its job, or the new deleting state preserves the running
// job for the worker to continue.
func (s *Store) CancelTenantDeleteJob(ctx context.Context, tenantID string) (cancelled bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "cancel_tenant_delete_job", start, &err)
	if strings.TrimSpace(tenantID) == "" {
		return false, fmt.Errorf("tenant id is required")
	}
	res, err := s.db.ExecContext(ctx, `DELETE j FROM tenant_delete_jobs j
		LEFT JOIN tenants t ON t.id = j.tenant_id
		WHERE j.tenant_id = ? AND j.state = ?
			AND (t.id IS NULL OR t.status <> ?)`, tenantID, TenantDeleteJobRunning, TenantDeleting)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	return n > 0, nil
}

func (s *Store) TenantDeleteJobExists(ctx context.Context, tenantID string) (out bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "tenant_delete_job_exists", start, &err)
	if tenantID == "" {
		return false, fmt.Errorf("tenant id is required")
	}
	row := s.db.QueryRowContext(ctx, `SELECT 1 FROM tenant_delete_jobs WHERE tenant_id = ? LIMIT 1`, tenantID)
	var one int
	if err := row.Scan(&one); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (s *Store) ListDueTenantDeleteJobs(ctx context.Context, now time.Time, limit int) (out []TenantDeleteJob, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_due_tenant_delete_jobs", start, &err)
	if limit <= 0 {
		limit = 25
	}
	rows, err := s.db.QueryContext(ctx, `SELECT tenant_id, namespace_id, backend, bucket, prefix,
			state, attempts, COALESCE(last_error, ''), not_before, deleted_objects,
			aborted_multipart_uploads, created_at, updated_at, completed_at
		FROM tenant_delete_jobs
		WHERE state = ? AND not_before <= ?
		ORDER BY not_before ASC, created_at ASC
		LIMIT ?`, TenantDeleteJobPending, now.UTC(), limit)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var job TenantDeleteJob
		var completedAt sql.NullTime
		if err := rows.Scan(&job.TenantID, &job.NamespaceID, &job.Backend, &job.Bucket, &job.Prefix,
			&job.State, &job.Attempts, &job.LastError, &job.NotBefore, &job.DeletedObjects,
			&job.AbortedMultipartUploads, &job.CreatedAt, &job.UpdatedAt, &completedAt); err != nil {
			return nil, err
		}
		if completedAt.Valid {
			t := completedAt.Time.UTC()
			job.CompletedAt = &t
		}
		out = append(out, job)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func (s *Store) RecoverStaleTenantDeleteJobs(ctx context.Context, before time.Time) (n int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "recover_stale_tenant_delete_jobs", start, &err)
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_delete_jobs
		SET state = ?, not_before = ?, last_error = ?, updated_at = ?
		WHERE state = ? AND updated_at < ?`,
		TenantDeleteJobPending, time.Now().UTC(), "recovered stale running delete job", time.Now().UTC(),
		TenantDeleteJobRunning, before.UTC())
	if err != nil {
		return 0, err
	}
	n, _ = res.RowsAffected()
	return n, nil
}

func (s *Store) MarkTenantDeleteJobRunning(ctx context.Context, tenantID string) (updated bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_tenant_delete_job_running", start, &err)
	now := time.Now().UTC()
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_delete_jobs
		SET state = ?, updated_at = ?
		WHERE tenant_id = ? AND state = ? AND not_before <= ?`,
		TenantDeleteJobRunning, now, tenantID, TenantDeleteJobPending, now)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	return n > 0, nil
}

func (s *Store) RetryTenantDeleteJob(ctx context.Context, tenantID string, notBefore time.Time, lastError string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "retry_tenant_delete_job", start, &err)
	_, err = s.db.ExecContext(ctx, `UPDATE tenant_delete_jobs
		SET state = ?, not_before = ?, attempts = attempts + 1, last_error = ?, updated_at = ?
		WHERE tenant_id = ? AND state = ?`,
		TenantDeleteJobPending, notBefore.UTC(), nullStr(lastError), time.Now().UTC(), tenantID, TenantDeleteJobRunning)
	return err
}

func (s *Store) MarkTenantDeleted(ctx context.Context, tenantID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_tenant_deleted", start, &err)
	now := time.Now().UTC()
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		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
}

func (s *Store) MarkTenantDeleteJobDeleted(ctx context.Context, tenantID string, deletedObjects, abortedMultipartUploads int64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_tenant_delete_job_deleted", start, &err)
	now := time.Now().UTC()
	_, err = s.db.ExecContext(ctx, `UPDATE tenant_delete_jobs
		SET state = ?, deleted_objects = ?, aborted_multipart_uploads = ?, completed_at = ?,
			last_error = NULL, updated_at = ?
		WHERE tenant_id = ?`,
		TenantDeleteJobDeleted, deletedObjects, abortedMultipartUploads, now, now, tenantID)
	return err
}

func (s *Store) FinalizeTenantDelete(ctx context.Context, tenantID, namespaceID string, deletedObjects, abortedMultipartUploads int64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "finalize_tenant_delete", start, &err)
	now := time.Now().UTC()
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		res, err := tx.ExecContext(ctx, `UPDATE tenant_delete_jobs
			SET state = ?, deleted_objects = ?, aborted_multipart_uploads = ?, completed_at = ?,
				last_error = NULL, updated_at = ?
			WHERE tenant_id = ? AND state = ?`,
			TenantDeleteJobDeleted, deletedObjects, abortedMultipartUploads, now, now, tenantID, TenantDeleteJobRunning)
		if err != nil {
			return err
		}
		if err := requireAffected(res); err != nil {
			return err
		}
		if namespaceID != "" {
			res, err = tx.ExecContext(ctx, `UPDATE storage_namespaces SET state = ?, updated_at = ? WHERE namespace_id = ?`,
				StorageNamespaceDeleted, now, namespaceID)
			if err != nil {
				return err
			}
			if err := requireAffected(res); err != nil {
				return err
			}
		}
		res, err = tx.ExecContext(ctx, `UPDATE tenants SET status = ?, updated_at = ? WHERE id = ? AND status = ?`,
			TenantDeleted, now, tenantID, TenantDeleting)
		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
		}
		if _, err := tx.ExecContext(ctx, `DELETE p FROM tenant_placements p
			JOIN fs_registry f ON f.fs_id = p.fs_id
			WHERE f.tenant_id = ? AND p.status = ?`, tenantID, PlacementStatusDeleting); err != nil {
			return err
		}
		return insertTenantNotifyTx(ctx, tx, tenantID, TenantNotifyWorkMetricsCleanup)
	})
	if err != nil {
		return err
	}
	s.apiKeys.evictTenant(tenantID)
	return nil
}

func requireAffected(res sql.Result) error {
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

func getStorageNamespaceTx(ctx context.Context, tx *sql.Tx, id string) (*StorageNamespace, error) {
	row := tx.QueryRowContext(ctx, `SELECT namespace_id, owner_tenant_id, backend, bucket, prefix,
		append_log_binding_state, append_log_bucket, append_log_prefix, append_log_region,
		state, created_at, updated_at
		FROM storage_namespaces WHERE namespace_id = ?`, id)
	var ns StorageNamespace
	if err := row.Scan(&ns.ID, &ns.OwnerTenantID, &ns.Backend, &ns.Bucket, &ns.Prefix,
		&ns.AppendLogBindingState, &ns.AppendLogBucket, &ns.AppendLogPrefix, &ns.AppendLogRegion,
		&ns.State, &ns.CreatedAt, &ns.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	return &ns, nil
}

type ObjectGCCandidateInput struct {
	NamespaceID    string
	StorageRef     string
	StorageRefHash string
	ContentLayout  string
	Reason         ObjectGCCandidateReason
	SourceTenantID string
	SourceFileID   string
	NotBefore      time.Time
}

type ObjectGCCandidate struct {
	NamespaceID    string
	StorageRef     string
	StorageRefHash string
	ContentLayout  string
	Reason         ObjectGCCandidateReason
	SourceTenantID string
	SourceFileID   string
	NotBefore      time.Time
	State          ObjectGCCandidateState
	Attempts       int
	LastError      string
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

func objectGCCandidateContentLayout(layout string) string {
	if layout == "" {
		return "single"
	}
	return layout
}

func (s *Store) EnqueueObjectGCCandidate(ctx context.Context, c *ObjectGCCandidateInput) (err error) {
	return s.enqueueObjectGCCandidate(ctx, c, ObjectGCCandidatePending)
}

// PrepareFailedWriteObjectGCCandidate establishes fenced ownership before an
// S3 mutation can materialize an object.
func (s *Store) PrepareFailedWriteObjectGCCandidate(ctx context.Context, c *ObjectGCCandidateInput) (err error) {
	if c == nil || c.Reason != ObjectGCReasonFailedWrite {
		return fmt.Errorf("failed-write object GC candidate required")
	}
	return s.enqueueObjectGCCandidate(ctx, c, ObjectGCCandidatePreparing)
}

func (s *Store) enqueueObjectGCCandidate(ctx context.Context, c *ObjectGCCandidateInput, failedWriteState ObjectGCCandidateState) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "enqueue_object_gc_candidate", start, &err)
	if c == nil {
		return fmt.Errorf("object gc candidate is required")
	}
	if c.NamespaceID == "" {
		return fmt.Errorf("namespace id is required")
	}
	if c.StorageRef == "" {
		return fmt.Errorf("storage ref is required")
	}
	if c.StorageRefHash == "" {
		return fmt.Errorf("storage ref hash is required")
	}
	if c.Reason == "" {
		return fmt.Errorf("object gc reason is required")
	}
	notBefore := c.NotBefore.UTC()
	if notBefore.IsZero() {
		notBefore = time.Now().UTC()
	}
	now := time.Now().UTC()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			_ = tx.Rollback()
		}
	}()
	var currentReason ObjectGCCandidateReason
	var currentState ObjectGCCandidateState
	stateErr := tx.QueryRowContext(ctx, `SELECT reason, state FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? FOR UPDATE`, c.NamespaceID, c.StorageRefHash).Scan(&currentReason, &currentState)
	if stateErr == nil {
		lifecycleTakesOverPublish := currentReason == ObjectGCReasonFailedWrite &&
			currentState == ObjectGCCandidatePublishing &&
			(c.Reason == ObjectGCReasonOverwrite || c.Reason == ObjectGCReasonFileDelete || c.Reason == ObjectGCReasonForkDelete)
		if currentState == ObjectGCCandidateDeleting || currentState == ObjectGCCandidatePreparing ||
			(currentState == ObjectGCCandidatePublishing && !lifecycleTakesOverPublish) {
			return fmt.Errorf("object GC candidate is being finalized or deleted")
		}
	}
	if stateErr != nil && !errors.Is(stateErr, sql.ErrNoRows) {
		return stateErr
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO object_gc_candidates
		(namespace_id, storage_ref, storage_ref_hash, content_layout, reason, source_tenant_id, source_file_id, not_before, state, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE
			storage_ref = VALUES(storage_ref),
			content_layout = VALUES(content_layout),
			not_before = CASE WHEN VALUES(reason) = 'failed_write'
				THEN VALUES(not_before)
				WHEN reason = VALUES(reason)
				THEN LEAST(not_before, VALUES(not_before))
				ELSE VALUES(not_before) END,
			reason = VALUES(reason),
			source_tenant_id = VALUES(source_tenant_id),
			source_file_id = VALUES(source_file_id),
			state = CASE WHEN VALUES(reason) = 'failed_write' THEN VALUES(state)
				WHEN state = 'deleted' THEN state ELSE 'pending' END,
			last_error = CASE WHEN VALUES(reason) = 'failed_write' THEN NULL ELSE last_error END,
			deleted_at = CASE WHEN VALUES(reason) = 'failed_write' THEN NULL ELSE deleted_at END,
			updated_at = VALUES(updated_at)`,
		c.NamespaceID, c.StorageRef, c.StorageRefHash, objectGCCandidateContentLayout(c.ContentLayout), c.Reason, c.SourceTenantID, c.SourceFileID,
		notBefore, failedWriteState, now, now)
	if err != nil {
		return err
	}
	return tx.Commit()
}

// RetryFailedWriteObjectGCCandidate releases producer ownership after a known
// publication failure so the GC worker can reconcile the object after grace.
func (s *Store) RetryFailedWriteObjectGCCandidate(ctx context.Context, namespaceID, storageRefHash, storageRef string, notBefore time.Time, lastError string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "retry_failed_write_object_gc_candidate", start, &err)
	if namespaceID == "" || storageRefHash == "" || storageRef == "" {
		return nil
	}
	_, err = s.db.ExecContext(ctx, `UPDATE object_gc_candidates
		SET state = ?, not_before = ?, last_error = ?, attempts = attempts + 1, updated_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state IN (?, ?)`,
		ObjectGCCandidatePending, notBefore.UTC(), nullStr(lastError), time.Now().UTC(),
		namespaceID, storageRefHash, storageRef, ObjectGCReasonFailedWrite,
		ObjectGCCandidatePreparing, ObjectGCCandidatePublishing)
	return err
}

// BeginFailedWriteObjectGCCandidatePublish acquires or refreshes the producer
// publish lease. A deleting row cannot be reclaimed by a producer.
func (s *Store) BeginFailedWriteObjectGCCandidatePublish(ctx context.Context, namespaceID, storageRefHash, storageRef string) (begun bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "begin_failed_write_object_gc_publish", start, &err)
	if namespaceID == "" || storageRefHash == "" || storageRef == "" {
		return false, nil
	}
	now := time.Now().UTC()
	res, err := s.db.ExecContext(ctx, `UPDATE object_gc_candidates
		SET state = ?, updated_at = GREATEST(updated_at + INTERVAL 1000 MICROSECOND, ?)
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state IN (?, ?)`,
		ObjectGCCandidatePublishing, now, namespaceID, storageRefHash, storageRef,
		ObjectGCReasonFailedWrite, ObjectGCCandidatePreparing, ObjectGCCandidatePublishing)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	return n == 1, err
}

// DeleteFailedWriteObjectGCCandidate removes only the provisional candidate
// created before a direct S3 write. A later overwrite/delete may reuse the
// same storage ref key, so cleanup must not remove that real lifecycle row.
func (s *Store) DeleteFailedWriteObjectGCCandidate(ctx context.Context, namespaceID, storageRefHash, storageRef string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_failed_write_object_gc_candidate", start, &err)
	if namespaceID == "" || storageRefHash == "" || storageRef == "" {
		return nil
	}
	_, err = s.db.ExecContext(ctx, `DELETE FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ? AND reason = ?
		AND state IN (?, ?, ?)`, namespaceID, storageRefHash, storageRef, ObjectGCReasonFailedWrite,
		ObjectGCCandidatePending, ObjectGCCandidatePreparing, ObjectGCCandidatePublishing)
	return err
}

// DeletePendingObjectGCCandidate cancels a lifecycle candidate that was
// registered before a replacement metadata CAS. Only a still-pending row may
// be removed: a claimed, failed, or deleted candidate belongs to the worker
// and must remain durable for reconciliation.
func (s *Store) DeletePendingObjectGCCandidate(ctx context.Context, namespaceID, storageRefHash, storageRef string, reason ObjectGCCandidateReason) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_pending_object_gc_candidate", start, &err)
	if namespaceID == "" || storageRefHash == "" || storageRef == "" || reason == "" {
		return nil
	}
	_, err = s.db.ExecContext(ctx, `DELETE FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state = ?`,
		namespaceID, storageRefHash, storageRef, reason, ObjectGCCandidatePending)
	return err
}

// DeleteFailedWriteObjectGCCandidateIfCurrent removes a provisional row only
// when it is still the exact snapshot observed by the worker.
func (s *Store) DeleteFailedWriteObjectGCCandidateIfCurrent(ctx context.Context, c ObjectGCCandidate) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_current_failed_write_object_gc_candidate", start, &err)
	_, err = s.db.ExecContext(ctx, `DELETE FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state IN (?, ?, ?) AND not_before = ? AND updated_at = ?`,
		c.NamespaceID, c.StorageRefHash, c.StorageRef, ObjectGCReasonFailedWrite,
		ObjectGCCandidatePending, ObjectGCCandidatePreparing, ObjectGCCandidateDeleting,
		c.NotBefore.UTC(), c.UpdatedAt.UTC())
	return err
}

func (s *Store) ListDueObjectGCCandidates(ctx context.Context, now time.Time, limit int) (out []ObjectGCCandidate, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_due_object_gc_candidates", start, &err)
	if limit <= 0 {
		limit = 100
	}
	rows, err := s.db.QueryContext(ctx, `SELECT namespace_id, storage_ref, storage_ref_hash, content_layout, reason,
			source_tenant_id, source_file_id, not_before, state, attempts, COALESCE(last_error, ''),
			created_at, updated_at
		FROM object_gc_candidates
		WHERE not_before <= ? AND (state = ? OR
			(state = ? AND updated_at <= ?) OR
			(state = ? AND updated_at <= ?))
		ORDER BY not_before ASC
		LIMIT ?`, now.UTC(), ObjectGCCandidatePending, ObjectGCCandidateDeleting,
		now.UTC().Add(-time.Hour), ObjectGCCandidatePreparing, now.UTC().Add(-staleObjectGCPreparingAge), limit)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var c ObjectGCCandidate
		if err := rows.Scan(&c.NamespaceID, &c.StorageRef, &c.StorageRefHash, &c.ContentLayout, &c.Reason,
			&c.SourceTenantID, &c.SourceFileID, &c.NotBefore, &c.State, &c.Attempts, &c.LastError,
			&c.CreatedAt, &c.UpdatedAt); err != nil {
			return nil, err
		}
		out = append(out, c)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

// ClaimObjectGCCandidate marks the exact listed row deleting. This fences
// stale workers before they touch S3; a failed delete can retry the claim.
func (s *Store) ClaimObjectGCCandidate(ctx context.Context, c *ObjectGCCandidate) (claimed bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "claim_object_gc_candidate", start, &err)
	if c == nil || c.UpdatedAt.IsZero() {
		return false, nil
	}
	claimAt := time.Now().UTC().Truncate(time.Millisecond)
	if !claimAt.After(c.UpdatedAt) {
		claimAt = c.UpdatedAt.Add(time.Millisecond)
	}
	expectedState := c.State
	if expectedState != ObjectGCCandidatePending && expectedState != ObjectGCCandidatePreparing && expectedState != ObjectGCCandidateDeleting {
		return false, nil
	}
	res, err := s.db.ExecContext(ctx, `UPDATE object_gc_candidates
		SET state = ?, updated_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state = ? AND updated_at = ?`,
		ObjectGCCandidateDeleting, claimAt, c.NamespaceID, c.StorageRefHash, c.StorageRef, c.Reason,
		expectedState, c.UpdatedAt.UTC())
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	if err != nil || n == 0 {
		return false, err
	}
	c.UpdatedAt = claimAt
	c.State = ObjectGCCandidateDeleting
	return true, nil
}

func (s *Store) NamespaceHasNonDeletedFork(ctx context.Context, namespaceID string) (out bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "namespace_has_non_deleted_fork", start, &err)
	if namespaceID == "" {
		return false, fmt.Errorf("namespace id is required")
	}
	row := s.db.QueryRowContext(ctx, `SELECT 1 FROM tenants
		WHERE storage_namespace_id = ? AND kind = ? AND status <> ?
		LIMIT 1`, namespaceID, TenantKindFork, TenantDeleted)
	var one int
	if err := row.Scan(&one); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (s *Store) PostponeObjectGCCandidateIfCurrent(ctx context.Context, c ObjectGCCandidate, notBefore time.Time, lastError string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "postpone_current_object_gc_candidate", start, &err)
	_, err = s.db.ExecContext(ctx, `UPDATE object_gc_candidates
		SET not_before = ?, last_error = ?, updated_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state IN (?, ?, ?) AND updated_at = ?`,
		notBefore.UTC(), nullStr(lastError), time.Now().UTC(), c.NamespaceID, c.StorageRefHash,
		c.StorageRef, c.Reason, ObjectGCCandidatePending, ObjectGCCandidatePreparing, ObjectGCCandidateDeleting, c.UpdatedAt.UTC())
	return err
}

func (s *Store) RetryObjectGCCandidateIfCurrent(ctx context.Context, c ObjectGCCandidate, notBefore time.Time, lastError string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "retry_current_object_gc_candidate", start, &err)
	_, err = s.db.ExecContext(ctx, `UPDATE object_gc_candidates
		SET state = ?, not_before = ?, last_error = ?, attempts = attempts + 1, updated_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state IN (?, ?, ?) AND updated_at = ?`,
		ObjectGCCandidatePending, notBefore.UTC(), nullStr(lastError), time.Now().UTC(), c.NamespaceID, c.StorageRefHash,
		c.StorageRef, c.Reason, ObjectGCCandidatePending, ObjectGCCandidatePreparing, ObjectGCCandidateDeleting, c.UpdatedAt.UTC())
	return err
}

func (s *Store) MarkObjectGCCandidateDeletedIfCurrent(ctx context.Context, c ObjectGCCandidate) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_current_object_gc_candidate_deleted", start, &err)
	now := time.Now().UTC()
	_, err = s.db.ExecContext(ctx, `UPDATE object_gc_candidates
		SET state = ?, deleted_at = ?, updated_at = ?, last_error = NULL
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?
		AND reason = ? AND state = ? AND updated_at = ?`,
		ObjectGCCandidateDeleted, now, now, c.NamespaceID, c.StorageRefHash, c.StorageRef,
		c.Reason, ObjectGCCandidateDeleting, c.UpdatedAt.UTC())
	return err
}

// UpdateTenantSchemaVersion records the tenant DB schema version after a
// successful ensure/repair cycle.  Callers should treat failures as best-
// effort: log the error but do not fail the overall operation.
func (s *Store) UpdateTenantSchemaVersion(ctx context.Context, id string, version int) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_tenant_schema_version", start, &err)
	res, err := s.db.ExecContext(ctx,
		`UPDATE tenants SET schema_version = ?, updated_at = ? WHERE id = ?`,
		version, time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	s.apiKeys.evictTenant(id)
	return nil
}

func observeMeta(ctx context.Context, op string, start time.Time, errp *error) {
	result := "ok"
	if errp != nil && *errp != nil {
		switch {
		case errors.Is(*errp, ErrNotFound), errors.Is(*errp, sql.ErrNoRows):
			result = "not_found"
		case errors.Is(*errp, ErrDuplicate):
			result = "duplicate"
		case errors.Is(*errp, ErrSharedDBCapacityExhausted):
			// Capacity races are an expected allocator miss. The caller either
			// selects another pool or reports the unrecovered wave failure.
			result = "capacity_exhausted"
		case errors.Is(*errp, sql.ErrConnDone):
			// Connection closed during shutdown — not an unexpected failure.
			result = "conn_closed"
		default:
			if strings.Contains((*errp).Error(), "database is closed") {
				result = "conn_closed"
			} else {
				result = "error"
			}
		}
		switch result {
		case "conn_closed":
			// Connection closed during shutdown — suppress the noisy log and
			// only record the metric below.
		case "capacity_exhausted":
			// Expected allocator outcome; the caller retries another candidate
			// or reports an unrecovered wave failure.
		case "not_found", "duplicate":
			logger.Warn(ctx, "meta_op_failed", zap.String("operation", op), zap.String("result", result), zap.String("detail", (*errp).Error()))
		case "error":
			logger.Error(ctx, "meta_op_failed", zap.String("operation", op), zap.String("result", result), zap.Error(*errp))
		}
	}
	metrics.RecordOperation("meta", op, result, time.Since(start))
}

func boolToInt(v bool) int {
	if v {
		return 1
	}
	return 0
}

func tenantKindForInsert(t *Tenant) TenantKind {
	if t.Kind == "" {
		return TenantKindLive
	}
	return t.Kind
}

func tenantS3EncryptionModeForInsert(t *Tenant) S3EncryptionMode {
	if t.S3EncryptionMode == "" {
		return S3EncryptionModeInherit
	}
	return t.S3EncryptionMode
}

func tenantS3BucketKeyEnabledForInsert(t *Tenant) bool {
	if t.S3BucketKeyEnabled == nil {
		return true
	}
	return *t.S3BucketKeyEnabled
}

func tenantPoolBindingStatusForInsert(status TenantPoolBindingStatus) TenantPoolBindingStatus {
	if status == "" {
		return TenantPoolBindingUsed
	}
	return status
}

func tenantPoolStatusForInsert(status TenantPoolStatus) TenantPoolStatus {
	if status == "" {
		return TenantPoolActive
	}
	return status
}

func scanTenantPoolRow(row tenantBindingScanner) (*TenantPool, error) {
	var rec TenantPool
	var orgID sql.NullString
	if err := row.Scan(&rec.PoolID, &orgID, &rec.Size, &rec.Status, &rec.CreatedAt, &rec.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	if orgID.Valid {
		rec.OrganizationID = orgID.String
	}
	return &rec, nil
}

func boolPtr(v bool) *bool {
	return &v
}

func nullStr(v string) any {
	if v == "" {
		return nil
	}
	return v
}

func nullableBytes(v []byte) any {
	if len(v) == 0 {
		return nil
	}
	return v
}

func isDuplicateEntry(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	return strings.Contains(msg, "Duplicate entry") || strings.Contains(msg, "UNIQUE constraint failed")
}

// ---------------------------------------------------------------------------
// Pod registry
// ---------------------------------------------------------------------------

// PodRegistryStatus is the lifecycle state of a pod in pod_registry.
type PodRegistryStatus string

const (
	PodRegistryActive PodRegistryStatus = "active"
	PodRegistryStale  PodRegistryStatus = "stale"
)

// PodRow mirrors a row in pod_registry.
type PodRow struct {
	PodID  string
	Addr   string
	Status PodRegistryStatus
}

// UpsertPod registers or refreshes this pod's heartbeat in pod_registry. On
// first insert the row is created with status='active'; on subsequent calls
// the addr, last_heartbeat, and status are updated (reviving a stale pod).
func (s *Store) UpsertPod(ctx context.Context, podID, addr string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_pod", start, &err)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO pod_registry (pod_id, addr, last_heartbeat, status)
		 VALUES (?, ?, NOW(3), 'active')
		 ON DUPLICATE KEY UPDATE addr = VALUES(addr), last_heartbeat = NOW(3), status = 'active'`,
		podID, addr)
	return err
}

// ListActivePods returns all active pods excluding selfID. Used by the shard
// resolver to build the active pod ring and by tests to verify registration.
func (s *Store) ListActivePods(ctx context.Context, selfID string) (out []PodRow, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_active_pods", start, &err)
	rows, err := s.db.QueryContext(ctx,
		`SELECT pod_id, addr, status FROM pod_registry WHERE status = 'active' AND pod_id != ?`,
		selfID)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var rec PodRow
		if err = rows.Scan(&rec.PodID, &rec.Addr, &rec.Status); err != nil {
			return nil, err
		}
		out = append(out, rec)
	}
	return out, rows.Err()
}

// MarkStalePods marks pods with heartbeats older than the threshold as stale.
// Leader-gated: prevents writers from pushing notifications to dead pods.
func (s *Store) MarkStalePods(ctx context.Context, before time.Time) (n int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_stale_pods", start, &err)
	res, err := s.db.ExecContext(ctx,
		`UPDATE pod_registry SET status = 'stale' WHERE last_heartbeat < ? AND status = 'active'`,
		before)
	if err != nil {
		return 0, err
	}
	n, err = res.RowsAffected()
	if err != nil {
		return 0, err
	}
	return n, nil
}

// DeletePod removes a pod and its subscriptions from the registry. Used when a
// pod is decommissioned. Best-effort: stale pod cleanup handles the common case.
// DeletePod removes a pod and its subscriptions from the registry. Both
// deletions run in a single transaction so a partial failure doesn't leave
// orphaned subscription rows. Used when a pod is decommissioned.
func (s *Store) DeletePod(ctx context.Context, podID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_pod", start, &err)
	return s.InTx(ctx, func(tx *sql.Tx) error {
		if _, err := tx.ExecContext(ctx, `DELETE FROM pod_subscriptions WHERE pod_id = ?`, podID); err != nil {
			return err
		}
		_, err := tx.ExecContext(ctx, `DELETE FROM pod_registry WHERE pod_id = ?`, podID)
		return err
	})
}

// ---------------------------------------------------------------------------
// Pod subscriptions
// ---------------------------------------------------------------------------

// UpsertPodSubscriptions records the set of tenant IDs for which podID has
// active SSE subscribers. Each (pod_id, tenant_id) pair is upserted with the
// current timestamp. Callers should call PrunePodSubscriptions to remove
// tenants that are no longer active.
func (s *Store) UpsertPodSubscriptions(ctx context.Context, podID string, tenantIDs []string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_pod_subscriptions", start, &err)
	if len(tenantIDs) == 0 {
		return nil
	}
	return s.InTx(ctx, func(tx *sql.Tx) error {
		for _, tenantID := range tenantIDs {
			if _, err := tx.ExecContext(ctx,
				`INSERT INTO pod_subscriptions (pod_id, tenant_id)
				 VALUES (?, ?)
				 ON DUPLICATE KEY UPDATE updated_at = NOW(3)`,
				podID, tenantID); err != nil {
				return err
			}
		}
		return nil
	})
}

// PrunePodSubscriptions removes pod_subscriptions rows for podID whose
// tenant_id is not in keepSet. This lets a pod prune tenants whose SSE
// subscribers have all disconnected. keepSet may be empty (prunes all).
func (s *Store) PrunePodSubscriptions(ctx context.Context, podID string, keepSet []string) (n int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "prune_pod_subscriptions", start, &err)
	if len(keepSet) == 0 {
		res, err := s.db.ExecContext(ctx,
			`DELETE FROM pod_subscriptions WHERE pod_id = ?`, podID)
		if err != nil {
			return 0, err
		}
		return res.RowsAffected()
	}
	// Build a NOT IN clause with placeholders. keepSet is bounded by the number
	// of tenants with active SSE connections on this pod (typically small).
	placeholders := make([]string, len(keepSet))
	args := make([]any, 0, len(keepSet)+1)
	args = append(args, podID)
	for i, t := range keepSet {
		placeholders[i] = "?"
		args = append(args, t)
	}
	query := fmt.Sprintf(
		`DELETE FROM pod_subscriptions WHERE pod_id = ? AND tenant_id NOT IN (%s)`,
		strings.Join(placeholders, ","))
	res, err := s.db.ExecContext(ctx, query, args...)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// ListPodSubscriptions returns the tenant IDs for which podID has active SSE
// subscribers. Used by the route cache builder to compute which peers care
// about each tenant.
func (s *Store) ListPodSubscriptions(ctx context.Context, podID string) (out []string, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_pod_subscriptions", start, &err)
	rows, err := s.db.QueryContext(ctx,
		`SELECT tenant_id FROM pod_subscriptions WHERE pod_id = ?`, podID)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var tenantID string
		if err = rows.Scan(&tenantID); err != nil {
			return nil, err
		}
		out = append(out, tenantID)
	}
	return out, rows.Err()
}

// ListAllPodSubscriptions returns all (pod_id, tenant_id) pairs from
// pod_subscriptions. Used by the route cache builder to construct a
// tenant→pods reverse index for push notification routing.
func (s *Store) ListAllPodSubscriptions(ctx context.Context) (out []PodSubscriptionRow, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_all_pod_subscriptions", start, &err)
	rows, err := s.db.QueryContext(ctx,
		`SELECT pod_id, tenant_id FROM pod_subscriptions`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var rec PodSubscriptionRow
		if err = rows.Scan(&rec.PodID, &rec.TenantID); err != nil {
			return nil, err
		}
		out = append(out, rec)
	}
	return out, rows.Err()
}

// PodSubscriptionRow mirrors a row in pod_subscriptions.
type PodSubscriptionRow struct {
	PodID    string
	TenantID string
}

// DeletePodSubscriptions removes all subscription rows for podID. Called when
// a pod is decommissioned or marked stale by the leader sweeper.
func (s *Store) DeletePodSubscriptions(ctx context.Context, podID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_pod_subscriptions", start, &err)
	_, err = s.db.ExecContext(ctx, `DELETE FROM pod_subscriptions WHERE pod_id = ?`, podID)
	return err
}

// ListStalePods returns pod IDs with status='stale'. Used by the leader sweeper
// to clean up subscriptions for dead pods.
func (s *Store) ListStalePods(ctx context.Context) (out []string, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_stale_pods", start, &err)
	rows, err := s.db.QueryContext(ctx,
		`SELECT pod_id FROM pod_registry WHERE status = 'stale'`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var podID string
		if err = rows.Scan(&podID); err != nil {
			return nil, err
		}
		out = append(out, podID)
	}
	return out, rows.Err()
}

// DeleteSubscriptionsForStalePods deletes pod_subscriptions rows whose pod is
// currently stale, using a subquery join so the stale check and the delete are
// atomic. This avoids a TOCTOU race where a pod recovers to active between
// ListStalePods and DeletePodSubscriptions, which would delete subscriptions for
// a live pod. Returns the number of deleted rows.
func (s *Store) DeleteSubscriptionsForStalePods(ctx context.Context) (n int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_subscriptions_for_stale_pods", start, &err)
	res, err := s.db.ExecContext(ctx,
		`DELETE ps FROM pod_subscriptions ps
		 INNER JOIN pod_registry pr ON ps.pod_id = pr.pod_id
		 WHERE pr.status = 'stale'`)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// ---------------------------------------------------------------------------
// Unified tenant notify outbox
// ---------------------------------------------------------------------------

// TenantNotifyRow mirrors a row in tenant_notify_outbox. Each row is a
// lightweight work signal: tenant_id identifies the tenant and work_mask is a
// bitmask of work types to dispatch (SSE, semantic, file_gc, metrics cleanup,
// or API-key cache cleanup).
type TenantNotifyRow struct {
	ID             uint64
	TenantID       string
	TiDBCloudOrgID string
	WorkMask       int
	CreatedAt      time.Time
}

// InsertTenantNotify writes a unified work signal to tenant_notify_outbox.
// This is best-effort: if the INSERT fails, the work is still durable in the
// per-tenant TiDB and will be recovered by the safety-net scan. The central
// meta DB is always provisioned so this write never wakes a serverless tenant.
func (s *Store) InsertTenantNotify(ctx context.Context, tenantID string, workMask int) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_tenant_notify", start, &err)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_notify_outbox (tenant_id, work_mask) VALUES (?, ?)`,
		tenantID, workMask)
	return err
}

// UpsertDeleteReconcilePending durably records that the tenant's
// delete-orphan reconciliation sweep is owed and should run no earlier than
// dueAt. One row per tenant; the write is MONOTONIC — it creates the row when
// absent and otherwise keeps the later of the stored and new due times, so no
// write (pre-delete crash-recovery lead, throttle extension, sweep-failure
// retry) can ever shorten an in-flight delete's protection. The row is
// deleted by DeleteDeleteReconcilePending once a sweep succeeds, and is
// re-dispatched by the tenant outbox poller's pending scan until then, so a
// throttled or failed reconcile survives pod restarts.
func (s *Store) UpsertDeleteReconcilePending(ctx context.Context, tenantID string, dueAt time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_delete_reconcile_pending", start, &err)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO delete_reconcile_pending (tenant_id, due_at) VALUES (?, ?)
		 ON DUPLICATE KEY UPDATE due_at = GREATEST(due_at, VALUES(due_at))`,
		tenantID, dueAt.UTC())
	return err
}

// InsertDeleteReconcilePendingIfAbsent records the tenant's pending
// delete-orphan reconciliation marker only when none exists (INSERT IGNORE).
// Every pod that reads a WorkDeleteReconcile outbox row — owner or not — uses
// this to make the request durable before any pod advances its cursor, and
// without disturbing a due time that a throttled worker already extended.
func (s *Store) InsertDeleteReconcilePendingIfAbsent(ctx context.Context, tenantID string, dueAt time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_delete_reconcile_pending_if_absent", start, &err)
	_, err = s.db.ExecContext(ctx,
		`INSERT IGNORE INTO delete_reconcile_pending (tenant_id, due_at) VALUES (?, ?)`,
		tenantID, dueAt.UTC())
	return err
}

// RefreshDeleteReconcilePending is the post-delete refresh: it guarantees a
// pending row exists (inserting dueAt if absent) and moves an EXPIRED row's
// due time to dueAt, but NEVER pulls a future-dated row earlier. A
// future-dated row is another in-flight delete's crash-recovery protection;
// overwriting it with due=now would let a sweep clear that delete's marker
// before it completes. Immediacy for the completed delete comes from the
// in-process worker kick, not from this row.
func (s *Store) RefreshDeleteReconcilePending(ctx context.Context, tenantID string, dueAt time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "refresh_delete_reconcile_pending", start, &err)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO delete_reconcile_pending (tenant_id, due_at) VALUES (?, ?)
		 ON DUPLICATE KEY UPDATE due_at = IF(due_at <= ?, ?, due_at)`,
		tenantID, dueAt.UTC(), dueAt.UTC(), dueAt.UTC())
	return err
}

// RearmDeleteReconcilePending schedules the tenant's follow-up
// delete-reconcile round at dueAt after a sweep that repaired rows or
// stopped mid-pass, in ONE atomic statement: a missing row is inserted with
// dueAt, an owed row (due_at <= cutoff, where cutoff is the sweep's start
// time) is moved to dueAt, and a row due after cutoff — written by a NEWER
// delete while the sweep was running — keeps its own due time. The
// atomicity matters because the piggyback sweep is gated on this marker: a
// clear-then-insert sequence could lose the follow-up round to a crash or
// transient error between the two writes.
func (s *Store) RearmDeleteReconcilePending(ctx context.Context, tenantID string, dueAt, cutoff time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "rearm_delete_reconcile_pending", start, &err)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO delete_reconcile_pending (tenant_id, due_at) VALUES (?, ?)
		 ON DUPLICATE KEY UPDATE due_at = IF(due_at <= ?, VALUES(due_at), due_at)`,
		tenantID, dueAt.UTC(), cutoff.UTC())
	return err
}

// ExtendDeleteReconcilePending moves an EXISTING pending row's due time to
// dueAt, but never EARLIER than its current due time (monotonic GREATEST). It
// never inserts: callers use it to delay an owed sweep (throttle extension)
// without re-creating a marker for work already done or stripping another
// in-flight delete's future-dated crash-recovery protection, and the UPDATE is
// atomic so a concurrent clear cannot be followed by a stale re-insert.
// extended reports whether a row was modified; a missing row or a no-op
// (dueAt <= current due) both report false.
func (s *Store) ExtendDeleteReconcilePending(ctx context.Context, tenantID string, dueAt time.Time) (extended bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "extend_delete_reconcile_pending", start, &err)
	res, err := s.db.ExecContext(ctx,
		`UPDATE delete_reconcile_pending SET due_at = GREATEST(due_at, ?) WHERE tenant_id = ?`,
		dueAt.UTC(), tenantID)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

// DeleteDeleteReconcilePending removes the tenant's pending delete-orphan
// reconciliation marker after a successful sweep. A missing row is a no-op.
func (s *Store) DeleteDeleteReconcilePending(ctx context.Context, tenantID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_delete_reconcile_pending", start, &err)
	_, err = s.db.ExecContext(ctx,
		`DELETE FROM delete_reconcile_pending WHERE tenant_id = ?`, tenantID)
	return err
}

// DeleteDeleteReconcilePendingBefore removes the tenant's pending marker only
// when it was due no later than cutoff. A sweep uses its own start time as the
// cutoff so a marker written by a NEWER delete while the sweep was running
// survives and is still re-dispatched by the poller scan.
func (s *Store) DeleteDeleteReconcilePendingBefore(ctx context.Context, tenantID string, cutoff time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_delete_reconcile_pending_before", start, &err)
	_, err = s.db.ExecContext(ctx,
		`DELETE FROM delete_reconcile_pending WHERE tenant_id = ? AND due_at <= ?`,
		tenantID, cutoff.UTC())
	return err
}

// GetDeleteReconcilePending returns the tenant's pending delete-reconcile
// marker due time, or found=false when no marker exists. The per-tenant
// worker's piggyback reconcile uses it as the gate for the periodic
// backstop sweep: no marker means no delete is owed a reconciliation, so
// the full-table scan is skipped entirely.
func (s *Store) GetDeleteReconcilePending(ctx context.Context, tenantID string) (dueAt time.Time, found bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_delete_reconcile_pending", start, &err)
	err = s.db.QueryRowContext(ctx,
		`SELECT due_at FROM delete_reconcile_pending WHERE tenant_id = ?`, tenantID).Scan(&dueAt)
	if errors.Is(err, sql.ErrNoRows) {
		return time.Time{}, false, nil
	}
	if err != nil {
		return time.Time{}, false, err
	}
	return dueAt, true, nil
}

// ListDeleteReconcilePendingDue returns tenant IDs whose pending
// delete-orphan reconciliation sweep is due (due_at <= now), ordered by due
// time, up to limit. This is the durable recovery input for the tenant outbox
// poller: unlike the once-consumed outbox rows, pending rows survive restarts
// and are re-kicked until the sweep succeeds.
func (s *Store) ListDeleteReconcilePendingDue(ctx context.Context, now time.Time, limit int) (out []string, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_delete_reconcile_pending_due", start, &err)
	if limit <= 0 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx,
		`SELECT tenant_id FROM delete_reconcile_pending WHERE due_at <= ? ORDER BY due_at, tenant_id LIMIT ?`,
		now.UTC(), limit)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var tenantID string
		if err := rows.Scan(&tenantID); err != nil {
			return nil, err
		}
		out = append(out, tenantID)
	}
	return out, rows.Err()
}

func insertTenantNotifyTx(ctx context.Context, tx *sql.Tx, tenantID string, workMask int) error {
	_, err := tx.ExecContext(ctx,
		`INSERT INTO tenant_notify_outbox (tenant_id, work_mask) VALUES (?, ?)`,
		tenantID, workMask)
	return err
}

// InsertTenantNotifyTx appends a tenant outbox notification in the caller's
// transaction so a configuration mutation and its cache invalidation cannot
// become visible independently.
func InsertTenantNotifyTx(ctx context.Context, tx *sql.Tx, tenantID string, workMask int) error {
	return insertTenantNotifyTx(ctx, tx, tenantID, workMask)
}

// TenantNotifyEntry is one tenant's pending work signal for
// InsertTenantNotifyBatch.
type TenantNotifyEntry struct {
	TenantID string
	WorkMask int
}

// insertTenantNotifyBatchChunkSize caps the number of rows per multi-row
// INSERT statement.
const insertTenantNotifyBatchChunkSize = 1000

// InsertTenantNotifyBatch writes many unified work signals as multi-row
// INSERTs (chunked at insertTenantNotifyBatchChunkSize rows per statement).
// It exists because single-row InsertTenantNotify calls dominated metadb
// write load under stress: the server's notify coalescer OR-merges signals
// per tenant and flushes them here in one statement per flush window.
// Same best-effort semantics as InsertTenantNotify; each chunk commits
// atomically, so poller cursor semantics (id > cursor ORDER BY id) are
// unaffected.
func (s *Store) InsertTenantNotifyBatch(ctx context.Context, entries []TenantNotifyEntry) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_tenant_notify_batch", start, &err)
	for from := 0; from < len(entries); from += insertTenantNotifyBatchChunkSize {
		to := from + insertTenantNotifyBatchChunkSize
		if to > len(entries) {
			to = len(entries)
		}
		if err := s.insertTenantNotifyChunk(ctx, entries[from:to]); err != nil {
			return err
		}
	}
	return nil
}

func (s *Store) insertTenantNotifyChunk(ctx context.Context, entries []TenantNotifyEntry) error {
	var sb strings.Builder
	sb.WriteString(`INSERT INTO tenant_notify_outbox (tenant_id, work_mask) VALUES `)
	args := make([]any, 0, len(entries)*2)
	for i, e := range entries {
		if i > 0 {
			sb.WriteByte(',')
		}
		sb.WriteString("(?,?)")
		args = append(args, e.TenantID, e.WorkMask)
	}
	_, err := s.db.ExecContext(ctx, sb.String(), args...)
	return err
}

// ListTenantNotifySince returns outbox rows with id > afterID, ordered by id,
// up to limit. The consumer advances its cursor to the last row's id so
// subsequent calls do not re-read the same rows.
func (s *Store) ListTenantNotifySince(ctx context.Context, afterID uint64, limit int) (out []TenantNotifyRow, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenant_notify_since", start, &err)
	if limit <= 0 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx,
		`SELECT o.id, o.tenant_id,
		        COALESCE(CASE WHEN t.provider = ? THEN d.org_id ELSE b.organization_id END, ''),
		        o.work_mask, o.created_at
		 FROM tenant_notify_outbox o
		 LEFT JOIN tenants t ON t.id = o.tenant_id
		 LEFT JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = o.tenant_id
		 LEFT JOIN fs_registry f ON f.tenant_id = o.tenant_id
		 LEFT JOIN tenant_placements p ON p.fs_id = f.fs_id
		 LEFT JOIN db_pool d ON d.db_id = p.db_id
		 WHERE o.id > ?
		 ORDER BY o.id
		 LIMIT ?`,
		tidbCloudNativeSharedProvider, afterID, limit)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var rec TenantNotifyRow
		if err = rows.Scan(&rec.ID, &rec.TenantID, &rec.TiDBCloudOrgID, &rec.WorkMask, &rec.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, rec)
	}
	return out, rows.Err()
}

// MaxTenantNotifyID returns the current maximum id in tenant_notify_outbox, or
// 0 if the table is empty. Used by a pod on first launch to skip historical rows
// (the pod never owned work before its first start).
func (s *Store) MaxTenantNotifyID(ctx context.Context) (out uint64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "max_tenant_notify_id", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT COALESCE(MAX(id), 0) FROM tenant_notify_outbox`)
	if err = row.Scan(&out); err != nil {
		return 0, err
	}
	return out, nil
}

// tenantOutboxCursorFreshnessBound is the maximum age of a cursor's
// updated_at for that cursor to hold back outbox pruning. A pod whose
// heartbeat is healthy but whose outbox poller is stuck stops refreshing its
// cursor; ignoring cursors older than this bound prevents such a stalled pod
// from pinning MIN(last_id) forever and growing the 1-hour outbox unboundedly.
const tenantOutboxCursorFreshnessBound = 10 * time.Minute

// DeleteTenantNotifyBefore prunes outbox rows older than the given threshold.
// Leader-gated; retention is relative to DB insert time (created_at). Rows are
// only pruned up to the minimum last_id across pods' cursors so a lagging pod
// that has not yet processed a row does not lose its work signal. Cursors
// whose updated_at is older than tenantOutboxCursorFreshnessBound are ignored
// when computing that floor: SSE notify rows are lossy hints, and pruning past
// a stalled pod's cursor is safe — its online clients simply STALL (no new
// wake-ups) until they reconnect and recover via replay (or, if enabled, the
// per-connection liveness poll detects the signal loss). When no fresh
// cursors remain (e.g. before any pod has registered), pruning falls back to
// age alone.
func (s *Store) DeleteTenantNotifyBefore(ctx context.Context, before time.Time) (n int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_tenant_notify_before", start, &err)
	freshSince := start.Add(-tenantOutboxCursorFreshnessBound)
	res, err := s.db.ExecContext(ctx, `DELETE FROM tenant_notify_outbox
WHERE created_at < ?
AND (
  NOT EXISTS (SELECT 1 FROM tenant_outbox_cursor WHERE updated_at >= ?)
  OR id <= (SELECT MIN(last_id) FROM tenant_outbox_cursor WHERE updated_at >= ?)
)`, before, freshSince, freshSince)
	if err != nil {
		return 0, err
	}
	n, err = res.RowsAffected()
	if err != nil {
		return 0, err
	}
	return n, nil
}

// ---------------------------------------------------------------------------
// Tenant outbox cursor
// ---------------------------------------------------------------------------

// TenantOutboxCursorRow mirrors a row in tenant_outbox_cursor.
type TenantOutboxCursorRow struct {
	PodID     string
	LastID    uint64
	UpdatedAt time.Time
}

// GetTenantOutboxCursor returns the persisted cursor (last_id) for podID, or
// ErrNotFound if no row exists (first launch). The pod uses this on restart to
// resume processing without skipping work.
func (s *Store) GetTenantOutboxCursor(ctx context.Context, podID string) (out *TenantOutboxCursorRow, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tenant_outbox_cursor", start, &err)
	row := s.db.QueryRowContext(ctx,
		`SELECT pod_id, last_id, updated_at FROM tenant_outbox_cursor WHERE pod_id = ?`, podID)
	var rec TenantOutboxCursorRow
	if err = row.Scan(&rec.PodID, &rec.LastID, &rec.UpdatedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	return &rec, nil
}

// UpsertTenantOutboxCursor persists (or refreshes) the cursor for podID. The
// poller flushes its in-memory last_id every few seconds so a restart resumes
// from the last processed row. updated_at advances only when last_id advances:
// the DeleteTenantNotifyBefore freshness bound uses updated_at to detect
// stalled pods, and a pod whose poll query keeps failing (while its cursor
// flush still succeeds) must be allowed to go stale. A healthy caught-up pod
// also goes stale during zero-traffic periods and falls out of the prune
// floor — safe, because age-only pruning then applies and the 1-hour age gate
// leaves it an hour of rows to resume from.
func (s *Store) UpsertTenantOutboxCursor(ctx context.Context, podID string, lastID uint64) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "upsert_tenant_outbox_cursor", start, &err)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_outbox_cursor (pod_id, last_id) VALUES (?, ?)
		 ON DUPLICATE KEY UPDATE updated_at = IF(VALUES(last_id) > last_id, CURRENT_TIMESTAMP(3), updated_at),
		                         last_id = VALUES(last_id)`,
		podID, lastID)
	// NOTE: assignment order is load-bearing. MySQL/TiDB evaluate ON
	// DUPLICATE KEY UPDATE assignments left to right, so the updated_at
	// comparison must run BEFORE last_id is overwritten — otherwise
	// VALUES(last_id) > last_id compares the new value against itself and
	// updated_at never refreshes.
	return err
}

// DeleteTenantOutboxCursor removes the cursor row for podID. Used when a pod
// is decommissioned so its cursor doesn't linger.
// DeleteTenantOutboxCursor deletes the cursor row for the given pod, but only
// if the pod is currently stale in pod_registry. This prevents a TOCTOU race
// where a pod recovers between ListStalePods and the cursor delete — a
// recovered pod retains its cursor so its last_id is not lost. The conditional
// join mirrors DeleteSubscriptionsForStalePods.
func (s *Store) DeleteTenantOutboxCursor(ctx context.Context, podID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_tenant_outbox_cursor", start, &err)
	_, err = s.db.ExecContext(ctx,
		`DELETE FROM tenant_outbox_cursor WHERE pod_id = ?
		 AND EXISTS (SELECT 1 FROM pod_registry WHERE pod_id = ? AND status = 'stale')`,
		podID, podID)
	return err
}

// ---------------------------------------------------------------------------
// Shared maintenance state
// ---------------------------------------------------------------------------

// ClaimSharedMaintenanceRun atomically claims the right to run the named
// cluster-wide maintenance task (e.g. the shared-pool fs_events sweep).
// Exactly one caller across all pods wins per minInterval window: the UPDATE
// only succeeds when the stored last_run_at is older than the interval, and
// the winner stamps its own run in the same statement. The row is created
// lazily via INSERT IGNORE (seeded at the epoch so the first claim always
// wins). Callers should keep a cheap local pre-filter so the common case
// never reaches the meta DB.
func (s *Store) ClaimSharedMaintenanceRun(ctx context.Context, name string, minInterval time.Duration) (claimed bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "claim_shared_maintenance_run", start, &err)
	if _, err = s.db.ExecContext(ctx,
		`INSERT IGNORE INTO shared_maintenance_state (name, last_run_at) VALUES (?, ?)`,
		name, time.Unix(0, 0).UTC()); err != nil {
		return false, fmt.Errorf("seed shared maintenance state %q: %w", name, err)
	}
	res, execErr := s.db.ExecContext(ctx,
		`UPDATE shared_maintenance_state SET last_run_at = NOW(3)
		 WHERE name = ? AND last_run_at < DATE_SUB(NOW(3), INTERVAL ? SECOND)`,
		name, int64(minInterval/time.Second))
	if execErr != nil {
		err = fmt.Errorf("claim shared maintenance %q: %w", name, execErr)
		return false, err
	}
	n, rowsErr := res.RowsAffected()
	if rowsErr != nil {
		err = fmt.Errorf("claim shared maintenance %q rows affected: %w", name, rowsErr)
		return false, err
	}
	return n == 1, nil
}

// ListAllActivePodIDs returns the pod_id of every active pod in pod_registry,
// ordered by pod_id. Used by the shard resolver to build the active pod ring.
func (s *Store) ListAllActivePodIDs(ctx context.Context) (out []string, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_all_active_pod_ids", start, &err)
	rows, err := s.db.QueryContext(ctx,
		`SELECT pod_id FROM pod_registry WHERE status = 'active' ORDER BY pod_id`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var podID string
		if err = rows.Scan(&podID); err != nil {
			return nil, err
		}
		out = append(out, podID)
	}
	return out, rows.Err()
}