package meta

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

	mysql "github.com/go-sql-driver/mysql"

	"github.com/mem9-ai/drive9/pkg/metrics"
)

func TestDefaultMaxStorageBytesDefault(t *testing.T) {
	if got := DefaultMaxStorageBytes(); got != int64(50*(1<<30)) {
		t.Fatalf("default = %d, want 50 GiB", got)
	}
}

func TestSetQuotaConfigPatchWaitsForConcurrentExtractConfigDisable(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "quota-extract-disable-race"
	insertFreeQuotaTestTenant(t, s, tenantID, TenantActive, tidbCloudNativeProvider)
	if err := s.UpsertTenantExtractConfig(ctx, &TenantExtractConfig{
		TenantID: tenantID, MediaType: ExtractMediaTypeImage, Enabled: true,
		APIBase: "https://extract.example.com", APIKeyCipher: []byte("cipher"), Model: "model",
	}); err != nil {
		t.Fatal(err)
	}

	disableTx, err := s.DB().BeginTx(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = disableTx.Rollback() }()
	var lockedTenantID string
	if err := disableTx.QueryRowContext(ctx, `SELECT id FROM tenants WHERE id = ? FOR UPDATE`, tenantID).Scan(&lockedTenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := disableTx.ExecContext(ctx, `UPDATE tenant_extract_configs SET enabled = 0 WHERE tenant_id = ? AND media_type = ?`,
		tenantID, ExtractMediaTypeImage); err != nil {
		t.Fatal(err)
	}

	limit := int64(123)
	result := make(chan error, 1)
	started := make(chan struct{})
	go func() {
		close(started)
		result <- s.SetQuotaConfigPatch(context.Background(), tenantID,
			QuotaConfigPatch{MaxMediaLLMFiles: &limit})
	}()
	<-started
	select {
	case err := <-result:
		t.Fatalf("quota update returned before concurrent disable committed: %v", err)
	case <-time.After(100 * time.Millisecond):
	}
	if err := disableTx.Commit(); err != nil {
		t.Fatal(err)
	}
	if err := <-result; !errors.Is(err, ErrTenantExtractConfigNotEnabled) {
		t.Fatalf("quota update error = %v, want disabled extract config", err)
	}
	cfg, err := s.GetQuotaConfig(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxMediaLLMFiles == limit {
		t.Fatalf("media LLM limit committed despite concurrent disable: %d", cfg.MaxMediaLLMFiles)
	}
}

func TestSetDefaultMaxStorageBytes(t *testing.T) {
	orig := DefaultMaxStorageBytes()
	defer func() { SetDefaultMaxStorageBytes(orig) }()

	SetDefaultMaxStorageBytes(1 << 30)
	if got := DefaultMaxStorageBytes(); got != int64(1<<30) {
		t.Fatalf("got %d, want 1 GiB", got)
	}
}

func TestSetDefaultMaxStorageBytesRejectsZero(t *testing.T) {
	orig := DefaultMaxStorageBytes()
	defer func() { SetDefaultMaxStorageBytes(orig) }()

	SetDefaultMaxStorageBytes(0)
	if got := DefaultMaxStorageBytes(); got != orig {
		t.Fatalf("zero val changed default from %d to %d", orig, got)
	}
}

func TestDefaultMaxFileSizeBytesDefault(t *testing.T) {
	if got := DefaultMaxFileSizeBytes(); got != int64(10*(1<<30)) {
		t.Fatalf("default = %d, want 10 GiB", got)
	}
}

func TestDefaultMaxFileCountDefaultAndOverride(t *testing.T) {
	original := DefaultMaxFileCount()
	defer SetDefaultMaxFileCount(original)

	SetDefaultMaxFileCount(0)
	if got := DefaultMaxFileCount(); got != 0 {
		t.Fatalf("default max file count = %d, want unlimited 0", got)
	}
	SetDefaultMaxFileCount(1234)
	if got := DefaultMaxFileCount(); got != 1234 {
		t.Fatalf("default max file count = %d, want 1234", got)
	}
	SetDefaultMaxFileCount(-1)
	if got := DefaultMaxFileCount(); got != 1234 {
		t.Fatalf("negative override changed max file count to %d", got)
	}
}

func TestSetDefaultMaxFileSizeBytes(t *testing.T) {
	orig := DefaultMaxFileSizeBytes()
	defer func() { SetDefaultMaxFileSizeBytes(orig) }()

	SetDefaultMaxFileSizeBytes(2 << 30)
	if got := DefaultMaxFileSizeBytes(); got != int64(2<<30) {
		t.Fatalf("got %d, want 2 GiB", got)
	}
}

func TestSetQuotaConfigPatchUpdatesLLMFileLimitsOnly(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "tenant-llm-limit-patch"
	insertFreeQuotaTestTenant(t, s, tenantID, TenantActive, tidbCloudNativeProvider)
	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID: tenantID, MaxStorageBytes: 100, MaxFileSizeBytes: 200, MaxFileCount: 300,
		MaxMediaLLMFiles: 400, MaxVideoLLMFiles: 500, MaxMonthlyCostMC: 600,
	}); err != nil {
		t.Fatal(err)
	}
	for _, mediaType := range []string{ExtractMediaTypeImage, ExtractMediaTypeVideo} {
		if err := s.UpsertTenantExtractConfig(ctx, &TenantExtractConfig{
			TenantID: tenantID, MediaType: mediaType, Enabled: true,
			APIBase: "https://extract.example.com", APIKeyCipher: []byte("cipher"), Model: "model",
		}); err != nil {
			t.Fatal(err)
		}
	}
	mediaLimit, videoLimit := int64(41), int64(51)
	if err := s.SetQuotaConfigPatch(ctx, tenantID, QuotaConfigPatch{
		MaxMediaLLMFiles: &mediaLimit,
		MaxVideoLLMFiles: &videoLimit,
	}); err != nil {
		t.Fatal(err)
	}
	cfg, err := s.GetQuotaConfig(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxMediaLLMFiles != mediaLimit || cfg.MaxVideoLLMFiles != videoLimit {
		t.Fatalf("LLM limits = %+v", cfg)
	}
	if cfg.MaxStorageBytes != 100 || cfg.MaxFileSizeBytes != 200 || cfg.MaxFileCount != 300 || cfg.MaxMonthlyCostMC != 600 {
		t.Fatalf("unrelated limits changed: %+v", cfg)
	}
}

func TestSetQuotaConfigPatchRequiresEnabledTenantExtractConfig(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "tenant-llm-limit-without-extract-config"
	insertFreeQuotaTestTenant(t, s, tenantID, TenantActive, tidbCloudNativeProvider)
	mediaLimit := int64(41)

	err := s.SetQuotaConfigPatch(ctx, tenantID, QuotaConfigPatch{MaxMediaLLMFiles: &mediaLimit})
	if !errors.Is(err, ErrTenantExtractConfigNotEnabled) {
		t.Fatalf("SetQuotaConfigPatch error = %v, want enabled tenant extract config error", err)
	}
}

func TestGetQuotaConfigUsesConfiguredDefaultStorageBytes(t *testing.T) {
	orig := DefaultMaxStorageBytes()
	defer func() { SetDefaultMaxStorageBytes(orig) }()

	SetDefaultMaxStorageBytes(12345)
	s := newControlStore(t)

	cfg, err := s.GetQuotaConfig(context.Background(), "tenant-without-config")
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxStorageBytes != 12345 {
		t.Errorf("MaxStorageBytes = %d, want configured default 12345", cfg.MaxStorageBytes)
	}
	if cfg.MaxFileSizeBytes != DefaultMaxFileSizeBytes() {
		t.Errorf("MaxFileSizeBytes = %d, want default %d", cfg.MaxFileSizeBytes, DefaultMaxFileSizeBytes())
	}
	if cfg.MaxFileCount != DefaultMaxFileCount() {
		t.Errorf("MaxFileCount = %d, want default %d", cfg.MaxFileCount, DefaultMaxFileCount())
	}
	if cfg.MaxMediaLLMFiles != 100 {
		t.Errorf("MaxMediaLLMFiles = %d, want default 100", cfg.MaxMediaLLMFiles)
	}
	if cfg.MaxVideoLLMFiles != 30 {
		t.Errorf("MaxVideoLLMFiles = %d, want default 30", cfg.MaxVideoLLMFiles)
	}
	if cfg.MaxMonthlyCostMC != 0 {
		t.Errorf("MaxMonthlyCostMC = %d, want default 0", cfg.MaxMonthlyCostMC)
	}
}

func TestTenantQuotaConfigDDLUsesDefaultMediaLimit(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "tenant-ddl-default-media-limit"

	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_quota_config (tenant_id) VALUES (?)`, tenantID); err != nil {
		t.Fatal(err)
	}
	var got int64
	if err := s.DB().QueryRowContext(ctx,
		`SELECT max_media_llm_files FROM tenant_quota_config WHERE tenant_id = ?`, tenantID).Scan(&got); err != nil {
		t.Fatal(err)
	}
	if got != 100 {
		t.Fatalf("max_media_llm_files = %d, want DDL default 100", got)
	}
}

func TestGetQuotaConfigUsesDefaultFileSizeForExistingZeroRow(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_quota_config
		(tenant_id, max_storage_bytes, max_file_size_bytes, max_file_count,
		 max_media_llm_files, max_video_llm_files, max_monthly_cost_mc)
		VALUES (?, ?, 0, 0, ?, ?, 0)`, "tenant-zero-file-size", DefaultMaxStorageBytes(),
		DefaultMaxMediaLLMFiles(), DefaultMaxVideoLLMFiles()); err != nil {
		t.Fatalf("insert quota row: %v", err)
	}

	cfg, err := s.GetQuotaConfig(ctx, "tenant-zero-file-size")
	if err != nil {
		t.Fatalf("GetQuotaConfig: %v", err)
	}
	if cfg.MaxFileSizeBytes != DefaultMaxFileSizeBytes() {
		t.Fatalf("MaxFileSizeBytes = %d, want default %d", cfg.MaxFileSizeBytes, DefaultMaxFileSizeBytes())
	}
}

func TestListTenantQuotaSnapshotsUsesStoredAndDefaultValues(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	storedID := "tenant-quota-snapshot-stored"
	defaultID := "tenant-quota-snapshot-default"
	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID: storedID, MaxStorageBytes: 100, MaxFileSizeBytes: 10, MaxFileCount: 7,
		MaxMediaLLMFiles: 8, MaxVideoLLMFiles: 9,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.SetQuotaCounters(ctx, storedID, 11, 0, 12); err != nil {
		t.Fatal(err)
	}

	snapshots, err := s.ListTenantQuotaSnapshots(ctx, []string{storedID, defaultID})
	if err != nil {
		t.Fatal(err)
	}
	stored := snapshots[storedID]
	if stored.Config.MaxStorageBytes != 100 || stored.Config.MaxFileSizeBytes != 10 || stored.Config.MaxFileCount != 7 {
		t.Fatalf("stored config = %#v", stored.Config)
	}
	if stored.Config.MaxMediaLLMFiles != 8 || stored.Config.MaxVideoLLMFiles != 9 {
		t.Fatalf("stored LLM limits = %#v", stored.Config)
	}
	if stored.Usage.StorageBytes != 11 || stored.Usage.FileCount != 12 {
		t.Fatalf("stored usage = %#v", stored.Usage)
	}
	defaults := snapshots[defaultID]
	if defaults.Config.MaxStorageBytes != DefaultMaxStorageBytes() || defaults.Config.MaxFileSizeBytes != DefaultMaxFileSizeBytes() || defaults.Config.MaxFileCount != DefaultMaxFileCount() {
		t.Fatalf("default config = %#v", defaults.Config)
	}
	if defaults.Usage.StorageBytes != 0 || defaults.Usage.ReservedBytes != 0 || defaults.Usage.FileCount != 0 {
		t.Fatalf("default usage = %#v", defaults.Usage)
	}
}

func TestSetQuotaStorageBytesUpdatesStorageOnly(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()

	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID:         "tenant-patch",
		MaxStorageBytes:  100,
		MaxFileSizeBytes: 101,
		MaxFileCount:     102,
		MaxMediaLLMFiles: 200,
		MaxVideoLLMFiles: 210,
		MaxMonthlyCostMC: 300,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.SetQuotaStorageBytes(ctx, "tenant-patch", 999); err != nil {
		t.Fatal(err)
	}
	cfg, err := s.GetQuotaConfig(ctx, "tenant-patch")
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxStorageBytes != 999 || cfg.MaxFileSizeBytes != 101 || cfg.MaxFileCount != 102 || cfg.MaxMediaLLMFiles != 200 || cfg.MaxVideoLLMFiles != 210 || cfg.MaxMonthlyCostMC != 300 {
		t.Errorf("cfg = %+v, want storage=999 file_size=101 file_count=102 media=200 video=210 monthly=300", cfg)
	}
}

func TestSetQuotaConfigWithoutSpendingLimitPreservesExistingValue(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	spendingLimit := int64(12_000)

	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID:               "tenant-preserve-spending-limit",
		MaxStorageBytes:        100,
		MaxFileSizeBytes:       101,
		MaxFileCount:           102,
		TiDBCloudSpendingLimit: &spendingLimit,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID:         "tenant-preserve-spending-limit",
		MaxStorageBytes:  200,
		MaxFileSizeBytes: 201,
		MaxFileCount:     202,
	}); err != nil {
		t.Fatal(err)
	}

	cfg, err := s.GetQuotaConfig(ctx, "tenant-preserve-spending-limit")
	if err != nil {
		t.Fatal(err)
	}
	if cfg.TiDBCloudSpendingLimit == nil || *cfg.TiDBCloudSpendingLimit != spendingLimit {
		t.Fatalf("spending limit = %#v, want %d", cfg.TiDBCloudSpendingLimit, spendingLimit)
	}
}

func TestSetQuotaStorageBytesInsertsInternalDefaults(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()

	if err := s.SetQuotaStorageBytes(ctx, "tenant-patch-insert", 12345); err != nil {
		t.Fatal(err)
	}
	cfg, err := s.GetQuotaConfig(ctx, "tenant-patch-insert")
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxStorageBytes != 12345 || cfg.MaxFileSizeBytes != DefaultMaxFileSizeBytes() || cfg.MaxFileCount != 0 || cfg.MaxMediaLLMFiles != 100 || cfg.MaxVideoLLMFiles != 30 || cfg.MaxMonthlyCostMC != 0 {
		t.Errorf("cfg = %+v, want storage patch with internal defaults", cfg)
	}
}

func TestSetQuotaConfigPatchUpdatesExternalFieldsOnly(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()

	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID:         "tenant-external-patch",
		MaxStorageBytes:  100,
		MaxFileSizeBytes: 200,
		MaxFileCount:     300,
		MaxMediaLLMFiles: 400,
		MaxVideoLLMFiles: 410,
		MaxMonthlyCostMC: 500,
	}); err != nil {
		t.Fatal(err)
	}
	fileSize := int64(222)
	fileCount := int64(333)
	if err := s.SetQuotaConfigPatch(ctx, "tenant-external-patch", QuotaConfigPatch{
		MaxFileSizeBytes: &fileSize,
		MaxFileCount:     &fileCount,
	}); err != nil {
		t.Fatal(err)
	}
	cfg, err := s.GetQuotaConfig(ctx, "tenant-external-patch")
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxStorageBytes != 100 || cfg.MaxFileSizeBytes != 222 || cfg.MaxFileCount != 333 || cfg.MaxMediaLLMFiles != 400 || cfg.MaxVideoLLMFiles != 410 || cfg.MaxMonthlyCostMC != 500 {
		t.Fatalf("cfg = %+v", cfg)
	}
}

func TestAtomicReserveAndInsertUploadBootstrapsUsageRow(t *testing.T) {
	orig := DefaultMaxStorageBytes()
	defer func() { SetDefaultMaxStorageBytes(orig) }()

	SetDefaultMaxStorageBytes(100)
	s := newControlStore(t)
	ctx := context.Background()

	if err := s.AtomicReserveAndInsertUpload(ctx, &UploadReservation{
		TenantID:       "tenant-without-usage-row",
		UploadID:       "upload-1",
		ReservedBytes:  40,
		FileCountDelta: 1,
		TargetPath:     "/large.bin",
		ExpiresAt:      time.Now().Add(time.Hour),
	}); err != nil {
		t.Fatal(err)
	}

	usage, err := s.GetQuotaUsage(ctx, "tenant-without-usage-row")
	if err != nil {
		t.Fatal(err)
	}
	if usage.StorageBytes != 0 || usage.ReservedBytes != 40 || usage.FileCount != 1 {
		t.Fatalf("usage = %+v, want storage=0 reserved=40 file_count=1", usage)
	}

	reservation, err := s.GetUploadReservation(ctx, "tenant-without-usage-row", "upload-1")
	if err != nil {
		t.Fatal(err)
	}
	if reservation.Status != "active" || reservation.ReservedBytes != 40 || reservation.FileCountDelta != 1 {
		t.Fatalf("reservation = %+v, want active reserved=40", reservation)
	}
}

func TestAtomicReserveAndInsertUploadStorageQuotaExceededExistingRow(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "tenant-reserve-storage-limit"

	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID:         tenantID,
		MaxStorageBytes:  100,
		MaxFileSizeBytes: DefaultMaxFileSizeBytes(),
		MaxFileCount:     0,
		MaxMediaLLMFiles: DefaultMaxMediaLLMFiles(),
		MaxVideoLLMFiles: DefaultMaxVideoLLMFiles(),
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.IncrStorageBytes(ctx, tenantID, 80); err != nil {
		t.Fatal(err)
	}

	err := s.AtomicReserveAndInsertUpload(ctx, &UploadReservation{
		TenantID:       tenantID,
		UploadID:       "upload-over-storage",
		ReservedBytes:  30,
		FileCountDelta: 1,
		TargetPath:     "/over.bin",
		ExpiresAt:      time.Now().Add(time.Hour),
	})
	if !errors.Is(err, ErrStorageQuotaExceeded) {
		t.Fatalf("err = %v, want ErrStorageQuotaExceeded", err)
	}

	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.StorageBytes != 80 || usage.ReservedBytes != 0 || usage.FileCount != 0 {
		t.Fatalf("usage = %+v, want storage=80 reserved=0 file_count=0", usage)
	}
	if _, err := s.GetUploadReservation(ctx, tenantID, "upload-over-storage"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("reservation err = %v, want ErrNotFound", err)
	}
}

func TestAtomicReserveAndInsertUploadFileCountQuotaExceededExistingRow(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "tenant-reserve-file-limit"

	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID:         tenantID,
		MaxStorageBytes:  1 << 30,
		MaxFileSizeBytes: DefaultMaxFileSizeBytes(),
		MaxFileCount:     2,
		MaxMediaLLMFiles: DefaultMaxMediaLLMFiles(),
		MaxVideoLLMFiles: DefaultMaxVideoLLMFiles(),
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.IncrFileCount(ctx, tenantID, 2); err != nil {
		t.Fatal(err)
	}

	err := s.AtomicReserveAndInsertUpload(ctx, &UploadReservation{
		TenantID:       tenantID,
		UploadID:       "upload-over-files",
		ReservedBytes:  10,
		FileCountDelta: 1,
		TargetPath:     "/over.bin",
		ExpiresAt:      time.Now().Add(time.Hour),
	})
	if !errors.Is(err, ErrFileCountQuotaExceeded) {
		t.Fatalf("err = %v, want ErrFileCountQuotaExceeded", err)
	}

	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.StorageBytes != 0 || usage.ReservedBytes != 0 || usage.FileCount != 2 {
		t.Fatalf("usage = %+v, want storage=0 reserved=0 file_count=2", usage)
	}
	if _, err := s.GetUploadReservation(ctx, tenantID, "upload-over-files"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("reservation err = %v, want ErrNotFound", err)
	}
}

func TestAtomicReserveAndInsertUploadQuotaExceededDoesNotBootstrapUsageRow(t *testing.T) {
	orig := DefaultMaxStorageBytes()
	defer func() { SetDefaultMaxStorageBytes(orig) }()
	SetDefaultMaxStorageBytes(50)

	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "tenant-reserve-no-bootstrap-on-exceeded"

	err := s.AtomicReserveAndInsertUpload(ctx, &UploadReservation{
		TenantID:       tenantID,
		UploadID:       "upload-too-big",
		ReservedBytes:  80,
		FileCountDelta: 1,
		TargetPath:     "/too-big.bin",
		ExpiresAt:      time.Now().Add(time.Hour),
	})
	if !errors.Is(err, ErrStorageQuotaExceeded) {
		t.Fatalf("err = %v, want ErrStorageQuotaExceeded", err)
	}
	if got := quotaUsageRowExists(t, s, tenantID); got {
		t.Fatal("quota-exceeded path created tenant_quota_usage row, want none")
	}
	if _, err := s.GetUploadReservation(ctx, tenantID, "upload-too-big"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("reservation err = %v, want ErrNotFound", err)
	}
}

func TestAtomicReserveAndInsertUploadConcurrentExistingTenantNoDeadlock(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const (
		tenantID = "tenant-reserve-concurrent"
		n        = 8
	)
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.SetQuotaConfig(ctx, &QuotaConfig{
		TenantID:         tenantID,
		MaxStorageBytes:  1 << 30,
		MaxFileSizeBytes: DefaultMaxFileSizeBytes(),
		MaxFileCount:     1000,
		MaxMediaLLMFiles: DefaultMaxMediaLLMFiles(),
		MaxVideoLLMFiles: DefaultMaxVideoLLMFiles(),
	}); err != nil {
		t.Fatal(err)
	}

	beforeRetry := reserveUploadLockConflictRetryCount(t)
	var (
		wg      sync.WaitGroup
		mu      sync.Mutex
		okBytes int64
		okFiles int64
	)
	wg.Add(n)
	for i := 0; i < n; i++ {
		go func() {
			defer wg.Done()
			r := &UploadReservation{
				TenantID:       tenantID,
				UploadID:       fmt.Sprintf("upload-%d", i),
				ReservedBytes:  int64(1024 * (i + 1)),
				FileCountDelta: 1,
				TargetPath:     fmt.Sprintf("/f-%d.bin", i),
				ExpiresAt:      time.Now().Add(time.Hour),
			}
			err := s.AtomicReserveAndInsertUpload(ctx, r)
			if err != nil {
				if errors.Is(err, ErrStorageQuotaExceeded) || errors.Is(err, ErrFileCountQuotaExceeded) {
					return
				}
				t.Errorf("reserve upload-%d: %v", i, err)
				return
			}
			mu.Lock()
			okBytes += r.ReservedBytes
			okFiles += r.FileCountDelta
			mu.Unlock()
		}()
	}
	wg.Wait()

	afterRetry := reserveUploadLockConflictRetryCount(t)
	if afterRetry != beforeRetry {
		t.Fatalf("lock_conflict_retry increased from %g to %g; S→X deadlock still occurring", beforeRetry, afterRetry)
	}
	if okBytes == 0 {
		t.Fatal("all concurrent reserves failed")
	}
	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.ReservedBytes != okBytes || usage.FileCount != okFiles {
		t.Fatalf("usage reserved=%d file_count=%d, want reserved=%d file_count=%d",
			usage.ReservedBytes, usage.FileCount, okBytes, okFiles)
	}
}

func quotaUsageRowExists(t *testing.T, s *Store, tenantID string) bool {
	t.Helper()
	var id string
	err := s.DB().QueryRow(`SELECT tenant_id FROM tenant_quota_usage WHERE tenant_id = ?`, tenantID).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return false
	}
	if err != nil {
		t.Fatalf("lookup tenant_quota_usage %s: %v", tenantID, err)
	}
	return true
}

func reserveUploadLockConflictRetryCount(t *testing.T) float64 {
	t.Helper()
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	const prefix = `drive9_service_operations_total{component="central_quota",operation="reserve_upload",result="lock_conflict_retry"} `
	for _, line := range strings.Split(rec.Body.String(), "\n") {
		if !strings.HasPrefix(line, prefix) {
			continue
		}
		var v float64
		if _, err := fmt.Sscanf(strings.TrimSpace(strings.TrimPrefix(line, prefix)), "%f", &v); err != nil {
			t.Fatalf("parse lock_conflict_retry metric %q: %v", line, err)
		}
		return v
	}
	return 0
}

func TestRetryMetaLockConflictRetriesDeadlock(t *testing.T) {
	var calls int
	err := retryMetaLockConflict(context.Background(), "tenant-retry-metric", "reserve_upload", "test_op", func() error {
		calls++
		if calls < 3 {
			return errors.New("Error 1213 (40001): Deadlock found when trying to get lock; try restarting transaction")
		}
		return nil
	})
	if err != nil {
		t.Fatalf("retryMetaLockConflict returned error: %v", err)
	}
	if calls != 3 {
		t.Fatalf("calls = %d, want 3", calls)
	}

	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	if !strings.Contains(rec.Body.String(), `drive9_service_operations_total{component="central_quota",operation="reserve_upload",result="lock_conflict_retry"} 2`) {
		t.Fatalf("missing lock conflict retry metric:\n%s", rec.Body.String())
	}
}

func TestRetryMetaLockConflictDoesNotRetryBusinessError(t *testing.T) {
	var calls int
	err := retryMetaLockConflict(context.Background(), "tenant-business-error", "reserve_upload", "test_op", func() error {
		calls++
		return ErrStorageQuotaExceeded
	})
	if !errors.Is(err, ErrStorageQuotaExceeded) {
		t.Fatalf("err = %v, want ErrStorageQuotaExceeded", err)
	}
	if calls != 1 {
		t.Fatalf("calls = %d, want 1", calls)
	}
}

func TestRetryMetaLockConflictExhaustedReturnsBusy(t *testing.T) {
	var calls int
	lockErr := errors.New("Error 1213 (40001): Deadlock found when trying to get lock; try restarting transaction")
	err := retryMetaLockConflict(context.Background(), "tenant-busy", "reserve_upload", "test_op", func() error {
		calls++
		return lockErr
	})
	if !errors.Is(err, ErrQuotaReservationBusy) {
		t.Fatalf("err = %v, want ErrQuotaReservationBusy", err)
	}
	if !errors.Is(err, lockErr) {
		t.Fatalf("err = %v, want wrapped lock conflict error", err)
	}
	if calls != metaLockConflictRetryAttempts {
		t.Fatalf("calls = %d, want %d", calls, metaLockConflictRetryAttempts)
	}
}

func TestIsMetaLockConflictErrorIncludesTiDBWriteConflict(t *testing.T) {
	tests := []struct {
		name string
		err  error
		want bool
	}{
		{
			name: "tidb write conflict mysql error",
			err:  &mysql.MySQLError{Number: 9007, SQLState: [5]byte{'H', 'Y', '0', '0', '0'}, Message: "Write conflict, txnStartTS=123, conflictStartTS=456"},
			want: true,
		},
		{
			name: "mysql deadlock",
			err:  &mysql.MySQLError{Number: 1213, SQLState: [5]byte{'4', '0', '0', '0', '1'}, Message: "Deadlock found when trying to get lock; try restarting transaction"},
			want: true,
		},
		{
			name: "mysql lock wait timeout",
			err:  &mysql.MySQLError{Number: 1205, SQLState: [5]byte{'H', 'Y', '0', '0', '0'}, Message: "Lock wait timeout exceeded; try restarting transaction"},
			want: true,
		},
		{
			name: "sqlstate serialization failure",
			err:  &mysql.MySQLError{Number: 1105, SQLState: [5]byte{'4', '0', '0', '0', '1'}, Message: "transaction aborted"},
			want: true,
		},
		{
			name: "tidb write conflict db-shaped fallback",
			err:  errors.New("ERROR 9007 (HY000): Write conflict, txnStartTS=123, conflictStartTS=456"),
			want: true,
		},
		{
			name: "unrelated 9007",
			err:  errors.New("ERROR 9007 quota unavailable"),
			want: false,
		},
		{
			name: "tidb mysql error without write conflict",
			err:  &mysql.MySQLError{Number: 9007, SQLState: [5]byte{'H', 'Y', '0', '0', '0'}, Message: "quota unavailable"},
			want: false,
		},
		{
			name: "unrelated 40001 string",
			err:  errors.New("quota request 40001 rejected by policy"),
			want: false,
		},
		{
			name: "unrelated write conflict string",
			err:  errors.New("application write conflict in cache layer"),
			want: false,
		},
		{
			name: "wrapped non-db context with 40001",
			err:  fmt.Errorf("quota admission 40001: %w", errors.New("application rejected write")),
			want: false,
		},
		{
			name: "wrapped non-db context with write conflict",
			err:  fmt.Errorf("quota admission write conflict: %w", errors.New("application rejected write")),
			want: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := isMetaLockConflictError(tt.err); got != tt.want {
				t.Fatalf("isMetaLockConflictError(%v) = %v, want %v", tt.err, got, tt.want)
			}
		})
	}
}

func TestRetryMetaLockConflictDoesNotRetryFalsePositiveLockText(t *testing.T) {
	tests := []struct {
		name string
		err  error
	}{
		{
			name: "unrelated 40001 string",
			err:  errors.New("quota request 40001 rejected by policy"),
		},
		{
			name: "unrelated write conflict string",
			err:  errors.New("application write conflict in cache layer"),
		},
		{
			name: "wrapped non-db context with 40001",
			err:  fmt.Errorf("quota admission 40001: %w", errors.New("application rejected write")),
		},
		{
			name: "wrapped non-db context with write conflict",
			err:  fmt.Errorf("quota admission write conflict: %w", errors.New("application rejected write")),
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var calls int
			err := retryMetaLockConflict(context.Background(), "tenant-false-positive", "reserve_upload", "test_op", func() error {
				calls++
				return tt.err
			})
			if !errors.Is(err, tt.err) {
				t.Fatalf("err = %v, want original error %v", err, tt.err)
			}
			if errors.Is(err, ErrQuotaReservationBusy) {
				t.Fatalf("err = %v, want no busy fail-open sentinel", err)
			}
			if calls != 1 {
				t.Fatalf("calls = %d, want 1", calls)
			}
		})
	}
}

func TestRetryMetaLockConflictHonorsContextDuringBackoff(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	var calls int
	err := retryMetaLockConflict(ctx, "tenant-canceled", "reserve_upload", "test_op", func() error {
		calls++
		cancel()
		return errors.New("SQLSTATE 40001 serialization failure")
	})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("err = %v, want context.Canceled", err)
	}
	if calls != 1 {
		t.Fatalf("calls = %d, want 1", calls)
	}
}