package tenant

import (
	"container/list"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/encrypt"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/migrate"
	"github.com/mem9-ai/drive9/pkg/mysqlutil"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/semantic"
	"github.com/mem9-ai/drive9/pkg/tenant/schema"
	"go.uber.org/zap"
)

type PoolConfig struct {
	MaxTenants         int
	S3Dir              string
	PublicURL          string
	S3Bucket           string
	S3Region           string
	S3Prefix           string
	S3RoleARN          string
	S3Endpoint         string
	S3ForcePathStyle   bool
	S3AccessKeyID      string
	S3SecretAccessKey  string
	S3SessionToken     string
	S3EncryptionPolicy meta.S3EncryptionPolicy
	S3ExpressBucket    string
	S3ExpressPrefix    string

	BackendOptions backend.Options

	// SkipTiDBSchemaCheck disables the TiDB auto-embedding schema
	// ensure/validate steps during Acquire. Used in tests that run
	// against plain MySQL but need a TiDB-class provider for vault.
	SkipTiDBSchemaCheck bool

	// DisableDatabaseAutoEmbedding makes newly resolved NULL tenant embedding
	// profiles use fts_only mode. Once a tenant profile has an embedding_mode,
	// the tenant-level mode is the source of truth.
	DisableDatabaseAutoEmbedding bool

	// LeaderChecker, when set, gates per-tenant FileGCWorker to run only on
	// the leader pod. When nil, FileGCWorker starts unconditionally (single-pod).
	//
	// FileGC reacts to leadership transitions and races safely with concurrent
	// backend creation: the server calls StartAllFileGC on leadership gain and
	// StopAllFileGC on loss. These set a pool-owned fileGCEnabled flag under p.mu,
	// and Acquire starts FileGC on a newly inserted backend according to that
	// flag (read under p.mu), so a backend created concurrently with a transition
	// resolves to the post-transition state instead of a moment-in-time
	// IsLeader() snapshot.
	LeaderChecker LeaderChecker

	// IdleTimeout controls how long a standalone cached backend can stay in the
	// warm cache without activity before the idle reaper evicts it. Shared
	// tenant entries intentionally bypass this per-tenant TTL; they are reclaimed
	// by capacity pressure or together when their physical shared DB has had no
	// real activity for defaultSharedDBHandleIdleTTL. 0 disables standalone
	// idle eviction; shared physical-DB group reaping remains active.
	//
	// For standalone tenants, "activity" includes foreground requests and
	// tenant-specific durable work. Background work refreshes this idle TTL but
	// does not promote the entry in the capacity LRU.
	//
	// AcquireCached (safety-net scan) does NOT refresh the idle timer on
	// acquire or release, so warm-only periodic observation cannot keep an
	// idle tenant warm forever. Object-GC may cold-open a tenant for a due
	// candidate and intentionally keeps it warm for one idle-TTL window
	// per due attempt.
	IdleTimeout time.Duration

	// IdleReapInterval is how often the reaper scans standalone entries and
	// activity-idle shared physical-DB groups. It defaults to
	// defaultTenantPoolIdleReapInterval when not set.
	IdleReapInterval time.Duration

	// SharedDBForcePlaintext, when true, opens shared-schema DB handles without
	// TLS regardless of the meta-persisted SharedDB.TLSMode. Set once at Pool
	// construction for LocalClustersAPI (Docker/Podman TiDB has no TLS). Do not
	// re-read process env on each open — a leaked DRIVE9_TIDBCLOUD_CLUSTERS_BACKEND
	// must not silently strip TLS from production connections.
	SharedDBForcePlaintext bool
}

// LeaderChecker reports whether the current process is the leader. Used by
// the tenant pool to gate per-tenant background workers (e.g. FileGCWorker).
type LeaderChecker interface {
	IsLeader() bool
}

// ActiveTenantStores snapshots the (tenantID, store) pairs currently cached in
// the pool. It is intended for leader-only background maintenance (e.g. the
// fs_events cleanup goroutine) that should operate only on tenants with a
// live backend — fs_events rows only accumulate for tenants that have served
// traffic and thus have a cached backend. Unlike ListTenantsByStatus +
// Acquire, this does NOT open backends for dormant tenants (which would
// exhaust connections and fail for tenants whose credentials are not
// materialized), so it is cheap and safe to call on every cleanup tick.
// The returned stores are pinned only for the duration of the snapshot call;
// callers must not retain them past the cleanup operation (a concurrent
// retirement may close the underlying DB once this returns). In practice the
// leader cleanup runs briefly and the entry's refs prevent close mid-use only
// while Acquire is held — for a read-only COUNT/DELETE that tolerates a stale
// reference error, this snapshot is sufficient and matches the existing
// StartAllFileGC iteration pattern.
func (p *Pool) ActiveTenantStores() []TenantStoreEntry {
	if p == nil {
		return nil
	}
	p.mu.Lock()
	out := make([]TenantStoreEntry, 0, len(p.items))
	for _, e := range p.items {
		if e.store == nil || e.retired {
			continue
		}
		out = append(out, TenantStoreEntry{TenantID: e.tenantID, Store: e.store})
	}
	p.mu.Unlock()
	return out
}

// TenantStoreEntry is a (tenantID, store) pair returned by ActiveTenantStores.
type TenantStoreEntry struct {
	TenantID string
	Store    *datastore.Store
}

type Pool struct {
	mu                  sync.Mutex
	cfg                 PoolConfig
	enc                 encrypt.Encryptor
	metaStore           *meta.Store // central server DB for quota operations; nil disables central quota
	items               map[string]*entry
	order               *list.List
	maxSize             int
	tenantWorkNotifier  atomic.Pointer[func(tenantID, tidbCloudOrgID string, workMask int)]
	embeddingResolverMu sync.RWMutex
	embeddingResolver   func(context.Context, string) (backend.AppEmbeddingRuntime, bool, error)
	// deleteReconcilePersister, when set, durably records that a pool backend's
	// tenant owes a delete-orphan reconciliation sweep before a recursive
	// directory delete starts (see backend.SetDeleteReconcilePersister).
	deleteReconcilePersister atomic.Pointer[func(tenantID string) error]
	// fileGCEnabled is retained for backward compatibility but no longer used:
	// FileGC is now kick-driven through the unified tenant worker, not a
	// per-backend goroutine.
	fileGCEnabled bool
	idleTimeout   time.Duration
	reapInterval  time.Duration
	reapStop      context.CancelFunc
	reapWG        sync.WaitGroup
	// sharedDBs caches one *sql.DB handle per shared-schema database (keyed by
	// db_pool.id). All tenants placed on the same shared DB share its
	// handle; each Acquire still gets its own Store (carrying that tenant's
	// fs_id scope) over the shared handle.
	sharedMu         sync.Mutex
	sharedDBs        map[int64]*sql.DB
	sharedDBRefs     map[int64]int
	sharedDBLastUsed map[int64]time.Time
	sharedDBLabels   map[int64]sharedDBMetricLabels
	sharedDBOpenMu   map[int64]*sync.Mutex
	sharedDBClosed   bool
}

// SetEmbeddingRuntimeResolver wires cold-backend app embedding resolution.
// The callback is never invoked on a pool cache hit.
func (p *Pool) SetEmbeddingRuntimeResolver(resolve func(context.Context, string) (backend.AppEmbeddingRuntime, bool, error)) {
	if p == nil {
		return
	}
	p.embeddingResolverMu.Lock()
	p.embeddingResolver = resolve
	p.embeddingResolverMu.Unlock()
}

// SkipTiDBSchemaCheck reports whether this pool's caller explicitly disabled
// TiDB schema checks, primarily for tests using a TiDB-class provider against
// a fixture that cannot support the provider's optional DDL.
func (p *Pool) SkipTiDBSchemaCheck() bool {
	return p != nil && p.cfg.SkipTiDBSchemaCheck
}

func (p *Pool) resolveAppEmbeddingRuntime(ctx context.Context, tenantID string) (backend.AppEmbeddingRuntime, bool, error) {
	p.embeddingResolverMu.RLock()
	resolve := p.embeddingResolver
	p.embeddingResolverMu.RUnlock()
	if resolve == nil {
		return backend.AppEmbeddingRuntime{}, false, nil
	}
	return resolve(ctx, tenantID)
}

// sharedDBMetricLabels remembers the metric identity of a cached shared DB
// handle so ref-count gauges and series cleanup do not need a meta lookup.
type sharedDBMetricLabels struct {
	orgID string
	uuid  string
}

type tenantAutoEmbeddingProfile struct {
	schemaProfile schema.TiDBAutoEmbeddingProfile
	provider      schema.TiDBAutoEmbeddingProviderConfig
	mode          string
	modeWasNull   bool
}

var (
	applyTiDBAutoEmbeddingProviderConfig    = schema.ApplyTiDBAutoEmbeddingProviderConfig
	ensureTiDBSchemaForAutoEmbeddingProfile = schema.EnsureTiDBSchemaForAutoEmbeddingProfile
	ensureTiDBSchemaForFTSOnlyProfile       = schema.EnsureTiDBSchemaForFTSOnlyProfile
	ensureSharedDBSchema                    = schema.EnsureSharedSchema
	ErrSharedDBSchemaEnsureBusy             = errors.New("shared schema ensure already running")
	errAppendLogConfigurationUnavailable    = errors.New("append-log configuration is unavailable")
	defaultTenantPoolDrainTimeout           = 30 * time.Second
	defaultTenantPoolMaxTenants             = 1024
	defaultTenantPoolIdleReapInterval       = 2 * time.Minute
	defaultSharedDBHandleIdleTTL            = 10 * time.Minute
)

const (
	schemaLockReleaseTimeout        = 5 * time.Second
	schemaLockForegroundWaitSeconds = 1
	// schemaOpDetachedTimeout bounds detached contexts used for one-time
	// Acquire-time schema convergence work (standalone tenant schema
	// ensure/repair plus version bookkeeping, split-tables migration, and
	// shared physical-DB migration). It must be generous enough for
	// multi-pass DDL repair plus storage_ref_hash/path_hash backfills on a
	// cold or large database.
	schemaOpDetachedTimeout = 10 * time.Minute
)

// detachSchemaOp returns a bounded context detached from caller cancellation
// for one-time schema convergence work. Context values (trace IDs, tenant IDs)
// are preserved; only cancellation and the caller's deadline are dropped.
func detachSchemaOp(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.WithoutCancel(ctx), schemaOpDetachedTimeout)
}

func ensureTiDBSchemaForEmbeddingMode(ctx context.Context, db *sql.DB, mode string, profile schema.TiDBAutoEmbeddingProfile) error {
	tidbMode, err := TiDBEmbeddingModeForTenantMode(mode)
	if err != nil {
		return err
	}
	switch tidbMode {
	case schema.TiDBEmbeddingModeAuto:
		return ensureTiDBSchemaForAutoEmbeddingProfile(ctx, db, profile)
	case schema.TiDBEmbeddingModeFTSOnly:
		return ensureTiDBSchemaForFTSOnlyProfile(ctx, db, profile)
	default:
		return schema.EnsureTiDBSchemaForEmbeddingModeProfile(ctx, db, tidbMode, profile)
	}
}

// withSchemaAdvisoryLock serializes schema work on the target DB itself via
// GET_LOCK(lockName). desc identifies the lock in error messages. busyErr is
// required (and returned) when waitForLock is false and the lock is already
// held. The lock is
// session-owned; if that target connection is lost mid-DDL, TiDB releases it
// and a later attempt can overlap the still-finishing statement.
// Lock acquisition and waiting run on the caller's context so a foreground
// caller keeps its deadline while another migration is in flight; fn receives
// that same context, and callers running one-time convergence work inside fn
// should detach it themselves (see detachSchemaOp).
func withSchemaAdvisoryLock(
	ctx context.Context,
	db *sql.DB,
	lockName, desc string,
	waitForLock bool,
	busyErr error,
	fn func(context.Context) error,
) (err error) {
	if !waitForLock && busyErr == nil {
		return fmt.Errorf("%s: fail-fast lock acquisition requires a busyErr", desc)
	}
	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = conn.Close() }()
	// MySQL honors a zero GET_LOCK timeout as non-blocking. TiDB clamps its
	// minimum timeout to one second, so the provisioning path is fail-fast but
	// can wait for approximately one second when the same physical DB is busy.
	waitSeconds := 0
	if waitForLock {
		waitSeconds = schemaLockForegroundWaitSeconds
	}
	for {
		var acquired sql.NullInt64
		if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", lockName, waitSeconds).Scan(&acquired); err != nil {
			return fmt.Errorf("acquire %s: %w", desc, err)
		}
		if !acquired.Valid {
			return fmt.Errorf("acquire %s returned NULL", desc)
		}
		if acquired.Int64 == 1 {
			break
		}
		if !waitForLock {
			return busyErr
		}
		if err := ctx.Err(); err != nil {
			return fmt.Errorf("wait for %s: %w", desc, err)
		}
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), schemaLockReleaseTimeout)
		defer cancel()
		var released sql.NullInt64
		releaseErr := conn.QueryRowContext(releaseCtx, "SELECT RELEASE_LOCK(?)", lockName).Scan(&released)
		if releaseErr != nil {
			err = errors.Join(err, fmt.Errorf("release %s: %w", desc, releaseErr))
			return
		}
		if !released.Valid || released.Int64 != 1 {
			err = errors.Join(err, fmt.Errorf("%s was not held by current connection", desc))
		}
	}()
	return fn(ctx)
}

// withSharedDBSchemaAdvisoryLock serializes schema ensure on the target
// physical DB itself. It does not retain a MetaDB connection, and distinct
// db_pool rows use distinct lock names so their DDL remains concurrent.
func withSharedDBSchemaAdvisoryLock(
	ctx context.Context,
	db *sql.DB,
	dbID int64,
	waitForLock bool,
	fn func(context.Context) error,
) error {
	lockName := fmt.Sprintf("drive9:shared-schema:%d", dbID)
	desc := fmt.Sprintf("shared schema advisory lock for db pool %d", dbID)
	busyErr := fmt.Errorf("%w for db pool %d", ErrSharedDBSchemaEnsureBusy, dbID)
	return withSchemaAdvisoryLock(ctx, db, lockName, desc, waitForLock, busyErr, fn)
}

// withTenantSchemaAdvisoryLock serializes Acquire-time schema ensure/repair on
// the standalone tenant's own physical DB, so concurrent cold opens (same pod
// or across pods) cannot start overlapping detached DDL/backfill runs for one
// tenant. It always waits for the lock; the wait runs on the caller's context
// so a request keeps its deadline while another migration is in flight.
func withTenantSchemaAdvisoryLock(
	ctx context.Context,
	db *sql.DB,
	tenantID string,
	fn func(context.Context) error,
) error {
	lockName := tenantSchemaLockName(tenantID)
	desc := fmt.Sprintf("tenant schema advisory lock for tenant %s", tenantID)
	return withSchemaAdvisoryLock(ctx, db, lockName, desc, true, nil, fn)
}

// tenantSchemaLockName derives the MySQL/TiDB named lock for a tenant schema
// migration. MySQL caps lock names at 64 characters while tenant IDs are
// schema-valid up to 64 bytes, so the ID is hashed the same way as the other
// lock-name helpers (see tenantPoolDatabaseLockName): a fixed prefix plus a
// collision-resistant digest keeps every valid tenant ID within the limit.
func tenantSchemaLockName(tenantID string) string {
	sum := sha256.Sum256([]byte(tenantID))
	// 21-byte prefix + 40 hex chars = 61 bytes, below the 64-byte limit.
	return "drive9:tenant-schema:" + hex.EncodeToString(sum[:20])
}

func durationMs(d time.Duration) float64 {
	return float64(d.Microseconds()) / 1000.0
}

func poolAcquireTimingFields(tenantID string, coldOpen bool, createBackendDurationMs float64, totalDuration time.Duration) []zap.Field {
	fields := []zap.Field{
		zap.String("tenant_id", tenantID),
		zap.Bool("cold_open", coldOpen),
	}
	if coldOpen {
		fields = append(fields, zap.Float64("create_backend_ms", createBackendDurationMs))
	}
	fields = append(fields, zap.Float64("total_ms", durationMs(totalDuration)))
	return fields
}

func (p *Pool) resolveTenantEmbeddingMode(persisted string) (mode string, wasNull bool, err error) {
	return meta.ResolveTenantEmbeddingMode(persisted, p.cfg.DisableDatabaseAutoEmbedding)
}

type entry struct {
	tenantID           string
	storageNamespaceID string
	s3EncryptionPolicy meta.S3EncryptionPolicy
	backend            *backend.Dat9Backend
	store              *datastore.Store
	sharedDBLease      *sharedDBHandleLease
	elem               *list.Element
	refs               int
	retired            bool
	lastUsed           time.Time // refreshed by real activity through Acquire; never by AcquireCached
	sharedDBID         int64     // db_pool.id for shared-schema tenants; zero for standalone tenants
}

// sharedDBHandleLease pins one physical shared-DB handle generation while a
// caller constructs or uses a tenant-scoped Store. Release is generation-safe:
// a delayed release from an invalidated handle cannot decrement refs belonging
// to a newly opened handle with the same db_pool.id.
type sharedDBHandleLease struct {
	pool *Pool
	dbID int64
	db   *sql.DB
	once sync.Once
}

func (l *sharedDBHandleLease) DB() *sql.DB {
	if l == nil {
		return nil
	}
	return l.db
}

func (l *sharedDBHandleLease) DBID() int64 {
	if l == nil {
		return 0
	}
	return l.dbID
}

func (l *sharedDBHandleLease) Release() {
	if l == nil || l.pool == nil || l.dbID <= 0 || l.db == nil {
		return
	}
	l.once.Do(func() {
		l.pool.sharedMu.Lock()
		defer l.pool.sharedMu.Unlock()
		if current, ok := l.pool.sharedDBs[l.dbID]; !ok || current != l.db {
			return
		}
		if refs := l.pool.sharedDBRefs[l.dbID]; refs > 1 {
			l.pool.sharedDBRefs[l.dbID] = refs - 1
		} else {
			delete(l.pool.sharedDBRefs, l.dbID)
		}
		l.pool.recordSharedDBCacheTenantsLocked(l.dbID)
	})
}

func NewPool(cfg PoolConfig, enc encrypt.Encryptor) *Pool {
	max := cfg.MaxTenants
	if max <= 0 {
		max = defaultTenantPoolMaxTenants
	}
	idleTimeout := cfg.IdleTimeout
	reapInterval := cfg.IdleReapInterval
	if reapInterval <= 0 {
		reapInterval = defaultTenantPoolIdleReapInterval
	}
	metrics.RecordGauge("tenant_pool", "cached_backends", 0)
	metrics.RecordGauge("tenant_pool", "max_backends", float64(max))
	return &Pool{
		cfg:              cfg,
		enc:              enc,
		items:            map[string]*entry{},
		order:            list.New(),
		maxSize:          max,
		sharedDBs:        map[int64]*sql.DB{},
		sharedDBRefs:     map[int64]int{},
		sharedDBLastUsed: map[int64]time.Time{},
		sharedDBLabels:   map[int64]sharedDBMetricLabels{},
		sharedDBOpenMu:   map[int64]*sync.Mutex{},
		// No LeaderChecker means single-pod mode: FileGC runs unconditionally.
		fileGCEnabled: cfg.LeaderChecker == nil,
		idleTimeout:   idleTimeout,
		reapInterval:  reapInterval,
	}
}

// SetMetaStore sets the central server DB store for quota operations.
// Must be called before any backend is created via Acquire.
func (p *Pool) SetMetaStore(ms *meta.Store) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.metaStore = ms
}

// SetTenantWorkNotifier registers a callback invoked with the tenant ID, TiDB
// Cloud org ID, and work mask whenever one of this pool's backends commits
// durable work (semantic task, file_gc task, quota outbox row), so the tenant
// worker can claim the new work immediately instead of waiting for a periodic
// scan.
func (p *Pool) SetTenantWorkNotifier(fn func(tenantID, tidbCloudOrgID string, workMask int)) {
	if p == nil {
		return
	}
	p.tenantWorkNotifier.Store(&fn)
	if fn == nil {
		return
	}
	// A backend can be warm before the tenant worker registers its notifier
	// (notably during startup resume). Kick those cached tenants once so a
	// pending storage-tier scan does not wait for a future cold-open.
	p.mu.Lock()
	warm := make([]struct{ tenantID, orgID string }, 0, len(p.items))
	for _, e := range p.items {
		if e == nil || e.retired || e.backend == nil {
			continue
		}
		warm = append(warm, struct{ tenantID, orgID string }{tenantID: e.tenantID, orgID: e.backend.TiDBCloudOrgID()})
	}
	p.mu.Unlock()
	for _, item := range warm {
		fn(item.tenantID, item.orgID, backend.BackendWorkStorageTierScan)
	}
}

func (p *Pool) wireTenantWorkNotifier(b *backend.Dat9Backend, tenantID string) {
	// Resolve the notifier at call time, not wire time, so backends created
	// before SetTenantWorkNotifier (e.g. during tenant resume on startup) still
	// notify once the tenant worker registers.
	b.SetWorkEnqueuedNotifier(func(workMask int, tidbCloudOrgID string) {
		if fn := p.tenantWorkNotifier.Load(); fn != nil && *fn != nil {
			(*fn)(tenantID, tidbCloudOrgID, workMask)
		}
	})
}

func (p *Pool) wireBackendAndScheduleEmbeddingBackfill(ctx context.Context, b *backend.Dat9Backend, tenantID string) error {
	p.wireTenantWorkNotifier(b, tenantID)
	p.wireDeleteReconcilePersister(b, tenantID)
	if _, err := b.EnqueueEmbeddingBackfillTask(ctx); err != nil {
		return fmt.Errorf("enqueue embedding backfill: %w", err)
	}
	return nil
}

// SetDeleteReconcilePersister registers a callback that durably records the
// tenant's owed delete-orphan reconciliation sweep. It is invoked by pool
// backends before a recursive directory delete and must return nil only after
// the record is durable; an error aborts the delete (fail closed).
func (p *Pool) SetDeleteReconcilePersister(fn func(tenantID string) error) {
	if p == nil {
		return
	}
	p.deleteReconcilePersister.Store(&fn)
}

func (p *Pool) wireDeleteReconcilePersister(b *backend.Dat9Backend, tenantID string) {
	// Resolve at call time (same rationale as wireTenantWorkNotifier): backends
	// created before SetDeleteReconcilePersister still get the durable pre-write
	// once the server registers it.
	b.SetDeleteReconcilePersister(func() error {
		if fn := p.deleteReconcilePersister.Load(); fn != nil && *fn != nil {
			return (*fn)(tenantID)
		}
		return nil
	})
}

// Acquire returns a backend pinned for real tenant activity, including request
// traffic and durable background work. It may cold-open the tenant, promotes
// cache hits in the capacity LRU, and refreshes shared physical-DB activity.
// The returned release callback must be called when the caller is done.
func (p *Pool) Acquire(ctx context.Context, t *meta.Tenant) (out *backend.Dat9Backend, release func(), err error) {
	start := time.Now()
	defer observePool(ctx, "acquire", t.ID, &err, start)

	if t.Status != meta.TenantActive {
		logger.Warn(ctx, "tenant_pool_get_skipped_inactive",
			zap.String("tenant_id", t.ID),
			zap.String("status", string(t.Status)))
		p.Invalidate(t.ID)
		return nil, nil, fmt.Errorf("tenant status: %s", t.Status)
	}

	s3EncryptionPolicy := t.S3EncryptionPolicy()
	storageNamespaceID := t.StorageNamespaceID
	var toClose []*entry
	p.mu.Lock()
	if e, ok := p.items[t.ID]; ok {
		if e.s3EncryptionPolicy == s3EncryptionPolicy && (storageNamespaceID == "" || e.storageNamespaceID == storageNamespaceID) {
			e.refs++
			p.order.MoveToFront(e.elem)
			p.touchSharedDBActivityLocked(e.sharedDBID, time.Now())
			e.lastUsed = time.Now()
			p.mu.Unlock()
			metrics.RecordOperation("tenant_pool", "cache_lookup", "hit", 0)
			totalDuration := time.Since(start)
			logger.InfoOpenPoolTiming(ctx, "tenant_pool_acquire_timing", totalDuration,
				poolAcquireTimingFields(t.ID, false, 0, totalDuration)...)
			return e.backend, p.makeRelease(e), nil
		}
		if removed := p.removeLocked(e.elem, "replace"); removed != nil {
			toClose = append(toClose, removed)
		}
		p.mu.Unlock()
		for _, retired := range toClose {
			p.closeEntry(retired)
		}
		toClose = nil
	} else {
		p.mu.Unlock()
	}

	createBackendStart := time.Now()
	b, st, tidbCloudOrgID, sharedLease, err := p.createBackend(ctx, t)
	if err != nil {
		// Cold-open failure: a tenant TiDB open was attempted but failed.
		// Record so alerts can detect a scan path that is churning cold opens.
		metrics.RecordTenantOperationWithOrg(t.ID, tidbCloudOrgID, "user_db_access", "acquire_cold_open", "error", time.Since(createBackendStart))
		return nil, nil, err
	}
	sharedDBID := sharedLease.DBID()
	createBackendDurationMs := float64(time.Since(createBackendStart).Microseconds()) / 1000.0
	// Cold-open success: a tenant TiDB was opened from scratch (cache miss).
	// This is the canonical "a serverless TiDB was woken up" signal — the
	// primary billing-storm indicator. PR #660 eliminated periodic scans that
	// caused cold opens at scale; this metric guards against regressions.
	metrics.RecordTenantOperationWithOrg(t.ID, tidbCloudOrgID, "user_db_access", "acquire_cold_open", "ok", time.Since(createBackendStart))

	p.mu.Lock()
	if e, ok := p.items[t.ID]; ok {
		if e.s3EncryptionPolicy == s3EncryptionPolicy && (t.StorageNamespaceID == "" || e.storageNamespaceID == t.StorageNamespaceID) {
			e.refs++
			p.order.MoveToFront(e.elem)
			p.touchSharedDBActivityLocked(e.sharedDBID, time.Now())
			e.lastUsed = time.Now()
			p.mu.Unlock()
			b.Close()
			_ = st.Close()
			sharedLease.Release()
			metrics.RecordOperation("tenant_pool", "cache_lookup", "hit", 0)
			totalDuration := time.Since(start)
			logger.InfoOpenPoolTiming(ctx, "tenant_pool_acquire_timing", totalDuration,
				poolAcquireTimingFields(t.ID, true, createBackendDurationMs, totalDuration)...)
			return e.backend, p.makeRelease(e), nil
		}
		if removed := p.removeLocked(e.elem, "replace"); removed != nil {
			toClose = append(toClose, removed)
		}
	}
	e := &entry{tenantID: t.ID, storageNamespaceID: t.StorageNamespaceID, s3EncryptionPolicy: s3EncryptionPolicy, backend: b, store: st, sharedDBLease: sharedLease, refs: 1, lastUsed: time.Now(), sharedDBID: sharedDBID}
	e.elem = p.order.PushFront(e)
	p.items[t.ID] = e
	p.touchSharedDBActivityLocked(sharedDBID, time.Now())
	for p.order.Len() > p.maxSize {
		oldest := p.order.Back()
		if oldest != nil {
			if removed := p.removeLocked(oldest, "evict"); removed != nil {
				toClose = append(toClose, removed)
			}
		}
	}
	p.recordCachedBackendCountLocked()
	// FileGC is now kick-driven through the unified tenant worker; no
	// per-backend goroutine is started here.
	p.mu.Unlock()
	for _, retired := range toClose {
		p.closeEntry(retired)
	}
	metrics.RecordOperation("tenant_pool", "cache_lookup", "miss", 0)
	p.scheduleStorageTierScan(ctx, t, tidbCloudOrgID)
	totalDuration := time.Since(start)
	logger.InfoOpenPoolTiming(ctx, "tenant_pool_acquire_timing", totalDuration,
		poolAcquireTimingFields(t.ID, true, createBackendDurationMs, totalDuration)...)
	return b, p.makeRelease(e), nil
}

func (p *Pool) scheduleStorageTierScan(ctx context.Context, t *meta.Tenant, tidbCloudOrgID string) {
	if p == nil || p.metaStore == nil || t == nil {
		return
	}
	created, err := p.metaStore.EnsureStorageTierScanJob(ctx, t.ID)
	if err != nil {
		logger.Warn(ctx, "storage_tier_scan_job_ensure_failed", zap.String("tenant_id", t.ID), zap.Error(err))
		return
	}
	if !created {
		job, err := p.metaStore.GetStorageTierScanJob(ctx, t.ID)
		if err != nil || job.FullScanCompletedAt != nil {
			return
		}
	}
	if fn := p.tenantWorkNotifier.Load(); fn != nil && *fn != nil {
		(*fn)(t.ID, tidbCloudOrgID, backend.BackendWorkStorageTierScan)
	}
}

func (p *Pool) Invalidate(tenantID string) {
	var toClose *entry
	p.mu.Lock()
	if e, ok := p.items[tenantID]; ok {
		toClose = p.removeLocked(e.elem, "invalidate")
	}
	p.mu.Unlock()
	p.closeEntry(toClose)
}

func (p *Pool) WaitTenantIdle(ctx context.Context, tenantID string) error {
	ctx, cancel := withTenantPoolDrainTimeout(ctx)
	defer cancel()
	ticker := time.NewTicker(50 * time.Millisecond)
	defer ticker.Stop()
	for {
		p.mu.Lock()
		e, ok := p.items[tenantID]
		refs := 0
		if ok {
			refs = e.refs
		}
		idle := !ok || e.refs == 0
		p.mu.Unlock()
		if idle {
			return nil
		}
		select {
		case <-ctx.Done():
			logger.Warn(ctx, "tenant_pool_wait_idle_timeout",
				zap.String("tenant_id", tenantID),
				zap.Int("refs", refs),
				zap.Error(ctx.Err()))
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

// AcquireCached pins a cached backend without creating one on cache miss. It
// is intended for the safety-net scan and other background work that may
// process already-active tenants but must not wake dormant tenant TiDBs.
//
// If the tenant's S3 encryption policy or storage namespace has changed since
// the backend was cached, AcquireCached returns false (the stale backend is
// not usable). In this rare case the safety-net scan skips the tenant — its
// expired lease recovery is deferred until the next write kick triggers a
// fresh Acquire with the updated policy. This is acceptable because policy
// changes are rare and the tenant's tasks remain durable in its TiDB.
func (p *Pool) AcquireCached(t *meta.Tenant) (b *backend.Dat9Backend, release func(), ok bool) {
	return p.acquireCached(t, false)
}

// acquireCached pins an existing tenant backend without cold-opening it. When
// refreshActivity is true, the warm hit is real tenant activity: it promotes
// the capacity LRU and refreshes tenant and physical shared-DB idle state.
// AcquireCached passes false for safety-net observation; the local S3 gateway
// passes true after resolving the tenant from a presigned URL.
func (p *Pool) acquireCached(t *meta.Tenant, refreshActivity bool) (b *backend.Dat9Backend, release func(), ok bool) {
	if p == nil || t == nil {
		return nil, nil, false
	}
	if t.Status != meta.TenantActive {
		p.Invalidate(t.ID)
		return nil, nil, false
	}
	s3EncryptionPolicy := t.S3EncryptionPolicy()
	p.mu.Lock()
	e, exists := p.items[t.ID]
	if !exists {
		p.mu.Unlock()
		if !refreshActivity {
			// Cold-skip: the tenant is not cached (TiDB likely scaled to zero).
			// This is the warm-only invariant working as designed — the safety-net
			// must never open a cold tenant TiDB. Record so the invariant-violation
			// alert can confirm AcquireCached is staying warm-only.
			metrics.RecordTenantOperationWithOrg(t.ID, "", "user_db_access", "acquire_cached", "cold_skipped", 0)
		}
		return nil, nil, false
	}
	if e.s3EncryptionPolicy != s3EncryptionPolicy || (t.StorageNamespaceID != "" && e.storageNamespaceID != t.StorageNamespaceID) {
		p.mu.Unlock()
		if !refreshActivity {
			// Stale-skip: the cached backend's encryption policy / storage namespace
			// no longer matches the tenant. The safety-net defers to the next write
			// kick. Record so this rare edge case is observable.
			metrics.RecordTenantOperationWithOrg(t.ID, e.backend.TiDBCloudOrgID(), "user_db_access", "acquire_cached", "stale_skipped", 0)
		}
		return nil, nil, false
	}
	e.refs++
	if refreshActivity {
		now := time.Now()
		p.order.MoveToFront(e.elem)
		p.touchSharedDBActivityLocked(e.sharedDBID, now)
		e.lastUsed = now
	}
	p.mu.Unlock()
	metrics.RecordOperation("tenant_pool", "cache_lookup", "hit", 0)
	if refreshActivity {
		return e.backend, p.makeRelease(e), true
	}
	metrics.RecordTenantOperationWithOrg(t.ID, e.backend.TiDBCloudOrgID(), "user_db_access", "acquire_cached", "hit", 0)
	return e.backend, p.makeReleaseNoRefresh(e), true
}

func (p *Pool) InvalidateAndWait(ctx context.Context, tenantID string) error {
	ctx, cancel := withTenantPoolDrainTimeout(ctx)
	defer cancel()
	var retired *entry
	var toClose *entry
	p.mu.Lock()
	if e, ok := p.items[tenantID]; ok {
		retired = e
		toClose = p.removeLocked(e.elem, "invalidate")
	}
	p.mu.Unlock()
	p.closeEntry(toClose)
	if retired == nil || toClose != nil {
		return nil
	}
	ticker := time.NewTicker(50 * time.Millisecond)
	defer ticker.Stop()
	for {
		p.mu.Lock()
		idle := retired.refs == 0
		refs := retired.refs
		p.mu.Unlock()
		if idle {
			return nil
		}
		select {
		case <-ctx.Done():
			logger.Warn(ctx, "tenant_pool_invalidate_wait_timeout",
				zap.String("tenant_id", tenantID),
				zap.Int("refs", refs),
				zap.Error(ctx.Err()))
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

func (p *Pool) S3ForStorageNamespace(ctx context.Context, ns *meta.StorageNamespace) (out s3client.S3Client, err error) {
	start := time.Now()
	namespaceID := ""
	if ns != nil {
		namespaceID = ns.ID
	}
	defer observePool(ctx, "s3_for_storage_namespace", namespaceID, &err, start)
	if ns == nil {
		return nil, fmt.Errorf("storage namespace is required")
	}
	switch ns.Backend {
	case "s3":
		if p.cfg.S3Bucket == "" {
			return nil, fmt.Errorf("s3 bucket is not configured")
		}
		if ns.Bucket != "" && ns.Bucket != p.cfg.S3Bucket {
			return nil, fmt.Errorf("storage namespace bucket %q does not match configured bucket", ns.Bucket)
		}
		return s3client.New(ctx, s3client.AWSConfig{
			Region:          p.cfg.S3Region,
			Bucket:          p.cfg.S3Bucket,
			Prefix:          ns.Prefix,
			RoleARN:         p.cfg.S3RoleARN,
			Endpoint:        p.cfg.S3Endpoint,
			ForcePathStyle:  p.cfg.S3ForcePathStyle,
			AccessKeyID:     p.cfg.S3AccessKeyID,
			SecretAccessKey: p.cfg.S3SecretAccessKey,
			SessionToken:    p.cfg.S3SessionToken,
		})
	case "local":
		if p.cfg.S3Dir == "" {
			return nil, fmt.Errorf("local s3 dir is not configured")
		}
		localPrefix, err := cleanStorageNamespaceLocalPrefix(ns.Prefix)
		if err != nil {
			return nil, err
		}
		rootDir, err := filepath.Abs(p.cfg.S3Dir)
		if err != nil {
			return nil, fmt.Errorf("resolve local s3 dir: %w", err)
		}
		s3Dir := filepath.Join(rootDir, localPrefix)
		if rel, err := filepath.Rel(rootDir, s3Dir); err != nil {
			return nil, fmt.Errorf("resolve local storage namespace prefix: %w", err)
		} else if rel == ".." || strings.HasPrefix(rel, "../") {
			return nil, fmt.Errorf("storage namespace prefix escapes local s3 dir")
		}
		baseURL := strings.TrimRight(p.cfg.PublicURL, "/")
		if baseURL != "" {
			baseURL += "/s3/" + localPrefix
		}
		return s3client.NewLocal(s3Dir, baseURL)
	default:
		return nil, fmt.Errorf("unsupported storage backend %q", ns.Backend)
	}
}

// S3ExpressForAppendLogNamespace returns a Directory Bucket client scoped to
// one bound storage namespace's append-log objects. An unbound namespace has
// no Express lifecycle to clean up; a missing or mismatched configuration for
// a bound namespace is an error so tenant deletion retries without finalizing.
func (p *Pool) S3ExpressForAppendLogNamespace(ctx context.Context, ns *meta.StorageNamespace) (s3client.S3Client, error) {
	if ns == nil {
		return nil, fmt.Errorf("storage namespace is required")
	}
	switch ns.AppendLogBindingState {
	case "", meta.AppendLogBindingUnbound:
		return nil, nil
	case meta.AppendLogBindingBound:
		// Continue below.
	default:
		return nil, fmt.Errorf("unsupported append-log binding state %q", ns.AppendLogBindingState)
	}
	client, err := p.newBoundAppendLogClient(ctx, ns, true)
	if err != nil {
		return nil, err
	}
	return client, nil
}

func (p *Pool) appendLogClientForStorageNamespace(ctx context.Context, ns *meta.StorageNamespace) (s3client.AppendLogClient, error) {
	if ns == nil {
		return nil, fmt.Errorf("storage namespace is required")
	}
	bucket, prefix, region, ok := p.appendLogConfiguration()
	if !ok || p.metaStore == nil {
		return nil, nil
	}
	bound, err := p.metaStore.EnsureAppendLogStorageNamespaceBinding(ctx, ns.ID, bucket, prefix, region)
	if errors.Is(err, meta.ErrAppendLogBindingMismatch) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("bind append-log storage namespace: %w", err)
	}
	ns = bound
	client, err := p.newBoundAppendLogClient(ctx, ns, false)
	if errors.Is(err, errAppendLogConfigurationUnavailable) || errors.Is(err, meta.ErrAppendLogBindingMismatch) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return client, nil
}

func (p *Pool) newBoundAppendLogClient(ctx context.Context, ns *meta.StorageNamespace, namespaceScoped bool) (*s3client.AWSS3Client, error) {
	if ns == nil || ns.AppendLogBindingState != meta.AppendLogBindingBound {
		return nil, fmt.Errorf("append-log storage namespace is not bound")
	}
	bucket, prefix, region, ok := p.appendLogConfiguration()
	if !ok {
		return nil, errAppendLogConfigurationUnavailable
	}
	if ns.AppendLogBucket != bucket || ns.AppendLogPrefix != prefix || ns.AppendLogRegion != region {
		return nil, fmt.Errorf("namespace %q: %w", ns.ID, meta.ErrAppendLogBindingMismatch)
	}
	if namespaceScoped {
		if ns.ID == "" || strings.Contains(ns.ID, "/") {
			return nil, fmt.Errorf("invalid append-log storage namespace %q", ns.ID)
		}
		prefix += "append-log/" + ns.ID
	}
	return s3client.New(ctx, s3client.AWSConfig{
		Region:          region,
		Bucket:          bucket,
		Prefix:          prefix,
		RoleARN:         p.cfg.S3RoleARN,
		AccessKeyID:     p.cfg.S3AccessKeyID,
		SecretAccessKey: p.cfg.S3SecretAccessKey,
		SessionToken:    p.cfg.S3SessionToken,
	})
}

func (p *Pool) appendLogConfiguration() (bucket, prefix, region string, ok bool) {
	bucket = strings.TrimSpace(p.cfg.S3ExpressBucket)
	region = strings.TrimSpace(p.cfg.S3Region)
	prefix = meta.NormalizeAppendLogBasePrefix(p.cfg.S3ExpressPrefix)
	return bucket, prefix, region, bucket != "" && region != ""
}

func cleanStorageNamespaceLocalPrefix(prefix string) (string, error) {
	if filepath.IsAbs(prefix) {
		return "", fmt.Errorf("storage namespace prefix must be relative")
	}
	localPrefix := strings.Trim(prefix, "/")
	if localPrefix == "" {
		return "", fmt.Errorf("storage namespace prefix is required")
	}
	for _, part := range strings.Split(localPrefix, "/") {
		if part == "" || part == "." || part == ".." {
			return "", fmt.Errorf("invalid storage namespace prefix %q", prefix)
		}
	}
	return localPrefix, nil
}

func withTenantPoolDrainTimeout(ctx context.Context) (context.Context, context.CancelFunc) {
	if _, ok := ctx.Deadline(); ok {
		return ctx, func() {}
	}
	return context.WithTimeout(ctx, defaultTenantPoolDrainTimeout)
}

// Start launches the reaper goroutine. IdleTimeout=0 disables standalone
// tenant eviction, but shared physical-DB handles still need this reaper.
// The reaper is stopped by Close.
func (p *Pool) Start(ctx context.Context) {
	if p == nil || p.reapInterval <= 0 {
		return
	}
	workerCtx, cancel := context.WithCancel(ctx)
	p.reapStop = cancel
	p.reapWG.Add(1)
	go func() {
		defer p.reapWG.Done()
		p.reapLoop(workerCtx)
	}()
}

func (p *Pool) reapLoop(ctx context.Context) {
	if p.reapInterval <= 0 {
		return
	}
	ticker := time.NewTicker(p.reapInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			p.reapOnce(ctx)
		}
	}
}

func (p *Pool) reapOnce(ctx context.Context) {
	now := time.Now()
	var toClose []*entry
	if p.idleTimeout > 0 {
		p.mu.Lock()
		for _, e := range p.items {
			if e.retired || e.refs > 0 {
				continue
			}
			// Shared tenant backends are lightweight fs_id-scoped views over a
			// separately managed physical DB handle. Keep them warm until the
			// capacity LRU evicts them; applying the standalone idle TTL here turns
			// stable high-cardinality shared traffic into continuous cold opens.
			if e.sharedDBID > 0 {
				continue
			}
			if now.Sub(e.lastUsed) > p.idleTimeout {
				if removed := p.removeLocked(e.elem, "idle"); removed != nil {
					toClose = append(toClose, removed)
				}
			}
		}
		p.recordCachedBackendCountLocked()
		p.mu.Unlock()
	}
	for _, retired := range toClose {
		p.closeEntry(retired)
	}
	p.reapExpiredSharedDBTenantEntries(now)
	p.reapIdleSharedDBs(now)
}

func (p *Pool) Close() {
	// Stop the idle reaper first so it doesn't race with shutdown eviction.
	if p.reapStop != nil {
		p.reapStop()
		p.reapWG.Wait()
		p.reapStop = nil
	}
	toClose := make([]*entry, 0, p.order.Len())
	p.mu.Lock()
	for p.order.Len() > 0 {
		if removed := p.removeLocked(p.order.Back(), "shutdown"); removed != nil {
			toClose = append(toClose, removed)
		}
	}
	p.mu.Unlock()
	for _, retired := range toClose {
		p.closeEntry(retired)
	}
	type sharedDBToClose struct {
		id int64
		db *sql.DB
	}
	p.sharedMu.Lock()
	p.sharedDBClosed = true
	sharedToClose := make([]sharedDBToClose, 0, len(p.sharedDBs))
	for dbID, db := range p.sharedDBs {
		sharedToClose = append(sharedToClose, sharedDBToClose{id: dbID, db: db})
		delete(p.sharedDBs, dbID)
		delete(p.sharedDBRefs, dbID)
		delete(p.sharedDBLastUsed, dbID)
		p.deleteSharedDBCacheMetricsLocked(dbID)
	}
	p.sharedMu.Unlock()
	for _, item := range sharedToClose {
		if err := mysqlutil.CloseInstrumented(item.db); err != nil {
			logger.Warn(context.Background(), "shared_db_close_failed", zap.Int64("db_id", item.id), zap.Error(err))
		}
	}
}

func (p *Pool) Decrypt(ctx context.Context, cipher []byte) ([]byte, error) {
	return p.enc.Decrypt(ctx, cipher)
}

func (p *Pool) Encrypt(ctx context.Context, plain []byte) ([]byte, error) {
	return p.enc.Encrypt(ctx, plain)
}

// SupportsAsyncImageExtract reports whether tenant backends created by this
// pool carry the async image extraction runtime.
func (p *Pool) SupportsAsyncImageExtract() bool {
	return p != nil && p.cfg.BackendOptions.AsyncImageExtract.Enabled
}

// AutoSemanticTaskTypes returns the auto-backend durable semantic task types
// implied by PoolConfig.BackendOptions (async image and/or audio extract). This
// is a coarse routing hint for tenant list filtering before a backend is
// acquired; it does not include app-managed embed tasks. Nil means the pool
// contributes no auto semantic tasks. The returned slice must be treated as
// read-only.
//
// Image: viability matches backend.Options.configureOptions — when Enabled, a nil
// Image Extractor is replaced with NewBasicImageTextExtractor before workers start.
// Audio: Phase 2 requires Enabled and a non-nil Extractor (no implicit default);
// pool routing must stay aligned with Dat9Backend.SupportsAsyncAudioExtract.
func (p *Pool) AutoSemanticTaskTypes() []semantic.TaskType {
	if p == nil {
		return nil
	}
	var out []semantic.TaskType
	if backend.AsyncImageExtractWillWireRuntime(p.cfg.BackendOptions.AsyncImageExtract) {
		out = append(out, semantic.TaskTypeImgExtractText)
	}
	if backend.AsyncAudioExtractWillWireRuntime(p.cfg.BackendOptions.AsyncAudioExtract) {
		out = append(out, semantic.TaskTypeAudioExtractText)
	}
	if backend.AsyncVideoExtractWillWireRuntime(p.cfg.BackendOptions.AsyncVideoExtract) {
		out = append(out, semantic.TaskTypeVideoExtractVisual)
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

// LoadS3Backend resolves tenant metadata and pins an already-warm backend for
// one local presigned-URL request. Local mock-S3 URLs do not carry independent
// tenant authentication, so this path must never cold-open a tenant database.
// A warm hit is still real activity and refreshes the normal LRU/idle state.
func (p *Pool) LoadS3Backend(ctx context.Context, metaStore *meta.Store, tenantID string) (out *backend.Dat9Backend, release func()) {
	start := time.Now()
	var err error
	defer observePool(ctx, "load_s3_backend", tenantID, &err, start)

	tenant, err := metaStore.GetTenant(ctx, tenantID)
	if err != nil {
		if !errors.Is(err, meta.ErrNotFound) {
			logger.Error(ctx, "tenant_pool_load_s3_backend_failed", zap.String("tenant_id", tenantID), zap.Error(err))
		}
		return nil, nil
	}
	b, release, ok := p.acquireCached(tenant, true)
	if !ok {
		return nil, nil
	}
	return b, release
}

// fsIDForTenant resolves the tenant's internal fs_id from the meta DB,
// allocating one on first use. Returns (0, nil) when the pool has no meta
// store (tests, non-multi-tenant mode); the standalone SQL shape never emits
// fs_id, so 0 is inert there.
func (p *Pool) fsIDForTenant(ctx context.Context, t *meta.Tenant) (int64, error) {
	if p.metaStore == nil || t == nil || t.ID == "" {
		return 0, nil
	}
	fsID, err := p.metaStore.EnsureFsID(ctx, t.ID)
	if err != nil {
		return 0, fmt.Errorf("ensure fs_id for tenant %s: %w", t.ID, err)
	}
	return fsID, nil
}

func (p *Pool) tenantMetricTiDBCloudOrgID(ctx context.Context, t *meta.Tenant) string {
	if t == nil || strings.TrimSpace(t.ID) == "" || !UsesTiDBCloudNativeCredentials(t.Provider) {
		return defaultTenantMetricTiDBCloudOrgID
	}
	if orgID := strings.TrimSpace(t.TiDBCloudOrgID); orgID != "" {
		return orgID
	}
	if p == nil || p.metaStore == nil {
		return defaultTenantMetricTiDBCloudOrgID
	}
	if IsSharedSchemaProvider(t.Provider) {
		sharedDB, err := p.metaStore.GetSharedDBForTenant(ctx, t.ID)
		if err != nil {
			if !errors.Is(err, meta.ErrNotFound) {
				logger.Warn(ctx, "tenant_pool_metric_shared_org_lookup_failed",
					zap.String("tenant_id", t.ID),
					zap.Error(err))
			}
			return defaultTenantMetricTiDBCloudOrgID
		}
		orgID := strings.TrimSpace(sharedDB.TiDBCloudOrganizationID)
		if orgID == "" {
			return defaultTenantMetricTiDBCloudOrgID
		}
		return orgID
	}
	binding, err := p.metaStore.GetTenantTiDBCloudOrgBinding(ctx, t.ID)
	if err != nil {
		if !errors.Is(err, meta.ErrNotFound) {
			logger.Warn(ctx, "tenant_pool_metric_org_lookup_failed",
				zap.String("tenant_id", t.ID),
				zap.Error(err))
		}
		return defaultTenantMetricTiDBCloudOrgID
	}
	orgID := strings.TrimSpace(binding.OrganizationID)
	if orgID == "" {
		return defaultTenantMetricTiDBCloudOrgID
	}
	return orgID
}

// placementForTenant returns the tenant's placement row, or (nil, nil) when
// the tenant has none — which is the normal case for every standalone tenant
// today. Placement lookup errors are fatal to the cold open: silently falling
// back to standalone could route a shared tenant with the wrong schema shape.
func (p *Pool) placementForTenant(ctx context.Context, fsID int64) (*meta.TenantPlacement, error) {
	if p.metaStore == nil || fsID <= 0 {
		return nil, nil
	}
	placement, err := p.metaStore.GetTenantPlacement(ctx, fsID)
	if errors.Is(err, meta.ErrNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("resolve tenant placement for fs_id %d: %w", fsID, err)
	}
	return placement, nil
}

// leaseSharedDBLocked acquires one reference to the current physical handle.
// Caller must hold p.sharedMu.
func (p *Pool) leaseSharedDBLocked(dbID int64, db *sql.DB) *sharedDBHandleLease {
	p.sharedDBRefs[dbID]++
	p.sharedDBLastUsed[dbID] = time.Now()
	p.recordSharedDBCacheTenantsLocked(dbID)
	return &sharedDBHandleLease{pool: p, dbID: dbID, db: db}
}

// touchSharedDBActivityLocked refreshes physical shared-DB activity for real
// tenant use. Caller must hold p.mu; taking p.sharedMu here preserves the
// pool-wide lock order p.mu -> p.sharedMu.
func (p *Pool) touchSharedDBActivityLocked(dbID int64, now time.Time) {
	if dbID <= 0 {
		return
	}
	p.sharedMu.Lock()
	if _, ok := p.sharedDBs[dbID]; ok {
		p.sharedDBLastUsed[dbID] = now
	}
	p.sharedMu.Unlock()
}

// recordSharedDBCacheTenantsLocked emits the per-handle reference gauge. The
// count includes cached tenant ownership and short-lived handle reservations
// used while opening or maintaining a tenant view.
// Caller must hold p.sharedMu.
func (p *Pool) recordSharedDBCacheTenantsLocked(dbID int64) {
	labels, ok := p.sharedDBLabels[dbID]
	if !ok {
		return
	}
	metrics.RecordSharedDBPoolCacheTenants(labels.orgID, dbID, labels.uuid, int64(p.sharedDBRefs[dbID]))
}

// deleteSharedDBCacheMetricsLocked removes both cache series for a handle that
// is leaving the cache. Caller must hold p.sharedMu.
func (p *Pool) deleteSharedDBCacheMetricsLocked(dbID int64) {
	labels, ok := p.sharedDBLabels[dbID]
	if !ok {
		return
	}
	metrics.DeleteSharedDBPoolCacheHandles(labels.orgID, dbID, labels.uuid)
	metrics.DeleteSharedDBPoolCacheTenants(labels.orgID, dbID, labels.uuid)
	delete(p.sharedDBLabels, dbID)
}

// reapExpiredSharedDBTenantEntries retires every cached tenant view belonging
// to a physical shared DB whose activity TTL expired. Active entries leave
// the capacity LRU immediately but remain usable until their references drain.
//
// Expiration is revalidated while holding p.mu -> p.sharedMu. sharedMu is then
// released before walking the capacity LRU so opening or leasing unrelated
// physical handles is not blocked by a large tenant cache scan. Stores and
// per-entry metrics are handled outside both locks.
func (p *Pool) reapExpiredSharedDBTenantEntries(now time.Time) {
	p.sharedMu.Lock()
	candidates := make([]int64, 0)
	for dbID := range p.sharedDBs {
		lastUsed := p.sharedDBLastUsed[dbID]
		if !lastUsed.IsZero() && now.Sub(lastUsed) > defaultSharedDBHandleIdleTTL {
			candidates = append(candidates, dbID)
		}
	}
	p.sharedMu.Unlock()
	if len(candidates) == 0 {
		return
	}

	var toClose []*entry
	p.mu.Lock()
	p.sharedMu.Lock()
	expired := make(map[int64]struct{}, len(candidates))
	for _, dbID := range candidates {
		lastUsed := p.sharedDBLastUsed[dbID]
		_, exists := p.sharedDBs[dbID]
		if exists && !lastUsed.IsZero() && now.Sub(lastUsed) > defaultSharedDBHandleIdleTTL {
			expired[dbID] = struct{}{}
		}
	}
	p.sharedMu.Unlock()
	removedCount := 0
	for elem := p.order.Front(); elem != nil; {
		next := elem.Next()
		e := elem.Value.(*entry)
		if _, ok := expired[e.sharedDBID]; ok {
			removedCount++
			if removed := p.removeLockedWithoutMetrics(elem); removed != nil {
				toClose = append(toClose, removed)
			}
		}
		elem = next
	}
	cachedCount := len(p.items)
	p.mu.Unlock()
	if removedCount > 0 {
		metrics.RecordGauge("tenant_pool", "cached_backends", float64(cachedCount))
		for range removedCount {
			metrics.RecordOperation("tenant_pool", "remove", "shared_idle", 0)
		}
	}
	for _, retired := range toClose {
		p.closeEntry(retired)
	}
}

// reapIdleSharedDBs closes physical shared-DB handles only after the activity
// TTL has expired and all cached, active, or in-flight handle references have
// drained. The existing tenant-pool reaper invokes this; no separate worker is
// needed.
func (p *Pool) reapIdleSharedDBs(now time.Time) {
	type sharedDBToClose struct {
		id int64
		db *sql.DB
	}
	var toClose []sharedDBToClose
	p.sharedMu.Lock()
	for dbID, db := range p.sharedDBs {
		if p.sharedDBRefs[dbID] > 0 {
			continue
		}
		lastUsed := p.sharedDBLastUsed[dbID]
		if lastUsed.IsZero() || now.Sub(lastUsed) <= defaultSharedDBHandleIdleTTL {
			continue
		}
		toClose = append(toClose, sharedDBToClose{id: dbID, db: db})
		delete(p.sharedDBs, dbID)
		delete(p.sharedDBRefs, dbID)
		delete(p.sharedDBLastUsed, dbID)
		p.deleteSharedDBCacheMetricsLocked(dbID)
	}
	p.sharedMu.Unlock()
	for _, item := range toClose {
		if err := mysqlutil.CloseInstrumented(item.db); err != nil {
			logger.Warn(context.Background(), "shared_db_idle_close_failed", zap.Int64("db_id", item.id), zap.Error(err))
		}
	}
}

// sharedDBHandle returns the cached *sql.DB for a shared-schema database
// (db_pool.id), opening it on first use. A new handle gets one activity-idle
// grace window; real tenant or maintenance use extends it. Once expired, the
// reaper retires the handle's tenant entries and closes the handle after their
// active references drain.
func (p *Pool) sharedDBHandle(ctx context.Context, dbID int64) (*sharedDBHandleLease, error) {
	return p.sharedDBHandleWithSchemaLockWait(ctx, dbID, true)
}

func (p *Pool) sharedDBHandleWithSchemaLockWait(ctx context.Context, dbID int64, waitForSchemaLock bool) (*sharedDBHandleLease, error) {
	p.sharedMu.Lock()
	if db, ok := p.sharedDBs[dbID]; ok {
		lease := p.leaseSharedDBLocked(dbID, db)
		p.sharedMu.Unlock()
		return lease, nil
	}
	if p.sharedDBClosed {
		p.sharedMu.Unlock()
		return nil, fmt.Errorf("shared db pool is closed")
	}
	openMu := p.sharedDBOpenMu[dbID]
	if openMu == nil {
		openMu = &sync.Mutex{}
		p.sharedDBOpenMu[dbID] = openMu
	}
	p.sharedMu.Unlock()

	openMu.Lock()
	defer openMu.Unlock()
	p.sharedMu.Lock()
	if db, ok := p.sharedDBs[dbID]; ok {
		lease := p.leaseSharedDBLocked(dbID, db)
		p.sharedMu.Unlock()
		return lease, nil
	}
	if p.sharedDBClosed {
		p.sharedMu.Unlock()
		return nil, fmt.Errorf("shared db pool is closed")
	}
	p.sharedMu.Unlock()
	if p.metaStore == nil {
		return nil, fmt.Errorf("shared db %d: no meta store", dbID)
	}
	info, err := p.metaStore.GetSharedDB(ctx, dbID)
	if err != nil {
		return nil, fmt.Errorf("shared db %d: %w", dbID, err)
	}
	ctx = logger.WithContext(ctx, logger.FromContext(ctx).With(
		zap.Int64("db_pool_id", info.ID),
		zap.String("db_pool_uuid", info.UUID),
		zap.String("tidbcloud_org_id", info.TiDBCloudOrganizationID),
	))
	pass, err := p.enc.Decrypt(ctx, info.PasswordCipher)
	if err != nil {
		return nil, fmt.Errorf("shared db %d: decrypt password: %w", dbID, err)
	}
	query := "parseTime=true"
	// LocalClustersAPI TiDB has no TLS. Warm-pool batch once persisted tls=true and
	// blocked schema ensure / free-tenant activation; force plaintext when the
	// pool was constructed for the local backend (not via a live env re-read).
	tlsMode := info.TLSMode
	if p.cfg.SharedDBForcePlaintext {
		tlsMode = ""
	}
	if tlsMode != "" {
		query += "&tls=" + tlsMode
	}
	// An empty TLSMode means a plain connection (local/self-hosted databases);
	// shared DBs on TiDB Cloud are registered with tls=skip-verify or tls=true.
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?%s", info.User, string(pass), info.Host, info.Port, info.Name, query)
	// Bootstrap the schema before the data-plane handle exists, on a handle of
	// its own, so multi-second DDL and lock waits never land in the
	// role="shared" series that data-plane latency alerts read.
	if info.SchemaVersion != schema.CurrentSharedTiDBSchemaVersion {
		if err := p.ensureSharedDBSchemaOnce(ctx, dsn, info, dbID, waitForSchemaLock); err != nil {
			return nil, fmt.Errorf("shared db %d: %w", dbID, err)
		}
	}
	db, err := mysqlutil.OpenInstrumentedForSharedDB(ctx, dsn, info.UUID, info.TiDBCloudOrganizationID)
	if err != nil {
		return nil, fmt.Errorf("shared db %d: open: %w", dbID, err)
	}
	p.sharedMu.Lock()
	if p.sharedDBClosed {
		p.sharedMu.Unlock()
		_ = mysqlutil.CloseInstrumented(db)
		return nil, fmt.Errorf("shared db pool is closed")
	}
	p.sharedDBs[dbID] = db
	// A newly opened physical handle gets one activity-idle grace window.
	p.sharedDBLastUsed[dbID] = time.Now()
	p.sharedDBLabels[dbID] = sharedDBMetricLabels{orgID: info.TiDBCloudOrganizationID, uuid: info.UUID}
	metrics.RecordSharedDBPoolCacheHandles(info.TiDBCloudOrganizationID, dbID, info.UUID, 1)
	lease := p.leaseSharedDBLocked(dbID, db)
	p.sharedMu.Unlock()
	return lease, nil
}

// ensureSharedDBSchemaOnce applies the checked-in shared schema through a
// dedicated handle opened with mysqlutil.RoleSharedSchema and closes it before
// the data-plane handle is opened. The advisory lock runs on the same handle, so
// both the lock wait and the DDL are attributed to schema work.
func (p *Pool) ensureSharedDBSchemaOnce(ctx context.Context, dsn string, info *meta.SharedDB, dbID int64, waitForSchemaLock bool) error {
	schemaDB, err := schema.OpenSharedSchemaDB(ctx, dsn, info.UUID, info.TiDBCloudOrganizationID)
	if err != nil {
		return fmt.Errorf("open schema handle: %w", err)
	}
	defer func() { _ = mysqlutil.CloseInstrumented(schemaDB) }()
	// The lock acquisition/wait stays on the caller's context so a foreground
	// open keeps its deadline while another migration is in flight. Once the
	// lock is held, the migration itself runs detached: it is a one-time
	// convergence step per physical DB that can take minutes (30 tenant
	// tables plus backfills), and aborting it on caller disconnect would
	// leave the shared schema version unpersisted so every later open
	// retries the migration from scratch.
	return withSharedDBSchemaAdvisoryLock(ctx, schemaDB, dbID, waitForSchemaLock, func(lockCtx context.Context) error {
		migCtx, cancel := detachSchemaOp(lockCtx)
		defer cancel()
		// Another pod may have completed the migration between the initial
		// metadata read and this lock acquisition.
		current, err := p.metaStore.GetSharedDB(migCtx, dbID)
		if err != nil {
			return err
		}
		if current.SchemaVersion == schema.CurrentSharedTiDBSchemaVersion {
			return nil
		}
		if err := ensureSharedDBSchema(migCtx, schemaDB); err != nil {
			return fmt.Errorf("ensure schema: %w", err)
		}
		return p.metaStore.UpdateSharedDBSchemaVersion(migCtx, dbID, schema.CurrentSharedTiDBSchemaVersion)
	})
}

// EnsureSharedDBReady opens the shared physical DB through the normal cache
// path, which also compares and applies the checked-in shared schema version.
func (p *Pool) EnsureSharedDBReady(ctx context.Context, dbID int64) error {
	lease, err := p.sharedDBHandleWithSchemaLockWait(ctx, dbID, false)
	if err != nil {
		return err
	}
	lease.Release()
	return nil
}

// InvalidateSharedDB removes and closes one cached physical shared-DB handle.
// A later access reopens it from the current db_pool connection metadata.
func (p *Pool) InvalidateSharedDB(dbID int64) error {
	if dbID <= 0 {
		return fmt.Errorf("shared db id must be positive")
	}
	p.sharedMu.Lock()
	openMu := p.sharedDBOpenMu[dbID]
	if openMu == nil {
		openMu = &sync.Mutex{}
		p.sharedDBOpenMu[dbID] = openMu
	}
	p.sharedMu.Unlock()
	openMu.Lock()
	defer openMu.Unlock()
	p.sharedMu.Lock()
	db := p.sharedDBs[dbID]
	delete(p.sharedDBs, dbID)
	delete(p.sharedDBRefs, dbID)
	delete(p.sharedDBLastUsed, dbID)
	p.deleteSharedDBCacheMetricsLocked(dbID)
	p.sharedMu.Unlock()
	if db == nil {
		return nil
	}
	return mysqlutil.CloseInstrumented(db)
}

// PurgeSharedTenant deletes all rows belonging to fsID from the shared DB it
// is placed on, in bounded batches. It runs after the tenant's pool entry has
// been drained, so no backend is concurrently writing through the same scope.
func (p *Pool) PurgeSharedTenant(ctx context.Context, fsID, dbID int64) error {
	lease, err := p.sharedDBHandle(ctx, dbID)
	if err != nil {
		return err
	}
	defer lease.Release()
	return datastore.NewStoreWithDB(lease.DB(), datastore.SharedScopeWithDB(fsID, lease.DBID())).PurgeTenantData(ctx)
}

func (p *Pool) createBackend(ctx context.Context, t *meta.Tenant) (*backend.Dat9Backend, *datastore.Store, string, *sharedDBHandleLease, error) {
	start := time.Now()
	tidbCloudOrgID := p.tenantMetricTiDBCloudOrgID(ctx, t)
	var sharedLease *sharedDBHandleLease
	leaseTransferred := false
	defer func() {
		if sharedLease != nil && !leaseTransferred {
			sharedLease.Release()
		}
	}()
	opts := p.cfg.BackendOptions
	resolvedEncryptionPolicy, err := meta.ResolveS3EncryptionPolicy(p.cfg.S3EncryptionPolicy, t.S3EncryptionPolicy())
	if err != nil {
		return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("resolve s3 encryption policy: %w", err)
	}
	opts.TenantID = t.ID
	opts.TiDBCloudOrgID = tidbCloudOrgID
	opts.S3EncryptionPolicy = resolvedEncryptionPolicy
	opts.AppendLogWriteDisabled = t.Kind == meta.TenantKindFork
	fsID, err := p.fsIDForTenant(ctx, t)
	if err != nil {
		return nil, nil, tidbCloudOrgID, nil, err
	}
	placement, err := p.placementForTenant(ctx, fsID)
	if err != nil {
		return nil, nil, tidbCloudOrgID, nil, err
	}
	sharedTenant := placement != nil && placement.SchemaShape == meta.SchemaShapeShared
	if IsSharedSchemaProvider(t.Provider) && !sharedTenant {
		// The persisted provider says shared but the placement row is gone.
		// Opening a standalone store against the (connectionless) tenant row
		// would fail with a confusing empty-host DSN error — and if the row
		// still had coordinates it could even serve the wrong schema shape.
		// This happens only mid-delete (placement is removed before final
		// cleanup), so fail fast with a clear message instead.
		return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("tenant %s is shared-schema but has no placement row", t.ID)
	}

	decryptDurationMs := 0.0
	openStoreStart := time.Now()
	var store *datastore.Store
	if sharedTenant {
		// Shared-schema tenant: connect through the shared DB handle and skip
		// the per-tenant DSN/decrypt entirely — the tenant row carries no
		// connection coordinates for shared placements.
		sharedLease, err = p.sharedDBHandle(ctx, placement.DbID)
		if err != nil {
			return nil, nil, tidbCloudOrgID, nil, err
		}
		store = datastore.NewStoreWithDB(sharedLease.DB(), datastore.SharedScopeWithDB(fsID, sharedLease.DBID()))
	} else {
		decryptStart := time.Now()
		pass, err := p.enc.Decrypt(ctx, t.DBPasswordCipher)
		if err != nil {
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("decrypt db password: %w", err)
		}
		decryptDurationMs = float64(time.Since(decryptStart).Microseconds()) / 1000.0
		dsn := FormatTenantMySQLDSN(t.DBUser, string(pass), t.DBHost, t.DBPort, t.DBName, t.DBTLS, t.Provider)
		store, err = datastore.OpenForTenantScoped(ctx, dsn, t.ID, tidbCloudOrgID, datastore.StandaloneScope(fsID))
		if err != nil {
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("open datastore: %w", err)
		}
	}
	openStoreDurationMs := float64(time.Since(openStoreStart).Microseconds()) / 1000.0
	ensureSchemaDurationMs := 0.0
	migrateDurationMs := 0.0
	appManagedEmbedding := false
	if t.Provider == ProviderDB9 {
		ensureSchemaStart := time.Now()
		schemaCtx, cancel := detachSchemaOp(ctx)
		err := ensureDB9ContentLayoutOnAcquire(schemaCtx, store)
		cancel()
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("repair active db9 content layout schema: %w", err)
		}
		ensureSchemaDurationMs += float64(time.Since(ensureSchemaStart).Microseconds()) / 1000.0
	}
	if sharedTenant {
		// Shared schema has no generated columns, so database auto-embedding
		// is unavailable; schema is managed per physical DB, not per tenant,
		// so the Acquire-time ensure below is skipped entirely.
		opts.DatabaseAutoEmbedding = false
		opts.AppSemanticTasksEnabled = false
		// Narrow image-extract enablement: shared tenants keep only the
		// capability flag so the server-side extract config resolver can
		// execute img_extract_text with a per-tenant runtime. Image
		// extraction writes content_text and tags independently of
		// EMBED_TEXT, so a shared tenant with a custom config must not end
		// up in an "API accepts but runtime never runs" state. No
		// process-default extractors are wired for shared tenants.
		opts.AsyncImageExtract = backend.AsyncImageExtractOptions{Capability: opts.AsyncImageExtract.Capability}
		opts.AsyncAudioExtract = backend.AsyncAudioExtractOptions{Capability: opts.AsyncAudioExtract.Capability}
		opts.AsyncVideoExtract = backend.AsyncVideoExtractOptions{Capability: opts.AsyncVideoExtract.Capability}
		appManagedEmbedding = true
	} else if UsesTiDBAutoEmbedding(t.Provider) {
		autoEmbeddingProfile, err := p.autoEmbeddingProfileForTenant(ctx, t)
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("resolve tenant auto-embedding profile: %w", err)
		}
		if autoEmbeddingProfile.mode == meta.TenantEmbeddingModeAuto {
			opts.DatabaseAutoEmbedding = true
			opts.AppSemanticTasksEnabled = false
			opts.AppEmbeddingGeneration = 0
			opts.AppEmbeddingConfigRevision = 0
			opts.AppEmbeddingTenantManaged = false
			opts.QueryEmbedding = backend.QueryEmbeddingOptions{}
			if err := applyTiDBAutoEmbeddingProviderConfig(ctx, store.DB(), autoEmbeddingProfile.provider); err != nil {
				_ = store.Close()
				return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("apply tenant auto-embedding provider config: %w", err)
			}
		} else {
			opts.DatabaseAutoEmbedding = false
			opts.AppSemanticTasksEnabled = false
			// Effective fts_only tenants use resolver-backed extraction just like
			// shared tenants: preserve admission capability, but never inherit a
			// process-default provider runtime into the tenant backend.
			opts.AsyncImageExtract = backend.AsyncImageExtractOptions{Capability: opts.AsyncImageExtract.Capability}
			opts.AsyncAudioExtract = backend.AsyncAudioExtractOptions{Capability: opts.AsyncAudioExtract.Capability}
			opts.AsyncVideoExtract = backend.AsyncVideoExtractOptions{Capability: opts.AsyncVideoExtract.Capability}
			appManagedEmbedding = true
		}
		if !p.cfg.SkipTiDBSchemaCheck {
			targetSchemaVersion, err := TiDBTenantSchemaVersionForEmbeddingMode(autoEmbeddingProfile.mode, autoEmbeddingProfile.schemaProfile)
			if err != nil {
				_ = store.Close()
				return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("resolve tenant embedding schema version: %w", err)
			}
			if t.SchemaVersion != targetSchemaVersion {
				ensureSchemaStart := time.Now()
				// Serialize the detached ensure per physical tenant DB.
				// Without this fence, every concurrent cold open that
				// resolved the same stale SchemaVersion starts its own
				// detached ensure, and a burst of retries turns into a
				// DDL/backfill herd. The lock wait stays on the request
				// context so a caller keeps its deadline; once the lock is
				// held, re-read the durable version (another pod may have
				// finished while we waited) before running any DDL.
				err := withTenantSchemaAdvisoryLock(ctx, store.DB(), t.ID, func(lockCtx context.Context) error {
					// Detach after acquiring the lock (same shape as the
					// shared path): lock-wait time must not burn the
					// migration budget, and a caller that disconnected
					// while waiting must not fail the double-check or the
					// version persist — aborting here re-creates the
					// "backfill storage_ref_hash: context canceled" loop
					// where the new version never becomes durable.
					opCtx, cancel := detachSchemaOp(lockCtx)
					defer cancel()
					schemaCtx := schema.WithTenantID(opCtx, t.ID)
					if p.metaStore != nil {
						current, verErr := p.metaStore.GetTenant(opCtx, t.ID)
						if verErr != nil {
							return fmt.Errorf("re-read tenant schema version: %w", verErr)
						}
						if current.SchemaVersion == targetSchemaVersion {
							return nil
						}
					}
					if err := ensureTiDBSchemaForEmbeddingMode(schemaCtx, store.DB(), autoEmbeddingProfile.mode, autoEmbeddingProfile.schemaProfile); err != nil {
						return err
					}
					if p.metaStore != nil {
						// Record the tenant-profile-specific version only after the
						// schema has been confirmed to match that tenant's profile.
						// Version-matched opens trust this durable value and skip the
						// physical diff by default; out-of-band schema drift is detected
						// only when the target schema version changes, or through an
						// explicit validation/repair path. SkipTiDBSchemaCheck disables
						// the Acquire-time TiDB schema ensure/check entirely, mainly for
						// tests that use a TiDB-class provider against plain MySQL.
						updateSchemaVersionStart := time.Now()
						if verErr := p.metaStore.UpdateTenantSchemaVersion(opCtx, t.ID, targetSchemaVersion); verErr != nil {
							recordTenantSchemaVersionUpdateFailure(opCtx, t.ID, targetSchemaVersion, time.Since(updateSchemaVersionStart), verErr)
						}
					}
					return nil
				})
				if err != nil {
					_ = store.Close()
					return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("ensure tidb embedding schema: %w", err)
				}
				ensureSchemaDurationMs += float64(time.Since(ensureSchemaStart).Microseconds()) / 1000.0
			}
			if autoEmbeddingProfile.modeWasNull && p.metaStore != nil {
				// Single-row UPDATE, but still one-time convergence
				// bookkeeping: keep it off the (possibly canceled) request
				// context.
				modeCtx, modeCancel := detachSchemaOp(ctx)
				_, modeErr := p.metaStore.SetTenantAutoEmbeddingProfileModeIfNull(modeCtx, t.ID, autoEmbeddingProfile.mode)
				modeCancel()
				if modeErr != nil {
					_ = store.Close()
					return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("persist tenant embedding mode: %w", modeErr)
				}
			}
		}
	} else if t.Provider == ProviderLocal && !p.cfg.SkipTiDBSchemaCheck {
		targetSchemaVersion, err := schema.TiDBTenantSchemaVersionForEmbeddingModeProfile(schema.TiDBEmbeddingModeApp, schema.TiDBAutoEmbeddingProfile{})
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("resolve app-managed tenant schema version: %w", err)
		}
		if t.SchemaVersion != targetSchemaVersion {
			ensureSchemaStart := time.Now()
			err := withTenantSchemaAdvisoryLock(ctx, store.DB(), t.ID, func(lockCtx context.Context) error {
				opCtx, cancel := detachSchemaOp(lockCtx)
				defer cancel()
				schemaCtx := schema.WithTenantID(opCtx, t.ID)
				if p.metaStore != nil {
					current, verErr := p.metaStore.GetTenant(opCtx, t.ID)
					if verErr != nil {
						return fmt.Errorf("re-read tenant schema version: %w", verErr)
					}
					if current.SchemaVersion == targetSchemaVersion {
						return nil
					}
				}
				if err := schema.EnsureTiDBSchemaForMode(schemaCtx, store.DB(), schema.TiDBEmbeddingModeApp); err != nil {
					return err
				}
				if p.metaStore != nil {
					if verErr := p.metaStore.UpdateTenantSchemaVersion(opCtx, t.ID, targetSchemaVersion); verErr != nil {
						recordTenantSchemaVersionUpdateFailure(opCtx, t.ID, targetSchemaVersion, 0, verErr)
					}
				}
				return nil
			})
			if err != nil {
				_ = store.Close()
				return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("ensure app-managed tenant schema: %w", err)
			}
			ensureSchemaDurationMs += float64(time.Since(ensureSchemaStart).Microseconds()) / 1000.0
		}
	}
	if appManagedEmbedding {
		opts.AppSemanticTasksEnabled = false
		opts.AppEmbeddingGeneration = 0
		opts.AppEmbeddingConfigRevision = 0
		opts.AppEmbeddingTenantManaged = false
		opts.QueryEmbedding = backend.QueryEmbeddingOptions{}
		runtime, ready, err := p.resolveAppEmbeddingRuntime(ctx, t.ID)
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("resolve app embedding runtime: %w", err)
		}
		if ready {
			opts.AppSemanticTasksEnabled = true
			opts.QueryEmbedding = backend.QueryEmbeddingOptions{Client: runtime.Client}
			opts.AppEmbeddingGeneration = runtime.Generation
			opts.AppEmbeddingConfigRevision = runtime.ConfigRevision
			opts.AppEmbeddingTenantManaged = runtime.TenantManaged
		}
	}
	// Run split-tables migration if needed. Only for tenants that have the
	// legacy files table; new tenants skip it entirely.
	migrateStart := time.Now()
	if store.HasLegacyFiles() {
		// Same cancel-restart concern as the schema ensure above: each step
		// is a full-table INSERT IGNORE ... SELECT, and a client disconnect
		// rolling it back forces the next cold open to restart from
		// scratch. Detach with the same bounded budget.
		migCtx, migCancel := detachSchemaOp(ctx)
		err := p.migrateSplitTables(migCtx, store.DB(), t.Provider)
		migCancel()
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("migrate split tables: %w", err)
		}
	}
	migrateDurationMs = float64(time.Since(migrateStart).Microseconds()) / 1000.0
	if p.cfg.S3Bucket != "" {
		ns, err := p.resolveStorageNamespace(ctx, t, "s3", p.cfg.S3Bucket)
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, err
		}
		opts.StorageNamespaceID = ns.ID
		prefix := ns.Prefix
		s3ClientStart := time.Now()
		s3c, err := s3client.New(ctx, s3client.AWSConfig{
			Region:          p.cfg.S3Region,
			Bucket:          p.cfg.S3Bucket,
			Prefix:          prefix,
			RoleARN:         p.cfg.S3RoleARN,
			Endpoint:        p.cfg.S3Endpoint,
			ForcePathStyle:  p.cfg.S3ForcePathStyle,
			AccessKeyID:     p.cfg.S3AccessKeyID,
			SecretAccessKey: p.cfg.S3SecretAccessKey,
			SessionToken:    p.cfg.S3SessionToken,
		})
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("create aws s3 client: %w", err)
		}
		express, err := p.appendLogClientForStorageNamespace(ctx, ns)
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("create s3 express client: %w", err)
		}
		opts.AppendLogS3 = express
		s3ClientDurationMs := float64(time.Since(s3ClientStart).Microseconds()) / 1000.0
		smallInDB := SmallInDB(t.Provider)
		backendCreateStart := time.Now()
		b, err := backend.NewWithS3ModeAndOptions(store, s3c, smallInDB, opts)
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("create backend with s3 mode: %w", err)
		}
		backendCreateDurationMs := float64(time.Since(backendCreateStart).Microseconds()) / 1000.0
		totalDuration := time.Since(start)
		logger.InfoOpenPoolTiming(ctx, "tenant_pool_create_backend_timing", totalDuration,
			zap.String("tenant_id", t.ID),
			zap.String("provider", t.Provider),
			zap.String("storage_mode", "aws_s3"),
			zap.Float64("decrypt_db_password_ms", decryptDurationMs),
			zap.Float64("open_datastore_ms", openStoreDurationMs),
			zap.Float64("ensure_schema_ms", ensureSchemaDurationMs),
			zap.Float64("migrate_duration_ms", migrateDurationMs),
			zap.Float64("create_s3_client_ms", s3ClientDurationMs),
			zap.Float64("create_backend_ms", backendCreateDurationMs),
			zap.Float64("total_ms", durationMs(totalDuration)))
		p.wireQuotaStore(ctx, b, t.ID)
		if err := p.wireBackendAndScheduleEmbeddingBackfill(ctx, b, t.ID); err != nil {
			b.Close()
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, err
		}
		leaseTransferred = true
		return b, store, tidbCloudOrgID, sharedLease, nil
	}
	if p.cfg.S3Dir != "" {
		ns, err := p.resolveStorageNamespace(ctx, t, "local", "")
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, err
		}
		opts.StorageNamespaceID = ns.ID
		localPrefix := strings.Trim(ns.Prefix, "/")
		s3Dir := strings.TrimRight(p.cfg.S3Dir, "/") + "/" + localPrefix
		s3BaseURL := strings.TrimRight(p.cfg.PublicURL, "/") + "/s3/" + localPrefix
		s3ClientStart := time.Now()
		s3c, err := s3client.NewLocal(s3Dir, s3BaseURL)
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("create local s3 client: %w", err)
		}
		s3ClientDurationMs := float64(time.Since(s3ClientStart).Microseconds()) / 1000.0
		smallInDB := SmallInDB(t.Provider)
		backendCreateStart := time.Now()
		b, err := backend.NewWithS3ModeAndOptions(store, s3c, smallInDB, opts)
		if err != nil {
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("create backend with local s3 mode: %w", err)
		}
		backendCreateDurationMs := float64(time.Since(backendCreateStart).Microseconds()) / 1000.0
		totalDuration := time.Since(start)
		logger.InfoOpenPoolTiming(ctx, "tenant_pool_create_backend_timing", totalDuration,
			zap.String("tenant_id", t.ID),
			zap.String("provider", t.Provider),
			zap.String("storage_mode", "local_s3"),
			zap.Float64("decrypt_db_password_ms", decryptDurationMs),
			zap.Float64("open_datastore_ms", openStoreDurationMs),
			zap.Float64("ensure_schema_ms", ensureSchemaDurationMs),
			zap.Float64("migrate_duration_ms", migrateDurationMs),
			zap.Float64("create_s3_client_ms", s3ClientDurationMs),
			zap.Float64("create_backend_ms", backendCreateDurationMs),
			zap.Float64("total_ms", durationMs(totalDuration)))
		p.wireQuotaStore(ctx, b, t.ID)
		if err := p.wireBackendAndScheduleEmbeddingBackfill(ctx, b, t.ID); err != nil {
			b.Close()
			_ = store.Close()
			return nil, nil, tidbCloudOrgID, nil, err
		}
		leaseTransferred = true
		return b, store, tidbCloudOrgID, sharedLease, nil
	}
	backendCreateStart := time.Now()
	b, err := backend.NewWithOptions(store, opts)
	if err != nil {
		_ = store.Close()
		return nil, nil, tidbCloudOrgID, nil, fmt.Errorf("create backend: %w", err)
	}
	backendCreateDurationMs := float64(time.Since(backendCreateStart).Microseconds()) / 1000.0
	totalDuration := time.Since(start)
	logger.InfoOpenPoolTiming(ctx, "tenant_pool_create_backend_timing", totalDuration,
		zap.String("tenant_id", t.ID),
		zap.String("provider", t.Provider),
		zap.String("storage_mode", "db_only"),
		zap.Float64("decrypt_db_password_ms", decryptDurationMs),
		zap.Float64("open_datastore_ms", openStoreDurationMs),
		zap.Float64("ensure_schema_ms", ensureSchemaDurationMs),
		zap.Float64("migrate_duration_ms", migrateDurationMs),
		zap.Float64("create_s3_client_ms", 0),
		zap.Float64("create_backend_ms", backendCreateDurationMs),
		zap.Float64("total_ms", durationMs(totalDuration)))
	p.wireQuotaStore(ctx, b, t.ID)
	if err := p.wireBackendAndScheduleEmbeddingBackfill(ctx, b, t.ID); err != nil {
		b.Close()
		_ = store.Close()
		return nil, nil, tidbCloudOrgID, nil, err
	}
	leaseTransferred = true
	return b, store, tidbCloudOrgID, sharedLease, nil
}

func ensureDB9ContentLayoutOnAcquire(ctx context.Context, store *datastore.Store) error {
	if store == nil || store.DB() == nil {
		return fmt.Errorf("nil db9 datastore")
	}
	for i, stmt := range schema.DB9ContentLayoutRepairStatements() {
		if i == 0 && !store.HasLegacyFiles() {
			continue
		}
		if _, err := store.DB().ExecContext(ctx, stmt); err != nil {
			return fmt.Errorf("repair db9 content layout schema %q: %w", stmt, err)
		}
	}
	return nil
}

func (p *Pool) migrateSplitTables(ctx context.Context, db *sql.DB, provider string) error {
	// All current providers use MySQL/TiDB as the metadata store,
	// so we default to MySQL dialect.
	dialect := migrate.DialectMySQL
	m := migrate.NewSplitTablesMigratorWithDialect(db, dialect)
	_, err := m.Run(ctx)
	return err
}

func (p *Pool) autoEmbeddingProfileForTenant(ctx context.Context, t *meta.Tenant) (tenantAutoEmbeddingProfile, error) {
	if p.metaStore == nil || t == nil || t.ID == "" {
		out, err := defaultTenantAutoEmbeddingProfile()
		if err != nil {
			return tenantAutoEmbeddingProfile{}, err
		}
		mode, _, err := p.resolveTenantEmbeddingMode("")
		if err != nil {
			return tenantAutoEmbeddingProfile{}, err
		}
		out.mode = mode
		return out, nil
	}
	profile, err := p.metaStore.EnsureTenantAutoEmbeddingProfile(ctx, t.ID)
	if err != nil {
		return tenantAutoEmbeddingProfile{}, err
	}
	mode, modeWasNull, err := p.resolveTenantEmbeddingMode(profile.EmbeddingMode)
	if err != nil {
		return tenantAutoEmbeddingProfile{}, err
	}
	schemaProfile := schema.TiDBAutoEmbeddingProfile{
		Model:       profile.Model,
		Dimensions:  profile.Dimensions,
		OptionsJSON: profile.OptionsJSON,
	}
	apiKey := ""
	if len(profile.APIKeyCipher) > 0 {
		plain, err := p.enc.Decrypt(ctx, profile.APIKeyCipher)
		if err != nil {
			return tenantAutoEmbeddingProfile{}, fmt.Errorf("decrypt tenant auto-embedding api key: %w", err)
		}
		apiKey = string(plain)
	}
	return tenantAutoEmbeddingProfile{
		schemaProfile: schemaProfile,
		mode:          mode,
		modeWasNull:   modeWasNull,
		provider: schema.TiDBAutoEmbeddingProviderConfig{
			Model:   schemaProfile.Model,
			APIKey:  apiKey,
			APIBase: profile.APIBase,
		},
	}, nil
}

func defaultTenantAutoEmbeddingProfile() (tenantAutoEmbeddingProfile, error) {
	schemaProfile, err := schema.TiDBAutoEmbeddingProfileFromConfig(schema.TiDBAutoEmbeddingConfig{
		Model:      schema.DefaultTiDBAutoEmbeddingModel,
		Dimensions: schema.DefaultTiDBAutoEmbeddingDimensions,
	})
	if err != nil {
		return tenantAutoEmbeddingProfile{}, fmt.Errorf("build default tenant auto-embedding profile: %w", err)
	}
	return tenantAutoEmbeddingProfile{
		schemaProfile: schemaProfile,
		mode:          meta.TenantEmbeddingModeAuto,
		provider: schema.TiDBAutoEmbeddingProviderConfig{
			Model: schemaProfile.Model,
		},
	}, nil
}

func recordTenantSchemaVersionUpdateFailure(ctx context.Context, tenantID string, version int, d time.Duration, err error) {
	logger.Warn(ctx, "tenant_pool_update_schema_version_failed",
		zap.String("tenant_id", tenantID),
		zap.Int("version", version),
		zap.Error(err))
	metrics.RecordOperation("tenant_pool", "update_schema_version_failed", "error", d)
}

func (p *Pool) resolveStorageNamespace(ctx context.Context, t *meta.Tenant, backendName, bucket string) (*meta.StorageNamespace, error) {
	if p.metaStore == nil {
		return &meta.StorageNamespace{
			ID:            t.ID,
			OwnerTenantID: t.ID,
			Backend:       backendName,
			Bucket:        bucket,
			Prefix:        p.defaultStorageNamespacePrefix(t.ID, backendName),
			State:         meta.StorageNamespaceActive,
		}, nil
	}
	ns, err := p.metaStore.EnsureTenantStorageNamespace(ctx, t.ID, backendName, bucket, p.defaultStorageNamespacePrefix(t.ID, backendName))
	if err != nil {
		return nil, fmt.Errorf("resolve storage namespace: %w", err)
	}
	// Keep the tenant value in sync with the namespace persisted by
	// EnsureTenantStorageNamespace. Acquire uses this resolved ID to key the
	// cache entry created after backend construction.
	t.StorageNamespaceID = ns.ID
	return ns, nil
}

func (p *Pool) defaultStorageNamespacePrefix(tenantID, backendName string) string {
	if backendName == "s3" {
		prefix := strings.Trim(p.cfg.S3Prefix, "/")
		if prefix != "" {
			prefix += "/"
		}
		return prefix + tenantID + "/"
	}
	return tenantID + "/"
}

// wireQuotaStore sets the central quota store on a newly created backend.
// No-op when the pool's metaStore is nil (tests, non-multi-tenant mode).
func (p *Pool) wireQuotaStore(ctx context.Context, b *backend.Dat9Backend, tenantID string) {
	if p.metaStore == nil {
		return
	}
	adapter := NewMetaQuotaAdapter(p.metaStore)
	b.SetMetaQuotaStore(ctx, tenantID, adapter)
}

// removeLocked drops elem from the pool. reason distinguishes capacity eviction
// ("evict") from entry replacement ("replace"), explicit invalidation
// ("invalidate"), and pool shutdown ("shutdown") so the tenant_pool/remove metric
// doesn't conflate them — only result="evict" reflects real cache pressure.
func (p *Pool) removeLocked(elem *list.Element, reason string) *entry {
	e := p.removeLockedWithoutMetrics(elem)
	p.recordCachedBackendCountLocked()
	metrics.RecordOperation("tenant_pool", "remove", reason, 0)
	return e
}

// removeLockedWithoutMetrics removes one cache entry while p.mu is held. It is
// used by batch retirement so metrics can be coalesced and emitted after the
// request-path lock is released.
func (p *Pool) removeLockedWithoutMetrics(elem *list.Element) *entry {
	e := elem.Value.(*entry)
	p.order.Remove(elem)
	e.elem = nil
	delete(p.items, e.tenantID)
	e.retired = true
	if e.refs == 0 {
		return e
	}
	return nil
}

func (p *Pool) recordCachedBackendCountLocked() {
	metrics.RecordGauge("tenant_pool", "cached_backends", float64(len(p.items)))
}

func (p *Pool) makeRelease(e *entry) func() {
	var once sync.Once
	return func() {
		once.Do(func() {
			p.releaseEntry(e, true, true)
		})
	}
}

// makeReleaseNoRefresh is like makeRelease but does not refresh lastUsed on
// release. Used by AcquireCached so internal scans (safety-net) do not reset
// the idle timer.
func (p *Pool) makeReleaseNoRefresh(e *entry) func() {
	var once sync.Once
	return func() {
		once.Do(func() {
			p.releaseEntry(e, false, false)
		})
	}
}

func (p *Pool) releaseEntry(e *entry, refreshLastUsed, refreshSharedActivity bool) {
	if e == nil {
		return
	}
	var toClose *entry
	p.mu.Lock()
	if e.refs > 0 {
		e.refs--
	}
	if e.refs == 0 && !e.retired && refreshLastUsed {
		e.lastUsed = time.Now()
	}
	if !e.retired && refreshSharedActivity {
		p.touchSharedDBActivityLocked(e.sharedDBID, time.Now())
	}
	if e.refs == 0 && e.retired {
		toClose = e
	}
	p.mu.Unlock()
	if toClose != nil {
		p.closeEntry(toClose)
	}
}

func (p *Pool) closeEntry(e *entry) {
	if e == nil {
		return
	}
	// A retired-with-refs entry can outlive a replacement for the same
	// tenant. FileGC on the new entry may already be exporting gauges;
	// clearing here would hide an active drain until the next Claim.
	p.mu.Lock()
	current := p.items[e.tenantID]
	if current == nil || current == e {
		metrics.ClearFileGCClaimableBacklog(e.tenantID)
	}
	p.mu.Unlock()
	if e.backend != nil {
		e.backend.Close()
	}
	if e.store != nil {
		_ = e.store.Close()
	}
	e.sharedDBLease.Release()
}

type tenantPoolResult string

const (
	tenantPoolResultOK                   tenantPoolResult = "ok"
	tenantPoolResultNotFound             tenantPoolResult = "not_found"
	tenantPoolResultAuthFailed           tenantPoolResult = "auth_failed"
	tenantPoolResultUsageQuotaExhausted  tenantPoolResult = "usage_quota_exhausted"
	tenantPoolResultError                tenantPoolResult = "error"
	defaultTenantMetricTiDBCloudOrgID    string           = "guest"
	mysqlErrAccessDenied                 uint16           = 1045
	mysqlErrUsageQuotaExhaustedCandidate uint16           = 1105
)

func observePool(ctx context.Context, op, tenantID string, errp *error, start time.Time) {
	result := tenantPoolResultOK
	if errp != nil && *errp != nil {
		result = tenantPoolErrorResult(*errp)
		fields := []zap.Field{zap.String("operation", op), zap.String("tenant_id", tenantID), zap.String("result", string(result)), zap.Error(*errp)}
		if result == tenantPoolResultError {
			logger.Error(ctx, "tenant_pool_op_failed", fields...)
		} else {
			logger.Warn(ctx, "tenant_pool_op_failed", fields...)
		}
	}
	metrics.RecordOperation("tenant_pool", op, string(result), time.Since(start))
}

func tenantPoolErrorResult(err error) tenantPoolResult {
	switch {
	case err == nil:
		return tenantPoolResultOK
	case errors.Is(err, meta.ErrNotFound):
		return tenantPoolResultNotFound
	}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		switch mysqlErr.Number {
		case mysqlErrAccessDenied:
			return tenantPoolResultAuthFailed
		case mysqlErrUsageQuotaExhaustedCandidate:
			if strings.Contains(strings.ToLower(mysqlErr.Message), "usage quota") {
				return tenantPoolResultUsageQuotaExhausted
			}
		}
	}
	msg := strings.ToLower(err.Error())
	switch {
	case strings.Contains(msg, "please check your user name and password"):
		return tenantPoolResultAuthFailed
	case strings.Contains(msg, "due to the usage quota being exhausted"):
		return tenantPoolResultUsageQuotaExhausted
	default:
		return tenantPoolResultError
	}
}