package tenant

import (
	"context"
	"crypto/rand"
	"database/sql"
	"errors"
	"fmt"
	"net"
	"net/http/httptest"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"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/embedding"
	"github.com/mem9-ai/drive9/pkg/encrypt"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/semantic"
	"github.com/mem9-ai/drive9/pkg/tenant/schema"
	"github.com/mem9-ai/drive9/pkg/testtidb"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

func TestNewPoolUsesDefaultMaxTenants(t *testing.T) {
	pool := NewPool(PoolConfig{}, nil)
	if pool.maxSize != defaultTenantPoolMaxTenants {
		t.Fatalf("max size = %d, want %d", pool.maxSize, defaultTenantPoolMaxTenants)
	}
}

func TestSetTenantWorkNotifierKicksAlreadyWarmTenant(t *testing.T) {
	pool, tenantRecord := newTestPoolAndTenant(t, 1, "tenant-warm-before-notifier")
	_, release, err := pool.Acquire(context.Background(), tenantRecord)
	if err != nil {
		t.Fatalf("warm tenant: %v", err)
	}
	release()

	kicked := make(chan int, 1)
	pool.SetTenantWorkNotifier(func(tenantID, _ string, workMask int) {
		if tenantID == tenantRecord.ID {
			kicked <- workMask
		}
	})

	select {
	case workMask := <-kicked:
		if workMask != backend.BackendWorkStorageTierScan {
			t.Fatalf("work mask = %d, want storage-tier scan", workMask)
		}
	case <-time.After(time.Second):
		t.Fatal("warm tenant was not kicked when notifier was installed")
	}
}

func TestSharedDBHandleIdleTTLIsTenMinutes(t *testing.T) {
	if defaultSharedDBHandleIdleTTL != 10*time.Minute {
		t.Fatalf("shared DB handle idle TTL = %v, want 10m", defaultSharedDBHandleIdleTTL)
	}
}

func TestNewPoolRecordsBackendCacheCapacityMetrics(t *testing.T) {
	for _, tt := range []struct {
		name       string
		maxTenants int
		wantMax    int
	}{
		{name: "default capacity", wantMax: defaultTenantPoolMaxTenants},
		{name: "configured capacity", maxTenants: 7, wantMax: 7},
	} {
		t.Run(tt.name, func(t *testing.T) {
			NewPool(PoolConfig{MaxTenants: tt.maxTenants}, nil)

			rec := httptest.NewRecorder()
			metrics.WritePrometheus(rec)
			text := rec.Body.String()
			if !strings.Contains(text, `drive9_service_gauge{component="tenant_pool",name="cached_backends",tenant_id="",tidbcloud_org_id="guest"} 0.000000`) {
				t.Fatalf("missing initialized cached backend gauge:\n%s", text)
			}
			wantMax := `drive9_service_gauge{component="tenant_pool",name="max_backends",tenant_id="",tidbcloud_org_id="guest"} ` + strconv.Itoa(tt.wantMax) + `.000000`
			if !strings.Contains(text, wantMax) {
				t.Fatalf("missing max backend gauge %q:\n%s", wantMax, text)
			}
		})
	}
}

func TestPoolAcquireRecordsCachedBackendGauge(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-metric-acquire")
	_, release, err := pool.Acquire(context.Background(), tenant)
	if err != nil {
		t.Fatal(err)
	}
	release()
	assertCachedBackendGauge(t, 1)
}

func TestPoolInvalidateRecordsCachedBackendGauge(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-metric-invalidate")
	_, release, err := pool.Acquire(context.Background(), tenant)
	if err != nil {
		t.Fatal(err)
	}
	release()
	assertCachedBackendGauge(t, 1)

	pool.Invalidate(tenant.ID)
	assertCachedBackendGauge(t, 0)
}

func TestLoadS3BackendPinsUseUntilRelease(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-local-s3-pin")
	metaStore, err := meta.OpenContext(context.Background(), testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	tenant.S3EncryptionMode = meta.S3EncryptionModeInherit
	if err := metaStore.InsertTenant(context.Background(), tenant); err != nil {
		t.Fatal(err)
	}
	pool.SetMetaStore(metaStore)

	if b, release := pool.LoadS3Backend(context.Background(), metaStore, tenant.ID); b != nil || release != nil {
		if release != nil {
			release()
		}
		t.Errorf("cold LoadS3Backend returned backend=%t release=%t, want both false", b != nil, release != nil)
	}
	_, warmRelease, err := pool.Acquire(context.Background(), tenant)
	if err != nil {
		t.Fatal(err)
	}
	warmRelease()
	oldLastUsed := time.Now().Add(-time.Hour)
	pool.mu.Lock()
	pool.items[tenant.ID].lastUsed = oldLastUsed
	pool.mu.Unlock()

	b, release := pool.LoadS3Backend(context.Background(), metaStore, tenant.ID)
	if b == nil || release == nil {
		t.Fatal("LoadS3Backend did not return a pinned backend")
	}
	pool.mu.Lock()
	lastUsed := pool.items[tenant.ID].lastUsed
	pool.mu.Unlock()
	if !lastUsed.After(oldLastUsed) {
		t.Errorf("warm LoadS3Backend lastUsed=%v, want after %v", lastUsed, oldLastUsed)
	}
	store := b.Store()
	pool.Invalidate(tenant.ID)
	assertStoreOpen(t, store)
	release()
	assertStoreClosed(t, store)
}

func TestPoolInvalidateSharedDBClosesOnlySelectedHandle(t *testing.T) {
	pool := NewPool(PoolConfig{}, nil)
	t.Cleanup(pool.Close)
	db1, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	db2, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := db1.Ping(); err != nil {
		t.Fatal(err)
	}
	if err := db2.Ping(); err != nil {
		t.Fatal(err)
	}
	pool.sharedDBs[1] = db1
	pool.sharedDBs[2] = db2

	if err := pool.InvalidateSharedDB(1); err != nil {
		t.Fatalf("InvalidateSharedDB: %v", err)
	}
	if _, ok := pool.sharedDBs[1]; ok {
		t.Fatal("invalidated shared DB remains cached")
	}
	if _, ok := pool.sharedDBs[2]; !ok {
		t.Fatal("unrelated shared DB was removed")
	}
	if err := db1.Ping(); err == nil {
		t.Fatal("invalidated shared DB handle remains open")
	}
	if err := db2.Ping(); err != nil {
		t.Fatalf("unrelated shared DB handle was closed: %v", err)
	}
	if err := pool.InvalidateSharedDB(1); err != nil {
		t.Fatalf("second InvalidateSharedDB: %v", err)
	}
	if _, ok := pool.sharedDBRefs[1]; ok {
		t.Fatal("invalidation left refs for an invalidated shared DB")
	}
	if _, ok := pool.sharedDBLastUsed[1]; ok {
		t.Fatal("invalidation left last-used state for an invalidated shared DB")
	}
}

func TestSharedDBHandleLeasePreventsIdleReapDuringColdOpen(t *testing.T) {
	pool := NewPool(PoolConfig{}, nil)
	t.Cleanup(pool.Close)
	handle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := handle.Ping(); err != nil {
		t.Fatal(err)
	}
	const dbID int64 = 41
	pool.sharedDBs[dbID] = handle
	pool.sharedDBLastUsed[dbID] = time.Now().Add(-defaultSharedDBHandleIdleTTL - time.Second)

	lease, err := pool.sharedDBHandle(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	// Force the activity timestamp back past the TTL after acquisition. The
	// explicit lease, rather than a timestamp grace period, must protect the
	// in-flight cold open from the reaper.
	pool.sharedMu.Lock()
	pool.sharedDBLastUsed[dbID] = time.Now().Add(-defaultSharedDBHandleIdleTTL - time.Second)
	pool.sharedMu.Unlock()

	pool.reapIdleSharedDBs(time.Now())
	if err := lease.DB().Ping(); err != nil {
		t.Fatalf("leased shared handle closed during cold open: %v", err)
	}
	if got := pool.sharedDBRefs[dbID]; got != 1 {
		t.Fatalf("shared handle refs while leased = %d, want 1", got)
	}

	lease.Release()
	pool.reapIdleSharedDBs(time.Now())
	if err := handle.Ping(); err == nil {
		t.Fatal("expired shared handle remains open after lease release")
	}
}

func TestSharedDBHandleLeaseRefreshesRealActivity(t *testing.T) {
	pool := NewPool(PoolConfig{}, nil)
	t.Cleanup(pool.Close)
	handle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := handle.Ping(); err != nil {
		t.Fatal(err)
	}
	const dbID int64 = 43
	pool.sharedDBs[dbID] = handle
	pool.sharedDBLastUsed[dbID] = time.Now().Add(-defaultSharedDBHandleIdleTTL - time.Second)

	lease, err := pool.sharedDBHandle(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	lease.Release()
	pool.reapIdleSharedDBs(time.Now())
	if err := handle.Ping(); err != nil {
		t.Fatalf("recently used shared handle was reaped: %v", err)
	}
}

func TestSharedDBHandleLeaseReleaseDoesNotAffectReopenedGeneration(t *testing.T) {
	pool := NewPool(PoolConfig{}, nil)
	t.Cleanup(pool.Close)
	oldHandle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := oldHandle.Ping(); err != nil {
		t.Fatal(err)
	}
	const dbID int64 = 42
	pool.sharedDBs[dbID] = oldHandle
	pool.sharedDBLastUsed[dbID] = time.Now()
	oldLease, err := pool.sharedDBHandle(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}

	if err := pool.InvalidateSharedDB(dbID); err != nil {
		t.Fatal(err)
	}
	newHandle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := newHandle.Ping(); err != nil {
		t.Fatal(err)
	}
	pool.sharedMu.Lock()
	pool.sharedDBs[dbID] = newHandle
	pool.sharedDBRefs[dbID] = 1
	pool.sharedDBLastUsed[dbID] = time.Now()
	pool.sharedMu.Unlock()

	oldLease.Release()
	pool.sharedMu.Lock()
	refs := pool.sharedDBRefs[dbID]
	current := pool.sharedDBs[dbID]
	pool.sharedMu.Unlock()
	if refs != 1 {
		t.Fatalf("new shared handle refs after old lease release = %d, want 1", refs)
	}
	if current != newHandle {
		t.Fatal("old lease release replaced the current shared handle")
	}
}

func TestSharedDBHandleIdleReapWaitsForReferencedHandlesToDrain(t *testing.T) {
	pool := NewPool(PoolConfig{IdleTimeout: 5 * time.Minute}, nil)
	t.Cleanup(pool.Close)
	now := time.Now()

	openDB := func() *sql.DB {
		t.Helper()
		db, err := sql.Open("mysql", testDSN)
		if err != nil {
			t.Fatal(err)
		}
		if err := db.Ping(); err != nil {
			t.Fatal(err)
		}
		return db
	}
	expired := openDB()
	recent := openDB()
	referenced := openDB()
	pool.sharedDBs[1] = expired
	pool.sharedDBs[2] = recent
	pool.sharedDBs[3] = referenced
	pool.sharedDBLastUsed[1] = now.Add(-defaultSharedDBHandleIdleTTL - time.Second)
	pool.sharedDBLastUsed[2] = now.Add(-5 * time.Minute)
	pool.sharedMu.Lock()
	referencedLease := pool.leaseSharedDBLocked(3, referenced)
	pool.sharedMu.Unlock()
	pool.sharedDBLastUsed[3] = now.Add(-defaultSharedDBHandleIdleTTL - time.Second)

	pool.reapIdleSharedDBs(now)

	if _, ok := pool.sharedDBs[1]; ok {
		t.Fatal("expired unreferenced shared handle remains cached")
	}
	if err := expired.Ping(); err == nil {
		t.Fatal("expired unreferenced shared handle remains open")
	}
	if _, ok := pool.sharedDBs[2]; !ok {
		t.Fatal("recent shared handle was evicted at tenant 5m TTL")
	}
	if _, ok := pool.sharedDBs[3]; !ok {
		t.Fatal("referenced shared handle was evicted")
	}

	referencedLease.Release()
	pool.reapIdleSharedDBs(now)
	if _, ok := pool.sharedDBs[3]; ok {
		t.Fatal("activity-idle shared handle remains cached after references drained")
	}
	if err := referenced.Ping(); err == nil {
		t.Fatal("activity-idle shared handle remains open after references drained")
	}
}

func TestSharedDBHandleIdleReapRunsWhenTenantIdleEvictionDisabled(t *testing.T) {
	p := NewPool(PoolConfig{IdleTimeout: 0}, nil)
	t.Cleanup(p.Close)
	db, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := db.Ping(); err != nil {
		t.Fatal(err)
	}
	p.sharedDBs[1] = db
	p.sharedDBLastUsed[1] = time.Now().Add(-defaultSharedDBHandleIdleTTL - time.Second)

	p.reapOnce(context.Background())

	if _, ok := p.sharedDBs[1]; ok {
		t.Fatal("expired unreferenced shared handle remains cached when tenant idle eviction is disabled")
	}
	if err := db.Ping(); err == nil {
		t.Fatal("expired unreferenced shared handle remains open when tenant idle eviction is disabled")
	}
}

func registerSharedDBForCacheMetrics(t *testing.T, orgID string) (*meta.Store, *Pool, int64, string) {
	t.Helper()
	metaStore, err := meta.OpenContext(context.Background(), testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	dsnCfg, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, portText, err := net.SplitHostPort(dsnCfg.Addr)
	if err != nil {
		t.Fatal(err)
	}
	port, err := strconv.Atoi(portText)
	if err != nil {
		t.Fatal(err)
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	passwordCipher, err := enc.Encrypt(context.Background(), []byte(dsnCfg.Passwd))
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := metaStore.RegisterSharedDB(context.Background(), &meta.SharedDB{
		TiDBCloudOrganizationID: orgID, Host: host, Port: port,
		User: dsnCfg.User, PasswordCipher: passwordCipher, Name: dsnCfg.DBName,
		MaxTenants: 100, Status: meta.SharedDBStatusActive,
	})
	if err != nil {
		t.Fatal(err)
	}
	previousEnsure := ensureSharedDBSchema
	ensureSharedDBSchema = func(ctx context.Context, db *sql.DB) error {
		return db.PingContext(ctx)
	}
	t.Cleanup(func() { ensureSharedDBSchema = previousEnsure })
	pool := NewPool(PoolConfig{}, enc)
	pool.SetMetaStore(metaStore)
	t.Cleanup(pool.Close)
	if err := pool.EnsureSharedDBReady(context.Background(), dbID); err != nil {
		t.Fatal(err)
	}
	info, err := metaStore.GetSharedDB(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	return metaStore, pool, dbID, info.UUID
}

func assertSharedDBCacheMetric(t *testing.T, series string, wantPresent bool) {
	t.Helper()
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if wantPresent && !strings.Contains(text, series) {
		t.Fatalf("missing shared DB cache metric %q:\n%s", series, text)
	}
	if !wantPresent && strings.Contains(text, series) {
		t.Fatalf("stale shared DB cache metric %q:\n%s", series, text)
	}
}

func TestSharedDBCacheMetricsTrackHandleLifecycle(t *testing.T) {
	_, pool, dbID, uuid := registerSharedDBForCacheMetrics(t, "org-cache-metrics")
	dbPoolID := strconv.FormatInt(dbID, 10)
	labels := `db_pool_id="` + dbPoolID + `",db_pool_uuid="` + uuid + `",tidbcloud_org_id="org-cache-metrics"`
	handleSeries := `drive9_shared_db_pool_cache_handles{` + labels + `}`
	tenantsSeries := func(value string) string {
		return `drive9_shared_db_pool_cache_tenants{` + labels + `} ` + value
	}

	assertSharedDBCacheMetric(t, handleSeries+` 1.000000`, true)
	assertSharedDBCacheMetric(t, tenantsSeries(`0.000000`), true)

	firstLease, err := pool.sharedDBHandle(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	assertSharedDBCacheMetric(t, tenantsSeries(`1.000000`), true)
	secondLease, err := pool.sharedDBHandle(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	assertSharedDBCacheMetric(t, tenantsSeries(`2.000000`), true)
	firstLease.Release()
	assertSharedDBCacheMetric(t, tenantsSeries(`1.000000`), true)
	secondLease.Release()
	assertSharedDBCacheMetric(t, tenantsSeries(`0.000000`), true)

	if err := pool.InvalidateSharedDB(dbID); err != nil {
		t.Fatal(err)
	}
	assertSharedDBCacheMetric(t, handleSeries, false)
	assertSharedDBCacheMetric(t, `drive9_shared_db_pool_cache_tenants{`+labels+`}`, false)
}

func TestSharedDBCacheMetricsDeletedOnIdleReap(t *testing.T) {
	_, pool, dbID, uuid := registerSharedDBForCacheMetrics(t, "org-cache-reap")
	dbPoolID := strconv.FormatInt(dbID, 10)
	labels := `db_pool_id="` + dbPoolID + `",db_pool_uuid="` + uuid + `",tidbcloud_org_id="org-cache-reap"`
	handleSeries := `drive9_shared_db_pool_cache_handles{` + labels + `}`
	assertSharedDBCacheMetric(t, handleSeries+` 1.000000`, true)
	assertSharedDBCacheMetric(t, `drive9_shared_db_pool_cache_tenants{`+labels+`} 0.000000`, true)

	pool.sharedMu.Lock()
	pool.sharedDBLastUsed[dbID] = time.Now().Add(-defaultSharedDBHandleIdleTTL - time.Second)
	pool.sharedMu.Unlock()
	pool.reapIdleSharedDBs(time.Now())

	assertSharedDBCacheMetric(t, handleSeries, false)
	assertSharedDBCacheMetric(t, `drive9_shared_db_pool_cache_tenants{`+labels+`}`, false)
}

func TestSharedDBCacheMetricsDeletedOnPoolClose(t *testing.T) {
	_, pool, dbID, uuid := registerSharedDBForCacheMetrics(t, "org-cache-close")
	dbPoolID := strconv.FormatInt(dbID, 10)
	labels := `db_pool_id="` + dbPoolID + `",db_pool_uuid="` + uuid + `",tidbcloud_org_id="org-cache-close"`
	handleSeries := `drive9_shared_db_pool_cache_handles{` + labels + `}`
	assertSharedDBCacheMetric(t, handleSeries+` 1.000000`, true)
	assertSharedDBCacheMetric(t, `drive9_shared_db_pool_cache_tenants{`+labels+`} 0.000000`, true)

	pool.Close()

	assertSharedDBCacheMetric(t, handleSeries, false)
	assertSharedDBCacheMetric(t, `drive9_shared_db_pool_cache_tenants{`+labels+`}`, false)
}

func TestSharedDBSchemaIsCheckedLazilyAndRetriedAfterFailure(t *testing.T) {
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	dsnCfg, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, portText, err := net.SplitHostPort(dsnCfg.Addr)
	if err != nil {
		t.Fatal(err)
	}
	port, err := strconv.Atoi(portText)
	if err != nil {
		t.Fatal(err)
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	passwordCipher, err := enc.Encrypt(context.Background(), []byte(dsnCfg.Passwd))
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := metaStore.RegisterSharedDB(context.Background(), &meta.SharedDB{
		TiDBCloudOrganizationID: "org-lazy-schema", Host: host, Port: port,
		User: dsnCfg.User, PasswordCipher: passwordCipher, Name: dsnCfg.DBName,
		MaxTenants: 100, Status: meta.SharedDBStatusActive,
	})
	if err != nil {
		t.Fatal(err)
	}
	pool := NewPool(PoolConfig{}, enc)
	pool.SetMetaStore(metaStore)
	t.Cleanup(pool.Close)

	wantErr := errors.New("schema apply failed")
	ensureCalls := 0
	previousEnsure := ensureSharedDBSchema
	ensureSharedDBSchema = func(ctx context.Context, db *sql.DB) error {
		ensureCalls++
		if err := db.PingContext(ctx); err != nil {
			return err
		}
		if ensureCalls == 1 {
			return wantErr
		}
		return nil
	}
	t.Cleanup(func() { ensureSharedDBSchema = previousEnsure })

	if err := pool.EnsureSharedDBReady(context.Background(), dbID); !errors.Is(err, wantErr) {
		t.Fatalf("first EnsureSharedDBReady = %v, want %v", err, wantErr)
	}
	if _, ok := pool.sharedDBs[dbID]; ok {
		t.Fatal("failed schema ensure cached the shared DB handle")
	}
	info, err := metaStore.GetSharedDB(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	if info.SchemaVersion != 0 {
		t.Fatalf("schema version after failure = %d, want 0", info.SchemaVersion)
	}

	if err := pool.EnsureSharedDBReady(context.Background(), dbID); err != nil {
		t.Fatalf("second EnsureSharedDBReady: %v", err)
	}
	if err := pool.EnsureSharedDBReady(context.Background(), dbID); err != nil {
		t.Fatalf("cached EnsureSharedDBReady: %v", err)
	}
	if ensureCalls != 2 {
		t.Fatalf("schema ensure calls = %d, want 2", ensureCalls)
	}
	info, err = metaStore.GetSharedDB(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	if info.SchemaVersion != schema.CurrentSharedTiDBSchemaVersion {
		t.Fatalf("schema version = %d, want %d", info.SchemaVersion, schema.CurrentSharedTiDBSchemaVersion)
	}

	if err := pool.InvalidateSharedDB(dbID); err != nil {
		t.Fatal(err)
	}
	if err := pool.EnsureSharedDBReady(context.Background(), dbID); err != nil {
		t.Fatalf("reopen current schema: %v", err)
	}
	if ensureCalls != 2 {
		t.Fatalf("schema ensure calls after current-version reopen = %d, want 2", ensureCalls)
	}
}

func TestSharedDBSchemaEnsureUsesCrossPoolAdvisoryLock(t *testing.T) {
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	dsnCfg, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, portText, err := net.SplitHostPort(dsnCfg.Addr)
	if err != nil {
		t.Fatal(err)
	}
	port, err := strconv.Atoi(portText)
	if err != nil {
		t.Fatal(err)
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	passwordCipher, err := enc.Encrypt(context.Background(), []byte(dsnCfg.Passwd))
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := metaStore.RegisterSharedDB(context.Background(), &meta.SharedDB{
		TiDBCloudOrganizationID: "org-schema-advisory-lock", Host: host, Port: port,
		User: dsnCfg.User, PasswordCipher: passwordCipher, Name: dsnCfg.DBName,
		MaxTenants: 100, Status: meta.SharedDBStatusActive,
	})
	if err != nil {
		t.Fatal(err)
	}
	firstPool := NewPool(PoolConfig{}, enc)
	firstPool.SetMetaStore(metaStore)
	t.Cleanup(firstPool.Close)
	secondPool := NewPool(PoolConfig{}, enc)
	secondPool.SetMetaStore(metaStore)
	t.Cleanup(secondPool.Close)

	var ensureCalls atomic.Int32
	firstEnsureStarted := make(chan struct{})
	releaseFirstEnsure := make(chan struct{})
	var releaseFirstEnsureOnce sync.Once
	releaseFirst := func() { releaseFirstEnsureOnce.Do(func() { close(releaseFirstEnsure) }) }
	previousEnsure := ensureSharedDBSchema
	// Register the global-hook restore first so the later release-and-drain
	// cleanup runs before it on every failure path.
	t.Cleanup(func() { ensureSharedDBSchema = previousEnsure })
	ensureSharedDBSchema = func(ctx context.Context, db *sql.DB) error {
		if ensureCalls.Add(1) == 1 {
			close(firstEnsureStarted)
			<-releaseFirstEnsure
		}
		return db.PingContext(ctx)
	}

	firstDone := make(chan error, 1)
	firstExited := make(chan struct{})
	t.Cleanup(func() {
		releaseFirst()
		select {
		case <-firstExited:
		case <-time.After(time.Second):
			t.Error("first shared schema ensure did not exit during cleanup")
		}
	})
	go func() {
		defer close(firstExited)
		firstDone <- firstPool.EnsureSharedDBReady(context.Background(), dbID)
	}()
	<-firstEnsureStarted
	secondDone := make(chan error, 1)
	go func() { secondDone <- secondPool.EnsureSharedDBReady(context.Background(), dbID) }()
	var secondErr error
	select {
	case secondErr = <-secondDone:
	case <-time.After(3 * time.Second):
		releaseFirst()
		<-firstDone
		t.Fatal("second shared schema ensure waited instead of returning a non-blocking lock conflict")
	}
	if !errors.Is(secondErr, ErrSharedDBSchemaEnsureBusy) {
		t.Fatalf("second shared schema ensure error = %v, want non-blocking advisory lock conflict", secondErr)
	}
	if got := ensureCalls.Load(); got != 1 {
		t.Fatalf("concurrent schema ensure calls = %d, want 1", got)
	}
	waitCtx, cancelWait := context.WithTimeout(context.Background(), 20*time.Millisecond)
	_, waitErr := secondPool.sharedDBHandle(waitCtx, dbID)
	cancelWait()
	if !errors.Is(waitErr, context.DeadlineExceeded) {
		t.Fatalf("foreground shared DB open deadline error = %v, want context deadline", waitErr)
	}
	foregroundDone := make(chan error, 1)
	go func() {
		_, err := secondPool.sharedDBHandle(context.Background(), dbID)
		foregroundDone <- err
	}()
	select {
	case err := <-foregroundDone:
		t.Fatalf("foreground shared DB open returned before active schema ensure completed: %v", err)
	case <-time.After(25 * time.Millisecond):
	}
	releaseFirst()
	if err := <-firstDone; err != nil {
		t.Fatalf("first shared schema ensure: %v", err)
	}
	select {
	case err := <-foregroundDone:
		if err != nil {
			t.Fatalf("foreground shared DB open after schema ensure: %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("foreground shared DB open did not resume after schema ensure completed")
	}
	if got := ensureCalls.Load(); got != 1 {
		t.Fatalf("schema ensure calls after foreground current-version retry = %d, want 1", got)
	}
}

// Shared schema migration must not inherit caller cancellation once it holds
// the advisory lock: aborting the DDL on client disconnect leaves the shared
// schema version unpersisted, so every later open retries the migration. The
// migration now runs on a detached context and converges exactly once.
func TestSharedDBSchemaEnsureSurvivesCallerCancel(t *testing.T) {
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	dsnCfg, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, portText, err := net.SplitHostPort(dsnCfg.Addr)
	if err != nil {
		t.Fatal(err)
	}
	port, err := strconv.Atoi(portText)
	if err != nil {
		t.Fatal(err)
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	passwordCipher, err := enc.Encrypt(context.Background(), []byte(dsnCfg.Passwd))
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := metaStore.RegisterSharedDB(context.Background(), &meta.SharedDB{
		TiDBCloudOrganizationID: "org-schema-detach-cancel", Host: host, Port: port,
		User: dsnCfg.User, PasswordCipher: passwordCipher, Name: dsnCfg.DBName,
		MaxTenants: 100, Status: meta.SharedDBStatusActive,
	})
	if err != nil {
		t.Fatal(err)
	}
	pool := NewPool(PoolConfig{}, enc)
	pool.SetMetaStore(metaStore)
	t.Cleanup(pool.Close)

	ensureStarted := make(chan struct{})
	releaseEnsure := make(chan struct{})
	ensureCalls := 0
	var ensureCtxErr error
	previousEnsure := ensureSharedDBSchema
	ensureSharedDBSchema = func(ctx context.Context, db *sql.DB) error {
		ensureCalls++
		close(ensureStarted)
		<-releaseEnsure
		ensureCtxErr = ctx.Err()
		return nil
	}
	t.Cleanup(func() { ensureSharedDBSchema = previousEnsure })

	callCtx, cancelCall := context.WithCancel(context.Background())
	firstDone := make(chan error, 1)
	go func() { firstDone <- pool.EnsureSharedDBReady(callCtx, dbID) }()
	select {
	case <-ensureStarted:
	case <-time.After(10 * time.Second):
		t.Fatal("schema ensure did not start")
	}
	cancelCall()
	close(releaseEnsure)
	// The first call may still fail after the migration: opening the
	// data-plane handle intentionally stays on the (now canceled) caller
	// context. What matters is that the migration itself converged.
	<-firstDone
	if ensureCtxErr != nil {
		t.Fatalf("shared schema ensure context err = %v, want nil (detached from caller cancellation)", ensureCtxErr)
	}
	info, err := metaStore.GetSharedDB(context.Background(), dbID)
	if err != nil {
		t.Fatal(err)
	}
	if info.SchemaVersion != schema.CurrentSharedTiDBSchemaVersion {
		t.Fatalf("schema version = %d, want %d (persisted despite caller cancel)", info.SchemaVersion, schema.CurrentSharedTiDBSchemaVersion)
	}
	if err := pool.EnsureSharedDBReady(context.Background(), dbID); err != nil {
		t.Fatalf("post-convergence EnsureSharedDBReady: %v", err)
	}
	if ensureCalls != 1 {
		t.Fatalf("schema ensure calls = %d, want 1 (no retry after caller cancel)", ensureCalls)
	}
}

func TestSharedDBColdOpensDoNotBlockDifferentDBPools(t *testing.T) {
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	dsnCfg, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, portText, err := net.SplitHostPort(dsnCfg.Addr)
	if err != nil {
		t.Fatal(err)
	}
	port, err := strconv.Atoi(portText)
	if err != nil {
		t.Fatal(err)
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	passwordCipher, err := enc.Encrypt(context.Background(), []byte(dsnCfg.Passwd))
	if err != nil {
		t.Fatal(err)
	}
	register := func(orgID string) int64 {
		t.Helper()
		dbID, err := metaStore.RegisterSharedDB(context.Background(), &meta.SharedDB{
			TiDBCloudOrganizationID: orgID, Host: host, Port: port,
			User: dsnCfg.User, PasswordCipher: passwordCipher, Name: dsnCfg.DBName,
			MaxTenants: 100, Status: meta.SharedDBStatusActive,
		})
		if err != nil {
			t.Fatal(err)
		}
		return dbID
	}
	dbID1 := register("org-concurrent-open-1")
	dbID2 := register("org-concurrent-open-2")
	pool := NewPool(PoolConfig{}, enc)
	pool.SetMetaStore(metaStore)
	t.Cleanup(pool.Close)

	var ensureCalls atomic.Int32
	firstEnsureStarted := make(chan struct{})
	releaseFirstEnsure := make(chan struct{})
	previousEnsure := ensureSharedDBSchema
	ensureSharedDBSchema = func(ctx context.Context, db *sql.DB) error {
		if ensureCalls.Add(1) == 1 {
			close(firstEnsureStarted)
			<-releaseFirstEnsure
		}
		return db.PingContext(ctx)
	}
	t.Cleanup(func() { ensureSharedDBSchema = previousEnsure })

	firstDone := make(chan error, 1)
	go func() { firstDone <- pool.EnsureSharedDBReady(context.Background(), dbID1) }()
	<-firstEnsureStarted
	secondDone := make(chan error, 1)
	go func() { secondDone <- pool.EnsureSharedDBReady(context.Background(), dbID2) }()

	var secondErr error
	secondBlocked := false
	select {
	case secondErr = <-secondDone:
	case <-time.After(time.Second):
		secondBlocked = true
	}
	close(releaseFirstEnsure)
	if err := <-firstDone; err != nil {
		t.Fatalf("first shared DB open: %v", err)
	}
	if secondBlocked {
		secondErr = <-secondDone
	}
	if secondErr != nil {
		t.Fatalf("second shared DB open: %v", secondErr)
	}
	if secondBlocked {
		t.Fatal("cold open for one shared DB blocked an unrelated shared DB")
	}
}

func assertCachedBackendGauge(t *testing.T, want int) {
	t.Helper()
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	wantMetric := `drive9_service_gauge{component="tenant_pool",name="cached_backends",tenant_id="",tidbcloud_org_id="guest"} ` + strconv.Itoa(want) + `.000000`
	if !strings.Contains(rec.Body.String(), wantMetric) {
		t.Fatalf("missing cached backend gauge %q:\n%s", wantMetric, rec.Body.String())
	}
}

func TestNewPoolUsesDocumentedDefaultIdleReapInterval(t *testing.T) {
	pool := NewPool(PoolConfig{IdleTimeout: time.Minute}, nil)
	if pool.reapInterval != 2*time.Minute {
		t.Fatalf("reap interval = %s, want 2m", pool.reapInterval)
	}
}

func TestPoolAcquireTimingFieldsClassifyColdOpen(t *testing.T) {
	tests := []struct {
		name              string
		coldOpen          bool
		createBackendMs   float64
		wantCreateBackend bool
	}{
		{name: "hot cache hit", coldOpen: false},
		{name: "cold open inserted", coldOpen: true, createBackendMs: 12.5, wantCreateBackend: true},
		{name: "cold open discarded by cache race", coldOpen: true, createBackendMs: 15.25, wantCreateBackend: true},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			core, recorded := observer.New(zap.InfoLevel)
			zap.New(core).Info("timing", poolAcquireTimingFields("tenant-a", tt.coldOpen, tt.createBackendMs, 20*time.Millisecond)...)

			entries := recorded.FilterMessage("timing").All()
			if len(entries) != 1 {
				t.Fatalf("entries = %d, want 1", len(entries))
			}
			fields := entries[0].ContextMap()
			if fields["tenant_id"] != "tenant-a" {
				t.Errorf("tenant_id = %v, want tenant-a", fields["tenant_id"])
			}
			if _, ok := fields["cache_hit"]; ok {
				t.Errorf("cache_hit field should be omitted: %#v", fields)
			}
			if fields["cold_open"] != tt.coldOpen {
				t.Errorf("cold_open = %v, want %v", fields["cold_open"], tt.coldOpen)
			}
			if _, ok := fields["create_backend_ms"]; ok != tt.wantCreateBackend {
				t.Errorf("create_backend_ms present = %v, want %v", ok, tt.wantCreateBackend)
			}
		})
	}
}

func TestTenantMetricTiDBCloudOrgIDUsesPreResolvedTenantOrg(t *testing.T) {
	pool := NewPool(PoolConfig{}, nil)
	got := pool.tenantMetricTiDBCloudOrgID(context.Background(), &meta.Tenant{
		ID:             "tenant-pre-resolved-org",
		Status:         meta.TenantActive,
		Provider:       ProviderTiDBCloudNative,
		TiDBCloudOrgID: "org-pre-resolved",
	})
	if got != "org-pre-resolved" {
		t.Fatalf("org = %q, want org-pre-resolved", got)
	}
}

func TestTenantMetricTiDBCloudOrgIDUsesSharedDBPool(t *testing.T) {
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	ctx := context.Background()
	now := time.Now().UTC()
	tenantID := "tenant-shared-metric-org"
	if err := metaStore.InsertTenant(ctx, &meta.Tenant{
		ID: tenantID, Status: meta.TenantActive, Kind: meta.TenantKindLive,
		Provider: ProviderTiDBCloudNativeShared, DBPasswordCipher: []byte{}, DBTLS: true,
		SchemaVersion: 1, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	fsID, err := metaStore.EnsureFsID(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := metaStore.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-metric", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_db",
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := metaStore.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatal(err)
	}
	pool := NewPool(PoolConfig{}, nil)
	pool.SetMetaStore(metaStore)
	t.Cleanup(pool.Close)

	got := pool.tenantMetricTiDBCloudOrgID(ctx, &meta.Tenant{
		ID: tenantID, Status: meta.TenantActive, Provider: ProviderTiDBCloudNativeShared,
	})
	if got != "org-shared-metric" {
		t.Fatalf("org = %q, want org-shared-metric", got)
	}
}

func TestPoolAcquireInvalidateDefersCloseUntilRelease(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-a")
	ctx := context.Background()

	b1, release1, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store1 := b1.Store()
	assertStoreOpen(t, store1)

	pool.Invalidate(tenant.ID)
	assertStoreOpen(t, store1)

	b2, release2, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	if b1 == b2 {
		t.Fatal("expected acquire after invalidate to create a new backend")
	}
	assertStoreOpen(t, b2.Store())

	release1()
	assertStoreClosed(t, store1)
	assertStoreOpen(t, b2.Store())
	release2()
}

func TestCleanStorageNamespaceLocalPrefixRejectsUnsafeSegments(t *testing.T) {
	tests := []struct {
		name    string
		prefix  string
		want    string
		wantErr bool
	}{
		{name: "normal", prefix: "tenant/abc", want: "tenant/abc"},
		{name: "trim", prefix: "tenant/abc/", want: "tenant/abc"},
		{name: "empty", prefix: "", wantErr: true},
		{name: "absolute", prefix: "/tmp/tenant", wantErr: true},
		{name: "parent", prefix: "tenant/../other", wantErr: true},
		{name: "dot", prefix: "tenant/./other", wantErr: true},
		{name: "empty segment", prefix: "tenant//other", wantErr: true},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			got, err := cleanStorageNamespaceLocalPrefix(tc.prefix)
			if tc.wantErr {
				if err == nil {
					t.Fatalf("cleanStorageNamespaceLocalPrefix(%q) error = nil, want error", tc.prefix)
				}
				return
			}
			if err != nil {
				t.Fatalf("cleanStorageNamespaceLocalPrefix(%q): %v", tc.prefix, err)
			}
			if got != tc.want {
				t.Fatalf("cleanStorageNamespaceLocalPrefix(%q) = %q, want %q", tc.prefix, got, tc.want)
			}
		})
	}
}

func TestPoolAcquireEvictionRetiresPinnedEntry(t *testing.T) {
	pool, tenantA := newTestPoolAndTenant(t, 1, "tenant-a")
	ctx := context.Background()
	bA, releaseA, err := pool.Acquire(ctx, tenantA)
	if err != nil {
		t.Fatal(err)
	}
	storeA := bA.Store()
	assertStoreOpen(t, storeA)

	tenantB := cloneTenantForID(t, pool, tenantA, "tenant-b")
	bB, releaseB, err := pool.Acquire(ctx, tenantB)
	if err != nil {
		t.Fatal(err)
	}
	releaseB()
	if bB == nil {
		t.Fatal("expected second backend")
	}
	assertStoreOpen(t, bB.Store())
	assertStoreOpen(t, storeA)

	releaseA()
	assertStoreClosed(t, storeA)
	assertStoreOpen(t, bB.Store())
}

func TestPoolAcquireReloadsS3EncryptionPolicyForNextWrite(t *testing.T) {
	globalKeyID := "arn:aws:kms:ap-southeast-1:123456789012:key/pool-test"
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants: 2,
		S3Dir:      t.TempDir(),
		PublicURL:  "http://localhost:9091",
		S3EncryptionPolicy: meta.S3EncryptionPolicy{
			Mode:             meta.S3EncryptionModeSSEKMS,
			KMSKeyID:         globalKeyID,
			BucketKeyEnabled: true,
		},
		// Force these small fixtures through S3 so this test exercises the
		// per-tenant encryption policy instead of DB-inline storage.
		BackendOptions: backend.Options{InlineThreshold: 1},
	}, "tenant-a")
	disabledBucketKey := false
	tenant.S3EncryptionMode = meta.S3EncryptionModeNone
	tenant.S3BucketKeyEnabled = &disabledBucketKey
	ctx := context.Background()

	b1, release1, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	defer release1()
	if _, _, err := b1.WriteCtxIfRevisionWithTagsResult(ctx, "/before.bin", []byte("before"), 0, filesystem.WriteFlagCreate|filesystem.WriteFlagTruncate, -1, nil, ""); err != nil {
		t.Fatalf("write before policy change: %v", err)
	}
	before, err := b1.Store().Stat(ctx, "/before.bin")
	if err != nil {
		t.Fatalf("stat before: %v", err)
	}
	if before.File.StorageEncryptionMode != datastore.StorageEncryptionNone {
		t.Fatalf("before encryption mode=%q, want %q", before.File.StorageEncryptionMode, datastore.StorageEncryptionNone)
	}

	changed := *tenant
	changed.S3EncryptionMode = meta.S3EncryptionModeSSEKMS
	b2, release2, err := pool.Acquire(ctx, &changed)
	if err != nil {
		t.Fatal(err)
	}
	defer release2()
	if b1 == b2 {
		t.Fatal("expected tenant policy change to recreate cached backend")
	}
	if _, _, err := b2.WriteCtxIfRevisionWithTagsResult(ctx, "/after.bin", []byte("after"), 0, filesystem.WriteFlagCreate|filesystem.WriteFlagTruncate, -1, nil, ""); err != nil {
		t.Fatalf("write after policy change: %v", err)
	}
	after, err := b2.Store().Stat(ctx, "/after.bin")
	if err != nil {
		t.Fatalf("stat after: %v", err)
	}
	if after.File.StorageEncryptionMode != datastore.StorageEncryptionSSEKMS {
		t.Fatalf("after encryption mode=%q, want %q", after.File.StorageEncryptionMode, datastore.StorageEncryptionSSEKMS)
	}
	if after.File.StorageEncryptionKeyID != globalKeyID {
		t.Fatalf("after encryption key=%q, want %q", after.File.StorageEncryptionKeyID, globalKeyID)
	}
}

func newTestPoolAndTenant(t *testing.T, maxTenants int, tenantID string) (*Pool, *meta.Tenant) {
	t.Helper()
	return newTestPoolAndTenantWithConfig(t, PoolConfig{MaxTenants: maxTenants}, tenantID)
}

func newTestPoolAndTenantWithConfig(t *testing.T, cfg PoolConfig, tenantID string) (*Pool, *meta.Tenant) {
	t.Helper()
	initTenantPoolSchema(t, testDSN)
	resetStore, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, resetStore.DB())
	if err := resetStore.Close(); err != nil {
		t.Fatal(err)
	}

	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	pool := NewPool(cfg, enc)
	t.Cleanup(func() { pool.Close() })
	return pool, cloneTenantForID(t, pool, nil, tenantID)
}

func initTenantPoolSchema(t *testing.T, dsn string) {
	t.Helper()
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = db.Close() }()

	stmts := []string{
		`CREATE TABLE IF NOT EXISTS file_nodes (node_id VARCHAR(64) PRIMARY KEY, path TEXT NOT NULL, path_hash VARCHAR(64) NOT NULL DEFAULT '', parent_path TEXT NOT NULL, parent_path_hash VARCHAR(64) NOT NULL DEFAULT '', name VARCHAR(255) NOT NULL, is_directory BOOLEAN NOT NULL DEFAULT FALSE, file_id VARCHAR(64), inode_id VARCHAR(64), created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3))`,
		`CREATE UNIQUE INDEX idx_path ON file_nodes(path_hash)`,
		`CREATE INDEX idx_parent ON file_nodes(parent_path_hash, name)`,
		`CREATE INDEX idx_file_id ON file_nodes(file_id)`,
		`CREATE INDEX idx_inode_id ON file_nodes(inode_id)`,
		`CREATE TABLE IF NOT EXISTS files (file_id VARCHAR(64) PRIMARY KEY, storage_type VARCHAR(32) NOT NULL, content_layout VARCHAR(32) NOT NULL DEFAULT 'single', storage_ref TEXT NOT NULL, storage_ref_hash VARCHAR(64) NOT NULL DEFAULT '', content_blob LONGBLOB, content_type VARCHAR(255), size_bytes BIGINT NOT NULL DEFAULT 0, checksum_sha256 VARCHAR(128), revision BIGINT NOT NULL DEFAULT 1, status VARCHAR(32) NOT NULL DEFAULT 'PENDING', source_id VARCHAR(255), content_text LONGTEXT, description LONGTEXT, embedding LONGTEXT, embedding_revision BIGINT, description_embedding LONGTEXT, description_embedding_revision BIGINT, storage_encryption_mode VARCHAR(16) NOT NULL DEFAULT 'legacy', storage_encryption_key_id VARCHAR(256) NOT NULL DEFAULT '', created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), confirmed_at DATETIME(3), expires_at DATETIME(3))`,
		`CREATE INDEX idx_status ON files(status, created_at)`,
		`CREATE INDEX idx_files_storage_ref_hash ON files(storage_ref_hash)`,
		`CREATE TABLE IF NOT EXISTS inodes (inode_id VARCHAR(64) PRIMARY KEY, size_bytes BIGINT NOT NULL DEFAULT 0, revision BIGINT NOT NULL DEFAULT 1, mode INT NOT NULL DEFAULT 420, status VARCHAR(32) NOT NULL DEFAULT 'PENDING', created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), mtime DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), confirmed_at DATETIME(3), expires_at DATETIME(3))`,
		`CREATE INDEX idx_inodes_status ON inodes(status, created_at)`,
		`CREATE TABLE IF NOT EXISTS contents (inode_id VARCHAR(64) PRIMARY KEY, storage_type VARCHAR(32) NOT NULL, content_layout VARCHAR(32) NOT NULL DEFAULT 'single', storage_ref TEXT NOT NULL, storage_ref_hash VARCHAR(64) NOT NULL DEFAULT '', storage_encryption_mode VARCHAR(16) NOT NULL DEFAULT 'legacy', storage_encryption_key_id VARCHAR(256) NOT NULL DEFAULT '', content_blob LONGBLOB, content_type VARCHAR(255), checksum_sha256 VARCHAR(128), source_id VARCHAR(255))`,
		`CREATE INDEX idx_contents_storage_ref_hash ON contents(storage_ref_hash)`,
		`CREATE TABLE IF NOT EXISTS semantic (inode_id VARCHAR(64) PRIMARY KEY, content_text LONGTEXT, description LONGTEXT, embedding LONGTEXT, embedding_revision BIGINT, description_embedding LONGTEXT, description_embedding_revision BIGINT)`,
		`CREATE TABLE IF NOT EXISTS file_tags (file_id VARCHAR(64) NOT NULL, inode_id VARCHAR(64), tag_key VARCHAR(255) NOT NULL, tag_value VARCHAR(255) NOT NULL DEFAULT '', source VARCHAR(32) NOT NULL DEFAULT 'user', PRIMARY KEY (file_id, tag_key))`,
		`CREATE INDEX idx_kv ON file_tags(tag_key, tag_value)`,
		`CREATE TABLE IF NOT EXISTS uploads (upload_id VARCHAR(64) PRIMARY KEY, file_id VARCHAR(64) NOT NULL, inode_id VARCHAR(64), target_kind VARCHAR(16) NOT NULL DEFAULT 'file', layer_id VARCHAR(64) NOT NULL DEFAULT '', target_path TEXT NOT NULL, target_path_hash VARCHAR(64) NOT NULL DEFAULT '', base_inode_id VARCHAR(64) NOT NULL DEFAULT '', mode INT UNSIGNED NOT NULL DEFAULT 420, s3_upload_id VARCHAR(255) NOT NULL, s3_key VARCHAR(2048) NOT NULL, total_size BIGINT NOT NULL, part_size BIGINT NOT NULL, parts_total INT NOT NULL, expected_revision BIGINT, status VARCHAR(32) NOT NULL DEFAULT 'UPLOADING', fingerprint_sha256 VARCHAR(128), idempotency_key VARCHAR(255), description LONGTEXT, storage_encryption_mode VARCHAR(16) NOT NULL DEFAULT 'none', storage_encryption_key_id VARCHAR(256) 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), expires_at DATETIME(3) NOT NULL, active_target_path_hash VARCHAR(64) AS (CASE WHEN status = 'UPLOADING' THEN target_path_hash ELSE NULL END) VIRTUAL)`,
		`CREATE INDEX idx_upload_path ON uploads(target_path_hash, status)`,
		`CREATE UNIQUE INDEX idx_idempotency ON uploads(idempotency_key)`,
		`CREATE UNIQUE INDEX idx_uploads_active ON uploads(active_target_path_hash)`,
		`CREATE TABLE IF NOT EXISTS semantic_tasks (task_id VARCHAR(64) PRIMARY KEY, task_type VARCHAR(32) NOT NULL, resource_id VARCHAR(64) NOT NULL, resource_version BIGINT NOT NULL, status VARCHAR(20) NOT NULL, attempt_count INT NOT NULL DEFAULT 0, max_attempts INT NOT NULL DEFAULT 5, receipt VARCHAR(128), leased_at DATETIME(3), lease_until DATETIME(3), available_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), payload_json JSON, last_error TEXT, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), completed_at DATETIME(3))`,
		`CREATE UNIQUE INDEX uk_task_resource_version ON semantic_tasks(task_type, resource_id, resource_version)`,
		`CREATE INDEX idx_task_claim ON semantic_tasks(status, available_at, lease_until, created_at)`,
		`CREATE INDEX idx_task_claim_type ON semantic_tasks(status, task_type, available_at, created_at, task_id)`,
		`CREATE TABLE IF NOT EXISTS file_gc_tasks (task_id VARCHAR(64) PRIMARY KEY, file_id VARCHAR(64) NOT NULL, inode_id VARCHAR(64), storage_type VARCHAR(32) NOT NULL, content_layout VARCHAR(32) NOT NULL DEFAULT 'single', storage_ref TEXT NOT NULL, size_bytes BIGINT NOT NULL DEFAULT 0, content_type VARCHAR(255), status VARCHAR(20) NOT NULL, attempt_count INT NOT NULL DEFAULT 0, max_attempts INT NOT NULL DEFAULT 0, receipt VARCHAR(128), leased_at DATETIME(3), lease_until DATETIME(3), available_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), last_error TEXT, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), completed_at DATETIME(3))`,
		`CREATE UNIQUE INDEX uk_file_gc_file_id ON file_gc_tasks(file_id)`,
		`CREATE UNIQUE INDEX uk_file_gc_inode_id ON file_gc_tasks(inode_id)`,
		`CREATE INDEX idx_file_gc_claim ON file_gc_tasks(status, available_at, lease_until, created_at)`,
	}
	for _, stmt := range stmts {
		if _, err := db.Exec(stmt); err != nil {
			msg := err.Error()
			if strings.Contains(msg, "Duplicate key name") || strings.Contains(msg, "already exists") {
				continue
			}
			t.Fatal(err)
		}
	}
}

func cloneTenantForID(t *testing.T, pool *Pool, src *meta.Tenant, tenantID string) *meta.Tenant {
	t.Helper()
	parsed, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, port := "127.0.0.1", 3306
	if parsed.Addr != "" {
		h, p, _ := strings.Cut(parsed.Addr, ":")
		if h != "" {
			host = h
		}
		if p != "" {
			_, _ = fmt.Sscanf(p, "%d", &port)
		}
	}
	passwd := parsed.Passwd
	if src != nil {
		plain, err := pool.Decrypt(context.Background(), src.DBPasswordCipher)
		if err != nil {
			t.Fatal(err)
		}
		passwd = string(plain)
	}
	passCipher, err := pool.Encrypt(context.Background(), []byte(passwd))
	if err != nil {
		t.Fatal(err)
	}
	appSchemaVersion, err := schema.TiDBTenantSchemaVersionForEmbeddingModeProfile(schema.TiDBEmbeddingModeApp, schema.TiDBAutoEmbeddingProfile{})
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	// The test fixture is a TiDB app-schema database, not PostgreSQL/DB9.
	provider := ProviderLocal
	if src != nil {
		provider = src.Provider
	}
	return &meta.Tenant{
		ID:               tenantID,
		Status:           meta.TenantActive,
		DBHost:           host,
		DBPort:           port,
		DBUser:           parsed.User,
		DBPasswordCipher: passCipher,
		DBName:           parsed.DBName,
		DBTLS:            false,
		Provider:         provider,
		SchemaVersion:    appSchemaVersion,
		CreatedAt:        now,
		UpdatedAt:        now,
	}
}

func assertStoreOpen(t *testing.T, store *datastore.Store) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := store.DB().PingContext(ctx); err != nil {
		t.Fatalf("expected store to remain open: %v", err)
	}
}

func assertStoreClosed(t *testing.T, store *datastore.Store) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := store.DB().PingContext(ctx); err == nil {
		t.Fatal("expected store to be closed")
	}
}

func TestPoolAutoSemanticTaskTypes(t *testing.T) {
	var p *Pool
	if p.AutoSemanticTaskTypes() != nil {
		t.Fatal("nil pool should return nil task types")
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	disabled := NewPool(PoolConfig{BackendOptions: backend.Options{}}, enc)
	defer disabled.Close()
	if disabled.AutoSemanticTaskTypes() != nil {
		t.Fatal("async image extract disabled should yield nil")
	}
	enabled := NewPool(PoolConfig{BackendOptions: backend.Options{
		AsyncImageExtract: backend.AsyncImageExtractOptions{Enabled: true},
	}}, enc)
	defer enabled.Close()
	got := enabled.AutoSemanticTaskTypes()
	if len(got) != 1 || got[0] != semantic.TaskTypeImgExtractText {
		t.Fatalf("got %#v, want [img_extract_text]", got)
	}

	audioOnly := NewPool(PoolConfig{BackendOptions: backend.Options{
		AsyncAudioExtract: backend.AsyncAudioExtractOptions{Enabled: true, Extractor: &poolDummyAudioExtractor{}},
	}}, enc)
	defer audioOnly.Close()
	gotAudio := audioOnly.AutoSemanticTaskTypes()
	if len(gotAudio) != 1 || gotAudio[0] != semantic.TaskTypeAudioExtractText {
		t.Fatalf("got %#v, want [audio_extract_text]", gotAudio)
	}

	both := NewPool(PoolConfig{BackendOptions: backend.Options{
		AsyncImageExtract: backend.AsyncImageExtractOptions{Enabled: true},
		AsyncAudioExtract: backend.AsyncAudioExtractOptions{Enabled: true, Extractor: &poolDummyAudioExtractor{}},
	}}, enc)
	defer both.Close()
	gotBoth := both.AutoSemanticTaskTypes()
	if len(gotBoth) != 2 || gotBoth[0] != semantic.TaskTypeImgExtractText || gotBoth[1] != semantic.TaskTypeAudioExtractText {
		t.Fatalf("got %#v, want [img_extract_text audio_extract_text]", gotBoth)
	}

	capabilityOnly := NewPool(PoolConfig{BackendOptions: backend.Options{
		AsyncAudioExtract: backend.AsyncAudioExtractOptions{Capability: true},
		AsyncVideoExtract: backend.AsyncVideoExtractOptions{Capability: true},
	}}, enc)
	defer capabilityOnly.Close()
	gotCapability := capabilityOnly.AutoSemanticTaskTypes()
	if len(gotCapability) != 2 || gotCapability[0] != semantic.TaskTypeAudioExtractText || gotCapability[1] != semantic.TaskTypeVideoExtractVisual {
		t.Fatalf("got %#v, want [audio_extract_text video_extract_visual]", gotCapability)
	}
}

func TestPoolCreateBackendSkipsSchemaEnsureWhenVersionMatches(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants: 2,
		BackendOptions: backend.Options{
			AppSemanticTasksEnabled:    true,
			AppEmbeddingGeneration:     5,
			AppEmbeddingConfigRevision: 8,
			QueryEmbedding:             backend.QueryEmbeddingOptions{Client: embedding.NopClient{}},
		},
	}, "tenant-skip-validate")
	tenant.Provider = ProviderTiDBZero
	tenant.SchemaVersion = schema.CurrentTiDBTenantSchemaVersion
	resolveCalls := 0
	pool.SetEmbeddingRuntimeResolver(func(context.Context, string) (backend.AppEmbeddingRuntime, bool, error) {
		resolveCalls++
		return backend.AppEmbeddingRuntime{}, false, nil
	})

	origEnsure := ensureTiDBSchemaForAutoEmbeddingProfile
	origApply := applyTiDBAutoEmbeddingProviderConfig
	ensureCalls := 0
	ensureTiDBSchemaForAutoEmbeddingProfile = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProfile) error {
		ensureCalls++
		return nil
	}
	applyTiDBAutoEmbeddingProviderConfig = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProviderConfig) error {
		return nil
	}
	t.Cleanup(func() {
		ensureTiDBSchemaForAutoEmbeddingProfile = origEnsure
		applyTiDBAutoEmbeddingProviderConfig = origApply
	})

	openedBackend, store, _, _, err := pool.createBackend(context.Background(), tenant)
	if err != nil {
		t.Fatalf("createBackend(): %v", err)
	}
	if openedBackend.SupportsAppEmbeddingTasks() {
		t.Fatal("native auto backend inherited the process-default app embedding runtime")
	}
	var backfills int
	if err := store.DB().QueryRow(`SELECT COUNT(*) FROM semantic_tasks WHERE task_type = ?`, semantic.TaskTypeEmbedBackfill).Scan(&backfills); err != nil {
		t.Fatal(err)
	}
	if backfills != 0 {
		t.Fatalf("native auto embedding backfill tasks=%d, want 0", backfills)
	}
	openedBackend.Close()
	if err := store.Close(); err != nil {
		t.Fatalf("close store: %v", err)
	}

	if ensureCalls != 0 {
		t.Fatalf("ensureTiDBSchemaForAutoEmbeddingProfile called %d times, want 0", ensureCalls)
	}
	if resolveCalls != 0 {
		t.Fatalf("app embedding resolver called %d times for native auto tenant, want 0", resolveCalls)
	}
}

func TestPoolCreateBackendEnsuresSchemaForTiDBCloudNative(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-native-ensure")
	tenant.Provider = ProviderTiDBCloudNative
	tenant.SchemaVersion = 0

	origEnsure := ensureTiDBSchemaForAutoEmbeddingProfile
	origApply := applyTiDBAutoEmbeddingProviderConfig
	ensureCalls := 0
	ensureTiDBSchemaForAutoEmbeddingProfile = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProfile) error {
		ensureCalls++
		return nil
	}
	applyTiDBAutoEmbeddingProviderConfig = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProviderConfig) error {
		return nil
	}
	t.Cleanup(func() {
		ensureTiDBSchemaForAutoEmbeddingProfile = origEnsure
		applyTiDBAutoEmbeddingProviderConfig = origApply
	})

	backend, store, _, _, err := pool.createBackend(context.Background(), tenant)
	if err != nil {
		t.Fatalf("createBackend(): %v", err)
	}
	backend.Close()
	if err := store.Close(); err != nil {
		t.Fatalf("close store: %v", err)
	}
	if ensureCalls != 1 {
		t.Fatalf("ensureTiDBSchemaForAutoEmbeddingProfile called %d times, want 1", ensureCalls)
	}
}

// Schema ensure must not inherit request cancellation: a client disconnect
// mid-ensure used to abort the repair, the new schema version was never
// persisted, and every subsequent cold open retried and failed the same way
// ("backfill storage_ref_hash: context canceled"). The ensure and its version
// bookkeeping now run on a detached context so the tenant converges even when
// the triggering request goes away.
func TestPoolCreateBackendSchemaEnsureSurvivesRequestCancel(t *testing.T) {
	metaStore, err := meta.OpenContext(context.Background(), testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-ensure-detach-cancel")
	tenant.Provider = ProviderTiDBZero
	tenant.SchemaVersion = 0
	if err := metaStore.InsertTenant(context.Background(), tenant); err != nil {
		t.Fatal(err)
	}
	pool.SetMetaStore(metaStore)

	origEnsure := ensureTiDBSchemaForAutoEmbeddingProfile
	origApply := applyTiDBAutoEmbeddingProviderConfig
	ensureStarted := make(chan struct{})
	releaseEnsure := make(chan struct{})
	ensureCalls := 0
	var ensureCtxErr error
	ensureTiDBSchemaForAutoEmbeddingProfile = func(ctx context.Context, _ *sql.DB, _ schema.TiDBAutoEmbeddingProfile) error {
		ensureCalls++
		close(ensureStarted)
		<-releaseEnsure
		ensureCtxErr = ctx.Err()
		return nil
	}
	applyTiDBAutoEmbeddingProviderConfig = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProviderConfig) error {
		return nil
	}
	t.Cleanup(func() {
		ensureTiDBSchemaForAutoEmbeddingProfile = origEnsure
		applyTiDBAutoEmbeddingProviderConfig = origApply
	})

	reqCtx, cancelReq := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		b, store, _, _, err := pool.createBackend(reqCtx, tenant)
		if err == nil {
			b.Close()
			_ = store.Close()
		}
		// A createBackend error after the ensure is acceptable here: the
		// remaining cold-open steps intentionally stay on the (now
		// canceled) request context. Only the schema work is detached.
	}()
	select {
	case <-ensureStarted:
	case <-time.After(10 * time.Second):
		t.Fatal("schema ensure did not start")
	}
	cancelReq()
	close(releaseEnsure)
	<-done
	if ensureCtxErr != nil {
		t.Fatalf("schema ensure context err = %v, want nil (detached from request cancellation)", ensureCtxErr)
	}
	// The version bookkeeping is the other half of the production loop: it
	// must persist despite the canceled request, or every cold open retries.
	current, err := metaStore.GetTenant(context.Background(), tenant.ID)
	if err != nil {
		t.Fatal(err)
	}
	if current.SchemaVersion == 0 {
		t.Fatal("tenant schema version not persisted after canceled request")
	}
	// The embedding-mode bookkeeping is part of the same contract: it must
	// also persist on a detached context.
	profile, err := metaStore.GetTenantAutoEmbeddingProfile(context.Background(), tenant.ID)
	if err != nil {
		t.Fatal(err)
	}
	if profile.EmbeddingMode == "" {
		t.Fatal("tenant embedding mode not persisted after canceled request")
	}
	// A fresh cold open with the same stale in-memory tenant row must not
	// re-run the ensure: the in-lock version double-check sees the durable
	// version and skips the DDL.
	b, store, _, _, err := pool.createBackend(context.Background(), tenant)
	if err != nil {
		t.Fatalf("second createBackend(): %v", err)
	}
	b.Close()
	_ = store.Close()
	if ensureCalls != 1 {
		t.Fatalf("schema ensure calls = %d, want 1 (no re-run after convergence)", ensureCalls)
	}
}

// Concurrent cold opens of the same stale-version tenant must not each start
// their own detached schema ensure: the tenant-DB advisory lock serializes
// them (same pod or across pods), and the waiter's in-lock version
// double-check skips the DDL once the first migration has converged.
func TestPoolCreateBackendSchemaEnsureSerializedAcrossPools(t *testing.T) {
	metaStore, err := meta.OpenContext(context.Background(), testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())

	pool1, tenant := newTestPoolAndTenant(t, 2, "tenant-ensure-serialized")
	tenant.Provider = ProviderTiDBZero
	tenant.SchemaVersion = 0
	if err := metaStore.InsertTenant(context.Background(), tenant); err != nil {
		t.Fatal(err)
	}
	pool1.SetMetaStore(metaStore)
	// The second pool must share pool1's encryptor: the tenant row's
	// DBPasswordCipher is only decryptable by the key that produced it.
	pool2 := NewPool(PoolConfig{MaxTenants: 2}, pool1.enc)
	pool2.SetMetaStore(metaStore)
	t.Cleanup(pool2.Close)

	origEnsure := ensureTiDBSchemaForAutoEmbeddingProfile
	origApply := applyTiDBAutoEmbeddingProviderConfig
	ensureStarted := make(chan struct{})
	releaseEnsure := make(chan struct{})
	var releaseEnsureOnce sync.Once
	release := func() { releaseEnsureOnce.Do(func() { close(releaseEnsure) }) }
	var ensureCalls atomic.Int32
	ensureTiDBSchemaForAutoEmbeddingProfile = func(ctx context.Context, db *sql.DB, _ schema.TiDBAutoEmbeddingProfile) error {
		if ensureCalls.Add(1) == 1 {
			close(ensureStarted)
			<-releaseEnsure
		}
		return db.PingContext(ctx)
	}
	applyTiDBAutoEmbeddingProviderConfig = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProviderConfig) error {
		return nil
	}
	t.Cleanup(func() {
		ensureTiDBSchemaForAutoEmbeddingProfile = origEnsure
		applyTiDBAutoEmbeddingProviderConfig = origApply
	})

	firstDone := make(chan error, 1)
	firstExited := make(chan struct{})
	// Release-and-drain first so a probe failure below cannot leave the
	// first goroutine blocked inside the ensure stub holding the tenant lock.
	t.Cleanup(func() {
		release()
		select {
		case <-firstExited:
		case <-time.After(time.Second):
			t.Error("first schema ensure did not exit during cleanup")
		}
	})
	go func() {
		defer close(firstExited)
		b, store, _, _, err := pool1.createBackend(context.Background(), tenant)
		if err == nil {
			b.Close()
			_ = store.Close()
		}
		firstDone <- err
	}()
	select {
	case <-ensureStarted:
	case <-time.After(10 * time.Second):
		t.Fatal("schema ensure did not start")
	}

	secondDone := make(chan error, 1)
	secondExited := make(chan struct{})
	go func() {
		defer close(secondExited)
		b, store, _, _, err := pool2.createBackend(context.Background(), tenant)
		if err == nil {
			b.Close()
			_ = store.Close()
		}
		secondDone <- err
	}()
	// The second cold open must wait on the tenant-DB advisory lock instead
	// of starting a second detached ensure.
	select {
	case <-secondExited:
		t.Fatalf("second cold open completed while the first schema ensure still held the tenant lock: err=%v", <-secondDone)
	case <-time.After(100 * time.Millisecond):
	}
	release()
	if err := <-firstDone; err != nil {
		t.Fatalf("first createBackend(): %v", err)
	}
	if err := <-secondDone; err != nil {
		t.Fatalf("second createBackend(): %v", err)
	}
	if got := ensureCalls.Load(); got != 1 {
		t.Fatalf("schema ensure calls = %d, want 1 (waiter skipped DDL after version double-check)", got)
	}
	current, err := metaStore.GetTenant(context.Background(), tenant.ID)
	if err != nil {
		t.Fatal(err)
	}
	if current.SchemaVersion == 0 {
		t.Fatal("tenant schema version not persisted after serialized migration")
	}
}

// MySQL/TiDB named locks cap at 64 characters while tenant IDs are
// schema-valid up to 64 bytes; the lock name must stay bounded for every
// valid tenant ID.
func TestTenantSchemaLockNameStaysWithinMySQLLimit(t *testing.T) {
	longID := strings.Repeat("t", 64)
	got := tenantSchemaLockName(longID)
	if len(got) >= 64 {
		t.Fatalf("lock name length = %d, want below 64", len(got))
	}
	if !strings.HasPrefix(got, "drive9:tenant-schema:") {
		t.Fatalf("lock name = %q, want drive9:tenant-schema: prefix", got)
	}
	if again := tenantSchemaLockName(longID); again != got {
		t.Fatalf("lock name not deterministic: %q vs %q", got, again)
	}
	if other := tenantSchemaLockName(strings.Repeat("u", 64)); other == got {
		t.Fatalf("distinct tenant IDs produced the same lock name %q", got)
	}
}

// A schema-valid 64-byte tenant ID must not fail cold-open schema ensure at
// GET_LOCK.
func TestPoolCreateBackendSchemaEnsureWithMaxLengthTenantID(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, strings.Repeat("t", 64))
	tenant.Provider = ProviderTiDBZero
	tenant.SchemaVersion = 0

	origEnsure := ensureTiDBSchemaForAutoEmbeddingProfile
	origApply := applyTiDBAutoEmbeddingProviderConfig
	ensureCalls := 0
	ensureTiDBSchemaForAutoEmbeddingProfile = func(ctx context.Context, db *sql.DB, _ schema.TiDBAutoEmbeddingProfile) error {
		ensureCalls++
		return db.PingContext(ctx)
	}
	applyTiDBAutoEmbeddingProviderConfig = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProviderConfig) error {
		return nil
	}
	t.Cleanup(func() {
		ensureTiDBSchemaForAutoEmbeddingProfile = origEnsure
		applyTiDBAutoEmbeddingProviderConfig = origApply
	})

	b, store, _, _, err := pool.createBackend(context.Background(), tenant)
	if err != nil {
		t.Fatalf("createBackend() with 64-byte tenant ID: %v", err)
	}
	b.Close()
	if err := store.Close(); err != nil {
		t.Fatalf("close store: %v", err)
	}
	if ensureCalls != 1 {
		t.Fatalf("ensureTiDBSchemaForAutoEmbeddingProfile called %d times, want 1", ensureCalls)
	}
}

func TestPoolCreateBackendRepairsFTSOnlySchemaWhenDatabaseAutoEmbeddingDisabled(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:                   2,
		DisableDatabaseAutoEmbedding: true,
		BackendOptions: backend.Options{
			AsyncImageExtract: backend.AsyncImageExtractOptions{Capability: true},
			AsyncAudioExtract: backend.AsyncAudioExtractOptions{Capability: true},
			AsyncVideoExtract: backend.AsyncVideoExtractOptions{Capability: true},
		},
	}, "tenant-disabled-ensure")
	tenant.Provider = ProviderTiDBZero
	tenant.SchemaVersion = 0
	resolveCalls := 0
	pool.SetEmbeddingRuntimeResolver(func(_ context.Context, tenantID string) (backend.AppEmbeddingRuntime, bool, error) {
		resolveCalls++
		if tenantID != tenant.ID {
			t.Fatalf("embedding resolver tenant=%q, want %q", tenantID, tenant.ID)
		}
		return backend.AppEmbeddingRuntime{Client: embedding.NopClient{}, Generation: 5, ConfigRevision: 8, TenantManaged: true}, true, nil
	})

	origEnsure := ensureTiDBSchemaForAutoEmbeddingProfile
	origEnsureFTS := ensureTiDBSchemaForFTSOnlyProfile
	origApply := applyTiDBAutoEmbeddingProviderConfig
	autoEnsureCalls := 0
	ftsEnsureCalls := 0
	applyCalls := 0
	ensureTiDBSchemaForAutoEmbeddingProfile = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProfile) error {
		autoEnsureCalls++
		return nil
	}
	ensureTiDBSchemaForFTSOnlyProfile = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProfile) error {
		ftsEnsureCalls++
		return nil
	}
	applyTiDBAutoEmbeddingProviderConfig = func(context.Context, *sql.DB, schema.TiDBAutoEmbeddingProviderConfig) error {
		applyCalls++
		return nil
	}
	t.Cleanup(func() {
		ensureTiDBSchemaForAutoEmbeddingProfile = origEnsure
		ensureTiDBSchemaForFTSOnlyProfile = origEnsureFTS
		applyTiDBAutoEmbeddingProviderConfig = origApply
	})

	b, store, _, _, err := pool.createBackend(context.Background(), tenant)
	if err != nil {
		t.Fatalf("createBackend(): %v", err)
	}
	defer func() {
		b.Close()
		_ = store.Close()
	}()
	if b.UsesDatabaseAutoEmbedding() {
		t.Fatal("backend runtime auto embedding enabled with DisableDatabaseAutoEmbedding=true")
	}
	if autoEnsureCalls != 0 {
		t.Fatalf("ensureTiDBSchemaForAutoEmbeddingProfile called %d times, want 0", autoEnsureCalls)
	}
	if ftsEnsureCalls != 1 {
		t.Fatalf("ensureTiDBSchemaForFTSOnlyProfile called %d times, want 1", ftsEnsureCalls)
	}
	if applyCalls != 0 {
		t.Fatalf("applyTiDBAutoEmbeddingProviderConfig called %d times, want 0", applyCalls)
	}
	if resolveCalls != 1 {
		t.Fatalf("app embedding resolver calls=%d, want 1", resolveCalls)
	}
	if !b.SupportsAppEmbeddingTasks() {
		t.Fatal("native effective fts_only backend must support app embedding")
	}
	if runtime, ok := b.AppEmbeddingRuntime(); !ok || !runtime.TenantManaged {
		t.Fatalf("native effective fts_only runtime=%+v ok=%v, want tenant-managed", runtime, ok)
	}
	if !b.SupportsImageExtractTasks() || !b.SupportsAudioExtractTasks() || !b.SupportsVideoExtractTasks() {
		t.Fatalf("native effective fts_only extraction capabilities image=%v audio=%v video=%v, want all true",
			b.SupportsImageExtractTasks(), b.SupportsAudioExtractTasks(), b.SupportsVideoExtractTasks())
	}
	for path, data := range map[string][]byte{
		"/media/admission.png": {0x89, 0x50, 0x4e, 0x47},
		"/media/admission.mp3": {0xff, 0xf3, 0x80},
		"/media/admission.mp4": {0x00, 0x00, 0x00, 0x18},
	} {
		if _, err := b.Write(path, data, 0, filesystem.WriteFlagCreate); err != nil {
			t.Fatalf("write %s: %v", path, err)
		}
	}
	for _, taskType := range []semantic.TaskType{
		semantic.TaskTypeImgExtractText,
		semantic.TaskTypeAudioExtractText,
		semantic.TaskTypeVideoExtractVisual,
	} {
		var count int
		if err := store.DB().QueryRow(`SELECT COUNT(*) FROM semantic_tasks WHERE task_type = ?`, taskType).Scan(&count); err != nil {
			t.Fatal(err)
		}
		if count == 0 {
			t.Fatalf("native fts_only upload did not enqueue %s", taskType)
		}
	}
	var backfills int
	if err := store.DB().QueryRow(`SELECT COUNT(*) FROM semantic_tasks
		WHERE task_type = ? AND resource_version = ?`, semantic.TaskTypeEmbedBackfill, 8).Scan(&backfills); err != nil {
		t.Fatal(err)
	}
	if backfills != 1 {
		t.Fatalf("native fts_only embedding backfill tasks=%d, want 1", backfills)
	}
}

// This test reads global metrics emitted by recordTenantSchemaVersionUpdateFailure
// via operationMetricValue, so it must stay non-parallel unless the metric state
// is isolated.
func TestRecordTenantSchemaVersionUpdateFailureRecordsMetric(t *testing.T) {
	recorder := httptest.NewRecorder()
	metrics.WritePrometheus(recorder)
	before := operationMetricValue(t, recorder.Body.String(), `component="tenant_pool",operation="update_schema_version_failed",result="error"`)

	recordTenantSchemaVersionUpdateFailure(context.Background(), "tenant-metric", 42, time.Millisecond, fmt.Errorf("meta unavailable"))

	recorder = httptest.NewRecorder()
	metrics.WritePrometheus(recorder)
	after := operationMetricValue(t, recorder.Body.String(), `component="tenant_pool",operation="update_schema_version_failed",result="error"`)
	if after != before+1 {
		t.Fatalf("expected update_schema_version_failed metric to increment by 1, before=%d after=%d", before, after)
	}
}

func TestTenantPoolErrorResultClassifiesExpectedDatabaseErrors(t *testing.T) {
	tests := []struct {
		name string
		err  error
		want tenantPoolResult
	}{
		{
			name: "tidb auth failed",
			err:  fmt.Errorf("open db: %w", &mysql.MySQLError{Number: 1045, Message: "Access denied for user"}),
			want: tenantPoolResultAuthFailed,
		},
		{
			name: "tidb auth failed legacy message",
			err:  fmt.Errorf("open db: Error 1045 (28000): Please check your user name and password and try again"),
			want: tenantPoolResultAuthFailed,
		},
		{
			name: "tidb usage quota exhausted",
			err:  fmt.Errorf("open db: %w", &mysql.MySQLError{Number: 1105, Message: "Due to the usage quota being exhausted, access to the cluster has been restricted"}),
			want: tenantPoolResultUsageQuotaExhausted,
		},
		{
			name: "tidb usage quota exhausted lowercase",
			err:  fmt.Errorf("open db: error 1105 (hy000): due to the usage quota being exhausted"),
			want: tenantPoolResultUsageQuotaExhausted,
		},
		{
			name: "not found",
			err:  meta.ErrNotFound,
			want: tenantPoolResultNotFound,
		},
		{
			name: "unexpected",
			err:  fmt.Errorf("open db: invalid connection"),
			want: tenantPoolResultError,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := tenantPoolErrorResult(tt.err); got != tt.want {
				t.Fatalf("tenantPoolErrorResult() = %q, want %q", got, tt.want)
			}
		})
	}
}

func operationMetricValue(t *testing.T, output, labels string) uint64 {
	t.Helper()
	prefix := `drive9_service_operations_total{` + labels + `} `
	for _, line := range strings.Split(output, "\n") {
		if !strings.HasPrefix(line, prefix) {
			continue
		}
		value, err := strconv.ParseUint(strings.TrimSpace(strings.TrimPrefix(line, prefix)), 10, 64)
		if err != nil {
			t.Fatalf("parse metric %q: %v", line, err)
		}
		return value
	}
	return 0
}

type poolDummyAudioExtractor struct{}

func (poolDummyAudioExtractor) ExtractAudioText(context.Context, backend.AudioExtractRequest) (string, backend.AudioExtractUsage, error) {
	return "", backend.AudioExtractUsage{}, nil
}

func TestIdleEviction(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:       2,
		IdleTimeout:      50 * time.Millisecond,
		IdleReapInterval: 10 * time.Millisecond,
	}, "tenant-idle")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()
	assertStoreOpen(t, store)
	release()

	// Wait for idle timeout to lapse, then reap manually.
	time.Sleep(60 * time.Millisecond)
	pool.reapOnce(ctx)

	assertStoreClosed(t, store)
	if _, ok := pool.items[tenant.ID]; ok {
		t.Fatal("expected entry to be evicted from pool")
	}
}

func TestIdleEvictionSkipsSharedTenantEntry(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:  2,
		IdleTimeout: time.Minute,
	}, "tenant-shared-idle")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()
	release()

	pool.mu.Lock()
	e := pool.items[tenant.ID]
	e.sharedDBID = 123
	e.lastUsed = time.Now().Add(-2 * time.Minute)
	pool.mu.Unlock()

	pool.reapOnce(ctx)

	assertStoreOpen(t, store)
	if _, ok := pool.items[tenant.ID]; !ok {
		t.Fatal("shared tenant entry was removed by idle reaper")
	}
}

func TestCapacityEvictionStillRemovesSharedTenantEntry(t *testing.T) {
	pool, first := newTestPoolAndTenant(t, 1, "tenant-shared-capacity-first")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, first)
	if err != nil {
		t.Fatal(err)
	}
	firstStore := b.Store()
	release()

	pool.mu.Lock()
	pool.items[first.ID].sharedDBID = 123
	pool.mu.Unlock()

	second := cloneTenantForID(t, pool, first, "tenant-shared-capacity-second")
	_, releaseSecond, err := pool.Acquire(ctx, second)
	if err != nil {
		t.Fatal(err)
	}
	releaseSecond()

	assertStoreClosed(t, firstStore)
	if _, ok := pool.items[first.ID]; ok {
		t.Fatal("shared tenant entry remained after capacity eviction")
	}
}

func TestAcquireCachedDoesNotRefreshSharedCapacityLRU(t *testing.T) {
	pool, busy := newTestPoolAndTenant(t, 2, "tenant-shared-lru-busy")
	ctx := context.Background()

	busyBackend, releaseBusy, err := pool.Acquire(ctx, busy)
	if err != nil {
		t.Fatal(err)
	}
	busyStore := busyBackend.Store()
	releaseBusy()

	idle := cloneTenantForID(t, pool, busy, "tenant-shared-lru-idle")
	idleBackend, releaseIdle, err := pool.Acquire(ctx, idle)
	if err != nil {
		t.Fatal(err)
	}
	idleStore := idleBackend.Store()
	releaseIdle()

	pool.mu.Lock()
	pool.items[busy.ID].sharedDBID = 123
	pool.items[idle.ID].sharedDBID = 123
	pool.mu.Unlock()

	// Real tenant activity makes busy the most recently used entry.
	_, releaseBusy, err = pool.Acquire(ctx, busy)
	if err != nil {
		t.Fatal(err)
	}
	releaseBusy()

	// The safety-net scan may pin idle, but must not rewrite capacity LRU order.
	_, releaseCached, ok := pool.AcquireCached(idle)
	if !ok {
		t.Fatal("expected AcquireCached to hit idle shared tenant")
	}
	releaseCached()

	third := cloneTenantForID(t, pool, busy, "tenant-shared-lru-third")
	_, releaseThird, err := pool.Acquire(ctx, third)
	if err != nil {
		t.Fatal(err)
	}
	releaseThird()

	pool.mu.Lock()
	_, busyCached := pool.items[busy.ID]
	_, idleCached := pool.items[idle.ID]
	pool.mu.Unlock()
	if !busyCached || idleCached {
		t.Errorf("capacity LRU after safety-net scan: busy cached=%t idle cached=%t, want true false", busyCached, idleCached)
	}
	assertStoreOpen(t, busyStore)
	assertStoreClosed(t, idleStore)
}

func TestAcquireCountsBackgroundWorkAsCapacityLRUActivity(t *testing.T) {
	pool, first := newTestPoolAndTenant(t, 2, "tenant-work-lru-first")
	ctx := context.Background()
	second := cloneTenantForID(t, pool, first, "tenant-work-lru-second")
	third := cloneTenantForID(t, pool, first, "tenant-work-lru-third")

	_, releaseFirst, err := pool.Acquire(ctx, first)
	if err != nil {
		t.Fatal(err)
	}
	releaseFirst()
	_, releaseSecond, err := pool.Acquire(ctx, second)
	if err != nil {
		t.Fatal(err)
	}
	releaseSecond()

	thirdBackend, releaseThird, err := pool.Acquire(ctx, third)
	if err != nil {
		t.Fatal(err)
	}
	thirdStore := thirdBackend.Store()
	pool.mu.Lock()
	_, firstCached := pool.items[first.ID]
	_, secondCached := pool.items[second.ID]
	_, thirdCached := pool.items[third.ID]
	pool.mu.Unlock()
	if firstCached || !secondCached || !thirdCached {
		t.Errorf("cache after work miss: first=%t second=%t third=%t, want false true true", firstCached, secondCached, thirdCached)
	}
	assertStoreOpen(t, thirdStore)
	releaseThird()
	assertStoreOpen(t, thirdStore)
}

func TestSharedDBIdleReapCascadesTenantEntries(t *testing.T) {
	pool, first := newTestPoolAndTenant(t, 3, "tenant-shared-cascade-first")
	ctx := context.Background()
	second := cloneTenantForID(t, pool, first, "tenant-shared-cascade-second")
	other := cloneTenantForID(t, pool, first, "tenant-shared-cascade-other")

	firstBackend, releaseFirst, err := pool.Acquire(ctx, first)
	if err != nil {
		t.Fatal(err)
	}
	releaseFirst()
	secondBackend, releaseSecond, err := pool.Acquire(ctx, second)
	if err != nil {
		t.Fatal(err)
	}
	releaseSecond()
	otherBackend, releaseOther, err := pool.Acquire(ctx, other)
	if err != nil {
		t.Fatal(err)
	}
	releaseOther()

	expiredHandle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	recentHandle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := expiredHandle.Ping(); err != nil {
		t.Fatal(err)
	}
	if err := recentHandle.Ping(); err != nil {
		t.Fatal(err)
	}
	const expiredDBID int64 = 101
	const recentDBID int64 = 202
	now := time.Now()
	pool.sharedMu.Lock()
	pool.sharedDBs[expiredDBID] = expiredHandle
	pool.sharedDBs[recentDBID] = recentHandle
	firstLease := pool.leaseSharedDBLocked(expiredDBID, expiredHandle)
	secondLease := pool.leaseSharedDBLocked(expiredDBID, expiredHandle)
	otherLease := pool.leaseSharedDBLocked(recentDBID, recentHandle)
	pool.sharedDBLastUsed[expiredDBID] = now.Add(-defaultSharedDBHandleIdleTTL - time.Second)
	pool.sharedDBLastUsed[recentDBID] = now
	pool.sharedMu.Unlock()
	pool.mu.Lock()
	pool.items[first.ID].sharedDBID = expiredDBID
	pool.items[first.ID].sharedDBLease = firstLease
	pool.items[second.ID].sharedDBID = expiredDBID
	pool.items[second.ID].sharedDBLease = secondLease
	pool.items[other.ID].sharedDBID = recentDBID
	pool.items[other.ID].sharedDBLease = otherLease
	pool.mu.Unlock()

	pool.reapOnce(ctx)

	pool.mu.Lock()
	_, firstCached := pool.items[first.ID]
	_, secondCached := pool.items[second.ID]
	_, otherCached := pool.items[other.ID]
	pool.mu.Unlock()
	if firstCached || secondCached || !otherCached {
		t.Errorf("cache after shared cascade: first=%t second=%t other=%t, want false false true", firstCached, secondCached, otherCached)
	}
	assertStoreClosed(t, firstBackend.Store())
	assertStoreClosed(t, secondBackend.Store())
	assertStoreOpen(t, otherBackend.Store())
	if err := expiredHandle.Ping(); err == nil {
		t.Error("expired physical shared handle remains open")
	}
	if err := recentHandle.Ping(); err != nil {
		t.Errorf("recent physical shared handle closed: %v", err)
	}
}

func TestSharedDBIdleReapDefersActiveEntryClose(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-shared-cascade-active")
	ctx := context.Background()
	backend, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	handle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := handle.Ping(); err != nil {
		t.Fatal(err)
	}
	const dbID int64 = 303
	now := time.Now()
	pool.sharedMu.Lock()
	pool.sharedDBs[dbID] = handle
	lease := pool.leaseSharedDBLocked(dbID, handle)
	pool.sharedDBLastUsed[dbID] = now.Add(-defaultSharedDBHandleIdleTTL - time.Second)
	pool.sharedMu.Unlock()
	pool.mu.Lock()
	pool.items[tenant.ID].sharedDBID = dbID
	pool.items[tenant.ID].sharedDBLease = lease
	pool.mu.Unlock()

	pool.reapOnce(ctx)
	pool.mu.Lock()
	_, cached := pool.items[tenant.ID]
	pool.mu.Unlock()
	if cached {
		t.Error("active shared entry remains in capacity LRU after cascade")
	}
	assertStoreOpen(t, backend.Store())
	if err := handle.Ping(); err != nil {
		t.Errorf("physical handle closed while tenant entry is active: %v", err)
	}

	release()
	assertStoreClosed(t, backend.Store())
	pool.reapIdleSharedDBs(now)
	if err := handle.Ping(); err == nil {
		t.Error("physical handle remains open after active entry release")
	}
}

func TestAcquireRefreshesSharedDBActivity(t *testing.T) {
	pool, tenant := newTestPoolAndTenant(t, 2, "tenant-shared-activity")
	ctx := context.Background()
	_, initialRelease, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	initialRelease()

	handle, err := sql.Open("mysql", testDSN)
	if err != nil {
		t.Fatal(err)
	}
	if err := handle.Ping(); err != nil {
		t.Fatal(err)
	}
	const dbID int64 = 404
	pool.sharedMu.Lock()
	pool.sharedDBs[dbID] = handle
	lease := pool.leaseSharedDBLocked(dbID, handle)
	pool.sharedDBLastUsed[dbID] = time.Now().Add(-defaultSharedDBHandleIdleTTL - time.Second)
	pool.sharedMu.Unlock()
	pool.mu.Lock()
	pool.items[tenant.ID].sharedDBID = dbID
	pool.items[tenant.ID].sharedDBLease = lease
	pool.mu.Unlock()

	_, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	release()
	pool.reapOnce(ctx)

	pool.mu.Lock()
	_, cached := pool.items[tenant.ID]
	pool.mu.Unlock()
	if !cached {
		t.Error("tenant cache entry reaped after real activity")
	}
	if err := handle.Ping(); err != nil {
		t.Errorf("active physical handle closed: %v", err)
	}
}

func TestAcquireRefreshesStandaloneIdleTTL(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:  2,
		IdleTimeout: 50 * time.Millisecond,
	}, "tenant-work-standalone-idle")
	ctx := context.Background()
	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	release()

	pool.mu.Lock()
	pool.items[tenant.ID].lastUsed = time.Now().Add(-time.Second)
	pool.mu.Unlock()
	workBackend, workRelease, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	if workBackend != b {
		t.Error("expected background work to reuse standalone backend")
	}
	workRelease()

	pool.reapOnce(ctx)
	pool.mu.Lock()
	_, cached := pool.items[tenant.ID]
	pool.mu.Unlock()
	if !cached {
		t.Error("standalone backend reaped immediately after background work")
	}
	assertStoreOpen(t, b.Store())
}

func TestCloseEntryPreservesTenantMetricsUntilRetentionExpiry(t *testing.T) {
	const tenantID = "tenant-retained-metrics-close-entry"
	metrics.DeleteTenantCounters(tenantID)
	t.Cleanup(func() { metrics.DeleteTenantCounters(tenantID) })

	metrics.RecordTenantRequestCountWithOrg(tenantID, "org-live-counter", "fs", "read", 200)
	metrics.RecordTenantInFlightWithOrg(tenantID, "org-live-counter", "fs", 2)
	recorder := httptest.NewRecorder()
	metrics.WritePrometheus(recorder)
	if !strings.Contains(recorder.Body.String(), `drive9_tenant_requests_total{`) ||
		!strings.Contains(recorder.Body.String(), `tenant_id="`+tenantID+`"`) {
		t.Fatalf("tenant request counter missing before close:\n%s", recorder.Body.String())
	}
	if !strings.Contains(recorder.Body.String(), `drive9_tenant_inflight_requests{`) ||
		!strings.Contains(recorder.Body.String(), `tenant_id="`+tenantID+`"`) {
		t.Fatalf("tenant inflight gauge missing before close:\n%s", recorder.Body.String())
	}

	pool := NewPool(PoolConfig{}, nil)
	pool.closeEntry(&entry{tenantID: tenantID})

	recorder = httptest.NewRecorder()
	metrics.WritePrometheus(recorder)
	if !metricHasTenantID(t, recorder.Body.String(), "drive9_tenant_requests_total", tenantID) {
		t.Fatalf("tenant request counter disappeared before retention expiry:\n%s", recorder.Body.String())
	}
	if !metricHasTenantID(t, recorder.Body.String(), "drive9_tenant_inflight_requests", tenantID) {
		t.Fatalf("tenant inflight gauge disappeared before retention expiry:\n%s", recorder.Body.String())
	}
}

func TestCloseEntryDoesNotClearFileGCGaugesForSupersededEntry(t *testing.T) {
	const tenantID = "tenant-filegc-superseded-close"
	const orgID = "org-filegc-superseded-close"
	metrics.ObserveFileGCClaimableBacklog(tenantID, orgID, 4, time.Hour)
	metrics.ObserveFileGCDrainHitCap(tenantID, orgID, true)
	t.Cleanup(func() { metrics.ClearFileGCClaimableBacklog(tenantID) })

	pool := NewPool(PoolConfig{}, nil)
	current := &entry{tenantID: tenantID}
	pool.mu.Lock()
	pool.items[tenantID] = current
	pool.mu.Unlock()

	pool.closeEntry(&entry{tenantID: tenantID, retired: true})
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !metricHasTenantID(t, text, "drive9_file_gc_claimable", tenantID) {
		t.Fatalf("superseded closeEntry cleared claimable for the live entry:\n%s", text)
	}
	if !metricHasTenantID(t, text, "drive9_file_gc_drain_hit_cap", tenantID) {
		t.Fatalf("superseded closeEntry cleared drain_hit_cap for the live entry:\n%s", text)
	}

	pool.mu.Lock()
	delete(pool.items, tenantID)
	pool.mu.Unlock()
	pool.closeEntry(current)
	rec = httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text = rec.Body.String()
	if metricHasTenantID(t, text, "drive9_file_gc_claimable", tenantID) {
		t.Fatalf("final closeEntry left claimable series:\n%s", rec.Body.String())
	}
}

func metricHasTenantID(t *testing.T, text, metric, tenantID string) bool {
	t.Helper()
	for _, line := range strings.Split(text, "\n") {
		if strings.HasPrefix(line, metric+"{") && strings.Contains(line, `tenant_id="`+tenantID+`"`) {
			return true
		}
	}
	return false
}

func TestIdleEvictionSkippedByRecentAcquire(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:       2,
		IdleTimeout:      50 * time.Millisecond,
		IdleReapInterval: 10 * time.Millisecond,
	}, "tenant-recent")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()
	release()

	// Re-acquire immediately — lastUsed is refreshed to now.
	b2, release2, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	if b2 != b {
		t.Fatal("expected same backend on cache hit")
	}
	release2()

	// Reap immediately — lastUsed is recent, should not evict.
	pool.reapOnce(ctx)
	assertStoreOpen(t, store)
	if _, ok := pool.items[tenant.ID]; !ok {
		t.Fatal("expected entry to remain in pool after recent acquire")
	}
}

func TestIdleEvictionNotRefreshedByAcquireCached(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:       2,
		IdleTimeout:      50 * time.Millisecond,
		IdleReapInterval: 10 * time.Millisecond,
	}, "tenant-cached")
	ctx := context.Background()

	// Cold-open via Acquire so the backend enters the pool.
	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()
	release()

	// Wait long enough that lastUsed is stale, but use AcquireCached
	// before reaping — AcquireCached must NOT refresh lastUsed.
	time.Sleep(60 * time.Millisecond)
	b2, release2, ok := pool.AcquireCached(tenant)
	if !ok {
		t.Fatal("expected AcquireCached to hit warm cache")
	}
	if b2 != b {
		t.Fatal("expected same backend from AcquireCached")
	}
	release2()

	// Reap — entry should be evicted because AcquireCached did not
	// refresh lastUsed.
	pool.reapOnce(ctx)
	assertStoreClosed(t, store)
	if _, ok := pool.items[tenant.ID]; ok {
		t.Fatal("expected entry to be evicted despite AcquireCached touch")
	}
}

func TestIdleEvictionRespectsRefs(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:       2,
		IdleTimeout:      50 * time.Millisecond,
		IdleReapInterval: 10 * time.Millisecond,
	}, "tenant-pinned")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()

	// Wait for idle timeout, but don't release — refs > 0 must prevent
	// eviction.
	time.Sleep(60 * time.Millisecond)
	pool.reapOnce(ctx)
	assertStoreOpen(t, store)
	if _, ok := pool.items[tenant.ID]; !ok {
		t.Fatal("expected pinned entry to remain in pool")
	}

	// Release refreshes lastUsed (user release), so entry stays warm
	// for another IdleTimeout window.
	release()
	pool.reapOnce(ctx)
	assertStoreOpen(t, store)

	// Now wait for idle timeout to lapse after release, then reap.
	time.Sleep(60 * time.Millisecond)
	pool.reapOnce(ctx)
	assertStoreClosed(t, store)
}

func TestIdleEvictionDisabled(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:  2,
		IdleTimeout: 0, // disabled
	}, "tenant-disabled")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()
	release()

	// Wait well beyond any idle timeout, but since IdleTimeout=0 the
	// reaper is disabled and reapOnce is a no-op (idleTimeout is 0 so
	// now.Sub(lastUsed) > 0 is always true — but Start won't have been
	// called). Call reapOnce directly to confirm it's a no-op.
	time.Sleep(20 * time.Millisecond)
	pool.reapOnce(ctx)
	assertStoreOpen(t, store)
	if _, ok := pool.items[tenant.ID]; !ok {
		t.Fatal("expected entry to remain when idle eviction is disabled")
	}
}

func TestIdleEvictionStartKeepsSharedHandleReaperWhenDisabled(t *testing.T) {
	pool := NewPool(PoolConfig{
		MaxTenants:  2,
		IdleTimeout: 0,
	}, nil)
	pool.Start(context.Background())
	t.Cleanup(pool.Close)
	if pool.reapStop == nil {
		t.Fatal("expected shared handle reaper to run when IdleTimeout=0")
	}
}

// TestIdleEvictionReleaseRefreshesLastUsed verifies that a long-running user
// request (held longer than IdleTimeout) is not immediately evicted on
// release. The release path must refresh lastUsed so the idle timer starts
// counting from "user finished", not "user started".
func TestIdleEvictionReleaseRefreshesLastUsed(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:       2,
		IdleTimeout:      50 * time.Millisecond,
		IdleReapInterval: 10 * time.Millisecond,
	}, "tenant-long")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()

	// Hold the backend longer than IdleTimeout. The reaper must skip it
	// because refs > 0.
	time.Sleep(80 * time.Millisecond)
	pool.reapOnce(ctx)
	assertStoreOpen(t, store)

	// Release now — lastUsed is refreshed. Immediate reap must NOT evict.
	release()
	pool.reapOnce(ctx)
	assertStoreOpen(t, store)
	if _, ok := pool.items[tenant.ID]; !ok {
		t.Fatal("expected entry to remain after release refreshed lastUsed")
	}

	// Wait for idle timeout to lapse after release, then reap.
	time.Sleep(60 * time.Millisecond)
	pool.reapOnce(ctx)
	assertStoreClosed(t, store)
}

// TestIdleEvictionAcquireCachedReleaseDoesNotRefresh verifies that
// AcquireCached's release does not refresh lastUsed. A safety-net scan
// touch must not reset the idle timer even on release.
func TestIdleEvictionAcquireCachedReleaseDoesNotRefresh(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:       2,
		IdleTimeout:      50 * time.Millisecond,
		IdleReapInterval: 10 * time.Millisecond,
	}, "tenant-cached-release")
	ctx := context.Background()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()
	release()

	// Wait until lastUsed is stale.
	time.Sleep(60 * time.Millisecond)

	// AcquireCached + release should not refresh lastUsed.
	_, release2, ok := pool.AcquireCached(tenant)
	if !ok {
		t.Fatal("expected AcquireCached to hit warm cache")
	}
	release2()

	// Reap — should evict because neither AcquireCached nor its release
	// refreshed lastUsed.
	pool.reapOnce(ctx)
	assertStoreClosed(t, store)
	if _, ok := pool.items[tenant.ID]; ok {
		t.Fatal("expected entry to be evicted — AcquireCached release must not refresh lastUsed")
	}
}

// TestIdleEvictionStartAutoEvictsAndCloseStops verifies the end-to-end
// lifecycle: Start launches the reaper, the ticker automatically evicts an
// idle entry, and Close stops the reaper cleanly.
func TestIdleEvictionStartAutoEvictsAndCloseStops(t *testing.T) {
	pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{
		MaxTenants:       2,
		IdleTimeout:      50 * time.Millisecond,
		IdleReapInterval: 10 * time.Millisecond,
	}, "tenant-e2e")
	ctx := context.Background()
	pool.Start(ctx)
	defer pool.Close()

	b, release, err := pool.Acquire(ctx, tenant)
	if err != nil {
		t.Fatal(err)
	}
	store := b.Store()
	release()

	// Wait for the ticker-driven reaper to auto-evict (within ~2s).
	deadline := time.Now().Add(2 * time.Second)
	evicted := false
	for time.Now().Before(deadline) {
		pool.mu.Lock()
		_, ok := pool.items[tenant.ID]
		pool.mu.Unlock()
		if !ok {
			evicted = true
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	if !evicted {
		t.Fatal("expected auto-eviction by ticker")
	}
	assertStoreClosed(t, store)
}