package meta

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

	"github.com/google/uuid"

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

func newControlStore(t *testing.T) *Store {
	t.Helper()
	s, err := Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = s.Close() })
	testtidb.ResetMetaDB(t, s.DB())
	return s
}

func insertSharedTenantPlacementForOrgTest(t *testing.T, s *Store, tenantID, organizationID string) {
	t.Helper()
	t.Cleanup(func() {
		testtidb.ResetMetaDB(t, s.DB())
	})
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID: tenantID, Status: TenantActive, Kind: TenantKindLive,
		Provider: tidbCloudNativeSharedProvider, DBPasswordCipher: []byte{}, DBTLS: true,
		SchemaVersion: 1, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	fsID, err := s.EnsureFsID(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := s.RegisterSharedDB(ctx, &SharedDB{
		TiDBCloudOrganizationID: organizationID, Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_db",
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertTenantPlacement(ctx, &TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: PlacementShared, SchemaShape: SchemaShapeShared,
	}); err != nil {
		t.Fatal(err)
	}
}

func TestMetaDBMetrics(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:               "metrics-meta-tenant",
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db_metrics",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}

	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_db_operations_total{operation="`) || !strings.Contains(text, `role="meta"`) {
		t.Fatalf("expected meta db operation metric in response: %s", text)
	}
	if !strings.Contains(text, `drive9_db_pool_registered{role="meta"}`) {
		t.Fatalf("expected meta db pool metric in response: %s", text)
	}
}

func TestDeleteFailedWriteObjectGCCandidatePreservesReenqueuedLifecycle(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-race-ns",
		StorageRef:     "blobs/gc-race",
		StorageRefHash: "gc-race-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      time.Now().UTC().Add(-time.Minute),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil || !begun {
		t.Fatalf("begin publish: begun=%v err=%v", begun, err)
	}
	input.Reason = ObjectGCReasonOverwrite
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if err := s.DeleteFailedWriteObjectGCCandidate(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil {
		t.Fatal(err)
	}
	candidates, err := s.ListDueObjectGCCandidates(ctx, time.Now().UTC(), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(candidates) != 1 || candidates[0].Reason != ObjectGCReasonOverwrite {
		t.Fatalf("candidates after provisional cleanup = %+v, want one overwrite candidate", candidates)
	}
}

func TestObjectGCCandidatePersistsContentLayoutAndCancelsPendingReplacement(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "append-log-gc-ns",
		StorageRef:     "append-log/file/generation",
		StorageRefHash: "append-log-gc-hash",
		ContentLayout:  "append_log",
		Reason:         ObjectGCReasonOverwrite,
		NotBefore:      time.Now().UTC().Add(-time.Minute),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	candidates, err := s.ListDueObjectGCCandidates(ctx, time.Now().UTC(), 10)
	if err != nil {
		t.Fatal(err)
	}
	var candidate *ObjectGCCandidate
	for i := range candidates {
		if candidates[i].NamespaceID == input.NamespaceID && candidates[i].StorageRefHash == input.StorageRefHash {
			candidate = &candidates[i]
			break
		}
	}
	if candidate == nil || candidate.ContentLayout != "append_log" {
		t.Fatalf("object GC candidates = %+v, want append_log candidate", candidates)
	}
	if err := s.DeletePendingObjectGCCandidate(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef, input.Reason); err != nil {
		t.Fatal(err)
	}
	candidates, err = s.ListDueObjectGCCandidates(ctx, time.Now().UTC(), 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, candidate := range candidates {
		if candidate.NamespaceID == input.NamespaceID && candidate.StorageRefHash == input.StorageRefHash {
			t.Fatalf("pending candidate was not cancelled: %+v", candidate)
		}
	}
}

func TestObjectGCLifecycleEnqueueTakesOverPublishingFailedWrite(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-publishing-takeover-ns",
		StorageRef:     "blobs/gc-publishing-takeover",
		StorageRefHash: "gc-publishing-takeover-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      time.Now().UTC().Add(time.Hour),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil || !begun {
		t.Fatalf("begin publish: begun=%v err=%v", begun, err)
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err == nil {
		t.Fatal("failed-write enqueue replaced an active publishing candidate")
	}

	input.Reason = ObjectGCReasonOverwrite
	input.NotBefore = time.Now().UTC().Add(-time.Minute)
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatalf("enqueue overwrite during failed-write publish: %v", err)
	}
	if err := s.DeleteFailedWriteObjectGCCandidate(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil {
		t.Fatal(err)
	}

	var reason ObjectGCCandidateReason
	var state ObjectGCCandidateState
	if err := s.DB().QueryRowContext(ctx, `SELECT reason, state FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		input.NamespaceID, input.StorageRefHash, input.StorageRef).Scan(&reason, &state); err != nil {
		t.Fatal(err)
	}
	if reason != ObjectGCReasonOverwrite || state != ObjectGCCandidatePending {
		t.Fatalf("candidate after publishing takeover and provisional cleanup: reason=%q state=%q, want overwrite pending", reason, state)
	}
}

func TestForkDeleteTakesOverPublishingFailedWrite(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-fork-publishing-takeover-ns",
		StorageRef:     "blobs/gc-fork-publishing-takeover",
		StorageRefHash: "gc-fork-publishing-takeover-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      time.Now().UTC().Add(time.Hour),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil || !begun {
		t.Fatalf("begin publish: begun=%v err=%v", begun, err)
	}
	input.Reason = ObjectGCReasonForkDelete
	input.NotBefore = time.Now().UTC().Add(24 * time.Hour)
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatalf("enqueue fork delete during failed-write publish: %v", err)
	}

	var reason ObjectGCCandidateReason
	var state ObjectGCCandidateState
	if err := s.DB().QueryRowContext(ctx, `SELECT reason, state FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		input.NamespaceID, input.StorageRefHash, input.StorageRef).Scan(&reason, &state); err != nil {
		t.Fatal(err)
	}
	if reason != ObjectGCReasonForkDelete || state != ObjectGCCandidatePending {
		t.Fatalf("candidate reason=%q state=%q, want fork_delete pending", reason, state)
	}
}

func TestObjectGCCandidateReasonChangeUsesNewGrace(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-grace-ns",
		StorageRef:     "blobs/gc-grace",
		StorageRefHash: "gc-grace-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(time.Hour),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil || !begun {
		t.Fatalf("begin publish: begun=%v err=%v", begun, err)
	}
	input.Reason = ObjectGCReasonOverwrite
	input.NotBefore = now.Add(24 * time.Hour)
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	var reason ObjectGCCandidateReason
	var notBefore time.Time
	if err := s.DB().QueryRowContext(ctx, `SELECT reason, not_before FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		input.NamespaceID, input.StorageRefHash, input.StorageRef).Scan(&reason, &notBefore); err != nil {
		t.Fatal(err)
	}
	if reason != ObjectGCReasonOverwrite {
		t.Fatalf("candidate reason = %q, want overwrite", reason)
	}
	if notBefore.Before(now.Add(23 * time.Hour)) {
		t.Fatalf("promoted candidate not_before = %s, want ordinary grace near 24h", notBefore)
	}
}

func TestFailedWritePublishCannotReclaimPendingRecovery(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-pending-recovery-ns",
		StorageRef:     "blobs/gc-pending-recovery",
		StorageRefHash: "gc-pending-recovery-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      time.Now().UTC().Add(time.Hour),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef)
	if err != nil {
		t.Fatal(err)
	}
	if begun {
		t.Fatal("producer reclaimed a failed-write candidate already released to pending")
	}
}

func TestFailedWriteReenqueueRefreshesOwnership(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-failed-write-refresh-ns",
		StorageRef:     "blobs/gc-failed-write-refresh",
		StorageRefHash: "gc-failed-write-refresh-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Hour),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE object_gc_candidates SET state = 'deleted', deleted_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		now, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil {
		t.Fatal(err)
	}
	input.NotBefore = now.Add(time.Hour)
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	var state ObjectGCCandidateState
	var notBefore time.Time
	var deletedAt *time.Time
	if err := s.DB().QueryRowContext(ctx, `SELECT state, not_before, deleted_at FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		input.NamespaceID, input.StorageRefHash, input.StorageRef).Scan(&state, &notBefore, &deletedAt); err != nil {
		t.Fatal(err)
	}
	if state != ObjectGCCandidatePending {
		t.Fatalf("candidate state = %q, want pending", state)
	}
	if notBefore.Before(now.Add(59 * time.Minute)) {
		t.Fatalf("refreshed candidate not_before = %s, want new 1h ownership", notBefore)
	}
	if deletedAt != nil {
		t.Fatalf("refreshed candidate deleted_at = %s, want NULL", *deletedAt)
	}
}

func TestStaleObjectGCWorkerMutationsDoNotOverwriteRenewedLease(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-stale-worker-ns",
		StorageRef:     "blobs/gc-stale-worker",
		StorageRefHash: "gc-stale-worker-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	var stale ObjectGCCandidate
	for _, candidate := range rows {
		if candidate.NamespaceID == input.NamespaceID && candidate.StorageRef == input.StorageRef {
			stale = candidate
			break
		}
	}
	if stale.StorageRef == "" {
		t.Fatalf("did not list target candidate: %+v", rows)
	}
	renewedNotBefore := now.Add(time.Hour)
	input.NotBefore = renewedNotBefore
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if err := s.PostponeObjectGCCandidateIfCurrent(ctx, stale, now.Add(2*time.Hour), "stale postpone"); err != nil {
		t.Fatal(err)
	}
	if err := s.RetryObjectGCCandidateIfCurrent(ctx, stale, now.Add(3*time.Hour), "stale retry"); err != nil {
		t.Fatal(err)
	}
	var state ObjectGCCandidateState
	var notBefore time.Time
	var attempts int
	var lastError sql.NullString
	if err := s.DB().QueryRowContext(ctx, `SELECT state, not_before, attempts, last_error
		FROM object_gc_candidates WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		input.NamespaceID, input.StorageRefHash, input.StorageRef).Scan(&state, &notBefore, &attempts, &lastError); err != nil {
		t.Fatal(err)
	}
	if state != ObjectGCCandidatePending || attempts != 0 || lastError.Valid {
		t.Fatalf("renewed candidate state=%q attempts=%d last_error=%q", state, attempts, lastError.String)
	}
	if notBefore.Before(renewedNotBefore.Add(-time.Second)) || notBefore.After(renewedNotBefore.Add(time.Second)) {
		t.Fatalf("renewed candidate not_before=%s, want %s", notBefore, renewedNotBefore)
	}
}

func TestObjectGCEnqueueRejectsDeletingClaim(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-deleting-claim-ns",
		StorageRef:     "blobs/gc-deleting-claim",
		StorageRefHash: "gc-deleting-claim-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	var candidate ObjectGCCandidate
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			candidate = row
			break
		}
	}
	if candidate.StorageRef == "" {
		t.Fatalf("did not list target candidate: %+v", rows)
	}
	claimed, err := s.ClaimObjectGCCandidate(ctx, &candidate)
	if err != nil || !claimed {
		t.Fatalf("claim candidate: claimed=%v err=%v", claimed, err)
	}
	input.NotBefore = now.Add(time.Hour)
	if err := s.EnqueueObjectGCCandidate(ctx, input); err == nil {
		t.Fatal("enqueue replaced an active deleting claim")
	}
	var state ObjectGCCandidateState
	var updatedAt time.Time
	if err := s.DB().QueryRowContext(ctx, `SELECT state, updated_at FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ? AND storage_ref = ?`,
		input.NamespaceID, input.StorageRefHash, input.StorageRef).Scan(&state, &updatedAt); err != nil {
		t.Fatal(err)
	}
	if state != ObjectGCCandidateDeleting || !updatedAt.Equal(candidate.UpdatedAt) {
		t.Fatalf("claimed candidate state=%q updated_at=%s, want deleting at %s", state, updatedAt, candidate.UpdatedAt)
	}
}

func TestFailedWritePublishCannotReplaceDeletingClaim(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-publish-claim-ns",
		StorageRef:     "blobs/gc-publish-claim",
		StorageRefHash: "gc-publish-claim-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	var candidate ObjectGCCandidate
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			candidate = row
			break
		}
	}
	claimed, err := s.ClaimObjectGCCandidate(ctx, &candidate)
	if err != nil || !claimed {
		t.Fatalf("claim candidate: claimed=%v err=%v", claimed, err)
	}
	begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef)
	if err != nil {
		t.Fatal(err)
	}
	if begun {
		t.Fatal("producer replaced an active deleting claim")
	}
}

func TestFailedWritePublishRefreshAlwaysAdvancesGeneration(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-publish-generation-ns",
		StorageRef:     "blobs/gc-publish-generation",
		StorageRefHash: "gc-publish-generation-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      time.Now().UTC().Add(-time.Minute),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	var before time.Time
	if err := s.DB().QueryRowContext(ctx, `SELECT updated_at FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ?`, input.NamespaceID, input.StorageRefHash).Scan(&before); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 2; i++ {
		begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef)
		if err != nil || !begun {
			t.Fatalf("begin publish %d: begun=%v err=%v", i, begun, err)
		}
		var after time.Time
		if err := s.DB().QueryRowContext(ctx, `SELECT updated_at FROM object_gc_candidates
			WHERE namespace_id = ? AND storage_ref_hash = ?`, input.NamespaceID, input.StorageRefHash).Scan(&after); err != nil {
			t.Fatal(err)
		}
		if !after.After(before) {
			t.Fatalf("publish refresh %d updated_at=%s, want after %s", i, after, before)
		}
		before = after
	}
}

func TestFailedWritePublishingCandidateIsNotDueForGC(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-publishing-fence-ns",
		StorageRef:     "blobs/gc-publishing-fence",
		StorageRefHash: "gc-publishing-fence-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil || !begun {
		t.Fatalf("begin publish: begun=%v err=%v", begun, err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			t.Fatal("publishing candidate was exposed to object GC")
		}
	}
}

func TestPreparedFailedWriteCandidateFencesS3Mutation(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-s3-mutation-fence-ns",
		StorageRef:     "blobs/gc-s3-mutation-fence",
		StorageRefHash: "gc-s3-mutation-fence-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Hour),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}

	var state ObjectGCCandidateState
	if err := s.DB().QueryRowContext(ctx, `SELECT state FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ?`, input.NamespaceID, input.StorageRefHash).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != ObjectGCCandidatePreparing {
		t.Fatalf("prepared candidate state=%q, want preparing before S3 mutation", state)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			t.Fatal("in-flight S3 mutation was exposed to object GC after its grace elapsed")
		}
	}
}

func TestStalePreparingCandidateBecomesRecoverable(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-stale-preparing-ns",
		StorageRef:     "blobs/gc-stale-preparing",
		StorageRefHash: "gc-stale-preparing-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE object_gc_candidates SET updated_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ?`, now.Add(-25*time.Hour), input.NamespaceID, input.StorageRefHash); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			if row.State != ObjectGCCandidatePreparing {
				t.Fatalf("recoverable candidate state=%q, want preparing", row.State)
			}
			return
		}
	}
	t.Fatal("stale preparing candidate was not exposed for recovery")
}

func TestFailedS3MutationReleasesFailedWriteCandidateForRecovery(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-s3-mutation-recovery-ns",
		StorageRef:     "blobs/gc-s3-mutation-recovery",
		StorageRefHash: "gc-s3-mutation-recovery-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(time.Hour),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if err := s.RetryFailedWriteObjectGCCandidate(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef, input.NotBefore, "S3 outcome requires reconciliation"); err != nil {
		t.Fatal(err)
	}

	rows, err := s.ListDueObjectGCCandidates(ctx, input.NotBefore.Add(time.Second), 100)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			if row.State != ObjectGCCandidatePending || row.LastError == "" {
				t.Fatalf("recoverable candidate=%+v, want pending with failure context", row)
			}
			return
		}
	}
	t.Fatal("failed S3 mutation did not leave recoverable object GC ownership")
}

func TestFailedWriteDeletingCandidateRecoversAfterOneHour(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-deleting-recovery-ns",
		StorageRef:     "blobs/gc-deleting-recovery",
		StorageRefHash: "gc-deleting-recovery-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	var candidate ObjectGCCandidate
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			candidate = row
			break
		}
	}
	claimed, err := s.ClaimObjectGCCandidate(ctx, &candidate)
	if err != nil || !claimed {
		t.Fatalf("claim candidate: claimed=%v err=%v", claimed, err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE object_gc_candidates SET updated_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ?`, now.Add(-2*time.Hour), input.NamespaceID, input.StorageRefHash); err != nil {
		t.Fatal(err)
	}
	rows, err = s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	found := false
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			found = true
			break
		}
	}
	if !found {
		t.Fatal("stale deleting candidate was not recovered after one hour")
	}
}

func TestStalePublishingCandidateIsNeverDueForGC(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-stale-publishing-ns",
		StorageRef:     "blobs/gc-stale-publishing",
		StorageRefHash: "gc-stale-publishing-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.PrepareFailedWriteObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	if begun, err := s.BeginFailedWriteObjectGCCandidatePublish(ctx, input.NamespaceID, input.StorageRefHash, input.StorageRef); err != nil || !begun {
		t.Fatalf("begin publishing: begun=%v err=%v", begun, err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE object_gc_candidates SET updated_at = ?
		WHERE namespace_id = ? AND storage_ref_hash = ?`, now.Add(-25*time.Hour), input.NamespaceID, input.StorageRefHash); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			t.Fatal("stale publishing candidate was exposed to object GC")
		}
	}

	var candidate ObjectGCCandidate
	if err := s.DB().QueryRowContext(ctx, `SELECT namespace_id, storage_ref, storage_ref_hash, reason,
			source_tenant_id, source_file_id, not_before, state, attempts, COALESCE(last_error, ''),
			created_at, updated_at
		FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ?`, input.NamespaceID, input.StorageRefHash).
		Scan(&candidate.NamespaceID, &candidate.StorageRef, &candidate.StorageRefHash, &candidate.Reason,
			&candidate.SourceTenantID, &candidate.SourceFileID, &candidate.NotBefore, &candidate.State,
			&candidate.Attempts, &candidate.LastError, &candidate.CreatedAt, &candidate.UpdatedAt); err != nil {
		t.Fatal(err)
	}
	claimed, err := s.ClaimObjectGCCandidate(ctx, &candidate)
	if err != nil {
		t.Fatal(err)
	}
	if claimed {
		t.Fatal("publishing candidate was claimed for physical deletion")
	}
}

func TestDeleteReachableFailedWriteDeletingSnapshot(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	input := &ObjectGCCandidateInput{
		NamespaceID:    "gc-reachable-deleting-ns",
		StorageRef:     "blobs/gc-reachable-deleting",
		StorageRefHash: "gc-reachable-deleting-hash",
		Reason:         ObjectGCReasonFailedWrite,
		NotBefore:      now.Add(-time.Minute),
	}
	if err := s.EnqueueObjectGCCandidate(ctx, input); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListDueObjectGCCandidates(ctx, now, 100)
	if err != nil {
		t.Fatal(err)
	}
	var candidate ObjectGCCandidate
	for _, row := range rows {
		if row.NamespaceID == input.NamespaceID && row.StorageRef == input.StorageRef {
			candidate = row
			break
		}
	}
	claimed, err := s.ClaimObjectGCCandidate(ctx, &candidate)
	if err != nil || !claimed {
		t.Fatalf("claim candidate: claimed=%v err=%v", claimed, err)
	}
	if err := s.DeleteFailedWriteObjectGCCandidateIfCurrent(ctx, candidate); err != nil {
		t.Fatal(err)
	}
	var count int
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM object_gc_candidates
		WHERE namespace_id = ? AND storage_ref_hash = ?`, input.NamespaceID, input.StorageRefHash).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatalf("candidate count=%d, want 0", count)
	}
}

func TestUpdateTenantDBCredentialIfComparesPreviousUser(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:               "credential-cas-tenant",
		Status:           TenantProvisioning,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "u1.root",
		DBPasswordCipher: []byte("root-cipher"),
		DBName:           "tenant_db",
		DBTLS:            true,
		Provider:         tidbCloudNativeProvider,
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}

	updated, err := s.UpdateTenantDBCredentialIf(context.Background(), "credential-cas-tenant", "other.root", "u1.tdc_fs_sys", []byte("system-cipher"))
	if err != nil {
		t.Fatal(err)
	}
	if updated {
		t.Fatal("credential update succeeded with mismatched previous user")
	}
	var dbUser string
	var passCipher []byte
	if err := s.DB().QueryRow("SELECT db_user, db_password FROM tenants WHERE id = ?", "credential-cas-tenant").Scan(&dbUser, &passCipher); err != nil {
		t.Fatal(err)
	}
	if dbUser != "u1.root" || string(passCipher) != "root-cipher" {
		t.Fatalf("tenant credential changed after failed CAS: user=%q pass=%q", dbUser, passCipher)
	}

	updated, err = s.UpdateTenantDBCredentialIf(context.Background(), "credential-cas-tenant", "u1.root", "u1.tdc_fs_sys", []byte("system-cipher"))
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("credential update did not match previous user")
	}
	if err := s.DB().QueryRow("SELECT db_user, db_password FROM tenants WHERE id = ?", "credential-cas-tenant").Scan(&dbUser, &passCipher); err != nil {
		t.Fatal(err)
	}
	if dbUser != "u1.tdc_fs_sys" || string(passCipher) != "system-cipher" {
		t.Fatalf("tenant credential = %q/%q, want system credential", dbUser, passCipher)
	}
}

func TestInsertAndResolveByAPIKeyHash(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	tenant := &Tenant{
		ID:               "t1",
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}
	if err := s.InsertTenant(context.Background(), tenant); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertTenantTiDBCloudOrgBinding(context.Background(), &TenantTiDBCloudOrgBinding{
		TenantID:       tenant.ID,
		OrganizationID: "org-resolved",
		ClusterID:      "cluster-resolved",
		CreatedAt:      now,
		UpdatedAt:      now,
	}); err != nil {
		t.Fatal(err)
	}
	key := &APIKey{
		ID:            "k1",
		TenantID:      tenant.ID,
		KeyName:       "default",
		JWTCiphertext: []byte("jwt-cipher"),
		JWTHash:       "hash1",
		TokenVersion:  1,
		Status:        APIKeyActive,
		IssuedAt:      now,
		CreatedAt:     now,
		UpdatedAt:     now,
	}
	if err := s.InsertAPIKey(context.Background(), key); err != nil {
		t.Fatal(err)
	}

	got, err := s.ResolveByAPIKeyHash(context.Background(), "hash1")
	if err != nil {
		t.Fatal(err)
	}
	if got.Tenant.ID != "t1" || got.APIKey.ID != "k1" {
		t.Fatalf("unexpected resolve result: tenant=%s key=%s", got.Tenant.ID, got.APIKey.ID)
	}
	if got.Tenant.Status != TenantActive {
		t.Fatalf("unexpected tenant status: %s", got.Tenant.Status)
	}
	if got.TiDBCloudOrgID != "org-resolved" {
		t.Fatalf("resolved org = %q, want org-resolved", got.TiDBCloudOrgID)
	}
	if got.Tenant.TiDBCloudOrgID != "org-resolved" {
		t.Fatalf("tenant org = %q, want org-resolved", got.Tenant.TiDBCloudOrgID)
	}
	if got.Tenant.S3EncryptionMode != S3EncryptionModeInherit {
		t.Fatalf("unexpected tenant encryption mode: %s", got.Tenant.S3EncryptionMode)
	}
	if got.Tenant.S3KMSKeyID != "" {
		t.Fatalf("unexpected tenant kms key: %q", got.Tenant.S3KMSKeyID)
	}
	if !got.Tenant.S3BucketKeyEnabledValue() {
		t.Fatal("tenant bucket key enabled = false, want true")
	}
	if got.APIKey.Status != APIKeyActive {
		t.Fatalf("unexpected key status: %s", got.APIKey.Status)
	}
	if got.APIKey.ScopeKind != APIKeyScopeKindOwner {
		t.Fatalf("unexpected key scope kind: %s", got.APIKey.ScopeKind)
	}

	badKey := *key
	badKey.ID = "bad-scope-kind"
	badKey.KeyName = "bad-scope-kind"
	badKey.JWTHash = "bad-scope-kind-hash"
	badKey.ScopeKind = APIKeyScopeKind("unknown")
	if err := s.InsertAPIKey(context.Background(), &badKey); err == nil {
		t.Fatal("InsertAPIKey with unknown scope kind error = nil, want error")
	}

	if err := s.RevokeAPIKey(context.Background(), tenant.ID, key.ID); err != nil {
		t.Fatalf("RevokeAPIKey active key error = %v, want nil", err)
	}
	revoked, err := s.ResolveByAPIKeyHash(context.Background(), "hash1")
	if err != nil {
		t.Fatal(err)
	}
	if revoked.APIKey.Status != APIKeyRevoked || revoked.APIKey.RevokedAt == nil {
		t.Fatalf("revoked API key = status %s revoked_at %v, want revoked timestamp", revoked.APIKey.Status, revoked.APIKey.RevokedAt)
	}
	if err := s.RevokeAPIKey(context.Background(), tenant.ID, key.ID); err != ErrNotFound {
		t.Fatalf("RevokeAPIKey already revoked error = %v, want ErrNotFound", err)
	}
	if err := s.RevokeAPIKey(context.Background(), "wrong-tenant", key.ID); err != ErrNotFound {
		t.Fatalf("RevokeAPIKey wrong tenant error = %v, want ErrNotFound", err)
	}

	key2 := &APIKey{
		ID:                 "k2",
		TenantID:           tenant.ID,
		KeyName:            "k2",
		JWTCiphertext:      []byte("jwt2"),
		JWTHash:            "hash2",
		TokenVersion:       2,
		Status:             APIKeyActive,
		ScopeKind:          APIKeyScopeKindOwner,
		IssuedByProvider:   "slock",
		IssuedBySubjectKey: "subject-1",
		IssuedAt:           now,
		CreatedAt:          now,
		UpdatedAt:          now,
	}
	if err := s.InsertAPIKey(context.Background(), key2); err != nil {
		t.Fatal(err)
	}
	key3 := *key2
	key3.ID = "k3"
	key3.KeyName = "k3"
	key3.JWTHash = "hash3"
	if err := s.InsertAPIKey(context.Background(), &key3); err != nil {
		t.Fatal(err)
	}
	if err := s.RevokeAPIKeysByIssuer(context.Background(), tenant.ID, "slock", "subject-1", key3.ID); err != nil {
		t.Fatalf("RevokeAPIKeysByIssuer error = %v, want nil", err)
	}
	issuerRevoked, err := s.ResolveByAPIKeyHash(context.Background(), "hash2")
	if err != nil {
		t.Fatal(err)
	}
	if issuerRevoked.APIKey.Status != APIKeyRevoked {
		t.Fatalf("issuer revoked key status = %s, want %s", issuerRevoked.APIKey.Status, APIKeyRevoked)
	}
	issuerKept, err := s.ResolveByAPIKeyHash(context.Background(), "hash3")
	if err != nil {
		t.Fatal(err)
	}
	if issuerKept.APIKey.Status != APIKeyActive {
		t.Fatalf("issuer kept key status = %s, want %s", issuerKept.APIKey.Status, APIKeyActive)
	}
}

func TestResolveByAPIKeyHashUsesSharedDBOrganization(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	tenantID := "shared-api-key-org"
	insertSharedTenantPlacementForOrgTest(t, s, tenantID, "org-shared-api-key")
	if err := s.InsertAPIKey(ctx, &APIKey{
		ID: "shared-api-key", TenantID: tenantID, KeyName: "default",
		JWTCiphertext: []byte("jwt-cipher"), JWTHash: "shared-api-key-hash", TokenVersion: 1,
		Status: APIKeyActive, ScopeKind: APIKeyScopeKindOwner,
		IssuedAt: now, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}

	got, err := s.ResolveByAPIKeyHash(ctx, "shared-api-key-hash")
	if err != nil {
		t.Fatal(err)
	}
	if got.TiDBCloudOrgID != "org-shared-api-key" {
		t.Fatalf("resolved org = %q, want org-shared-api-key", got.TiDBCloudOrgID)
	}
	if got.Tenant.TiDBCloudOrgID != "org-shared-api-key" {
		t.Fatalf("tenant org = %q, want org-shared-api-key", got.Tenant.TiDBCloudOrgID)
	}
}

func TestInsertAndGetExternalBinding(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:               "binding-tenant",
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db_binding",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}
	metadata := []byte(`{"server_id":"srv","sub":"sub"}`)
	if err := s.InsertExternalBinding(context.Background(), &ExternalBinding{
		Provider:     "slock",
		SubjectKey:   "srv:sub",
		TenantID:     "binding-tenant",
		MetadataJSON: metadata,
		CreatedAt:    now,
		UpdatedAt:    now,
	}); err != nil {
		t.Fatalf("InsertExternalBinding: %v", err)
	}
	got, err := s.GetExternalBinding(context.Background(), "slock", "srv:sub")
	if err != nil {
		t.Fatalf("GetExternalBinding: %v", err)
	}
	if got.Provider != "slock" || got.SubjectKey != "srv:sub" || got.TenantID != "binding-tenant" ||
		!strings.Contains(string(got.MetadataJSON), `"server_id": "srv"`) ||
		!strings.Contains(string(got.MetadataJSON), `"sub": "sub"`) {
		t.Fatalf("binding = %+v metadata=%s", got, string(got.MetadataJSON))
	}
	if err := s.InsertExternalBinding(context.Background(), got); !errors.Is(err, ErrDuplicate) {
		t.Fatalf("duplicate InsertExternalBinding error = %v, want ErrDuplicate", err)
	}
	if _, err := s.GetExternalBinding(context.Background(), "slock", "missing"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("missing GetExternalBinding error = %v, want ErrNotFound", err)
	}
}

func TestInsertAPIKeyStoresIssuedByMetadata(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	tenant := &Tenant{
		ID:               "issued-by-tenant",
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db_issued_by",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}
	if err := s.InsertTenant(context.Background(), tenant); err != nil {
		t.Fatal(err)
	}
	if err := s.InsertAPIKey(context.Background(), &APIKey{
		ID:                   "issued-key",
		TenantID:             tenant.ID,
		KeyName:              "slock",
		JWTCiphertext:        []byte("cipher"),
		JWTHash:              "hash-issued",
		TokenVersion:         1,
		Status:               APIKeyActive,
		ScopeKind:            APIKeyScopeKindOwner,
		IssuedByProvider:     "slock",
		IssuedBySubjectKey:   "srv:sub",
		IssuedByMetadataJSON: []byte(`{"type":"agent"}`),
		IssuedAt:             now,
		CreatedAt:            now,
		UpdatedAt:            now,
	}); err != nil {
		t.Fatalf("InsertAPIKey: %v", err)
	}
	got, err := s.ResolveByAPIKeyHash(context.Background(), "hash-issued")
	if err != nil {
		t.Fatalf("ResolveByAPIKeyHash: %v", err)
	}
	if got.APIKey.IssuedByProvider != "slock" || got.APIKey.IssuedBySubjectKey != "srv:sub" ||
		!strings.Contains(string(got.APIKey.IssuedByMetadataJSON), `"type": "agent"`) {
		t.Fatalf("issued-by metadata not round-tripped: %+v metadata=%s", got.APIKey, string(got.APIKey.IssuedByMetadataJSON))
	}
}

func TestWithExternalBindingLockSerializesCallbacks(t *testing.T) {
	s := newControlStore(t)
	var running int32
	var maxRunning int32
	var wg sync.WaitGroup
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			err := s.WithExternalBindingLock(context.Background(), "slock", "srv:sub", func(context.Context) error {
				n := atomic.AddInt32(&running, 1)
				for {
					max := atomic.LoadInt32(&maxRunning)
					if n <= max || atomic.CompareAndSwapInt32(&maxRunning, max, n) {
						break
					}
				}
				time.Sleep(20 * time.Millisecond)
				atomic.AddInt32(&running, -1)
				return nil
			})
			if err != nil {
				t.Errorf("WithExternalBindingLock: %v", err)
			}
		}()
	}
	wg.Wait()
	if maxRunning != 1 {
		t.Fatalf("max concurrent lock holders = %d, want 1", maxRunning)
	}
}

func TestWithTenantPoolLockSerializesCallbacks(t *testing.T) {
	s := newControlStore(t)
	var running int32
	var maxRunning int32
	var wg sync.WaitGroup
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			err := s.WithTenantPoolLock(context.Background(), "pool-lock-test", func(context.Context) error {
				n := atomic.AddInt32(&running, 1)
				for {
					max := atomic.LoadInt32(&maxRunning)
					if n <= max || atomic.CompareAndSwapInt32(&maxRunning, max, n) {
						break
					}
				}
				time.Sleep(20 * time.Millisecond)
				atomic.AddInt32(&running, -1)
				return nil
			})
			if err != nil {
				t.Errorf("WithTenantPoolLock: %v", err)
			}
		}()
	}
	wg.Wait()
	if maxRunning != 1 {
		t.Fatalf("max concurrent lock holders = %d, want 1", maxRunning)
	}
}

func TestTenantPoolDatabaseLockNameStaysWithinMySQLLimit(t *testing.T) {
	base := tenantPoolLockName("pool-lock-test")
	got := tenantPoolDatabaseLockName(base, "drive9_test_with_a_long_database_name")
	if len(got) >= 64 {
		t.Fatalf("lock name length = %d, want below 64", len(got))
	}
	if !strings.HasPrefix(got, base+":") {
		t.Fatalf("lock name = %q, want base prefix %q", got, base+":")
	}
	other := tenantPoolDatabaseLockName(base, "drive9_test_other")
	if got == other {
		t.Fatalf("database-scoped lock names should differ: %q", got)
	}
}

func TestInsertAPIKeyAllowsRepeatedKeyName(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	tenant := &Tenant{
		ID:               "repeat-key-name-tenant",
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}
	if err := s.InsertTenant(context.Background(), tenant); err != nil {
		t.Fatal(err)
	}
	for _, id := range []string{"k-repeat-a", "k-repeat-b"} {
		if err := s.InsertAPIKey(context.Background(), &APIKey{
			ID:            id,
			TenantID:      tenant.ID,
			KeyName:       "same-audit-label",
			JWTCiphertext: []byte("jwt-cipher-" + id),
			JWTHash:       "hash-" + id,
			TokenVersion:  1,
			Status:        APIKeyActive,
			IssuedAt:      now,
			CreatedAt:     now,
			UpdatedAt:     now,
		}); err != nil {
			t.Fatalf("InsertAPIKey %s: %v", id, err)
		}
	}
}

func TestInsertAndListAPIKeyFSScopes(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:               "scope-tenant",
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db_scopes",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:               "other-tenant",
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db_other_scopes",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}
	exp := now.Add(2 * time.Hour)
	if err := s.InsertAPIKey(context.Background(), &APIKey{
		ID:            "scope-key",
		TenantID:      "scope-tenant",
		KeyName:       "scoped",
		JWTCiphertext: []byte("jwt-cipher"),
		JWTHash:       "scope-hash",
		TokenVersion:  1,
		Status:        APIKeyActive,
		ScopeKind:     APIKeyScopeKindFS,
		IssuedAt:      now,
		ExpiresAt:     &exp,
		CreatedAt:     now,
		UpdatedAt:     now,
	}); err != nil {
		t.Fatal(err)
	}
	resolved, err := s.ResolveByAPIKeyHash(context.Background(), "scope-hash")
	if err != nil {
		t.Fatal(err)
	}
	if resolved.APIKey.ScopeKind != APIKeyScopeKindFS {
		t.Fatalf("resolved scope kind = %s, want %s", resolved.APIKey.ScopeKind, APIKeyScopeKindFS)
	}
	if err := s.InsertAPIKey(context.Background(), &APIKey{
		ID:            "other-key",
		TenantID:      "scope-tenant",
		KeyName:       "other-scoped",
		JWTCiphertext: []byte("jwt-cipher"),
		JWTHash:       "other-scope-hash",
		TokenVersion:  1,
		Status:        APIKeyActive,
		ScopeKind:     APIKeyScopeKindFS,
		IssuedAt:      now,
		ExpiresAt:     &exp,
		CreatedAt:     now,
		UpdatedAt:     now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.InsertAPIKey(context.Background(), &APIKey{
		ID:            "other-tenant-key",
		TenantID:      "other-tenant",
		KeyName:       "scoped",
		JWTCiphertext: []byte("jwt-cipher"),
		JWTHash:       "other-tenant-scope-hash",
		TokenVersion:  1,
		Status:        APIKeyActive,
		ScopeKind:     APIKeyScopeKindFS,
		IssuedAt:      now,
		ExpiresAt:     &exp,
		CreatedAt:     now,
		UpdatedAt:     now,
	}); err != nil {
		t.Fatal(err)
	}

	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "scope-tenant",
		APIKeyID: "scope-key",
		Prefix:   "/scratch/run-1",
		Ops:      "read,list",
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "scope-tenant",
		APIKeyID: "other-key",
		Prefix:   "/wrong-key",
		Ops:      "read",
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "other-tenant",
		APIKeyID: "scope-key",
		Prefix:   "/wrong-tenant",
		Ops:      "read",
	}); err != nil {
		t.Fatal(err)
	}

	got, err := s.ListAPIKeyFSScopes(context.Background(), "scope-tenant", "scope-key")
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 1 {
		t.Fatalf("scope count = %d, want 1: %#v", len(got), got)
	}
	if got[0].TenantID != "scope-tenant" || got[0].APIKeyID != "scope-key" || got[0].Prefix != "/scratch/run-1" || got[0].Ops != "read,list" {
		t.Fatalf("unexpected scope row: %#v", got[0])
	}

	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "scope-tenant",
		APIKeyID: "scope-key",
		Prefix:   "",
		Ops:      "read",
	}); err == nil {
		t.Fatal("InsertAPIKeyFSScope with empty prefix error = nil, want error")
	}
	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "scope-tenant",
		APIKeyID: "scope-key",
		Prefix:   ":",
		Ops:      "read",
	}); err == nil {
		t.Fatal("InsertAPIKeyFSScope with bare colon prefix error = nil, want error")
	}
	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "scope-tenant",
		APIKeyID: "scope-key",
		Prefix:   ":/",
		Ops:      "read",
	}); err != nil {
		t.Fatalf("InsertAPIKeyFSScope with explicit root prefix error = %v, want nil", err)
	}
	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "scope-tenant",
		APIKeyID: "scope-key",
		Prefix:   "/pseudo-root",
		Ops:      "pseudoroot",
	}); err != nil {
		t.Fatalf("InsertAPIKeyFSScope with pseudoroot error = %v, want nil", err)
	}
	if err := s.InsertAPIKeyFSScope(context.Background(), &APIKeyFSScope{
		TenantID: "scope-tenant",
		APIKeyID: "scope-key",
		Prefix:   "/bad-search",
		Ops:      "search",
	}); err == nil {
		t.Fatal("InsertAPIKeyFSScope with search-only ops error = nil, want error")
	}
}

func TestValidateFSScopeOpsAcceptsPseudoRootDirective(t *testing.T) {
	if err := validateFSScopeOps("pseudoroot"); err != nil {
		t.Fatalf("validateFSScopeOps(pseudoroot) error = %v", err)
	}
	if err := validateFSScopeOps("pseudoroot,search"); err == nil {
		t.Fatal("validateFSScopeOps(pseudoroot,search) error = nil, want search-without-read error")
	}
}

func TestGetTenantReadsS3EncryptionPolicy(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:                 "tenant-s3-policy",
		Status:             TenantActive,
		DBHost:             "127.0.0.1",
		DBPort:             4000,
		DBUser:             "root",
		DBPasswordCipher:   []byte("cipher"),
		DBName:             "tenant_db_s3_policy",
		DBTLS:              true,
		Provider:           "tidb_zero",
		SchemaVersion:      1,
		S3EncryptionMode:   S3EncryptionModeSSEKMS,
		S3BucketKeyEnabled: boolPtr(false),
		CreatedAt:          now,
		UpdatedAt:          now,
	}); err != nil {
		t.Fatal(err)
	}

	got, err := s.GetTenant(context.Background(), "tenant-s3-policy")
	if err != nil {
		t.Fatal(err)
	}
	if got.S3EncryptionMode != S3EncryptionModeSSEKMS {
		t.Fatalf("S3EncryptionMode = %q, want sse-kms", got.S3EncryptionMode)
	}
	if got.S3KMSKeyID != "" {
		t.Fatalf("S3KMSKeyID = %q, want empty", got.S3KMSKeyID)
	}
	if got.S3BucketKeyEnabledValue() {
		t.Fatal("S3BucketKeyEnabled = true, want false")
	}
}

func TestInsertTenantPreservesExplicitBucketKeyFalseWithEmptyMode(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:                 "tenant-explicit-bucket-key-false",
		Status:             TenantActive,
		DBHost:             "127.0.0.1",
		DBPort:             4000,
		DBUser:             "root",
		DBPasswordCipher:   []byte("cipher"),
		DBName:             "tenant_db_explicit_bucket_false",
		DBTLS:              true,
		Provider:           "tidb_zero",
		SchemaVersion:      1,
		S3BucketKeyEnabled: boolPtr(false),
		CreatedAt:          now,
		UpdatedAt:          now,
	}); err != nil {
		t.Fatal(err)
	}

	got, err := s.GetTenant(context.Background(), "tenant-explicit-bucket-key-false")
	if err != nil {
		t.Fatal(err)
	}
	if got.S3EncryptionMode != S3EncryptionModeInherit {
		t.Fatalf("S3EncryptionMode = %q, want inherit", got.S3EncryptionMode)
	}
	if got.S3BucketKeyEnabledValue() {
		t.Fatal("S3BucketKeyEnabled = true, want false")
	}
}

func TestUpdateTenantStatus(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:               "t2",
		Status:           TenantProvisioning,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_db2",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpdateTenantStatus(context.Background(), "t2", TenantSuspended); err != nil {
		t.Fatal(err)
	}

	row := s.DB().QueryRow(`SELECT status FROM tenants WHERE id = ?`, "t2")
	var status string
	if err := row.Scan(&status); err != nil {
		t.Fatal(err)
	}
	if status != string(TenantSuspended) {
		t.Fatalf("status=%s", status)
	}
}

func TestUpdateTenantStatusDeletingWritesMetricsCleanupOutbox(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	const tenantID = "tenant-status-cleanup-outbox"
	if err := s.InsertTenant(ctx, &Tenant{
		ID:               tenantID,
		Status:           TenantActive,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_status_cleanup_outbox",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}

	if err := s.UpdateTenantStatus(ctx, tenantID, TenantDeleting); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListTenantNotifySince(ctx, 0, 100)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.TenantID == tenantID && row.WorkMask&TenantNotifyWorkMetricsCleanup != 0 {
			return
		}
	}
	t.Fatalf("missing durable metrics-cleanup outbox row for tenant %s: %+v", tenantID, rows)
}

func TestFinalizeTenantDeleteUpdatesJobNamespaceAndTenant(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID:                 "delete-finalize-tenant",
		Status:             TenantDeleting,
		StorageNamespaceID: "delete-finalize-ns",
		DBHost:             "127.0.0.1",
		DBPort:             4000,
		DBUser:             "root",
		DBPasswordCipher:   []byte("cipher"),
		DBName:             "tenant_delete_finalize",
		DBTLS:              true,
		Provider:           tidbCloudNativeProvider,
		SchemaVersion:      1,
		CreatedAt:          now,
		UpdatedAt:          now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       "delete-finalize-tenant",
		OrganizationID: "org-delete-finalize",
		ClusterID:      "cluster-delete-finalize",
		CreatedAt:      now,
		UpdatedAt:      now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID:            "delete-finalize-ns",
		OwnerTenantID: "delete-finalize-tenant",
		Backend:       "s3",
		Bucket:        "bucket",
		Prefix:        "tenant/delete-finalize-tenant",
		State:         StorageNamespaceDeleting,
		CreatedAt:     now,
		UpdatedAt:     now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID:    "delete-finalize-tenant",
		NamespaceID: "delete-finalize-ns",
		Backend:     "s3",
		Bucket:      "bucket",
		Prefix:      "tenant/delete-finalize-tenant",
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeleteJobRunning(ctx, "delete-finalize-tenant")
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("MarkTenantDeleteJobRunning updated = false, want true")
	}
	dbID, err := s.RegisterSharedDB(ctx, &SharedDB{
		TiDBCloudOrganizationID: "org-delete-finalize", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_delete_finalize",
	})
	if err != nil {
		t.Fatal(err)
	}
	fsID, err := s.EnsureFsID(ctx, "delete-finalize-tenant")
	if err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertTenantPlacement(ctx, &TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: PlacementShared, SchemaShape: SchemaShapeShared,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, "UPDATE tenant_placements SET status = ? WHERE fs_id = ?", PlacementStatusDeleting, fsID); err != nil {
		t.Fatal(err)
	}
	seedTenantExtractConfig(t, s, "delete-finalize-tenant")
	if err := s.FinalizeTenantDelete(ctx, "delete-finalize-tenant", "delete-finalize-ns", 11, 2); err != nil {
		t.Fatal(err)
	}

	var jobState, nsState, tenantStatus string
	var deletedObjects, abortedMultipartUploads int64
	if err := s.DB().QueryRow(`SELECT state, deleted_objects, aborted_multipart_uploads FROM tenant_delete_jobs WHERE tenant_id = ?`, "delete-finalize-tenant").Scan(&jobState, &deletedObjects, &abortedMultipartUploads); err != nil {
		t.Fatal(err)
	}
	if jobState != string(TenantDeleteJobDeleted) || deletedObjects != 11 || abortedMultipartUploads != 2 {
		t.Fatalf("job = state %s deleted %d aborted %d, want deleted/11/2", jobState, deletedObjects, abortedMultipartUploads)
	}
	if err := s.DB().QueryRow(`SELECT state FROM storage_namespaces WHERE namespace_id = ?`, "delete-finalize-ns").Scan(&nsState); err != nil {
		t.Fatal(err)
	}
	if nsState != string(StorageNamespaceDeleted) {
		t.Fatalf("namespace state = %s, want %s", nsState, StorageNamespaceDeleted)
	}
	if err := s.DB().QueryRow(`SELECT status FROM tenants WHERE id = ?`, "delete-finalize-tenant").Scan(&tenantStatus); err != nil {
		t.Fatal(err)
	}
	if tenantStatus != string(TenantDeleted) {
		t.Fatalf("tenant status = %s, want %s", tenantStatus, TenantDeleted)
	}
	if _, err := s.GetTenantTiDBCloudOrgBinding(ctx, "delete-finalize-tenant"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("org binding err = %v, want %v", err, ErrNotFound)
	}
	if _, err := s.GetTenantPlacement(ctx, fsID); !errors.Is(err, ErrNotFound) {
		t.Fatalf("deleting placement after finalization = %v, want %v", err, ErrNotFound)
	}
	assertTenantExtractConfigWiped(t, s, "delete-finalize-tenant")
}

func TestMarkTenantDeletedWipesExtractConfigs(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	const tenantID = "mark-deleted-extract"
	if err := s.InsertTenant(ctx, &Tenant{
		ID:               tenantID,
		Status:           TenantDeleting,
		DBHost:           "127.0.0.1",
		DBPort:           4000,
		DBUser:           "root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tenant_mark_deleted_extract",
		DBTLS:            true,
		Provider:         "tidb_zero",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}
	seedTenantExtractConfig(t, s, tenantID)
	if err := s.MarkTenantDeleted(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	assertTenantExtractConfigWiped(t, s, tenantID)
}

func TestListTenantNotifySinceIncludesTiDBCloudOrgBinding(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	tenantID := "notify-binding-tenant"
	if err := s.InsertTenant(ctx, &Tenant{
		ID:                 tenantID,
		Status:             TenantActive,
		StorageNamespaceID: "notify-binding-ns",
		DBHost:             "127.0.0.1",
		DBPort:             4000,
		DBUser:             "root",
		DBPasswordCipher:   []byte("cipher"),
		DBName:             "tenant_notify_binding",
		DBTLS:              true,
		Provider:           tidbCloudNativeProvider,
		SchemaVersion:      1,
		CreatedAt:          now,
		UpdatedAt:          now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       tenantID,
		OrganizationID: "org-notify",
		ClusterID:      "cluster-notify",
		CreatedAt:      now,
		UpdatedAt:      now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.InsertTenantNotify(ctx, tenantID, 3); err != nil {
		t.Fatal(err)
	}

	rows, err := s.ListTenantNotifySince(ctx, 0, 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.TenantID != tenantID {
			continue
		}
		if row.TiDBCloudOrgID != "org-notify" {
			t.Fatalf("notify row org = %q, want org-notify", row.TiDBCloudOrgID)
		}
		return
	}
	t.Fatalf("missing notify row for tenant %s: %+v", tenantID, rows)
}

func TestListTenantNotifySinceUsesSharedDBOrganization(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DELETE FROM tenant_notify_outbox`); err != nil {
		t.Fatal(err)
	}
	tenantID := "notify-shared-tenant"
	insertSharedTenantPlacementForOrgTest(t, s, tenantID, "org-notify-shared")
	if err := s.InsertTenantNotify(ctx, tenantID, 3); err != nil {
		t.Fatal(err)
	}

	rows, err := s.ListTenantNotifySince(ctx, 0, 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 {
		t.Fatalf("notify row count = %d, want 1", len(rows))
	}
	if rows[0].TiDBCloudOrgID != "org-notify-shared" {
		t.Fatalf("notify row org = %q, want org-notify-shared", rows[0].TiDBCloudOrgID)
	}
}

func TestListPendingMutationsUsesSharedDBOrganization(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DELETE FROM quota_mutation_log`); err != nil {
		t.Fatal(err)
	}
	tenantID := "mutation-shared-tenant"
	insertSharedTenantPlacementForOrgTest(t, s, tenantID, "org-mutation-shared")
	id, err := s.InsertMutationLog(ctx, &MutationLogEntry{
		TenantID: tenantID, MutationType: "file_create", MutationData: []byte(`{"file_id":"f1"}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE quota_mutation_log SET created_at = ? WHERE id = ?`, time.Now().UTC().Add(-time.Minute), id); err != nil {
		t.Fatal(err)
	}

	rows, err := s.ListPendingMutations(ctx, 0, 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 {
		t.Fatalf("mutation row count = %d, want 1", len(rows))
	}
	if rows[0].TiDBCloudOrgID != "org-mutation-shared" {
		t.Fatalf("mutation row org = %q, want org-mutation-shared", rows[0].TiDBCloudOrgID)
	}
}

func TestObservePendingMutationsUsesSharedDBOrganization(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DELETE FROM quota_mutation_log`); err != nil {
		t.Fatal(err)
	}
	tenantID := "mutation-observe-shared-tenant"
	insertSharedTenantPlacementForOrgTest(t, s, tenantID, "org-mutation-observe-shared")
	oldest := time.Now().UTC().Add(-5 * time.Minute)
	firstID, err := s.InsertMutationLog(ctx, &MutationLogEntry{
		TenantID: tenantID, MutationType: "file_delete", MutationData: []byte(`{"file_id":"f1"}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.InsertMutationLog(ctx, &MutationLogEntry{
		TenantID: tenantID, MutationType: "file_delete", MutationData: []byte(`{"file_id":"f2"}`),
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE quota_mutation_log SET created_at = ? WHERE id = ?`, oldest, firstID); err != nil {
		t.Fatal(err)
	}

	rows, err := s.ObservePendingMutations(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range rows {
		if row.TenantID == tenantID {
			if row.TiDBCloudOrgID != "org-mutation-observe-shared" {
				t.Fatalf("mutation observation org = %q, want org-mutation-observe-shared", row.TiDBCloudOrgID)
			}
			if row.PendingCount != 2 {
				t.Fatalf("mutation observation pending count = %d, want 2", row.PendingCount)
			}
			if row.OldestPendingAgeSeconds < 4*60 || row.OldestPendingAgeSeconds > 6*60 {
				t.Fatalf("mutation observation oldest age = %.3fs, want about 5m", row.OldestPendingAgeSeconds)
			}
			return
		}
	}
	t.Fatalf("missing mutation observation for tenant %q", tenantID)
}

func TestListTenantsByStatus(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	for _, tc := range []struct {
		id     string
		status TenantStatus
	}{
		{id: "tp1", status: TenantProvisioning},
		{id: "tp2", status: TenantProvisioning},
		{id: "ta1", status: TenantActive},
	} {
		if err := s.InsertTenant(context.Background(), &Tenant{
			ID:               tc.id,
			Status:           tc.status,
			DBHost:           "127.0.0.1",
			DBPort:           4000,
			DBUser:           "root",
			DBPasswordCipher: []byte("cipher"),
			DBName:           "tenant_db",
			DBTLS:            true,
			Provider:         "tidb_zero",
			SchemaVersion:    1,
			CreatedAt:        now,
			UpdatedAt:        now,
		}); err != nil {
			t.Fatal(err)
		}
	}

	got, err := s.ListTenantsByStatus(context.Background(), TenantProvisioning, 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 2 {
		t.Fatalf("expected 2 provisioning tenants, got %d", len(got))
	}
	if got[0].Status != TenantProvisioning || got[1].Status != TenantProvisioning {
		t.Fatalf("unexpected statuses: %s, %s", got[0].Status, got[1].Status)
	}
}

func TestCountTenants(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC()
	for _, tc := range []struct {
		id     string
		status TenantStatus
	}{
		{id: "count-active", status: TenantActive},
		{id: "count-provisioning", status: TenantProvisioning},
		{id: "count-failed", status: TenantFailed},
		{id: "count-deleted", status: TenantDeleted},
	} {
		if err := s.InsertTenant(context.Background(), &Tenant{
			ID:               tc.id,
			Status:           tc.status,
			DBHost:           "127.0.0.1",
			DBPort:           4000,
			DBUser:           "root",
			DBPasswordCipher: []byte("cipher"),
			DBName:           "tenant_db",
			DBTLS:            true,
			Provider:         "tidb_zero",
			SchemaVersion:    1,
			CreatedAt:        now,
			UpdatedAt:        now,
		}); err != nil {
			t.Fatal(err)
		}
	}

	got, err := s.CountTenants(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	want := map[TenantStatus]int64{
		TenantPending:      0,
		TenantProvisioning: 1,
		TenantActive:       1,
		TenantFailed:       1,
		TenantSuspended:    0,
		TenantDeleting:     0,
		TenantDeleted:      1,
	}
	if len(got.Statuses) != len(want) {
		t.Errorf("status count length = %d, want %d: %+v", len(got.Statuses), len(want), got.Statuses)
	}
	for status, wantCount := range want {
		if got.Count(status) != wantCount {
			t.Errorf("count[%s] = %d, want %d", status, got.Count(status), wantCount)
		}
	}
}

func TestListTenantsByStatusAfterKeyset(t *testing.T) {
	s := newControlStore(t)
	now := time.Now().UTC().Truncate(time.Millisecond)
	for _, tc := range []struct {
		id        string
		status    TenantStatus
		createdAt time.Time
	}{
		{id: "active-a", status: TenantActive, createdAt: now},
		{id: "active-b", status: TenantActive, createdAt: now},
		{id: "active-c", status: TenantActive, createdAt: now.Add(time.Second)},
		{id: "provisioning-a", status: TenantProvisioning, createdAt: now},
	} {
		if err := s.InsertTenant(context.Background(), &Tenant{
			ID:               tc.id,
			Status:           tc.status,
			DBHost:           "127.0.0.1",
			DBPort:           4000,
			DBUser:           "root",
			DBPasswordCipher: []byte("cipher"),
			DBName:           "tenant_db",
			DBTLS:            true,
			Provider:         "tidb_zero",
			SchemaVersion:    1,
			CreatedAt:        tc.createdAt,
			UpdatedAt:        tc.createdAt,
		}); err != nil {
			t.Fatal(err)
		}
	}

	first, err := s.ListTenantsByStatusAfter(context.Background(), TenantActive, time.Time{}, "", 2)
	if err != nil {
		t.Fatal(err)
	}
	if len(first) != 2 || first[0].ID != "active-a" || first[1].ID != "active-b" {
		t.Fatalf("first page ids = %v, want active-a, active-b", tenantIDs(first))
	}

	second, err := s.ListTenantsByStatusAfter(context.Background(), TenantActive, first[1].CreatedAt, first[1].ID, 2)
	if err != nil {
		t.Fatal(err)
	}
	if len(second) != 1 || second[0].ID != "active-c" {
		t.Fatalf("second page ids = %v, want active-c", tenantIDs(second))
	}

	empty, err := s.ListTenantsByStatusAfter(context.Background(), TenantActive, second[0].CreatedAt, second[0].ID, 2)
	if err != nil {
		t.Fatal(err)
	}
	if len(empty) != 0 {
		t.Fatalf("tail page ids = %v, want empty", tenantIDs(empty))
	}
}

func tenantIDs(tenants []Tenant) []string {
	ids := make([]string, 0, len(tenants))
	for _, t := range tenants {
		ids = append(ids, t.ID)
	}
	return ids
}

func TestMetaSchemaSpecFromStatementsParsesNewTable(t *testing.T) {
	stmts := []string{
		`CREATE TABLE IF NOT EXISTS tenant_custom_events (
			event_id VARCHAR(64) PRIMARY KEY,
			tenant_id VARCHAR(64) NOT NULL,
			payload JSON,
			created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			INDEX idx_tenant_custom_events_tenant (tenant_id)
		)`,
	}
	spec, err := metaSchemaSpecFromStatements(stmts)
	if err != nil {
		t.Fatalf("metaSchemaSpecFromStatements: %v", err)
	}
	table := mustMetaTableSpec(t, spec, "tenant_custom_events")
	if _, ok := table.columns["tenant_id"]; !ok {
		t.Fatal("expected tenant_id in parsed columns")
	}
	if _, ok := table.indexes["idx_tenant_custom_events_tenant"]; !ok {
		t.Fatal("expected idx_tenant_custom_events_tenant in parsed indexes")
	}
}

func TestValidateMetaIdentifier(t *testing.T) {
	for _, id := range []string{"tenant_tidbcloud_org_bindings", "idx_tidbcloud_org_cluster", "A1_b2"} {
		if err := validateMetaIdentifier(id); err != nil {
			t.Fatalf("validateMetaIdentifier(%q): %v", id, err)
		}
	}
	for _, id := range []string{"", "tenant-table", "idx;drop", "idx name"} {
		if err := validateMetaIdentifier(id); err == nil {
			t.Fatalf("validateMetaIdentifier(%q) error = nil, want error", id)
		}
	}
}

func TestDiffMetaTableMetaReportsMissingColumnAndIndex(t *testing.T) {
	spec := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_api_keys")
	meta := metaTableMeta{
		tableName: "tenant_api_keys",
		columns: map[string]metaColumnMeta{
			"id":             {columnType: "varchar(64)"},
			"tenant_id":      {columnType: "varchar(64)"},
			"jwt_ciphertext": {columnType: "varbinary(4096)"},
			"jwt_hash":       {columnType: "varchar(128)"},
		},
	}
	createStmt := `CREATE TABLE tenant_api_keys (
		id VARCHAR(64) PRIMARY KEY,
		tenant_id VARCHAR(64) NOT NULL,
		jwt_ciphertext VARBINARY(4096) NOT NULL,
		jwt_hash VARCHAR(128) NOT NULL
	)`
	diffs := diffMetaTableMeta(spec, meta, createStmt)
	if !hasMetaDiff(diffs, metaSchemaDiffMissingColumn, "key_name") {
		t.Fatalf("expected missing key_name diff, got %#v", diffs)
	}
	if !hasMetaDiff(diffs, metaSchemaDiffMissingIndex, "idx_api_keys_tenant") {
		t.Fatalf("expected missing idx_api_keys_tenant diff, got %#v", diffs)
	}
}

func TestMetaSchemaSpecTracksPrimaryKeyConstraint(t *testing.T) {
	spec := mustMetaSpec(t)
	table := mustMetaTableSpec(t, spec, "tenant_quota_config")
	pk, ok := table.indexes["primary"]
	if !ok {
		t.Fatal("expected primary key constraint to be tracked in schema spec")
	}
	if !pk.isPrimary {
		t.Fatal("expected primary constraint marker")
	}
	for _, removed := range []string{"quota_limits_overridden"} {
		if _, ok := table.columns[removed]; ok {
			t.Fatalf("tenant_quota_config schema still tracks removed column %s", removed)
		}
	}
	for _, retained := range []string{"max_monthly_cost_mc", "tidbcloud_spending_limit", "tidbcloud_spending_limit_checked_at"} {
		if _, ok := table.columns[retained]; !ok {
			t.Fatalf("tenant_quota_config schema missing retained column %s", retained)
		}
	}
	checkedAt := table.columns["tidbcloud_spending_limit_checked_at"]
	if checkedAt.addSQL != "ALTER TABLE tenant_quota_config ADD COLUMN tidbcloud_spending_limit_checked_at DATETIME(3) NULL" {
		t.Fatalf("tidbcloud_spending_limit_checked_at addSQL = %q", checkedAt.addSQL)
	}
	videoLLM, ok := table.columns["max_video_llm_files"]
	if !ok {
		t.Fatal("tenant_quota_config schema missing max_video_llm_files")
	}
	if videoLLM.addSQL != "ALTER TABLE tenant_quota_config ADD COLUMN max_video_llm_files BIGINT NOT NULL DEFAULT 30" {
		t.Fatalf("max_video_llm_files addSQL = %q", videoLLM.addSQL)
	}
}

func TestMetaSchemaSpecIncludesTenantS3EncryptionColumns(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "tenants")
	tests := map[string]string{
		"s3_encryption_mode":    "ALTER TABLE tenants ADD COLUMN s3_encryption_mode VARCHAR(16) NOT NULL DEFAULT 'inherit'",
		"s3_kms_key_id":         "ALTER TABLE tenants ADD COLUMN s3_kms_key_id VARCHAR(256) NOT NULL DEFAULT ''",
		"s3_bucket_key_enabled": "ALTER TABLE tenants ADD COLUMN s3_bucket_key_enabled TINYINT(1) NOT NULL DEFAULT 1",
	}
	for column, wantAddSQL := range tests {
		spec, ok := table.columns[column]
		if !ok {
			t.Fatalf("missing %s in tenants schema spec", column)
		}
		if spec.addSQL != wantAddSQL {
			t.Fatalf("%s addSQL = %q, want %q", column, spec.addSQL, wantAddSQL)
		}
	}
}

func TestMetaSchemaSpecIncludesStorageTierAccounting(t *testing.T) {
	usage := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_quota_usage")
	for column, wantAddSQL := range map[string]string{
		"db_storage_bytes": "ALTER TABLE tenant_quota_usage ADD COLUMN db_storage_bytes BIGINT NULL",
		"s3_storage_bytes": "ALTER TABLE tenant_quota_usage ADD COLUMN s3_storage_bytes BIGINT NULL",
	} {
		spec, ok := usage.columns[column]
		if !ok {
			t.Fatalf("missing %s in tenant_quota_usage schema spec", column)
		}
		if spec.addSQL != wantAddSQL {
			t.Fatalf("%s addSQL = %q, want %q", column, spec.addSQL, wantAddSQL)
		}
	}

	fileMeta := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_file_meta")
	storageType, ok := fileMeta.columns["storage_type"]
	if !ok {
		t.Fatal("missing storage_type in tenant_file_meta schema spec")
	}
	if want := "ALTER TABLE tenant_file_meta ADD COLUMN storage_type VARCHAR(32) NULL"; storageType.addSQL != want {
		t.Fatalf("storage_type addSQL = %q, want %q", storageType.addSQL, want)
	}
	layerOwned, ok := fileMeta.columns["layer_owned"]
	if !ok {
		t.Fatal("missing layer_owned in tenant_file_meta schema spec")
	}
	if want := "ALTER TABLE tenant_file_meta ADD COLUMN layer_owned TINYINT(1) NOT NULL DEFAULT 0"; layerOwned.addSQL != want {
		t.Fatalf("layer_owned addSQL = %q, want %q", layerOwned.addSQL, want)
	}
	if _, ok := fileMeta.indexes["idx_file_meta_tenant_tier"]; !ok {
		t.Fatal("missing idx_file_meta_tenant_tier on tenant_file_meta")
	}

	layerMeta := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_layer_storage_meta")
	for _, column := range []string{
		"tenant_id", "layer_id", "path_hash", "entry_seq", "size_bytes", "storage_type", "created_at",
	} {
		if _, ok := layerMeta.columns[column]; !ok {
			t.Fatalf("missing %s in tenant_layer_storage_meta schema spec", column)
		}
	}

	scan := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_storage_tier_scan")
	for _, column := range []string{
		"tenant_id", "status", "scan_phase", "scan_cursor", "layer_path_hash", "layer_entry_seq", "watermark",
		"lease_owner", "lease_until", "attempt_count", "last_error",
		"full_scan_completed_at", "updated_at",
	} {
		if _, ok := scan.columns[column]; !ok {
			t.Fatalf("missing %s in tenant_storage_tier_scan schema spec", column)
		}
	}
}

func TestEnsureStorageTierScanJobCapturesWatermarkOnce(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log (tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_create', '{}', 'pending')`, "tier-job-tenant"); err != nil {
		t.Fatal(err)
	}
	created, err := s.EnsureStorageTierScanJob(ctx, "tier-job-tenant")
	if err != nil || !created {
		t.Fatalf("first ensure = created:%t err:%v", created, err)
	}
	job, err := s.GetStorageTierScanJob(ctx, "tier-job-tenant")
	if err != nil {
		t.Fatal(err)
	}
	if job.Watermark == 0 || job.Cursor != "" || job.FullScanCompletedAt != nil {
		t.Fatalf("initial job = %+v", job)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log (tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_create', '{}', 'pending')`, "tier-job-tenant"); err != nil {
		t.Fatal(err)
	}
	created, err = s.EnsureStorageTierScanJob(ctx, "tier-job-tenant")
	if err != nil || created {
		t.Fatalf("second ensure = created:%t err:%v", created, err)
	}
	again, err := s.GetStorageTierScanJob(ctx, "tier-job-tenant")
	if err != nil {
		t.Fatal(err)
	}
	if again.Watermark != job.Watermark {
		t.Fatalf("watermark changed from %d to %d", job.Watermark, again.Watermark)
	}
}

func TestStorageTierScanBackfillsMissingCentralFileMeta(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-missing-shadow"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.IncrStorageBytes(ctx, tenantID, 10); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute); err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner"); err != nil {
		t.Fatal(err)
	}

	err := s.ApplyStorageTierScanPage(ctx, tenantID, "owner", "", []StorageTierScanFile{{
		FileID: "missing-shadow", SizeBytes: 10, IsMedia: true, StorageType: StorageTypeDB9,
	}})
	if err != nil {
		t.Fatal(err)
	}
	fm, err := s.GetFileMeta(ctx, tenantID, "missing-shadow")
	if err != nil {
		t.Fatalf("backfilled central file metadata: %v", err)
	}
	if fm.SizeBytes != 10 || !fm.IsMedia || fm.StorageType == nil || *fm.StorageType != StorageTypeDB9 {
		t.Fatalf("backfilled file metadata = %+v", fm)
	}
	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.DBStorageBytes == nil || *usage.DBStorageBytes != 10 {
		t.Fatalf("db storage bytes = %v, want backfilled 10", usage.DBStorageBytes)
	}
}

func TestStorageTierScanLayerOwnedCoreDoesNotChangeStorageCounters(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-layer-owned-core"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.IncrStorageBytes(ctx, tenantID, 8); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_file_meta
		(tenant_id, file_id, size_bytes, is_media, storage_type, layer_owned)
		VALUES (?, 'layer-alias', 8, 0, NULL, 0)`, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute); err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner"); err != nil {
		t.Fatal(err)
	}

	if err := s.ApplyStorageTierScanPage(ctx, tenantID, "owner", "next", []StorageTierScanFile{{
		FileID: "layer-alias", SizeBytes: 8, StorageType: StorageTypeS3, LayerOwned: true,
	}}); err != nil {
		t.Fatal(err)
	}
	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.StorageBytes != 8 || usage.DBStorageBytes == nil || *usage.DBStorageBytes != 0 ||
		usage.S3StorageBytes == nil || *usage.S3StorageBytes != 0 {
		t.Fatalf("usage after core alias classification = %+v, want total/db9/s3 = 8/0/0", usage)
	}
	fm, err := s.GetFileMeta(ctx, tenantID, "layer-alias")
	if err != nil {
		t.Fatal(err)
	}
	if !fm.LayerOwned || fm.StorageType == nil || *fm.StorageType != StorageTypeS3 {
		t.Fatalf("layer alias metadata = %+v, want layer-owned s3", fm)
	}
}

func TestStorageTierScanMissingLayerOwnedShadowCountsLayerPayloadOnce(t *testing.T) {
	for _, preexistingLayerMeta := range []bool{false, true} {
		name := "without preexisting layer meta"
		if preexistingLayerMeta {
			name = "with preexisting layer meta"
		}
		t.Run(name, func(t *testing.T) {
			s := newControlStore(t)
			ctx := context.Background()
			tenantID := "tier-scan-missing-layer-alias"
			if preexistingLayerMeta {
				tenantID += "-existing"
			}
			if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
				t.Fatal(err)
			}
			if preexistingLayerMeta {
				if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_quota_usage
					SET storage_bytes = 10, db_storage_bytes = 0, s3_storage_bytes = 10
					WHERE tenant_id = ?`, tenantID); err != nil {
					t.Fatal(err)
				}
				if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_layer_storage_meta
					(tenant_id, layer_id, path_hash, entry_seq, size_bytes, storage_type)
					VALUES (?, 'layer-a', 'hash-a', 1, 10, 's3')`, tenantID); err != nil {
					t.Fatal(err)
				}
			}
			if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
				t.Fatal(err)
			}
			if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-core", time.Minute); err != nil || !claimed {
				t.Fatalf("claim core = %t, err = %v", claimed, err)
			}
			if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner-core"); err != nil {
				t.Fatal(err)
			}
			if err := s.ApplyStorageTierScanPage(ctx, tenantID, "owner-core", "", []StorageTierScanFile{{
				FileID: "missing-alias", SizeBytes: 10, StorageType: StorageTypeS3, LayerOwned: true,
			}}); err != nil {
				t.Fatal(err)
			}
			if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-layer", time.Minute); err != nil || !claimed {
				t.Fatalf("claim layer = %t, err = %v", claimed, err)
			}
			completed, err := s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner-layer", StorageTierLayerCursor{}, []StorageTierScanLayer{{
				StorageTierLayerCursor: StorageTierLayerCursor{LayerID: "layer-a", PathHash: "hash-a", EntrySeq: 1},
				SizeBytes:              10, StorageType: StorageTypeS3,
			}})
			if err != nil {
				t.Fatal(err)
			}
			if !completed {
				t.Fatal("scan did not complete after layer-owned shadow backfill")
			}
			usage, err := s.GetQuotaUsage(ctx, tenantID)
			if err != nil {
				t.Fatal(err)
			}
			if usage.StorageBytes != 10 || usage.S3StorageBytes == nil || *usage.S3StorageBytes != 10 {
				t.Fatalf("usage = %+v, want layer payload counted once as total/s3 = 10/10", usage)
			}
		})
	}
}

func TestStorageTierScanClassifiesExistingShadowWithoutChangingQuotaMetadata(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-existing-shadow"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.IncrStorageBytes(ctx, tenantID, 8); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_file_meta
		(tenant_id, file_id, size_bytes, is_media, storage_type) VALUES (?, ?, 8, 1, NULL)`, tenantID, "media-file"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute); err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner"); err != nil {
		t.Fatal(err)
	}

	err := s.ApplyStorageTierScanPage(ctx, tenantID, "owner", "", []StorageTierScanFile{{
		FileID: "media-file", StorageType: StorageTypeS3,
	}})
	if err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-layer", time.Minute); err != nil || !claimed {
		t.Fatalf("claim layer = %t, err = %v", claimed, err)
	}
	completed, err := s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner-layer", StorageTierLayerCursor{}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if !completed {
		t.Fatal("scan did not complete after classifying core and empty layer phases")
	}
	var size int64
	var isMedia bool
	var storageType string
	if err := s.DB().QueryRowContext(ctx, `SELECT size_bytes, is_media, storage_type
		FROM tenant_file_meta WHERE tenant_id = ? AND file_id = ?`, tenantID, "media-file").Scan(&size, &isMedia, &storageType); err != nil {
		t.Fatal(err)
	}
	if size != 8 || !isMedia || storageType != "s3" {
		t.Fatalf("central metadata = size:%d media:%t tier:%q, want 8/true/s3", size, isMedia, storageType)
	}
	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.S3StorageBytes == nil || *usage.S3StorageBytes != 8 {
		t.Fatalf("s3 storage bytes = %v, want 8", usage.S3StorageBytes)
	}
}

func TestStorageTierScanIncludesLayerPayloadsWithoutChangingCoreFileCounters(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-layer-storage"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.InTx(ctx, func(tx *sql.Tx) error {
		return s.IncrQuotaUsageCountersTx(tx, tenantID, 10, 0, 0, 1, 1, 0)
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_file_meta
		(tenant_id, file_id, size_bytes, is_media, storage_type) VALUES (?, 'core-file', 10, 1, 'db9')`, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-core", time.Minute); err != nil || !claimed {
		t.Fatalf("claim core = %t, err = %v", claimed, err)
	}
	if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner-core"); err != nil {
		t.Fatal(err)
	}

	if err := s.ApplyStorageTierScanPage(ctx, tenantID, "owner-core", "", nil); err != nil {
		t.Fatal(err)
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Phase != StorageTierScanPhaseLayer {
		t.Fatalf("phase = %q, want layer", job.Phase)
	}

	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-layer", time.Minute); err != nil || !claimed {
		t.Fatalf("claim layer = %t, err = %v", claimed, err)
	}
	layers := []StorageTierScanLayer{
		{StorageTierLayerCursor: StorageTierLayerCursor{LayerID: "layer-a", PathHash: "hash-a", EntrySeq: 1}, SizeBytes: 5, StorageType: StorageTypeDB9},
		{StorageTierLayerCursor: StorageTierLayerCursor{LayerID: "layer-a", PathHash: "hash-b", EntrySeq: 2}, SizeBytes: 7, StorageType: StorageTypeS3},
		{StorageTierLayerCursor: StorageTierLayerCursor{LayerID: "layer-a", PathHash: "hash-a", EntrySeq: 1}, SizeBytes: 5, StorageType: StorageTypeDB9},
	}
	completed, err := s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner-layer", StorageTierLayerCursor{}, layers)
	if err != nil {
		t.Fatal(err)
	}
	if !completed {
		t.Fatal("layer phase did not complete")
	}
	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.StorageBytes != 22 || usage.DBStorageBytes == nil || *usage.DBStorageBytes != 15 || usage.S3StorageBytes == nil || *usage.S3StorageBytes != 7 {
		t.Fatalf("usage = %+v, want total/db9/s3 = 22/15/7", usage)
	}
	if usage.FileCount != 1 || usage.MediaFileCount != 1 {
		t.Fatalf("core file counters changed: %+v", usage)
	}
}

func TestInsertLayerStorageMetaStoresUnknownTierAsNull(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	inserted := false
	if err := s.InTx(ctx, func(tx *sql.Tx) error {
		var err error
		inserted, err = s.InsertLayerStorageMetaIfAbsentTx(tx, &LayerStorageMeta{
			TenantID: "tenant-layer-unknown", LayerID: "layer-a", PathHash: "hash-a",
			EntrySeq: 1, SizeBytes: 23, StorageType: StorageType("future-tier"),
		})
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if !inserted {
		t.Fatal("layer shadow was not inserted")
	}
	var storageType sql.NullString
	if err := s.DB().QueryRowContext(ctx, `SELECT storage_type FROM tenant_layer_storage_meta
		WHERE tenant_id = ? AND layer_id = ? AND path_hash = ? AND entry_seq = ?`,
		"tenant-layer-unknown", "layer-a", "hash-a", 1).Scan(&storageType); err != nil {
		t.Fatal(err)
	}
	if storageType.Valid {
		t.Fatalf("storage_type = %q, want SQL NULL for unknown tier", storageType.String)
	}
}

func TestGetTierFileCountsFollowReadiness(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-file-counts"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	db9 := StorageTypeDB9
	s3 := StorageTypeS3
	if err := s.InTx(ctx, func(tx *sql.Tx) error {
		if err := s.UpsertFileMetaTx(tx, &FileMeta{TenantID: tenantID, FileID: "a", SizeBytes: 4, StorageType: &db9}); err != nil {
			return err
		}
		if err := s.UpsertFileMetaTx(tx, &FileMeta{TenantID: tenantID, FileID: "b", SizeBytes: 7, StorageType: &s3}); err != nil {
			return err
		}
		// Layer-owned aliases are counted as files (their bytes are not).
		return s.UpsertFileMetaTx(tx, &FileMeta{TenantID: tenantID, FileID: "c", SizeBytes: 9, StorageType: &s3, LayerOwned: true})
	}); err != nil {
		t.Fatal(err)
	}
	// file_count = 4 simulates one active upload reservation: the quota
	// admission counter claims the file before its tenant_file_meta row
	// exists. The gauge must report the confirmed count from the shadow
	// table (3), not the reservation-inflated quota counter.
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_quota_usage
		SET storage_bytes = 11, db_storage_bytes = 4, s3_storage_bytes = 7, file_count = 4
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}

	// Before the tier scan completes the split counters must stay nil even
	// though every file_meta row carries a storage_type; the confirmed count
	// is always returned.
	counts, err := s.GetTierFileCounts(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if counts.TierReady {
		t.Fatal("tier readiness without a completed scan")
	}
	if counts.FileCount != 3 {
		t.Fatalf("confirmed file count = %d, want 3 (active reservation must not leak in)", counts.FileCount)
	}
	if counts.DBFileCount != nil || counts.S3FileCount != nil {
		t.Fatalf("tier file counts before readiness = db9:%v s3:%v, want nil/nil",
			counts.DBFileCount, counts.S3FileCount)
	}

	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan
		SET status = 'completed', full_scan_completed_at = NOW(3)
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}
	counts, err = s.GetTierFileCounts(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !counts.TierReady {
		t.Fatal("completed, internally consistent tier state is not ready")
	}
	if counts.DBFileCount == nil || *counts.DBFileCount != 1 || counts.S3FileCount == nil || *counts.S3FileCount != 2 {
		t.Fatalf("tier file counts = db9:%v s3:%v, want 1/2", counts.DBFileCount, counts.S3FileCount)
	}
}

func TestStorageTierReadyIsRevokedByUnknownTierOrFailedMutation(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-ready-revoked"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_quota_usage
		SET storage_bytes = 0, db_storage_bytes = 0, s3_storage_bytes = 0
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan
		SET status = 'completed', full_scan_completed_at = NOW(3)
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}

	usage, err := s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !usage.StorageTierReady {
		t.Fatal("completed, internally consistent tier state is not ready")
	}

	// Readiness must enforce the shadow invariant directly rather than trust
	// only the completed marker: a zero-byte NULL row does not break the tier
	// counter equality and can be introduced by compatibility or repair tools.
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_file_meta
		(tenant_id, file_id, size_bytes, is_media, storage_type)
		VALUES (?, 'future-zero', 0, 0, NULL)`, tenantID); err != nil {
		t.Fatal(err)
	}
	usage, err = s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.StorageTierReady {
		t.Fatal("unknown storage tier did not revoke readiness")
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != StorageTierScanStatusCompleted || job.FullScanCompletedAt == nil {
		t.Fatalf("job = %+v, want read-time invariant to reject a stale completed marker", job)
	}

	db9 := StorageTypeDB9
	if err := s.InTx(ctx, func(tx *sql.Tx) error {
		return s.UpsertFileMetaTx(tx, &FileMeta{TenantID: tenantID, FileID: "future-zero", StorageType: &db9})
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan
		SET status = 'completed', full_scan_completed_at = NOW(3) WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}
	res, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
		(tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_create', '{}', 'pending')`, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	mutationID, err := res.LastInsertId()
	if err != nil {
		t.Fatal(err)
	}
	if err := s.IncrMutationRetry(ctx, mutationID, 1); err != nil {
		t.Fatal(err)
	}
	usage, err = s.GetQuotaUsage(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if usage.StorageTierReady {
		t.Fatal("failed storage mutation did not revoke readiness")
	}
}

func TestStorageTierReadinessInvalidationSkipsActiveScan(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-active-invalidation"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute); err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if err := s.InTx(ctx, func(tx *sql.Tx) error {
		return s.UpsertFileMetaTx(tx, &FileMeta{TenantID: tenantID, FileID: "unknown-active"})
	}); err != nil {
		t.Fatal(err)
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != StorageTierScanStatusRunning || job.LeaseOwner != "owner" || job.FullScanCompletedAt != nil {
		t.Fatalf("active scan was invalidated by an unclassified mutation: %+v", job)
	}

	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan
		SET status = 'completed', lease_owner = '', lease_until = NULL, full_scan_completed_at = NOW(3)
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.InTx(ctx, func(tx *sql.Tx) error {
		return s.UpsertFileMetaTx(tx, &FileMeta{TenantID: tenantID, FileID: "unknown-completed"})
	}); err != nil {
		t.Fatal(err)
	}
	job, err = s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != StorageTierScanStatusBlocked || job.FullScanCompletedAt != nil {
		t.Fatalf("completed scan retained readiness after an unclassified mutation: %+v", job)
	}
}

func TestStorageTierScanFailedMutationBlocksCompletion(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-failed-mutation"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_quota_usage
		SET storage_bytes = 0, db_storage_bytes = 0, s3_storage_bytes = 0
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
		(tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_create', '{}', 'failed')`, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute); err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan SET scan_phase = 'layer' WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}

	completed, err := s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner", StorageTierLayerCursor{}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if completed {
		t.Fatal("scan completed with a failed watermark mutation")
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != "blocked" || job.FullScanCompletedAt != nil {
		t.Fatalf("job = %+v, want blocked and incomplete", job)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "other-owner", time.Minute); err != nil || claimed {
		t.Fatalf("blocked claim = %t, err = %v", claimed, err)
	}
}

func TestBlockStorageTierScanRequeuesWhenObservedFailureWasAlreadyRepaired(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-concurrent-repair"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute); err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan
		SET scan_phase = 'layer', scan_cursor = 'layer-mid', layer_path_hash = 'path-mid', layer_entry_seq = 7,
		    full_scan_completed_at = NOW(3)
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}

	// The worker can reach BlockStorageTierScanJob from an earlier failed-state
	// read even though the final repair mutation has committed in the meantime.
	if err := s.BlockStorageTierScanJob(ctx, tenantID, "owner", "storage tier mutation failed"); err != nil {
		t.Fatal(err)
	}

	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != StorageTierScanStatusPending || job.Phase != StorageTierScanPhaseCore ||
		job.Cursor != "" || job.LayerPathHash != "" || job.LayerEntrySeq != 0 ||
		job.FullScanCompletedAt != nil {
		t.Fatalf("job = %+v, want pending full rescan after concurrent repair", job)
	}
	var signals int
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM tenant_notify_outbox
		WHERE tenant_id = ? AND work_mask = ?`, tenantID, TenantNotifyWorkStorageTierScan).Scan(&signals); err != nil {
		t.Fatal(err)
	}
	if signals != 1 {
		t.Fatalf("storage-tier recovery signals = %d, want 1", signals)
	}
}

func TestStorageTierScanInvariantMismatchBlocksWithoutRestart(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-invariant-mismatch"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_quota_usage
		SET storage_bytes = 10, db_storage_bytes = 3, s3_storage_bytes = 4
		WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute); err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan
		SET scan_phase = 'layer' WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}

	completed, err := s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner", StorageTierLayerCursor{}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if completed {
		t.Fatal("scan completed with inconsistent counters")
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != "blocked" || job.Phase != StorageTierScanPhaseLayer {
		t.Fatalf("job = %+v, want blocked at final cursor without restarting core", job)
	}
}

func TestMarkMutationAppliedRequeuesBlockedStorageTierScan(t *testing.T) {
	for _, tc := range []struct {
		name string
		mark func(*Store, *sql.Tx, int64) error
	}{
		{name: "single", mark: func(s *Store, tx *sql.Tx, id int64) error {
			return s.MarkMutationAppliedTx(tx, id)
		}},
		{name: "batch", mark: func(s *Store, tx *sql.Tx, id int64) error {
			return s.MarkMutationsAppliedTx(tx, []int64{id})
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			s := newControlStore(t)
			ctx := context.Background()
			tenantID := "tier-blocked-requeue-" + tc.name
			if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
				t.Fatal(err)
			}
			if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
				t.Fatal(err)
			}
			if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan
				SET status = 'blocked', scan_phase = 'layer', scan_cursor = 'layer-mid',
				    layer_path_hash = 'path-mid', layer_entry_seq = 7
				WHERE tenant_id = ?`, tenantID); err != nil {
				t.Fatal(err)
			}
			res, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
				(tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_overwrite', '{}', 'pending')`, tenantID)
			if err != nil {
				t.Fatal(err)
			}
			mutationID, err := res.LastInsertId()
			if err != nil {
				t.Fatal(err)
			}
			if err := s.InTx(ctx, func(tx *sql.Tx) error {
				return tc.mark(s, tx, mutationID)
			}); err != nil {
				t.Fatal(err)
			}

			job, err := s.GetStorageTierScanJob(ctx, tenantID)
			if err != nil {
				t.Fatal(err)
			}
			if job.Status != StorageTierScanStatusPending || job.Phase != StorageTierScanPhaseCore ||
				job.Cursor != "" || job.LayerPathHash != "" || job.LayerEntrySeq != 0 ||
				job.Watermark != mutationID || job.FullScanCompletedAt != nil {
				t.Fatalf("job = %+v, want pending full rescan after repair mutation", job)
			}
			var signals int
			if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM tenant_notify_outbox
				WHERE tenant_id = ? AND work_mask = ?`, tenantID, TenantNotifyWorkStorageTierScan).Scan(&signals); err != nil {
				t.Fatal(err)
			}
			if signals != 1 {
				t.Fatalf("storage-tier work signals = %d, want 1", signals)
			}

			res, err = s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
				(tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_overwrite', '{}', 'pending')`, tenantID)
			if err != nil {
				t.Fatal(err)
			}
			laterMutationID, err := res.LastInsertId()
			if err != nil {
				t.Fatal(err)
			}
			if err := s.InTx(ctx, func(tx *sql.Tx) error {
				return tc.mark(s, tx, laterMutationID)
			}); err != nil {
				t.Fatal(err)
			}
			if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM tenant_notify_outbox
				WHERE tenant_id = ? AND work_mask = ?`, tenantID, TenantNotifyWorkStorageTierScan).Scan(&signals); err != nil {
				t.Fatal(err)
			}
			if signals != 1 {
				t.Fatalf("storage-tier work signals after ordinary mutation = %d, want 1", signals)
			}
		})
	}
}

func TestMarkMutationAppliedKeepsScanBlockedWhileAnotherMutationFailed(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-blocked-failed-mutation"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE tenant_storage_tier_scan SET status = 'blocked' WHERE tenant_id = ?`, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
		(tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_create', '{}', 'failed')`, tenantID); err != nil {
		t.Fatal(err)
	}
	res, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
		(tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_overwrite', '{}', 'pending')`, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	mutationID, err := res.LastInsertId()
	if err != nil {
		t.Fatal(err)
	}
	if err := s.InTx(ctx, func(tx *sql.Tx) error { return s.MarkMutationAppliedTx(tx, mutationID) }); err != nil {
		t.Fatal(err)
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != StorageTierScanStatusBlocked || job.FullScanCompletedAt != nil {
		t.Fatalf("job = %+v, want blocked while a storage mutation is failed", job)
	}
}

func TestStorageTierScanDoesNotRestartForAppliedConcurrentMutation(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-watermark-advance"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if err := s.IncrStorageBytes(ctx, tenantID, 7); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_file_meta
		(tenant_id, file_id, size_bytes, is_media, storage_type) VALUES (?, ?, 7, 0, NULL)`, tenantID, "file-a"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	job, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner", time.Minute)
	if err != nil || !claimed {
		t.Fatalf("claim = %t, err = %v", claimed, err)
	}
	if job.Watermark != 0 {
		t.Fatalf("initial watermark = %d, want 0", job.Watermark)
	}
	if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner"); err != nil {
		t.Fatal(err)
	}
	res, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
		(tenant_id, mutation_type, mutation_data, status, applied_at)
		VALUES (?, 'file_overwrite', '{}', 'applied', NOW(3))`, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := res.LastInsertId(); err != nil {
		t.Fatal(err)
	}

	err = s.ApplyStorageTierScanPage(ctx, tenantID, "owner", "", []StorageTierScanFile{{
		FileID: "file-a", StorageType: StorageTypeDB9,
	}})
	if err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-layer", time.Minute); err != nil || !claimed {
		t.Fatalf("claim layer = %t, err = %v", claimed, err)
	}
	completed, err := s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner-layer", StorageTierLayerCursor{}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if !completed {
		t.Fatal("applied concurrent mutation caused an unnecessary full scan restart")
	}
	job, err = s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != StorageTierScanStatusCompleted || job.FullScanCompletedAt == nil {
		t.Fatalf("job = %+v, want completed without restarting core", job)
	}
}

func TestStorageTierScanWaitsAtLayerForConcurrentPendingMutation(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "tier-scan-concurrent-pending"
	if err := s.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.EnsureStorageTierScanJob(ctx, tenantID); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-core", time.Minute); err != nil || !claimed {
		t.Fatalf("claim core = %t, err = %v", claimed, err)
	}
	if err := s.InitializeStorageTierCounters(ctx, tenantID, "owner-core"); err != nil {
		t.Fatal(err)
	}
	if err := s.ApplyStorageTierScanPage(ctx, tenantID, "owner-core", "", nil); err != nil {
		t.Fatal(err)
	}
	res, err := s.DB().ExecContext(ctx, `INSERT INTO quota_mutation_log
		(tenant_id, mutation_type, mutation_data, status) VALUES (?, 'file_overwrite', '{}', 'pending')`, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	mutationID, err := res.LastInsertId()
	if err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-layer", time.Minute); err != nil || !claimed {
		t.Fatalf("claim layer = %t, err = %v", claimed, err)
	}
	completed, err := s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner-layer", StorageTierLayerCursor{}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if completed {
		t.Fatal("scan completed while a concurrent storage mutation was pending")
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != StorageTierScanStatusPending || job.Phase != StorageTierScanPhaseLayer {
		t.Fatalf("job = %+v, want pending at layer tail without restarting core", job)
	}
	if err := s.InTx(ctx, func(tx *sql.Tx) error {
		return s.MarkMutationAppliedTx(tx, mutationID)
	}); err != nil {
		t.Fatal(err)
	}
	if _, claimed, err := s.ClaimStorageTierScanJob(ctx, tenantID, "owner-layer-2", time.Minute); err != nil || !claimed {
		t.Fatalf("reclaim layer = %t, err = %v", claimed, err)
	}
	completed, err = s.ApplyStorageTierLayerScanPage(ctx, tenantID, "owner-layer-2", StorageTierLayerCursor{}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if !completed {
		t.Fatal("scan did not complete after the concurrent mutation was applied")
	}
}

func TestMetaSchemaSpecIncludesForkStorageNamespaceColumns(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "tenants")
	tests := map[string]string{
		"kind":                 "ALTER TABLE tenants ADD COLUMN kind VARCHAR(16) NOT NULL DEFAULT 'live'",
		"parent_tenant_id":     "ALTER TABLE tenants ADD COLUMN parent_tenant_id VARCHAR(64) NOT NULL DEFAULT ''",
		"storage_namespace_id": "ALTER TABLE tenants ADD COLUMN storage_namespace_id VARCHAR(64) NOT NULL DEFAULT ''",
	}
	for column, wantAddSQL := range tests {
		spec, ok := table.columns[column]
		if !ok {
			t.Fatalf("missing %s in tenants schema spec", column)
		}
		if spec.addSQL != wantAddSQL {
			t.Fatalf("%s addSQL = %q, want %q", column, spec.addSQL, wantAddSQL)
		}
	}
	if _, ok := table.indexes["idx_tenant_namespace"]; !ok {
		t.Fatal("tenants schema missing idx_tenant_namespace")
	}
	if _, ok := table.indexes["idx_tenant_parent"]; !ok {
		t.Fatal("tenants schema missing idx_tenant_parent")
	}
	_ = mustMetaTableSpec(t, mustMetaSpec(t), "storage_namespaces")
	_ = mustMetaTableSpec(t, mustMetaSpec(t), "tenant_delete_jobs")
	_ = mustMetaTableSpec(t, mustMetaSpec(t), "object_gc_candidates")
}

func TestMetaSchemaSpecIncludesAppendLogBindingColumns(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "storage_namespaces")
	tests := map[string]string{
		"append_log_binding_state": "ALTER TABLE storage_namespaces ADD COLUMN append_log_binding_state VARCHAR(16) NOT NULL DEFAULT 'unbound'",
		"append_log_bucket":        "ALTER TABLE storage_namespaces ADD COLUMN append_log_bucket VARCHAR(255) NOT NULL DEFAULT ''",
		"append_log_prefix":        "ALTER TABLE storage_namespaces ADD COLUMN append_log_prefix VARCHAR(2048) NOT NULL DEFAULT ''",
		"append_log_region":        "ALTER TABLE storage_namespaces ADD COLUMN append_log_region VARCHAR(64) NOT NULL DEFAULT ''",
	}
	for column, wantAddSQL := range tests {
		spec, ok := table.columns[column]
		if !ok {
			t.Fatalf("storage_namespaces missing %s", column)
		}
		if spec.addSQL != wantAddSQL {
			t.Fatalf("%s addSQL = %q, want %q", column, spec.addSQL, wantAddSQL)
		}
	}
}

func TestMetaSchemaSpecIncludesTenantStatusCreatedIDIndex(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "tenants")
	idx, ok := table.indexes["idx_tenant_status_created_id"]
	if !ok {
		t.Fatal("tenants schema missing idx_tenant_status_created_id")
	}
	wantCreateSQL := "CREATE INDEX idx_tenant_status_created_id ON tenants(status, created_at, id)"
	if idx.createSQL != wantCreateSQL {
		t.Errorf("idx_tenant_status_created_id createSQL = %q, want %q", idx.createSQL, wantCreateSQL)
	}

	observed := metaTableMeta{
		tableName: "tenants",
		columns: map[string]metaColumnMeta{
			"id":         {columnType: "varchar(64)"},
			"status":     {columnType: "varchar(20)"},
			"created_at": {columnType: "datetime(3)"},
		},
	}
	observedIndexes := map[string]struct{}{
		"primary":           {},
		"idx_tenant_status": {},
	}
	diffs := diffMetaTableMetaWithObservedIndexes(table, observed, "", observedIndexes)
	if !hasMetaDiff(diffs, metaSchemaDiffMissingIndex, "idx_tenant_status_created_id") {
		t.Errorf("expected missing idx_tenant_status_created_id diff, got %#v", diffs)
	}

	var indexDiff metaSchemaDiff
	for _, diff := range diffs {
		if diff.indexName == "idx_tenant_status_created_id" {
			indexDiff = diff
			break
		}
	}
	if indexDiff.repairSQL != wantCreateSQL {
		t.Errorf("idx_tenant_status_created_id repairSQL = %q, want %q", indexDiff.repairSQL, wantCreateSQL)
	}

	plans := plannedMetaSchemaRepairs([]metaSchemaDiff{indexDiff})
	if len(plans) != 1 || plans[0] != wantCreateSQL {
		t.Errorf("repair plans = %#v, want [%q]", plans, wantCreateSQL)
	}
}

func TestFinalizeTenantDeleteRollsBackWhenTenantStatusUpdateFails(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID:            "delete-ns-rollback",
		OwnerTenantID: "missing-delete-tenant",
		Backend:       "s3",
		Bucket:        "bucket",
		Prefix:        "tenant/missing-delete-tenant",
		State:         StorageNamespaceDeleting,
		CreatedAt:     now,
		UpdatedAt:     now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID:    "missing-delete-tenant",
		NamespaceID: "delete-ns-rollback",
		Backend:     "s3",
		Bucket:      "bucket",
		Prefix:      "tenant/missing-delete-tenant",
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeleteJobRunning(ctx, "missing-delete-tenant")
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("MarkTenantDeleteJobRunning updated = false, want true")
	}

	err = s.FinalizeTenantDelete(ctx, "missing-delete-tenant", "delete-ns-rollback", 7, 3)
	if !errors.Is(err, ErrNotFound) {
		t.Fatalf("FinalizeTenantDelete error = %v, want ErrNotFound", err)
	}

	var jobState string
	var deletedObjects int64
	if err := s.DB().QueryRow(`SELECT state, deleted_objects FROM tenant_delete_jobs WHERE tenant_id = ?`, "missing-delete-tenant").Scan(&jobState, &deletedObjects); err != nil {
		t.Fatal(err)
	}
	if jobState != string(TenantDeleteJobRunning) || deletedObjects != 0 {
		t.Fatalf("tenant delete job = state %s deleted_objects %d, want running/0", jobState, deletedObjects)
	}
	var nsState string
	if err := s.DB().QueryRow(`SELECT state FROM storage_namespaces WHERE namespace_id = ?`, "delete-ns-rollback").Scan(&nsState); err != nil {
		t.Fatal(err)
	}
	if nsState != string(StorageNamespaceDeleting) {
		t.Fatalf("namespace state = %s, want %s", nsState, StorageNamespaceDeleting)
	}
}

func TestMarkTenantDeleteJobRunningHonorsNotBefore(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	future := time.Now().UTC().Add(time.Hour)
	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID:    "delete-job-future",
		NamespaceID: "delete-ns-future",
		Backend:     "s3",
		Bucket:      "bucket",
		Prefix:      "tenant/delete-job-future",
		NotBefore:   future,
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeleteJobRunning(ctx, "delete-job-future")
	if err != nil {
		t.Fatal(err)
	}
	if updated {
		t.Fatal("MarkTenantDeleteJobRunning updated future job, want false")
	}
	var state string
	if err := s.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, "delete-job-future").Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != string(TenantDeleteJobPending) {
		t.Fatalf("job state = %s, want %s", state, TenantDeleteJobPending)
	}
}

func TestCancelTenantDeleteJobCanBeRearmedOnRetry(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "cancel-rearm-tenant"
	job := &TenantDeleteJob{
		TenantID: tenantID, NamespaceID: "cancel-rearm-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/cancel-rearm",
	}
	if err := s.EnqueueTenantDeleteJob(ctx, job); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeleteJobRunning(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("MarkTenantDeleteJobRunning updated = false, want true")
	}
	cancelled, err := s.CancelTenantDeleteJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !cancelled {
		t.Fatal("CancelTenantDeleteJob cancelled = false, want true")
	}
	var state string
	if err := s.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, tenantID).Scan(&state); !errors.Is(err, sql.ErrNoRows) {
		t.Fatalf("cancelled job query error = %v, want sql.ErrNoRows", err)
	}
	if err := s.EnqueueTenantDeleteJob(ctx, job); err != nil {
		t.Fatal(err)
	}
	armed, err := s.TenantDeleteJobArmed(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !armed {
		t.Fatal("re-enqueued cleanup job is not armed")
	}
}

func TestCancelTenantDeleteJobPreservesJobAfterDeleteResumes(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	tenantID := "cancel-resumed-delete-" + uuid.NewString()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID: tenantID, Status: TenantDeleting, Provider: tidbCloudNativeProvider,
		DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID: tenantID, NamespaceID: "cancel-resumed-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/cancel-resumed",
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeleteJobRunning(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("MarkTenantDeleteJobRunning updated = false, want true")
	}
	cancelled, err := s.CancelTenantDeleteJob(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if cancelled {
		t.Fatal("CancelTenantDeleteJob cancelled resumed delete job")
	}
	var state TenantDeleteJobState
	if err := s.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, tenantID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != TenantDeleteJobRunning {
		t.Fatalf("job state = %s, want %s", state, TenantDeleteJobRunning)
	}
}

func TestTenantDeleteJobArmedIgnoresCompletedJob(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const tenantID = "armed-completed-job"
	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID: tenantID, NamespaceID: "armed-completed-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/armed-completed",
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeleteJobRunning(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("MarkTenantDeleteJobRunning updated = false, want true")
	}
	if err := s.MarkTenantDeleteJobDeleted(ctx, tenantID, 1, 0); err != nil {
		t.Fatal(err)
	}
	armed, err := s.TenantDeleteJobArmed(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if armed {
		t.Fatal("completed cleanup job must not count as armed")
	}
}

func TestFinalizeTenantDeleteRequiresDeletingTenant(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	const tenantID = "finalize-status-guard"
	if err := s.InsertTenant(ctx, &Tenant{
		ID: tenantID, Status: TenantActive, StorageNamespaceID: "finalize-status-guard-ns",
		Provider: tidbCloudNativeProvider, DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: "finalize-status-guard-ns", OwnerTenantID: tenantID, Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/finalize-status-guard", State: StorageNamespaceDeleting,
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID: tenantID, NamespaceID: "finalize-status-guard-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/finalize-status-guard",
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeleteJobRunning(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("MarkTenantDeleteJobRunning updated = false, want true")
	}
	if err := s.FinalizeTenantDelete(ctx, tenantID, "finalize-status-guard-ns", 1, 0); !errors.Is(err, ErrNotFound) {
		t.Fatalf("FinalizeTenantDelete error = %v, want ErrNotFound", err)
	}
	row := s.DB().QueryRow(`SELECT status FROM tenants WHERE id = ?`, tenantID)
	var status string
	if err := row.Scan(&status); err != nil {
		t.Fatal(err)
	}
	if status != string(TenantActive) {
		t.Fatalf("tenant status = %s, want %s", status, TenantActive)
	}
	var state string
	if err := s.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, tenantID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != string(TenantDeleteJobRunning) {
		t.Fatalf("job state = %s, want %s", state, TenantDeleteJobRunning)
	}
}

func TestMarkTenantDeletingWithDeleteJobAtomic(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID: "atomic-delete", Status: TenantActive, Provider: tidbCloudNativeProvider,
		StorageNamespaceID: "atomic-delete-ns", DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: "atomic-delete-ns", OwnerTenantID: "atomic-delete", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/atomic-delete", State: StorageNamespaceActive,
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	delayed := &TenantDeleteJob{
		TenantID: "atomic-delete", NamespaceID: "atomic-delete-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/atomic-delete", NotBefore: now.Add(time.Hour),
	}
	updated, err := s.MarkTenantDeletingWithDeleteJob(ctx, "atomic-delete", TenantActive, delayed)
	if err != nil {
		t.Fatal(err)
	}
	if !updated {
		t.Fatal("MarkTenantDeletingWithDeleteJob updated = false, want true")
	}
	tenantRow, err := s.GetTenant(ctx, "atomic-delete")
	if err != nil {
		t.Fatal(err)
	}
	if tenantRow.Status != TenantDeleting {
		t.Fatalf("tenant status = %s, want %s", tenantRow.Status, TenantDeleting)
	}
	var nsState string
	if err := s.DB().QueryRow(`SELECT state FROM storage_namespaces WHERE namespace_id = ?`, "atomic-delete-ns").Scan(&nsState); err != nil {
		t.Fatal(err)
	}
	if nsState != string(StorageNamespaceDeleting) {
		t.Fatalf("namespace state = %s, want %s", nsState, StorageNamespaceDeleting)
	}
	armed, err := s.TenantDeleteJobArmed(ctx, "atomic-delete")
	if err != nil {
		t.Fatal(err)
	}
	if armed {
		t.Fatal("delayed job must not be armed")
	}
	exists, err := s.TenantDeleteJobExists(ctx, "atomic-delete")
	if err != nil {
		t.Fatal(err)
	}
	if !exists {
		t.Fatal("cleanup job missing after atomic mark")
	}

	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID: "atomic-delete", NamespaceID: "atomic-delete-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/atomic-delete",
	}); err != nil {
		t.Fatal(err)
	}
	armed, err = s.TenantDeleteJobArmed(ctx, "atomic-delete")
	if err != nil {
		t.Fatal(err)
	}
	if !armed {
		t.Fatal("re-enqueue with immediate not_before must arm the delayed job")
	}
}

func TestMarkTenantDeletingWithDeleteJobSkipsWrongStatus(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID: "atomic-delete-skip", Status: TenantDeleted, Provider: tidbCloudNativeProvider,
		StorageNamespaceID: "atomic-delete-skip-ns", DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: "atomic-delete-skip-ns", OwnerTenantID: "atomic-delete-skip", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/atomic-delete-skip", State: StorageNamespaceActive,
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeletingWithDeleteJob(ctx, "atomic-delete-skip", TenantActive, &TenantDeleteJob{
		TenantID: "atomic-delete-skip", NamespaceID: "atomic-delete-skip-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/atomic-delete-skip", NotBefore: now.Add(time.Hour),
	})
	if err != nil {
		t.Fatal(err)
	}
	if updated {
		t.Fatal("mark succeeded from a non-matching status")
	}
	exists, err := s.TenantDeleteJobExists(ctx, "atomic-delete-skip")
	if err != nil {
		t.Fatal(err)
	}
	if exists {
		t.Fatal("cleanup job created despite status guard mismatch")
	}
}

func TestRollbackTenantDeletingWithDeleteJob(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID: "rollback-delete", Status: TenantActive, Provider: tidbCloudNativeSharedProvider,
		StorageNamespaceID: "rollback-delete-ns", DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: "rollback-delete-ns", OwnerTenantID: "rollback-delete", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/rollback-delete", State: StorageNamespaceActive,
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeletingWithDeleteJob(ctx, "rollback-delete", TenantActive, &TenantDeleteJob{
		TenantID: "rollback-delete", NamespaceID: "rollback-delete-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/rollback-delete", NotBefore: now.Add(time.Hour),
	})
	if err != nil || !updated {
		t.Fatalf("mark: updated=%v err=%v", updated, err)
	}

	rolledBack, err := s.RollbackTenantDeletingWithDeleteJob(ctx, "rollback-delete", "rollback-delete-ns", TenantActive)
	if err != nil {
		t.Fatal(err)
	}
	if !rolledBack {
		t.Fatal("rollback rolledBack = false, want true")
	}
	tenantRow, err := s.GetTenant(ctx, "rollback-delete")
	if err != nil {
		t.Fatal(err)
	}
	if tenantRow.Status != TenantActive {
		t.Fatalf("tenant status after rollback = %s, want %s", tenantRow.Status, TenantActive)
	}
	exists, err := s.TenantDeleteJobExists(ctx, "rollback-delete")
	if err != nil {
		t.Fatal(err)
	}
	if exists {
		t.Fatal("cleanup job survived rollback")
	}
	var nsState string
	if err := s.DB().QueryRow(`SELECT state FROM storage_namespaces WHERE namespace_id = ?`, "rollback-delete-ns").Scan(&nsState); err != nil {
		t.Fatal(err)
	}
	if nsState != string(StorageNamespaceActive) {
		t.Fatalf("namespace state after rollback = %s, want %s", nsState, StorageNamespaceActive)
	}

	if _, err := s.MarkTenantDeletingWithDeleteJob(ctx, "rollback-delete", TenantActive, &TenantDeleteJob{
		TenantID: "rollback-delete", NamespaceID: "rollback-delete-ns", Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/rollback-delete", NotBefore: now.Add(time.Hour),
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpdateTenantStatus(ctx, "rollback-delete", TenantDeleted); err != nil {
		t.Fatal(err)
	}
	rolledBack, err = s.RollbackTenantDeletingWithDeleteJob(ctx, "rollback-delete", "rollback-delete-ns", TenantActive)
	if err != nil {
		t.Fatal(err)
	}
	if rolledBack {
		t.Fatal("rollback succeeded on a tenant already progressed past deleting")
	}
	exists, err = s.TenantDeleteJobExists(ctx, "rollback-delete")
	if err != nil {
		t.Fatal(err)
	}
	if !exists {
		t.Fatal("cleanup job dropped by a rollback that did not apply")
	}
}

func TestRollbackTenantDeletingWithDeleteJobRejectsClusterBackedTenant(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	const (
		tenantID    = "rollback-cluster-delete"
		namespaceID = "rollback-cluster-delete-ns"
	)
	if err := s.InsertTenant(ctx, &Tenant{
		ID: tenantID, Status: TenantDeleting, Provider: tidbCloudNativeProvider,
		StorageNamespaceID: namespaceID, DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: namespaceID, OwnerTenantID: tenantID, Backend: "s3", Bucket: "bucket",
		Prefix: "tenant/" + tenantID, State: StorageNamespaceDeleting, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.EnqueueTenantDeleteJob(ctx, &TenantDeleteJob{
		TenantID: tenantID, NamespaceID: namespaceID, Backend: "s3", Bucket: "bucket",
		Prefix: "tenant/" + tenantID, NotBefore: now.Add(time.Hour),
	}); err != nil {
		t.Fatal(err)
	}

	rolledBack, err := s.RollbackTenantDeletingWithDeleteJob(ctx, tenantID, namespaceID, TenantActive)
	if err != nil {
		t.Fatal(err)
	}
	if rolledBack {
		t.Fatal("rollback reactivated a cluster-backed tenant after cluster deletion handoff")
	}
	tenantRow, err := s.GetTenant(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if tenantRow.Status != TenantDeleting {
		t.Fatalf("tenant status = %s, want %s", tenantRow.Status, TenantDeleting)
	}
	if exists, err := s.TenantDeleteJobExists(ctx, tenantID); err != nil || !exists {
		t.Fatalf("cleanup job after rejected rollback: exists=%v err=%v, want true/nil", exists, err)
	}
}

func TestRollbackTenantDeletingWithRunningDeleteJobLeavesDeleteOwnedByWorker(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	const (
		tenantID    = "rollback-running-delete"
		namespaceID = "rollback-running-delete-ns"
	)
	if err := s.InsertTenant(ctx, &Tenant{
		ID: tenantID, Status: TenantActive, Provider: tidbCloudNativeSharedProvider,
		StorageNamespaceID: namespaceID, DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: namespaceID, OwnerTenantID: tenantID, Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/rollback-running-delete", State: StorageNamespaceActive,
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.MarkTenantDeletingWithDeleteJob(ctx, tenantID, TenantActive, &TenantDeleteJob{
		TenantID: tenantID, NamespaceID: namespaceID, Backend: "s3",
		Bucket: "bucket", Prefix: "tenant/rollback-running-delete", NotBefore: now.Add(-time.Minute),
	})
	if err != nil || !updated {
		t.Fatalf("mark: updated=%v err=%v", updated, err)
	}
	updated, err = s.MarkTenantDeleteJobRunning(ctx, tenantID)
	if err != nil || !updated {
		t.Fatalf("claim job: updated=%v err=%v", updated, err)
	}

	rolledBack, err := s.RollbackTenantDeletingWithDeleteJob(ctx, tenantID, namespaceID, TenantActive)
	if err != nil {
		t.Fatal(err)
	}
	if rolledBack {
		t.Fatal("rollback succeeded after cleanup worker claimed the job")
	}
	tenantRow, err := s.GetTenant(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if tenantRow.Status != TenantDeleting {
		t.Fatalf("tenant status = %s, want %s", tenantRow.Status, TenantDeleting)
	}
	var jobState, namespaceState string
	if err := s.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, tenantID).Scan(&jobState); err != nil {
		t.Fatal(err)
	}
	if jobState != string(TenantDeleteJobRunning) {
		t.Fatalf("job state = %s, want %s", jobState, TenantDeleteJobRunning)
	}
	if err := s.DB().QueryRow(`SELECT state FROM storage_namespaces WHERE namespace_id = ?`, namespaceID).Scan(&namespaceState); err != nil {
		t.Fatal(err)
	}
	if namespaceState != string(StorageNamespaceDeleting) {
		t.Fatalf("namespace state = %s, want %s", namespaceState, StorageNamespaceDeleting)
	}
}

func TestMarkTenantDeletingWithDeleteJobPreservesRunningWorkerOwnership(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	const (
		tenantID    = "remark-running-delete"
		namespaceID = "remark-running-delete-ns"
	)
	if err := s.InsertTenant(ctx, &Tenant{
		ID: tenantID, Status: TenantActive, Provider: tidbCloudNativeProvider,
		StorageNamespaceID: namespaceID, DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: namespaceID, OwnerTenantID: tenantID, Backend: "s3", Bucket: "bucket",
		Prefix: "tenant/remark-running-delete", State: StorageNamespaceActive,
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	job := &TenantDeleteJob{
		TenantID: tenantID, NamespaceID: namespaceID, Backend: "s3", Bucket: "bucket",
		Prefix: "tenant/remark-running-delete", NotBefore: now.Add(-time.Minute),
	}
	updated, err := s.MarkTenantDeletingWithDeleteJob(ctx, tenantID, TenantActive, job)
	if err != nil || !updated {
		t.Fatalf("initial mark: updated=%v err=%v", updated, err)
	}
	updated, err = s.MarkTenantDeleteJobRunning(ctx, tenantID)
	if err != nil || !updated {
		t.Fatalf("claim job: updated=%v err=%v", updated, err)
	}

	updated, err = s.MarkTenantDeletingWithDeleteJob(ctx, tenantID, TenantDeleting, job)
	if err != nil || !updated {
		t.Fatalf("re-mark deleting tenant: updated=%v err=%v", updated, err)
	}

	var state string
	if err := s.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, tenantID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != string(TenantDeleteJobRunning) {
		t.Fatalf("job state after re-mark = %s, want %s", state, TenantDeleteJobRunning)
	}
	if err := s.EnqueueTenantDeleteJob(ctx, job); err != nil {
		t.Fatal(err)
	}
	if err := s.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, tenantID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != string(TenantDeleteJobRunning) {
		t.Fatalf("job state after enqueue = %s, want %s", state, TenantDeleteJobRunning)
	}
}

func TestMetaSchemaSpecIncludesAPIKeyScopeTables(t *testing.T) {
	spec := mustMetaSpec(t)
	apiKeys := mustMetaTableSpec(t, spec, "tenant_api_keys")
	if _, ok := apiKeys.indexes["idx_api_keys_tenant_name"]; ok {
		t.Fatal("tenant_api_keys schema must not require key_name uniqueness")
	}
	scopeKind, ok := apiKeys.columns["scope_kind"]
	if !ok {
		t.Fatal("tenant_api_keys schema missing scope_kind")
	}
	if scopeKind.addSQL != "ALTER TABLE tenant_api_keys ADD COLUMN scope_kind VARCHAR(32) NOT NULL DEFAULT 'owner'" {
		t.Fatalf("scope_kind addSQL = %q", scopeKind.addSQL)
	}
	for column, wantAddSQL := range map[string]string{
		"issued_by_provider":      "ALTER TABLE tenant_api_keys ADD COLUMN issued_by_provider VARCHAR(64) NOT NULL DEFAULT ''",
		"issued_by_subject_key":   "ALTER TABLE tenant_api_keys ADD COLUMN issued_by_subject_key VARCHAR(512) NOT NULL DEFAULT ''",
		"issued_by_metadata_json": "ALTER TABLE tenant_api_keys ADD COLUMN issued_by_metadata_json JSON NULL",
	} {
		spec, ok := apiKeys.columns[column]
		if !ok {
			t.Fatalf("tenant_api_keys schema missing %s", column)
		}
		if spec.addSQL != wantAddSQL {
			t.Fatalf("%s addSQL = %q, want %q", column, spec.addSQL, wantAddSQL)
		}
	}

	scopes := mustMetaTableSpec(t, spec, "tenant_api_key_fs_scopes")
	for _, column := range []string{"tenant_id", "api_key_id", "prefix", "prefix_hash", "ops"} {
		if _, ok := scopes.columns[column]; !ok {
			t.Fatalf("tenant_api_key_fs_scopes schema missing %s", column)
		}
	}
	for _, index := range []string{"primary", "idx_fs_scopes_api_key", "idx_fs_scopes_tenant_key"} {
		if _, ok := scopes.indexes[index]; !ok {
			t.Fatalf("tenant_api_key_fs_scopes schema missing index %s", index)
		}
	}
}

func TestMetaSchemaSpecIncludesExternalBindings(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_external_bindings")
	for _, column := range []string{"provider", "subject_key", "tenant_id", "metadata_json", "created_at", "updated_at"} {
		if _, ok := table.columns[column]; !ok {
			t.Fatalf("tenant_external_bindings schema missing %s", column)
		}
	}
	for _, index := range []string{"uk_external_binding_subject", "idx_external_binding_tenant"} {
		if _, ok := table.indexes[index]; !ok {
			t.Fatalf("tenant_external_bindings schema missing index %s", index)
		}
	}
}

func TestMetaSchemaSpecIncludesOrgObjectBackends(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "org_object_backends")
	for _, column := range []string{
		"id", "organization_id", "name", "scheme", "bucket", "prefix", "endpoint",
		"sts_endpoint", "account_id", "secret_cipher", "external_id_cipher", "identity_hash",
	} {
		if _, ok := table.columns[column]; !ok {
			t.Fatalf("org_object_backends schema missing %s", column)
		}
	}
	if _, ok := table.indexes["uk_org_object_backend_identity"]; !ok {
		t.Fatal("org_object_backends schema missing uk_org_object_backend_identity")
	}
	if _, ok := table.indexes["uk_org_object_backend"]; ok {
		t.Fatal("org_object_backends must not keep uk_org_object_backend")
	}
}

func TestMetaSchemaSpecIncludesTenantObjectNamespaces(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_object_namespaces")
	for _, column := range []string{"tenant_id", "namespace_id", "created_at", "updated_at"} {
		if _, ok := table.columns[column]; !ok {
			t.Fatalf("tenant_object_namespaces schema missing %s", column)
		}
	}
}

func TestMetaSchemaSpecIncludesTiDBCloudOrgBindings(t *testing.T) {
	table := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_tidbcloud_org_bindings")
	for _, column := range []string{"tenant_id", "organization_id", "cluster_id", "branch_id", "created_at", "updated_at"} {
		if _, ok := table.columns[column]; !ok {
			t.Fatalf("tenant_tidbcloud_org_bindings schema missing %s", column)
		}
	}
	for _, index := range []string{"primary", "uk_tidbcloud_org_cluster_branch", "idx_tidbcloud_org_cluster", "idx_tidbcloud_org_created"} {
		if _, ok := table.indexes[index]; !ok {
			t.Fatalf("tenant_tidbcloud_org_bindings schema missing index %s", index)
		}
	}
}

func TestMetaSchemaSpecIncludesManagedSharedDBControlPlane(t *testing.T) {
	dbPool := mustMetaTableSpec(t, mustMetaSpec(t), "db_pool")
	for _, column := range []string{
		"db_id", "uuid", "org_id", "cluster_id", "provisioning_key", "cloud_provider", "region",
		"role", "db_host", "db_port", "db_user", "db_password", "db_name", "db_tls",
		"max_tenants", "tenant_count", "soft_cap_reached", "spending_limit", "schema_version", "status", "status_updated_at",
		"created_at", "updated_at",
	} {
		if _, ok := dbPool.columns[column]; !ok {
			t.Fatalf("db_pool schema missing %s", column)
		}
	}
	for _, index := range []string{
		"primary", "uk_db_pool_uuid", "uk_db_pool_cluster_id",
		"idx_db_pool_allocate", "idx_db_pool_provisioning_key", "idx_db_pool_role_status_id",
	} {
		if _, ok := dbPool.indexes[index]; !ok {
			t.Fatalf("db_pool schema missing index %s", index)
		}
	}
	pools := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_tidbcloud_pools")
	if _, ok := pools.indexes["idx_tidbcloud_pool_status_id"]; !ok {
		t.Fatal("tenant_tidbcloud_pools schema missing idx_tidbcloud_pool_status_id")
	}
	if _, ok := dbPool.indexes["uk_db_pool_endpoint"]; ok {
		t.Fatal("db_pool schema must not define endpoint uniqueness")
	}

	memberships := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_pool_memberships")
	for _, column := range []string{
		"tenant_id", "tidbcloud_organization_id", "pool_id", "pool_status", "used_at", "created_at", "updated_at",
	} {
		if _, ok := memberships.columns[column]; !ok {
			t.Fatalf("tenant_pool_memberships schema missing %s", column)
		}
	}
	if _, ok := memberships.columns["organization_id"]; ok {
		t.Fatal("tenant_pool_memberships must not use ambiguous organization_id column")
	}
	for _, index := range []string{"primary", "idx_tenant_pool_claim", "idx_tenant_pool_org_status"} {
		if _, ok := memberships.indexes[index]; !ok {
			t.Fatalf("tenant_pool_memberships schema missing index %s", index)
		}
	}
}

func TestDBPoolClusterIDIndexIsGloballyUnique(t *testing.T) {
	s := newControlStore(t)
	columns, err := loadMetaIndexColumns(context.Background(), s.DB(), "db_pool", "uk_db_pool_cluster_id")
	if err != nil {
		t.Fatalf("load uk_db_pool_cluster_id columns: %v", err)
	}
	if want := []string{"cluster_id"}; !sameStringSlice(columns, want) {
		t.Fatalf("uk_db_pool_cluster_id columns = %v, want %v", columns, want)
	}
}

func TestManagedSharedDBPeriodicScanIndexesMatchKeysetQueries(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	for _, tc := range []struct {
		table string
		index string
		want  []string
	}{
		{table: "db_pool", index: "idx_db_pool_role_status_id", want: []string{"role", "status", "db_id"}},
		{table: "tenant_tidbcloud_pools", index: "idx_tidbcloud_pool_status_id", want: []string{"status", "pool_id"}},
	} {
		columns, err := loadMetaIndexColumns(ctx, s.DB(), tc.table, tc.index)
		if err != nil {
			t.Fatalf("load %s columns: %v", tc.index, err)
		}
		if !sameStringSlice(columns, tc.want) {
			t.Fatalf("%s columns = %v, want %v", tc.index, columns, tc.want)
		}
	}
}

func TestMigrateKeepsLegacyClusterIDIndexWhenReplacementCannotBeCreated(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DROP TABLE db_pool`); err != nil {
		t.Fatalf("drop current db_pool: %v", err)
	}
	t.Cleanup(func() {
		_, _ = s.DB().ExecContext(context.Background(), `DROP TABLE IF EXISTS db_pool`)
		if err := s.migrate(); err != nil {
			t.Errorf("restore current db_pool schema: %v", err)
		}
	})
	if _, err := s.DB().ExecContext(ctx, `CREATE TABLE db_pool (
		db_id BIGINT AUTO_INCREMENT PRIMARY KEY,
		org_id VARCHAR(64) NOT NULL DEFAULT '',
		cluster_id VARCHAR(255) NULL,
		`+"`role`"+` VARCHAR(20) NOT NULL,
		db_host VARCHAR(255) NOT NULL,
		db_port INT NOT NULL,
		db_user VARCHAR(255) NOT NULL,
		db_password VARBINARY(2048) NOT NULL,
		db_name VARCHAR(255) NOT NULL,
		db_tls VARCHAR(32) NOT NULL DEFAULT '',
		max_tenants INT NOT NULL DEFAULT 0,
		tenant_count INT NOT NULL DEFAULT 0,
		status VARCHAR(20) NOT NULL DEFAULT 'active',
		created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
		updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
		UNIQUE INDEX uk_db_pool_cloud_resource (org_id, cluster_id)
	)`); err != nil {
		t.Fatalf("create legacy db_pool: %v", err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO db_pool
		(org_id, cluster_id, `+"`role`"+`, db_host, db_port, db_user, db_password, db_name)
		VALUES
		('org-a', 'duplicate-cluster', 'shared', 'shared.example.com', 4000, 'user-a', X'01', 'shared_db'),
		('org-b', 'duplicate-cluster', 'shared', 'shared.example.com', 4000, 'user-b', X'02', 'shared_db')`); err != nil {
		t.Fatalf("insert duplicate legacy cluster ids: %v", err)
	}

	if err := s.migrate(); err == nil {
		t.Fatal("migrate duplicate legacy cluster ids error = nil, want error")
	}
	legacyIndexExists, err := metaIndexExists(ctx, s.DB(), "db_pool", "uk_db_pool_cloud_resource")
	if err != nil {
		t.Fatalf("check uk_db_pool_cloud_resource: %v", err)
	}
	if !legacyIndexExists {
		t.Fatal("legacy uk_db_pool_cloud_resource was dropped before its replacement was created")
	}
}

func TestMigrateExpandsLegacyDBPoolForManagedProvisioning(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DROP TABLE db_pool`); err != nil {
		t.Fatalf("drop current db_pool: %v", err)
	}
	t.Cleanup(func() {
		_, _ = s.DB().ExecContext(context.Background(), `DROP TABLE IF EXISTS db_pool`)
		if err := s.migrate(); err != nil {
			t.Errorf("restore current db_pool schema: %v", err)
		}
	})
	if _, err := s.DB().ExecContext(ctx, `CREATE TABLE db_pool (
		db_id BIGINT AUTO_INCREMENT PRIMARY KEY,
		org_id VARCHAR(64) NOT NULL DEFAULT '',
		cluster_id VARCHAR(255) NULL,
		`+"`role`"+` VARCHAR(20) NOT NULL,
		db_host VARCHAR(255) NOT NULL,
		db_port INT NOT NULL,
		db_user VARCHAR(255) NOT NULL,
		db_password VARBINARY(2048) NOT NULL,
		db_name VARCHAR(255) NOT NULL,
		db_tls VARCHAR(32) NOT NULL DEFAULT '',
		max_tenants INT NOT NULL DEFAULT 0,
		tenant_count INT NOT NULL DEFAULT 0,
		status VARCHAR(20) NOT NULL DEFAULT 'active',
		created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
		updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
		UNIQUE INDEX uk_db_pool_endpoint (org_id, db_host, db_name),
		UNIQUE INDEX uk_db_pool_cloud_resource (org_id, cluster_id),
		INDEX idx_db_pool_org (org_id, status)
	)`); err != nil {
		t.Fatalf("create legacy db_pool: %v", err)
	}
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO db_pool
		(org_id, `+"`role`"+`, db_host, db_port, db_user, db_password, db_name, max_tenants)
		VALUES ('org-legacy', 'shared', 'legacy.example.com', 4000, 'root', X'01', 'legacy_db', 2)`); err != nil {
		t.Fatalf("insert legacy db_pool row: %v", err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE db_pool SET tenant_count = 2 WHERE org_id = 'org-legacy'`); err != nil {
		t.Fatalf("fill legacy db_pool row: %v", err)
	}

	if err := s.migrate(); err != nil {
		t.Fatalf("migrate legacy db_pool: %v", err)
	}

	for _, column := range []string{"org_id", "db_host", "db_port", "db_user", "db_password", "db_name"} {
		var nullable string
		if err := s.DB().QueryRowContext(ctx, `SELECT is_nullable
			FROM information_schema.columns
			WHERE table_schema = DATABASE() AND table_name = 'db_pool' AND column_name = ?`, column).Scan(&nullable); err != nil {
			t.Fatalf("load db_pool.%s nullability: %v", column, err)
		}
		if nullable != "YES" {
			t.Fatalf("db_pool.%s is_nullable = %q, want YES", column, nullable)
		}
	}
	for _, index := range []string{"uk_db_pool_uuid", "uk_db_pool_cluster_id", "idx_db_pool_allocate", "idx_db_pool_provisioning_key"} {
		exists, err := metaIndexExists(ctx, s.DB(), "db_pool", index)
		if err != nil {
			t.Fatalf("check %s: %v", index, err)
		}
		if !exists {
			t.Fatalf("db_pool index %s was not created", index)
		}
	}
	endpointIndexExists, err := metaIndexExists(ctx, s.DB(), "db_pool", "uk_db_pool_endpoint")
	if err != nil {
		t.Fatalf("check uk_db_pool_endpoint: %v", err)
	}
	if endpointIndexExists {
		t.Fatal("legacy uk_db_pool_endpoint was not dropped")
	}
	legacyCloudIndexExists, err := metaIndexExists(ctx, s.DB(), "db_pool", "uk_db_pool_cloud_resource")
	if err != nil {
		t.Fatalf("check uk_db_pool_cloud_resource: %v", err)
	}
	if legacyCloudIndexExists {
		t.Fatal("legacy uk_db_pool_cloud_resource was not dropped")
	}
	var dbPoolUUID, status string
	var softCapReached bool
	var spendingLimit *int64
	if err := s.DB().QueryRowContext(ctx, `SELECT uuid, status, spending_limit, soft_cap_reached FROM db_pool WHERE org_id = 'org-legacy'`).Scan(&dbPoolUUID, &status, &spendingLimit, &softCapReached); err != nil {
		t.Fatalf("load migrated legacy row: %v", err)
	}
	if _, err := uuid.Parse(dbPoolUUID); err != nil {
		t.Fatalf("migrated db_pool uuid = %q: %v", dbPoolUUID, err)
	}
	var uuidNullable string
	if err := s.DB().QueryRowContext(ctx, `SELECT is_nullable
		FROM information_schema.columns
		WHERE table_schema = DATABASE() AND table_name = 'db_pool' AND column_name = 'uuid'`).Scan(&uuidNullable); err != nil {
		t.Fatalf("load db_pool.uuid nullability: %v", err)
	}
	if uuidNullable != "NO" {
		t.Fatalf("db_pool.uuid is_nullable = %q, want NO", uuidNullable)
	}
	if status != "active" || spendingLimit != nil {
		t.Fatalf("legacy row status/spending_limit = %q/%v, want active/NULL", status, spendingLimit)
	}
	if !softCapReached {
		t.Fatal("legacy row at max_tenants must be backfilled with soft_cap_reached=true")
	}

	if _, err := s.DB().ExecContext(ctx, `DROP INDEX uk_db_pool_uuid ON db_pool`); err != nil {
		t.Fatalf("drop db_pool uuid index to simulate partial migration: %v", err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE db_pool SET uuid = UPPER(uuid) WHERE org_id = 'org-legacy'`); err != nil {
		t.Fatalf("uppercase db_pool uuid to simulate partial migration: %v", err)
	}
	if err := s.migrate(); err != nil {
		t.Fatalf("resume partial db_pool uuid migration: %v", err)
	}
	if err := s.migrate(); err != nil {
		t.Fatalf("repeat completed db_pool uuid migration: %v", err)
	}
	var uuidAfterRetries string
	if err := s.DB().QueryRowContext(ctx, `SELECT uuid FROM db_pool WHERE org_id = 'org-legacy'`).Scan(&uuidAfterRetries); err != nil {
		t.Fatalf("load db_pool uuid after repeated migration: %v", err)
	}
	if uuidAfterRetries != dbPoolUUID {
		t.Fatalf("db_pool uuid after repeated migration = %q, want stable %q", uuidAfterRetries, dbPoolUUID)
	}
	if exists, err := metaIndexExists(ctx, s.DB(), "db_pool", "uk_db_pool_uuid"); err != nil || !exists {
		t.Fatalf("db_pool uuid index after repeated migration = %t/%v, want true/nil", exists, err)
	}
}

func TestMigratePreservesManagedSharedDBSpendingLimit(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	limit := MaxTiDBCloudSpendingLimit
	id, err := s.CreateManagedSharedDBPool(ctx, &SharedDB{
		TiDBCloudOrganizationID: "org-spending-migration", ProvisioningKey: make([]byte, 32),
		CloudProvider: "aws", Region: "us-east-1", MaxTenants: 100, SpendingLimit: &limit,
	})
	if err != nil {
		t.Fatalf("CreateManagedSharedDBPool: %v", err)
	}
	const configuredLimit = int64(2_000_000)
	if _, err := s.DB().ExecContext(ctx, `UPDATE db_pool SET spending_limit = ? WHERE db_id = ?`, configuredLimit, id); err != nil {
		t.Fatalf("seed configured spending limit: %v", err)
	}

	if err := s.migrate(); err != nil {
		t.Fatalf("migrate configured spending limit: %v", err)
	}
	got, err := s.GetSharedDB(ctx, id)
	if err != nil {
		t.Fatalf("GetSharedDB: %v", err)
	}
	if got.SpendingLimit == nil || *got.SpendingLimit != configuredLimit {
		t.Fatalf("spending limit = %v, want %d", got.SpendingLimit, configuredLimit)
	}

	if err := s.migrate(); err != nil {
		t.Fatalf("repeat spending-limit migration: %v", err)
	}
}

func TestUpsertTenantTiDBCloudOrgBindingRejectsDuplicateLiveClusterBranch(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insertTiDBCloudBindingTenant(t, s, "binding-live-tenant", TenantKindLive, TenantActive, "cluster-shared", "", now)
	insertTiDBCloudBindingTenant(t, s, "binding-other-tenant", TenantKindLive, TenantActive, "cluster-shared", "", now)
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       "binding-live-tenant",
		OrganizationID: "org-1",
		ClusterID:      "cluster-shared",
		CreatedAt:      now,
		UpdatedAt:      now,
	}); err != nil {
		t.Fatalf("upsert first binding: %v", err)
	}
	err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       "binding-other-tenant",
		OrganizationID: "org-1",
		ClusterID:      "cluster-shared",
		CreatedAt:      now,
		UpdatedAt:      now,
	})
	if !errors.Is(err, ErrDuplicate) {
		t.Fatalf("duplicate cluster branch upsert err = %v, want ErrDuplicate", err)
	}
}

func TestUpsertTenantTiDBCloudOrgBindingRejectsSharedTenant(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insertTiDBCloudBindingTenant(t, s, "binding-shared-tenant", TenantKindLive, TenantActive, "", "", now)
	if err := s.UpdateTenantProvider(ctx, "binding-shared-tenant", "tidb_cloud_native_shared"); err != nil {
		t.Fatalf("mark tenant shared: %v", err)
	}
	err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID: "binding-shared-tenant", OrganizationID: "org-shared",
		ClusterID: "cluster-shared", CreatedAt: now, UpdatedAt: now,
	})
	if err == nil || !strings.Contains(err.Error(), "shared tenant") {
		t.Fatalf("shared binding error = %v, want shared tenant rejection", err)
	}
}

func TestUpsertTenantTiDBCloudOrgBindingAllowsSharedClusterAcrossBranches(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insertTiDBCloudBindingTenant(t, s, "binding-live-tenant", TenantKindLive, TenantActive, "cluster-shared", "", now)
	insertTiDBCloudBindingTenant(t, s, "binding-fork-tenant", TenantKindFork, TenantActive, "cluster-shared", "branch-1", now)
	for _, tenantID := range []string{"binding-live-tenant", "binding-fork-tenant"} {
		if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
			TenantID:       tenantID,
			OrganizationID: "org-1",
			ClusterID:      "cluster-shared",
			CreatedAt:      now,
			UpdatedAt:      now,
		}); err != nil {
			t.Fatalf("upsert binding for %s: %v", tenantID, err)
		}
	}
	forkBinding, err := s.GetTenantTiDBCloudOrgBinding(ctx, "binding-fork-tenant")
	if err != nil {
		t.Fatalf("get fork binding: %v", err)
	}
	if forkBinding.BranchID != "branch-1" {
		t.Fatalf("fork binding branch id = %q, want branch-1", forkBinding.BranchID)
	}
}

func TestUpsertTenantTiDBCloudOrgBindingIgnoresDeletedClusterBranch(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insertTiDBCloudBindingTenant(t, s, "binding-deleted-tenant", TenantKindLive, TenantDeleted, "cluster-reused", "", now)
	insertTiDBCloudBindingTenant(t, s, "binding-new-tenant", TenantKindLive, TenantActive, "cluster-reused", "", now)
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       "binding-deleted-tenant",
		OrganizationID: "org-1",
		ClusterID:      "cluster-reused",
		CreatedAt:      now,
		UpdatedAt:      now,
	}); err != nil {
		t.Fatalf("upsert deleted binding: %v", err)
	}
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       "binding-new-tenant",
		OrganizationID: "org-1",
		ClusterID:      "cluster-reused",
		CreatedAt:      now,
		UpdatedAt:      now,
	}); err != nil {
		t.Fatalf("upsert new binding after deleted tenant: %v", err)
	}
}

func TestEnsureNoDuplicateTiDBCloudOrgBindingTuplesReportsConflict(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	tenantIDs := []string{"binding-dup-1", "binding-dup-2", "binding-dup-3", "binding-dup-4"}
	if _, err := s.DB().ExecContext(ctx, `DROP INDEX uk_tidbcloud_org_cluster_branch ON tenant_tidbcloud_org_bindings`); err != nil {
		t.Fatalf("drop unique index: %v", err)
	}
	t.Cleanup(func() {
		_, _ = s.DB().ExecContext(context.Background(), `DELETE FROM tenant_tidbcloud_org_bindings WHERE tenant_id IN (?, ?, ?, ?)`, tenantIDs[0], tenantIDs[1], tenantIDs[2], tenantIDs[3])
		_, _ = s.DB().ExecContext(context.Background(), `DELETE FROM tenants WHERE id IN (?, ?, ?, ?)`, tenantIDs[0], tenantIDs[1], tenantIDs[2], tenantIDs[3])
		if err := ensureTiDBCloudOrgBindingUniqueIndex(context.Background(), s.DB()); err != nil {
			t.Fatalf("restore unique index: %v", err)
		}
	})
	insertTiDBCloudBindingTenant(t, s, "binding-dup-1", TenantKindLive, TenantActive, "cluster-dup-a", "", now)
	insertTiDBCloudBindingTenant(t, s, "binding-dup-2", TenantKindLive, TenantActive, "cluster-dup-a", "", now)
	insertTiDBCloudBindingTenant(t, s, "binding-dup-3", TenantKindLive, TenantActive, "cluster-dup-b", "branch-dup", now)
	insertTiDBCloudBindingTenant(t, s, "binding-dup-4", TenantKindLive, TenantActive, "cluster-dup-b", "branch-dup", now)
	for _, binding := range []struct {
		tenantID  string
		clusterID string
		branchID  string
	}{
		{tenantID: "binding-dup-1", clusterID: "cluster-dup-a"},
		{tenantID: "binding-dup-2", clusterID: "cluster-dup-a"},
		{tenantID: "binding-dup-3", clusterID: "cluster-dup-b", branchID: "branch-dup"},
		{tenantID: "binding-dup-4", clusterID: "cluster-dup-b", branchID: "branch-dup"},
	} {
		if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_tidbcloud_org_bindings
			(tenant_id, organization_id, cluster_id, branch_id, pool_id, pool_status, created_at, updated_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
			binding.tenantID, "org-1", binding.clusterID, binding.branchID, "", TenantPoolBindingUsed, now, now); err != nil {
			t.Fatalf("insert duplicate binding %s: %v", binding.tenantID, err)
		}
	}
	err := ensureNoDuplicateTiDBCloudOrgBindingTuples(ctx, s.DB())
	if err == nil {
		t.Fatal("ensureNoDuplicateTiDBCloudOrgBindingTuples error = nil, want duplicate error")
	}
	for _, want := range []string{"found 2 duplicate tuple(s)", "org-1", "cluster-dup-a", "cluster-dup-b", "branch-dup", "binding-dup-1", "binding-dup-2", "binding-dup-3", "binding-dup-4"} {
		if !strings.Contains(err.Error(), want) {
			t.Fatalf("duplicate error %q does not contain %q", err, want)
		}
	}
}

func TestBackfillTiDBCloudOrgBindingBranchIDsTrimsTenantBranchID(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insertTiDBCloudBindingTenant(t, s, "binding-spaced-branch-tenant", TenantKindFork, TenantActive, "cluster-spaced", " branch-1 ", now)
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_tidbcloud_org_bindings
		(tenant_id, organization_id, cluster_id, branch_id, pool_id, pool_status, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
		"binding-spaced-branch-tenant", "org-1", "cluster-spaced", "", "", TenantPoolBindingUsed, now, now); err != nil {
		t.Fatalf("insert raw binding: %v", err)
	}
	if err := backfillTiDBCloudOrgBindingBranchIDs(ctx, s.DB()); err != nil {
		t.Fatalf("backfill branch ids: %v", err)
	}
	binding, err := s.GetTenantTiDBCloudOrgBinding(ctx, "binding-spaced-branch-tenant")
	if err != nil {
		t.Fatalf("get binding: %v", err)
	}
	if binding.BranchID != "branch-1" {
		t.Fatalf("binding branch id = %q, want branch-1", binding.BranchID)
	}
}

func insertTiDBCloudBindingTenant(t *testing.T, s *Store, tenantID string, kind TenantKind, status TenantStatus, clusterID, branchID string, now time.Time) {
	t.Helper()
	if err := s.InsertTenant(context.Background(), &Tenant{
		ID:               tenantID,
		Status:           status,
		Kind:             kind,
		DBHost:           "db.example.com",
		DBPort:           4000,
		DBUser:           "u.root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tidbcloud_fs",
		DBTLS:            true,
		Provider:         tidbCloudNativeProvider,
		ClusterID:        clusterID,
		BranchID:         branchID,
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatalf("insert tenant %s: %v", tenantID, err)
	}
}

func TestClaimOldestFreeTenantPoolBindingRequiresActiveTenant(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.CreateTenantPool(ctx, &TenantPool{
		PoolID:         "pool-claim-active",
		OrganizationID: "org-claim-active",
		Size:           2,
		Status:         TenantPoolActive,
		CreatedAt:      now,
		UpdatedAt:      now,
	}); err != nil {
		t.Fatalf("create tenant pool: %v", err)
	}
	provisioningCreated := now.Add(-2 * time.Minute)
	if err := s.InsertTenant(ctx, &Tenant{
		ID:               "pool-tenant-provisioning",
		Status:           TenantProvisioning,
		Kind:             TenantKindLive,
		DBHost:           "db.example.com",
		DBPort:           4000,
		DBUser:           "u.root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tidbcloud_fs",
		DBTLS:            true,
		Provider:         tidbCloudNativeProvider,
		ClusterID:        "cluster-provisioning",
		SchemaVersion:    1,
		CreatedAt:        provisioningCreated,
		UpdatedAt:        provisioningCreated,
	}); err != nil {
		t.Fatalf("insert provisioning tenant: %v", err)
	}
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       "pool-tenant-provisioning",
		OrganizationID: "org-claim-active",
		ClusterID:      "cluster-provisioning",
		PoolID:         "pool-claim-active",
		PoolStatus:     TenantPoolBindingFree,
		CreatedAt:      provisioningCreated,
		UpdatedAt:      provisioningCreated,
	}); err != nil {
		t.Fatalf("upsert provisioning binding: %v", err)
	}
	if _, err := s.ClaimOldestFreeTenantPoolBinding(ctx, "org-claim-active"); !errors.Is(err, ErrNotFound) {
		t.Errorf("claim provisioning tenant err = %v, want ErrNotFound", err)
	}

	activeCreated := now.Add(-time.Minute)
	if err := s.InsertTenant(ctx, &Tenant{
		ID:               "pool-tenant-active",
		Status:           TenantActive,
		Kind:             TenantKindLive,
		DBHost:           "db.example.com",
		DBPort:           4000,
		DBUser:           "u.root",
		DBPasswordCipher: []byte("cipher"),
		DBName:           "tidbcloud_fs",
		DBTLS:            true,
		Provider:         tidbCloudNativeProvider,
		ClusterID:        "cluster-active",
		SchemaVersion:    1,
		CreatedAt:        activeCreated,
		UpdatedAt:        activeCreated,
	}); err != nil {
		t.Fatalf("insert active tenant: %v", err)
	}
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID:       "pool-tenant-active",
		OrganizationID: "org-claim-active",
		ClusterID:      "cluster-active",
		PoolID:         "pool-claim-active",
		PoolStatus:     TenantPoolBindingFree,
		CreatedAt:      activeCreated,
		UpdatedAt:      activeCreated,
	}); err != nil {
		t.Fatalf("upsert active binding: %v", err)
	}
	row, err := s.ClaimOldestFreeTenantPoolBinding(ctx, "org-claim-active")
	if err != nil {
		t.Errorf("claim active tenant: %v", err)
	} else if row.Tenant.ID != "pool-tenant-active" {
		t.Errorf("claimed tenant = %q, want pool-tenant-active", row.Tenant.ID)
	}
	provisioningBinding, err := s.GetTenantTiDBCloudOrgBinding(ctx, "pool-tenant-provisioning")
	if err != nil {
		t.Fatalf("get provisioning binding: %v", err)
	}
	if provisioningBinding.PoolStatus != TenantPoolBindingFree {
		t.Errorf("provisioning pool status = %s, want %s", provisioningBinding.PoolStatus, TenantPoolBindingFree)
	}
}

func TestClaimFreeTenantPoolBindingDoesNotFallThroughToNextCandidate(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.CreateTenantPool(ctx, &TenantPool{
		PoolID: "pool-exact-claim", OrganizationID: "org-exact-claim", Size: 2,
		Status: TenantPoolActive, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	for i, tenantID := range []string{"exact-claim-a", "exact-claim-b"} {
		createdAt := now.Add(time.Duration(i) * time.Second)
		clusterID := "cluster-" + tenantID
		insertTiDBCloudBindingTenant(t, s, tenantID, TenantKindLive, TenantActive, clusterID, "", createdAt)
		if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
			TenantID: tenantID, OrganizationID: "org-exact-claim", ClusterID: clusterID,
			PoolID: "pool-exact-claim", PoolStatus: TenantPoolBindingFree,
			CreatedAt: createdAt, UpdatedAt: createdAt,
		}); err != nil {
			t.Fatal(err)
		}
	}
	claimed, err := s.ClaimOldestFreeTenantPoolBinding(ctx, "org-exact-claim")
	if err != nil || claimed.Tenant.ID != "exact-claim-a" {
		t.Fatalf("claim candidate A = %+v, err=%v", claimed, err)
	}
	_, err = s.ClaimFreeTenantPoolBinding(ctx, "org-exact-claim", "exact-claim-a", QuotaConfigPatch{})
	if !errors.Is(err, ErrNotFound) {
		t.Fatalf("exact claim error = %v, want ErrNotFound", err)
	}
	binding, err := s.GetTenantTiDBCloudOrgBinding(ctx, "exact-claim-b")
	if err != nil || binding.PoolStatus != TenantPoolBindingFree {
		t.Fatalf("candidate B binding = %+v, err=%v, want free", binding, err)
	}
}

func TestClaimFreeTenantPoolBindingLLMQuotaRequiresEnabledTenantExtractConfig(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	const (
		poolID         = "pool-claim-llm-config"
		organizationID = "org-claim-llm-config"
		tenantID       = "tenant-claim-llm-config"
	)
	if err := s.CreateTenantPool(ctx, &TenantPool{
		PoolID: poolID, OrganizationID: organizationID, Size: 1,
		Status: TenantPoolActive, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	insertTiDBCloudBindingTenant(t, s, tenantID, TenantKindLive, TenantActive,
		"cluster-claim-llm-config", "", now)
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID: tenantID, OrganizationID: organizationID, ClusterID: "cluster-claim-llm-config",
		PoolID: poolID, PoolStatus: TenantPoolBindingFree, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	mediaLimit := int64(12)

	_, err := s.ClaimFreeTenantPoolBinding(ctx, organizationID, tenantID,
		QuotaConfigPatch{MaxMediaLLMFiles: &mediaLimit})
	if !errors.Is(err, ErrTenantExtractConfigNotEnabled) {
		t.Fatalf("claim error = %v, want enabled tenant extract config error", err)
	}
	binding, err := s.GetTenantTiDBCloudOrgBinding(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if binding.PoolStatus != TenantPoolBindingFree {
		t.Fatalf("pool status = %s, want free after rejected claim", binding.PoolStatus)
	}
	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)
	}
	if _, err := s.ClaimFreeTenantPoolBinding(ctx, organizationID, tenantID,
		QuotaConfigPatch{MaxMediaLLMFiles: &mediaLimit}); err != nil {
		t.Fatalf("claim with enabled extract config: %v", err)
	}
	cfg, err := s.GetQuotaConfig(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxMediaLLMFiles != mediaLimit {
		t.Fatalf("max media LLM files = %d, want %d", cfg.MaxMediaLLMFiles, mediaLimit)
	}
}

func TestClaimFreeTenantPoolBindingShadowWritesLegacyClassification(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insertCandidate := func(tenantID, organizationID string) {
		t.Helper()
		insertTiDBCloudBindingTenant(t, s, tenantID, TenantKindLive, TenantActive,
			"cluster-"+tenantID, "", now)
		if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
			TenantID: tenantID, OrganizationID: organizationID, ClusterID: "cluster-" + tenantID,
			PoolID: "pool-legacy-shadow", PoolStatus: TenantPoolBindingFree,
			CreatedAt: now, UpdatedAt: now,
		}); err != nil {
			t.Fatal(err)
		}
	}
	readLegacyLimit := func(tenantID string) sql.NullInt64 {
		t.Helper()
		var got sql.NullInt64
		if err := s.DB().QueryRowContext(ctx,
			"SELECT tidbcloud_spending_limit FROM tenant_quota_config WHERE tenant_id = ?", tenantID).
			Scan(&got); err != nil {
			t.Fatal(err)
		}
		return got
	}

	insertCandidate("tenant-free-shadow", "org-free-shadow")
	if _, err := s.ClaimFreeTenantPoolBindingWithMetadataForFreeOrganization(ctx,
		"org-free-shadow", "tenant-free-shadow", QuotaConfigPatch{}, TenantMetadata{}); err != nil {
		t.Fatal(err)
	}
	if got := readLegacyLimit("tenant-free-shadow"); !got.Valid || got.Int64 != 0 {
		t.Fatalf("free legacy spending limit = %+v, want 0", got)
	}

	insertCandidate("tenant-paid-shadow", "org-paid-shadow")
	if err := s.SetQuotaConfig(ctx, &QuotaConfig{TenantID: "tenant-paid-shadow"}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx,
		"UPDATE tenant_quota_config SET tidbcloud_spending_limit = 0 WHERE tenant_id = ?", "tenant-paid-shadow"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.ClaimFreeTenantPoolBindingWithMetadata(ctx,
		"org-paid-shadow", "tenant-paid-shadow", QuotaConfigPatch{}, TenantMetadata{}); err != nil {
		t.Fatal(err)
	}
	if got := readLegacyLimit("tenant-paid-shadow"); got.Valid {
		t.Fatalf("paid legacy spending limit = %+v, want NULL", got)
	}
}

func TestMarkFreeTenantPoolTenantDeletingKeepsClaimAndDeleteExclusive(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insert := func(tenantID string, poolStatus TenantPoolBindingStatus) {
		t.Helper()
		insertTiDBCloudBindingTenant(t, s, tenantID, TenantKindLive, TenantActive,
			"cluster-"+tenantID, "", now)
		if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
			TenantID: tenantID, OrganizationID: "org-delete-claim-exclusive",
			ClusterID: "cluster-" + tenantID, PoolID: "pool-delete-claim-exclusive",
			PoolStatus: poolStatus, CreatedAt: now, UpdatedAt: now,
		}); err != nil {
			t.Fatal(err)
		}
	}
	insert("delete-claim-used", TenantPoolBindingUsed)
	insert("delete-claim-free", TenantPoolBindingFree)

	updated, err := s.MarkFreeTenantPoolTenantDeleting(ctx, "delete-claim-used", TenantActive)
	if err != nil {
		t.Fatal(err)
	}
	if updated {
		t.Fatal("delete mark acquired a tenant already claimed by a user")
	}
	updated, err = s.MarkFreeTenantPoolTenantDeleting(ctx, "delete-claim-free", TenantActive)
	if err != nil || !updated {
		t.Fatalf("delete mark for free tenant: updated=%v err=%v", updated, err)
	}
	if _, err := s.ClaimFreeTenantPoolBinding(ctx, "org-delete-claim-exclusive", "delete-claim-free", QuotaConfigPatch{}); !errors.Is(err, ErrNotFound) {
		t.Fatalf("claim deleting tenant error = %v, want ErrNotFound", err)
	}
	binding, err := s.GetTenantTiDBCloudOrgBinding(ctx, "delete-claim-free")
	if err != nil {
		t.Fatal(err)
	}
	if binding.PoolStatus != TenantPoolBindingFree {
		t.Fatalf("deleting tenant binding = %s, want %s recovery handle", binding.PoolStatus, TenantPoolBindingFree)
	}
}

func TestPersistTiDBCloudTenantClusterReferenceCommitsReferenceAndBindingAtomically(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insertTiDBCloudBindingTenant(t, s, "early-bind-owner", TenantKindLive, TenantActive, "cluster-conflict", "", now)
	if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
		TenantID: "early-bind-owner", OrganizationID: "org-early-bind", ClusterID: "cluster-conflict",
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	for _, tenantID := range []string{"early-bind-success", "early-bind-rollback"} {
		if err := s.InsertTenant(ctx, &Tenant{
			ID: tenantID, Status: TenantPending, Kind: TenantKindLive,
			DBPasswordCipher: []byte{}, Provider: tidbCloudNativeProvider,
			SchemaVersion: 1, CreatedAt: now, UpdatedAt: now,
		}); err != nil {
			t.Fatal(err)
		}
	}

	if err := s.PersistTiDBCloudTenantClusterReference(ctx, "early-bind-success", "org-early-bind", &Tenant{
		Provider: tidbCloudNativeProvider, ClusterID: "cluster-success",
	}); err != nil {
		t.Fatalf("PersistTiDBCloudTenantClusterReference success: %v", err)
	}
	tenantRow, err := s.GetTenant(ctx, "early-bind-success")
	if err != nil || tenantRow.ClusterID != "cluster-success" || tenantRow.Status != TenantPending {
		t.Fatalf("early-bound tenant = %+v, err=%v", tenantRow, err)
	}
	binding, err := s.GetTenantTiDBCloudOrgBinding(ctx, "early-bind-success")
	if err != nil || binding.OrganizationID != "org-early-bind" || binding.ClusterID != "cluster-success" {
		t.Fatalf("early binding = %+v, err=%v", binding, err)
	}

	err = s.PersistTiDBCloudTenantClusterReference(ctx, "early-bind-rollback", "org-early-bind", &Tenant{
		Provider: tidbCloudNativeProvider, ClusterID: "cluster-conflict",
	})
	if err == nil {
		t.Fatal("conflicting early binding succeeded")
	}
	tenantRow, err = s.GetTenant(ctx, "early-bind-rollback")
	if err != nil || tenantRow.ClusterID != "" {
		t.Fatalf("rolled-back tenant = %+v, err=%v", tenantRow, err)
	}
	if _, err := s.GetTenantTiDBCloudOrgBinding(ctx, "early-bind-rollback"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("rolled-back binding error = %v, want ErrNotFound", err)
	}
}

func TestFinalizeTenantConnectionAndStatusCASIsAtomic(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID: "finalize-native", Status: TenantPending, Kind: TenantKindLive,
		DBPasswordCipher: []byte{}, Provider: tidbCloudNativeProvider, ClusterID: "cluster-finalize",
		SchemaVersion: 1, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	updated, err := s.FinalizeTenantConnection(ctx, "finalize-native", TenantPending, TenantProvisioning, &Tenant{
		DBHost: "db.example", DBPort: 4000, DBUser: "u1.root", DBPasswordCipher: []byte("cipher"),
		DBName: "tidbcloud_fs", DBTLS: true, Provider: tidbCloudNativeProvider, ClusterID: "cluster-finalize",
	})
	if err != nil || !updated {
		t.Fatalf("FinalizeTenantConnection updated=%v err=%v", updated, err)
	}
	row, err := s.GetTenant(ctx, "finalize-native")
	if err != nil || row.Status != TenantProvisioning || row.DBHost != "db.example" || row.DBUser != "u1.root" || string(row.DBPasswordCipher) != "cipher" {
		t.Fatalf("finalized tenant = %+v, err=%v", row, err)
	}

	updated, err = s.FinalizeTenantConnection(ctx, "finalize-native", TenantPending, TenantProvisioning, &Tenant{
		DBHost: "wrong.example", DBPort: 4000, DBUser: "wrong.root", DBPasswordCipher: []byte("wrong"),
		DBName: "tidbcloud_fs", DBTLS: true, Provider: tidbCloudNativeProvider, ClusterID: "cluster-finalize",
	})
	if err != nil || updated {
		t.Fatalf("second FinalizeTenantConnection updated=%v err=%v, want false/nil", updated, err)
	}
	row, err = s.GetTenant(ctx, "finalize-native")
	if err != nil || row.DBHost != "db.example" || row.DBUser != "u1.root" || string(row.DBPasswordCipher) != "cipher" {
		t.Fatalf("tenant changed after lost CAS = %+v, err=%v", row, err)
	}
}

func TestCountTenantPoolBindingsByStatusGroupsByPoolOrganizationAndStatus(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	for _, pool := range []TenantPool{
		{
			PoolID:         "pool-binding-counts-a",
			OrganizationID: "org-binding-counts-a",
			Size:           4,
			Status:         TenantPoolActive,
			CreatedAt:      now,
			UpdatedAt:      now,
		},
		{
			PoolID:         "pool-binding-counts-empty",
			OrganizationID: "org-binding-counts-empty",
			Size:           2,
			Status:         TenantPoolActive,
			CreatedAt:      now,
			UpdatedAt:      now,
		},
	} {
		if err := s.CreateTenantPool(ctx, &pool); err != nil {
			t.Fatalf("create tenant pool %s: %v", pool.PoolID, err)
		}
	}
	for _, tc := range []struct {
		tenantID string
		cluster  string
		status   TenantStatus
		pool     TenantPoolBindingStatus
	}{
		{tenantID: "binding-counts-free-1", cluster: "cluster-binding-counts-free-1", status: TenantActive, pool: TenantPoolBindingFree},
		{tenantID: "binding-counts-free-2", cluster: "cluster-binding-counts-free-2", status: TenantProvisioning, pool: TenantPoolBindingFree},
		{tenantID: "binding-counts-used-1", cluster: "cluster-binding-counts-used-1", status: TenantActive, pool: TenantPoolBindingUsed},
		{tenantID: "binding-counts-deleted-1", cluster: "cluster-binding-counts-deleted-1", status: TenantDeleted, pool: TenantPoolBindingFree},
	} {
		insertTiDBCloudBindingTenant(t, s, tc.tenantID, TenantKindLive, tc.status, tc.cluster, "", now)
		if err := s.UpsertTenantTiDBCloudOrgBinding(ctx, &TenantTiDBCloudOrgBinding{
			TenantID:       tc.tenantID,
			OrganizationID: "org-binding-counts-a",
			ClusterID:      tc.cluster,
			PoolID:         "pool-binding-counts-a",
			PoolStatus:     tc.pool,
			CreatedAt:      now,
			UpdatedAt:      now,
		}); err != nil {
			t.Fatalf("upsert binding %s: %v", tc.tenantID, err)
		}
	}
	for _, tc := range []struct {
		tenantID string
		status   TenantStatus
		pool     TenantPoolBindingStatus
	}{
		{tenantID: "binding-counts-shared-free", status: TenantActive, pool: TenantPoolBindingFree},
		{tenantID: "binding-counts-shared-deleted", status: TenantDeleted, pool: TenantPoolBindingUsed},
	} {
		insertTenantForPoolMembershipTest(t, s, tc.tenantID, tc.status, now)
		if err := s.UpsertTenantPoolMembership(ctx, &TenantPoolMembership{
			TenantID:                tc.tenantID,
			TiDBCloudOrganizationID: "org-binding-counts-a",
			PoolID:                  "pool-binding-counts-a",
			PoolStatus:              tc.pool,
			CreatedAt:               now,
			UpdatedAt:               now,
		}); err != nil {
			t.Fatalf("upsert shared membership %s: %v", tc.tenantID, err)
		}
	}

	counts, err := s.CountTenantPoolBindingsByStatus(ctx)
	if err != nil {
		t.Fatalf("count tenant pool bindings by status: %v", err)
	}
	got := map[string]int64{}
	for _, count := range counts {
		got[count.PoolID+"|"+count.OrganizationID+"|"+string(count.Status)] = count.Count
	}
	want := map[string]int64{
		"pool-binding-counts-a|org-binding-counts-a|free":         3,
		"pool-binding-counts-a|org-binding-counts-a|used":         1,
		"pool-binding-counts-empty|org-binding-counts-empty|free": 0,
		"pool-binding-counts-empty|org-binding-counts-empty|used": 0,
	}
	for key, wantCount := range want {
		if got[key] != wantCount {
			t.Errorf("count %s = %d, want %d; all counts=%v", key, got[key], wantCount, got)
		}
	}
}

func TestListTenantPoolsByStatusAfterUsesStableKeysetPagination(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	for _, pool := range []TenantPool{
		{PoolID: "pool-a", OrganizationID: "org-a", Size: 1, Status: TenantPoolActive, CreatedAt: now, UpdatedAt: now},
		{PoolID: "pool-b", OrganizationID: "org-b", Size: 2, Status: TenantPoolDeleting, CreatedAt: now, UpdatedAt: now},
		{PoolID: "pool-c", OrganizationID: "org-c", Size: 3, Status: TenantPoolActive, CreatedAt: now, UpdatedAt: now},
	} {
		if err := s.CreateTenantPool(ctx, &pool); err != nil {
			t.Fatalf("create tenant pool %s: %v", pool.PoolID, err)
		}
	}

	first, err := s.ListTenantPoolsByStatusAfter(ctx, TenantPoolActive, "", 1)
	if err != nil {
		t.Fatalf("first page: %v", err)
	}
	if len(first) != 1 || first[0].PoolID != "pool-a" {
		t.Fatalf("first page = %+v, want pool-a", first)
	}
	second, err := s.ListTenantPoolsByStatusAfter(ctx, TenantPoolActive, first[0].PoolID, 10)
	if err != nil {
		t.Fatalf("second page: %v", err)
	}
	if len(second) != 1 || second[0].PoolID != "pool-c" {
		t.Fatalf("second page = %+v, want pool-c", second)
	}
}

func TestDiffMetaTableMetaReportsMissingPrimaryKeyConstraint(t *testing.T) {
	spec := mustMetaTableSpec(t, mustMetaSpec(t), "tenant_quota_config")
	meta := metaTableMeta{
		tableName: "tenant_quota_config",
		columns: map[string]metaColumnMeta{
			"tenant_id":           {columnType: "varchar(64)"},
			"max_storage_bytes":   {columnType: "bigint"},
			"max_file_size_bytes": {columnType: "bigint"},
			"max_file_count":      {columnType: "bigint"},
			"max_media_llm_files": {columnType: "bigint"},
			"max_video_llm_files": {columnType: "bigint"},
			"max_monthly_cost_mc": {columnType: "bigint"},
			"created_at":          {columnType: "datetime(3)"},
			"updated_at":          {columnType: "datetime(3)"},
		},
	}
	createStmt := `CREATE TABLE tenant_quota_config (
		tenant_id VARCHAR(64) NOT NULL,
		max_storage_bytes BIGINT NOT NULL,
		max_file_size_bytes BIGINT NOT NULL,
		max_file_count BIGINT NOT NULL,
		max_media_llm_files BIGINT NOT NULL,
		max_video_llm_files BIGINT NOT NULL,
		max_monthly_cost_mc BIGINT NOT NULL,
		created_at DATETIME(3) NOT NULL,
		updated_at DATETIME(3) NOT NULL
	)`
	diffs := diffMetaTableMeta(spec, meta, createStmt)
	if !hasMetaDiff(diffs, metaSchemaDiffMissingIndex, "missing primary key constraint") {
		t.Fatalf("expected missing primary key diff, got %#v", diffs)
	}
}

func TestPlannedMetaSchemaRepairsSkipsUnsafeRepairs(t *testing.T) {
	diffs := []metaSchemaDiff{
		{kind: metaSchemaDiffMissingColumn, tableName: "tenant_api_keys", columnName: "must_fill", repairSQL: "ALTER TABLE tenant_api_keys ADD COLUMN must_fill BIGINT NOT NULL"},
		{kind: metaSchemaDiffMissingIndex, tableName: "tenant_api_keys", indexName: "uk_key_name", repairSQL: "CREATE UNIQUE INDEX uk_key_name ON tenant_api_keys(key_name)"},
		{kind: metaSchemaDiffMissingIndex, tableName: "tenant_api_keys", indexName: "idx_api_keys_tenant", repairSQL: "CREATE INDEX idx_api_keys_tenant ON tenant_api_keys(tenant_id, status)"},
	}

	plans := plannedMetaSchemaRepairs(diffs)
	if len(plans) != 1 {
		t.Fatalf("expected exactly one safe repair, got %#v", plans)
	}
	if plans[0] != "CREATE INDEX idx_api_keys_tenant ON tenant_api_keys(tenant_id, status)" {
		t.Fatalf("unexpected plan: %#v", plans)
	}
}

func mustMetaSpec(t *testing.T) metaSchemaSpec {
	t.Helper()
	spec, err := metaSchemaSpecFromStatements(metaInitSchemaStatements())
	if err != nil {
		t.Fatalf("meta schema spec: %v", err)
	}
	return spec
}

func mustMetaTableSpec(t *testing.T, spec metaSchemaSpec, tableName string) metaTableSpec {
	t.Helper()
	for _, table := range spec.tables {
		if table.name == tableName {
			return table
		}
	}
	t.Fatalf("missing table %q in meta schema spec", tableName)
	return metaTableSpec{}
}

func hasMetaDiff(diffs []metaSchemaDiff, kind metaSchemaDiffKind, contains string) bool {
	for _, diff := range diffs {
		if diff.kind != kind {
			continue
		}
		if strings.Contains(strings.ToLower(diff.detail), strings.ToLower(contains)) {
			return true
		}
	}
	return false
}

func TestColumnTypeMismatchSchemaVersionPlansRepair(t *testing.T) {
	spec := mustMetaSpec(t)
	tenantsSpec := mustMetaTableSpec(t, spec, "tenants")

	// Simulate tenants table with schema_version as INT (old type).
	observed := metaTableMeta{
		tableName: "tenants",
		columns:   map[string]metaColumnMeta{"schema_version": {columnType: "int"}},
	}
	diffs := diffMetaTableMetaWithObservedIndexes(tenantsSpec, observed, "", map[string]struct{}{})

	var typeDiff *metaSchemaDiff
	for i := range diffs {
		if diffs[i].kind == metaSchemaDiffColumnType && diffs[i].columnName == "schema_version" {
			typeDiff = &diffs[i]
			break
		}
	}
	if typeDiff == nil {
		t.Fatal("expected a column_type_mismatch diff for schema_version, got none")
		return
	}
	if typeDiff.repairSQL == "" {
		t.Fatal("expected repairSQL to be set for schema_version type mismatch")
	}

	plans := plannedMetaSchemaRepairs([]metaSchemaDiff{*typeDiff})
	if len(plans) != 1 {
		t.Fatalf("expected exactly one planned repair, got %#v", plans)
	}
	want := "ALTER TABLE tenants MODIFY COLUMN schema_version INT UNSIGNED NOT NULL DEFAULT 1"
	if plans[0] != want {
		t.Fatalf("unexpected repair plan:\n  got  %q\n  want %q", plans[0], want)
	}
}

func TestColumnTypeMismatchOtherColumnsNoRepair(t *testing.T) {
	spec := mustMetaSpec(t)
	tenantsSpec := mustMetaTableSpec(t, spec, "tenants")

	// Simulate a type mismatch on a column other than schema_version — no auto-repair expected.
	observed := metaTableMeta{
		tableName: "tenants",
		columns:   map[string]metaColumnMeta{"db_port": {columnType: "bigint"}},
	}
	diffs := diffMetaTableMetaWithObservedIndexes(tenantsSpec, observed, "", map[string]struct{}{})

	for _, d := range diffs {
		if d.kind == metaSchemaDiffColumnType && d.columnName != "schema_version" && d.repairSQL != "" {
			t.Errorf("unexpected repairSQL for non-schema_version column %q: %q", d.columnName, d.repairSQL)
		}
	}
}

func TestIsSafeModifyColumnRepairSQLAcceptsSchemaVersion(t *testing.T) {
	diff := metaSchemaDiff{
		tableName:  "tenants",
		columnName: "schema_version",
		repairSQL:  "ALTER TABLE tenants MODIFY COLUMN schema_version INT UNSIGNED NOT NULL DEFAULT 1",
	}
	if !isSafeModifyColumnRepairSQL(diff) {
		t.Fatal("expected isSafeModifyColumnRepairSQL to return true for schema_version repair")
	}
}

func TestIsSafeModifyColumnRepairSQLRejectsOtherCases(t *testing.T) {
	cases := []metaSchemaDiff{
		{tableName: "tenants", columnName: "db_port", repairSQL: "ALTER TABLE tenants MODIFY COLUMN db_port INT UNSIGNED NOT NULL"},
		{tableName: "tenant_api_keys", columnName: "schema_version", repairSQL: "ALTER TABLE tenant_api_keys MODIFY COLUMN schema_version INT UNSIGNED NOT NULL DEFAULT 1"},
		{tableName: "tenants", columnName: "schema_version", repairSQL: "ALTER TABLE tenants MODIFY COLUMN schema_version BIGINT NOT NULL DEFAULT 1"},
		{tableName: "tenants", columnName: "schema_version", repairSQL: ""},
		{tableName: "tenants", columnName: "schema_version", repairSQL: "ALTER TABLE tenants MODIFY COLUMN schema_version INT UNSIGNED"},
	}
	for _, diff := range cases {
		if isSafeModifyColumnRepairSQL(diff) {
			t.Errorf("isSafeModifyColumnRepairSQL(%q.%q sql=%q) = true, want false",
				diff.tableName, diff.columnName, diff.repairSQL)
		}
	}
}

func seedTenantExtractConfig(t *testing.T, s *Store, tenantID string) {
	t.Helper()
	if err := s.UpsertTenantExtractConfig(context.Background(), &TenantExtractConfig{
		TenantID: tenantID, MediaType: ExtractMediaTypeImage, Enabled: true,
		APIBase: "https://api.example.com", APIKeyCipher: []byte("cipher"), Model: "m",
	}); err != nil {
		t.Fatalf("seed extract config: %v", err)
	}
	if err := s.UpsertTenantEmbeddingConfig(context.Background(), &TenantEmbeddingConfig{
		TenantID: tenantID, Enabled: true, APIBase: "https://api.example.com/v1",
		APIKeyCipher: []byte("embed-cipher"), Model: "embed-model", Generation: 1, ConfigRevision: 1,
	}); err != nil {
		t.Fatalf("seed embedding config: %v", err)
	}
}

func assertTenantExtractConfigWiped(t *testing.T, s *Store, tenantID string) {
	t.Helper()
	if _, err := s.GetTenantExtractConfig(context.Background(), tenantID, ExtractMediaTypeImage); !errors.Is(err, ErrNotFound) {
		t.Fatalf("extract config after tenant deletion = %v, want %v", err, ErrNotFound)
	}
	if _, err := s.GetTenantEmbeddingConfig(context.Background(), tenantID); !errors.Is(err, ErrNotFound) {
		t.Fatalf("embedding config after tenant deletion = %v, want %v", err, ErrNotFound)
	}
}

func TestLockTenantForConfigUpdateTx(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	insert := func(tenantID string, status TenantStatus) {
		t.Helper()
		if err := s.InsertTenant(ctx, &Tenant{
			ID:               tenantID,
			Status:           status,
			DBHost:           "127.0.0.1",
			DBPort:           4000,
			DBUser:           "root",
			DBPasswordCipher: []byte("cipher"),
			DBName:           "db_" + tenantID,
			Provider:         "tidb_zero",
			SchemaVersion:    1,
			CreatedAt:        now,
			UpdatedAt:        now,
		}); err != nil {
			t.Fatal(err)
		}
	}
	insert("lock-cfg-active", TenantActive)
	insert("lock-cfg-pending", TenantPending)
	insert("lock-cfg-deleting", TenantDeleting)
	insert("lock-cfg-deleted", TenantDeleted)

	lock := func(tenantID string) error {
		return s.InTx(ctx, func(tx *sql.Tx) error {
			return LockTenantForConfigUpdateTx(ctx, tx, tenantID)
		})
	}
	for _, tenantID := range []string{"lock-cfg-active", "lock-cfg-pending"} {
		if err := lock(tenantID); err != nil {
			t.Errorf("lock(%s) = %v, want nil", tenantID, err)
		}
	}
	// Missing and deleting/deleted tenants fail closed with ErrNotFound so the
	// config write cannot re-insert a row for a tenant on its way out.
	for _, tenantID := range []string{"lock-cfg-missing", "lock-cfg-deleting", "lock-cfg-deleted"} {
		if err := lock(tenantID); !errors.Is(err, ErrNotFound) {
			t.Errorf("lock(%s) = %v, want %v", tenantID, err, ErrNotFound)
		}
	}
}

func TestTenantExtractConfigCRUD(t *testing.T) {
	s, cleanup := newTestStore(t)
	defer cleanup()
	ctx := context.Background()
	if _, err := s.GetTenantExtractConfig(ctx, "t1", ExtractMediaTypeImage); !errors.Is(err, ErrNotFound) {
		t.Fatalf("absent: %v", err)
	}
	cfg := &TenantExtractConfig{TenantID: "t1", MediaType: ExtractMediaTypeImage,
		Enabled: true, APIBase: "https://api.example.com", APIKeyCipher: []byte("cipher"), Model: "m", Prompt: "p"}
	if err := s.UpsertTenantExtractConfig(ctx, cfg); err != nil {
		t.Fatal(err)
	}
	got, err := s.GetTenantExtractConfig(ctx, "t1", ExtractMediaTypeImage)
	if err != nil {
		t.Fatal(err)
	}
	if !got.Enabled || got.APIBase != cfg.APIBase || string(got.APIKeyCipher) != "cipher" || got.Model != "m" || got.Prompt != "p" {
		t.Fatalf("got %+v", got)
	}
	// upsert replaces (off-marker shape: provider fields back to NULL)
	cfg.Prompt, cfg.Enabled = "", false
	cfg.APIBase, cfg.APIKeyCipher, cfg.Model = "", nil, ""
	if err := s.UpsertTenantExtractConfig(ctx, cfg); err != nil {
		t.Fatal(err)
	}
	got, _ = s.GetTenantExtractConfig(ctx, "t1", ExtractMediaTypeImage)
	if got.Enabled || got.APIBase != "" || got.APIKeyCipher != nil {
		t.Fatalf("off-marker got %+v", got)
	}
	// per-type isolation
	if _, err := s.GetTenantExtractConfig(ctx, "t1", ExtractMediaTypeVideo); !errors.Is(err, ErrNotFound) {
		t.Fatalf("video should be absent: %v", err)
	}
	if err := s.DeleteTenantExtractConfigs(ctx, "t1"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.GetTenantExtractConfig(ctx, "t1", ExtractMediaTypeImage); !errors.Is(err, ErrNotFound) {
		t.Fatalf("after delete: %v", err)
	}
	// invalid media type rejected at API layer
	if err := s.UpsertTenantExtractConfig(ctx, &TenantExtractConfig{TenantID: "t1", MediaType: "bogus"}); err == nil {
		t.Fatal("want error for bogus media type")
	}
}

// TestListPendingMutationsFairAcrossTenants pins the starvation fix: a
// high-volume tenant whose id sorts first must not crowd out later-sorting
// tenants' entries from a replay batch.
func TestListPendingMutationsFairAcrossTenants(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DELETE FROM quota_mutation_log`); err != nil {
		t.Fatal(err)
	}
	hot := "00-hot-tenant"
	cold := "zz-cold-tenant"
	insert := func(tenantID string, n int) {
		t.Helper()
		for i := 0; i < n; i++ {
			if _, err := s.InsertMutationLog(ctx, &MutationLogEntry{
				TenantID: tenantID, MutationType: "file_create", MutationData: []byte(`{"file_id":"f"}`),
			}); err != nil {
				t.Fatal(err)
			}
		}
	}
	insert(hot, 150)
	insert(cold, 5)
	if _, err := s.DB().ExecContext(ctx, `UPDATE quota_mutation_log SET created_at = ?`, time.Now().UTC().Add(-time.Minute)); err != nil {
		t.Fatal(err)
	}

	rows, err := s.ListPendingMutations(ctx, time.Minute, 100)
	if err != nil {
		t.Fatal(err)
	}
	counts := make(map[string]int)
	lastID := make(map[string]int64)
	for _, r := range rows {
		counts[r.TenantID]++
		if r.ID <= lastID[r.TenantID] {
			t.Fatalf("entries for %s not id-ordered: id %d after %d", r.TenantID, r.ID, lastID[r.TenantID])
		}
		lastID[r.TenantID] = r.ID
	}
	if counts[cold] != 5 {
		t.Fatalf("cold tenant entries = %d, want 5 (starved by hot tenant)", counts[cold])
	}
	if counts[hot] != 95 {
		t.Fatalf("hot tenant entries = %d, want 95 (share 50 + redistributed top-up)", counts[hot])
	}
	if len(rows) != 100 {
		t.Fatalf("batch size = %d, want full batch of 100", len(rows))
	}
}

// TestPurgeAppliedMutations covers the retention purge: only applied rows
// older than the cutoff are deleted, pending/failed/recent rows survive, and
// the batch limit is honored.
func TestPurgeAppliedMutations(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DELETE FROM quota_mutation_log`); err != nil {
		t.Fatal(err)
	}
	old := time.Now().UTC().Add(-8 * 24 * time.Hour)
	recent := time.Now().UTC().Add(-time.Hour)
	insert := func(tenantID, status string, createdAt time.Time) int64 {
		t.Helper()
		id, err := s.InsertMutationLog(ctx, &MutationLogEntry{
			TenantID: tenantID, MutationType: "file_create", MutationData: []byte(`{"file_id":"f"}`),
		})
		if err != nil {
			t.Fatal(err)
		}
		if _, err := s.DB().ExecContext(ctx,
			`UPDATE quota_mutation_log SET status = ?, created_at = ?, applied_at = ? WHERE id = ?`,
			status, createdAt, createdAt, id); err != nil {
			t.Fatal(err)
		}
		return id
	}
	for i := 0; i < 10; i++ {
		insert("tenant-old", "applied", old)
	}
	insert("tenant-recent", "applied", recent)
	insert("tenant-pending", "pending", old)
	insert("tenant-failed", "failed", old)
	// Created long ago but applied only recently (post-backlog convergence):
	// still fresh from an audit perspective and must survive the purge.
	lateApplied, err := s.InsertMutationLog(ctx, &MutationLogEntry{
		TenantID: "tenant-late", MutationType: "file_create", MutationData: []byte(`{"file_id":"f"}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx,
		`UPDATE quota_mutation_log SET status = 'applied', created_at = ?, applied_at = ? WHERE id = ?`,
		old, recent, lateApplied); err != nil {
		t.Fatal(err)
	}

	cutoff := time.Now().UTC().Add(-7 * 24 * time.Hour)
	deleted, err := s.PurgeAppliedMutations(ctx, cutoff, 4)
	if err != nil {
		t.Fatal(err)
	}
	if deleted != 4 {
		t.Fatalf("first purge deleted = %d, want 4 (batch limit)", deleted)
	}
	deleted, err = s.PurgeAppliedMutations(ctx, cutoff, 100)
	if err != nil {
		t.Fatal(err)
	}
	if deleted != 6 {
		t.Fatalf("second purge deleted = %d, want 6", deleted)
	}
	var remaining int
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM quota_mutation_log`).Scan(&remaining); err != nil {
		t.Fatal(err)
	}
	if remaining != 4 {
		t.Fatalf("remaining rows = %d, want 4 (recent applied + pending + failed + late-applied)", remaining)
	}
}

// TestListPendingMutationsFloorShareCoversAllTenants is the regression test
// for the ceil-share starvation found in review: with 32 selected tenants
// each holding >= 4 pending rows and limit 100, every tenant must appear in
// the batch (floor share 3) instead of the first 25 ceil-sharing tenants
// exhausting it.
func TestListPendingMutationsFloorShareCoversAllTenants(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DELETE FROM quota_mutation_log`); err != nil {
		t.Fatal(err)
	}
	const tenantCount = 32
	const perTenantPending = 4
	for i := 0; i < tenantCount; i++ {
		tenantID := fmt.Sprintf("floor-tenant-%02d", i)
		for j := 0; j < perTenantPending; j++ {
			if _, err := s.InsertMutationLog(ctx, &MutationLogEntry{
				TenantID: tenantID, MutationType: "file_create", MutationData: []byte(`{"file_id":"f"}`),
			}); err != nil {
				t.Fatal(err)
			}
		}
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE quota_mutation_log SET created_at = ?`, time.Now().UTC().Add(-time.Minute)); err != nil {
		t.Fatal(err)
	}

	rows, err := s.ListPendingMutations(ctx, time.Minute, 100)
	if err != nil {
		t.Fatal(err)
	}
	counts := make(map[string]int)
	for _, r := range rows {
		counts[r.TenantID]++
	}
	if len(counts) != tenantCount {
		t.Fatalf("tenants covered = %d, want %d (later tenants starved)", len(counts), tenantCount)
	}
	for i := 0; i < tenantCount; i++ {
		tenantID := fmt.Sprintf("floor-tenant-%02d", i)
		if counts[tenantID] < 3 {
			t.Fatalf("tenant %s got %d rows, want at least floor(100/32)=3", tenantID, counts[tenantID])
		}
	}
	if len(rows) != 100 {
		t.Fatalf("batch size = %d, want 100 (top-up should fill the batch)", len(rows))
	}
}

// TestPurgeAppliedMutationsEmptyTable pins the empty-table early return:
// aggregate functions over zero rows produce a NULL row, which must not
// surface as a scan error on every purge tick of a fresh cluster.
func TestPurgeAppliedMutationsEmptyTable(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	if _, err := s.DB().ExecContext(ctx, `DELETE FROM quota_mutation_log`); err != nil {
		t.Fatal(err)
	}
	deleted, err := s.PurgeAppliedMutations(ctx, time.Now().UTC(), 100)
	if err != nil {
		t.Fatalf("purge on empty table: %v", err)
	}
	if deleted != 0 {
		t.Fatalf("deleted = %d, want 0", deleted)
	}
}