package server

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/tenant/token"
	"github.com/mem9-ai/drive9/pkg/testtidb"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

func TestProcessTenantDeleteJobWaitsForSharedPurge(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-job-guard", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_job_guard", MaxTenants: 100,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatal(err)
	}
	ns, err := rt.meta.EnsureTenantStorageNamespace(ctx, rt.tenantID, "local", "", rt.tenantID+"/")
	if err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpdateTenantStatus(ctx, rt.tenantID, meta.TenantDeleting); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpdateStorageNamespaceState(ctx, ns.ID, meta.StorageNamespaceDeleting); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID: rt.tenantID, NamespaceID: ns.ID, Backend: ns.Backend,
		Bucket: ns.Bucket, Prefix: ns.Prefix,
	}); err != nil {
		t.Fatal(err)
	}

	rt.server.processTenantDeleteJobs(ctx)

	got, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != meta.TenantDeleting {
		t.Fatalf("tenant status = %s, want %s while shared placement is not deleting", got.Status, meta.TenantDeleting)
	}
	placement, err := rt.meta.GetTenantPlacement(ctx, fsID)
	if err != nil {
		t.Fatal(err)
	}
	if placement.Status == meta.PlacementStatusDeleting {
		t.Fatalf("placement status = %s, want purge-incomplete active placement", placement.Status)
	}
	var jobState string
	if err := rt.meta.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, rt.tenantID).Scan(&jobState); err != nil {
		t.Fatal(err)
	}
	if jobState != string(meta.TenantDeleteJobPending) {
		t.Fatalf("job state = %s, want %s for retry", jobState, meta.TenantDeleteJobPending)
	}
}

func TestProcessTenantDeleteJobRejectsMissingSharedPlacement(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	if _, err := rt.meta.EnsureFsID(ctx, rt.tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.EnsureTenantStorageNamespace(ctx, rt.tenantID, "local", "", rt.tenantID+"/"); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpdateTenantStatus(ctx, rt.tenantID, meta.TenantDeleting); err != nil {
		t.Fatal(err)
	}
	ns, err := rt.meta.GetStorageNamespace(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpdateStorageNamespaceState(ctx, ns.ID, meta.StorageNamespaceDeleting); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID: rt.tenantID, NamespaceID: ns.ID, Backend: ns.Backend,
		Bucket: ns.Bucket, Prefix: ns.Prefix,
	}); err != nil {
		t.Fatal(err)
	}

	rt.server.processTenantDeleteJobs(ctx)

	got, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != meta.TenantDeleting {
		t.Fatalf("tenant status = %s, want %s while shared placement is missing", got.Status, meta.TenantDeleting)
	}
	var jobState, namespaceState string
	if err := rt.meta.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, rt.tenantID).Scan(&jobState); err != nil {
		t.Fatal(err)
	}
	if jobState != string(meta.TenantDeleteJobPending) {
		t.Fatalf("job state = %s, want %s for retry", jobState, meta.TenantDeleteJobPending)
	}
	if err := rt.meta.DB().QueryRow(`SELECT state FROM storage_namespaces WHERE namespace_id = ?`, ns.ID).Scan(&namespaceState); err != nil {
		t.Fatal(err)
	}
	if namespaceState != string(meta.StorageNamespaceDeleting) {
		t.Fatalf("namespace state = %s, want %s", namespaceState, meta.StorageNamespaceDeleting)
	}
}

func TestProcessTenantDeleteJobFinalizesDeletingSharedPlacement(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-job-finalize", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_job_finalize", MaxTenants: 100,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, `UPDATE tenant_placements SET status = ? WHERE fs_id = ?`,
		meta.PlacementStatusDeleting, fsID); err != nil {
		t.Fatal(err)
	}
	ns, err := rt.meta.EnsureTenantStorageNamespace(ctx, rt.tenantID, "local", "", rt.tenantID+"/")
	if err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpdateTenantStatus(ctx, rt.tenantID, meta.TenantDeleting); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpdateStorageNamespaceState(ctx, ns.ID, meta.StorageNamespaceDeleting); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID: rt.tenantID, NamespaceID: ns.ID, Backend: ns.Backend,
		Bucket: ns.Bucket, Prefix: ns.Prefix,
	}); err != nil {
		t.Fatal(err)
	}

	rt.server.processTenantDeleteJobs(ctx)

	got, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != meta.TenantDeleted {
		t.Fatalf("tenant status = %s, want %s", got.Status, meta.TenantDeleted)
	}
	if _, err := rt.meta.GetTenantPlacement(ctx, fsID); !errors.Is(err, meta.ErrNotFound) {
		t.Fatalf("placement lookup error = %v, want ErrNotFound", err)
	}
	var jobState, namespaceState string
	if err := rt.meta.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, rt.tenantID).Scan(&jobState); err != nil {
		t.Fatal(err)
	}
	if jobState != string(meta.TenantDeleteJobDeleted) {
		t.Fatalf("job state = %s, want %s", jobState, meta.TenantDeleteJobDeleted)
	}
	if err := rt.meta.DB().QueryRow(`SELECT state FROM storage_namespaces WHERE namespace_id = ?`, ns.ID).Scan(&namespaceState); err != nil {
		t.Fatal(err)
	}
	if namespaceState != string(meta.StorageNamespaceDeleted) {
		t.Fatalf("namespace state = %s, want %s", namespaceState, meta.StorageNamespaceDeleted)
	}
}

func TestDeleteFreeSharedPoolTenantRejectsMissingPlacement(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	now := time.Now().UTC()
	const poolID = "pool-missing-placement"
	if err := rt.meta.CreateTenantPool(ctx, &meta.TenantPool{
		PoolID: poolID, OrganizationID: "org-missing-placement", Size: 1,
		Status: meta.TenantPoolDeleting, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.EnsureFsID(ctx, rt.tenantID); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.EnsureTenantStorageNamespace(ctx, rt.tenantID, "local", "", rt.tenantID+"/"); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpsertTenantPoolMembership(ctx, &meta.TenantPoolMembership{
		TenantID: rt.tenantID, TiDBCloudOrganizationID: "org-missing-placement", PoolID: poolID,
		PoolStatus: meta.TenantPoolBindingFree, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	tenantRow, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}

	owned, err := rt.server.deleteFreeSharedPoolTenant(ctx, tenantRow)
	if err == nil {
		t.Fatal("delete succeeded without a shared placement")
	}
	if !owned {
		t.Fatal("delete did not claim the free shared member before detecting the missing placement")
	}
	if _, err := rt.meta.GetTenantPoolMembership(ctx, rt.tenantID); err != nil {
		t.Fatalf("membership was removed despite missing placement: %v", err)
	}
	got, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != meta.TenantDeleting {
		t.Fatalf("tenant status = %s, want %s after missing placement", got.Status, meta.TenantDeleting)
	}
}

func TestDeleteFreeSharedPoolTenantDetachesMembershipForDeletingPlacement(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-placement-deleting", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_placement_deleting", MaxTenants: 100,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, `UPDATE tenant_placements SET status = ? WHERE fs_id = ?`,
		meta.PlacementStatusDeleting, fsID); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpsertTenantPoolMembership(ctx, &meta.TenantPoolMembership{
		TenantID: rt.tenantID, TiDBCloudOrganizationID: "org-shared-placement-deleting",
		PoolID: "pool-shared-placement-deleting", PoolStatus: meta.TenantPoolBindingFree,
		CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(),
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.EnsureTenantStorageNamespace(ctx, rt.tenantID, "local", "", rt.tenantID+"/"); err != nil {
		t.Fatal(err)
	}
	tenantRow, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}

	owned, err := rt.server.deleteFreeSharedPoolTenant(ctx, tenantRow)
	if err != nil || !owned {
		t.Fatalf("deleteFreeSharedPoolTenant: owned=%v err=%v", owned, err)
	}
	if _, err := rt.meta.GetTenantPoolMembership(ctx, rt.tenantID); !errors.Is(err, meta.ErrNotFound) {
		t.Fatalf("membership after deleting-placement handoff = %v, want ErrNotFound", err)
	}
}

func TestDeleteFreeSharedPoolTenantFinalizesEmptyNamespaceDeletingPlacement(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-empty-placement-deleting", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_empty_placement_deleting", MaxTenants: 100,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, `UPDATE tenant_placements SET status = ? WHERE fs_id = ?`,
		meta.PlacementStatusDeleting, fsID); err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, `UPDATE db_pool SET tenant_count = 1 WHERE db_id = ?`, dbID); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.UpsertTenantPoolMembership(ctx, &meta.TenantPoolMembership{
		TenantID: rt.tenantID, TiDBCloudOrganizationID: "org-shared-empty-placement-deleting",
		PoolID: "pool-shared-empty-placement-deleting", PoolStatus: meta.TenantPoolBindingFree,
		CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(),
	}); err != nil {
		t.Fatal(err)
	}
	tenantRow, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if tenantRow.StorageNamespaceID != "" {
		t.Fatalf("storage namespace = %q, want empty", tenantRow.StorageNamespaceID)
	}

	owned, err := rt.server.deleteFreeSharedPoolTenant(ctx, tenantRow)
	if err != nil || !owned {
		t.Fatalf("deleteFreeSharedPoolTenant: owned=%v err=%v", owned, err)
	}
	if _, err := rt.meta.GetTenantPlacement(ctx, fsID); !errors.Is(err, meta.ErrNotFound) {
		t.Fatalf("placement after delete = %v, want ErrNotFound", err)
	}
	if _, err := rt.meta.GetTenantPoolMembership(ctx, rt.tenantID); !errors.Is(err, meta.ErrNotFound) {
		t.Fatalf("membership after delete = %v, want ErrNotFound", err)
	}
	dbPool, err := rt.meta.GetSharedDB(ctx, dbID)
	if err != nil {
		t.Fatal(err)
	}
	if dbPool.TenantCount != 0 {
		t.Fatalf("db pool tenant_count = %d, want 0", dbPool.TenantCount)
	}
	got, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != meta.TenantDeleted {
		t.Fatalf("tenant status = %s, want %s", got.Status, meta.TenantDeleted)
	}
}

// TestSharedTenantDeleteRetriesFromDeletingPlacement covers the retry window
// after tenant data and capacity were released but before external cleanup was
// durably enqueued. The deleting placement retains authorization and lets the
// same shared-provider path enqueue cleanup without touching the cluster.
func TestSharedTenantDeleteRetriesFromDeletingPlacement(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("EnsureFsID: %v", err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-delete", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 100,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatalf("UpsertTenantPlacement: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE tenant_placements SET status = ? WHERE fs_id = ?", meta.PlacementStatusDeleting, fsID); err != nil {
		t.Fatalf("mark placement deleting: %v", err)
	}
	if err := rt.meta.UpsertStorageNamespace(ctx, &meta.StorageNamespace{
		ID: "namespace-delete-retry", OwnerTenantID: rt.tenantID, Backend: "local",
		Prefix: rt.tenantID + "/", State: meta.StorageNamespaceActive,
	}); err != nil {
		t.Fatalf("UpsertStorageNamespace: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE tenants SET storage_namespace_id = ? WHERE id = ?", "namespace-delete-retry", rt.tenantID); err != nil {
		t.Fatalf("set storage namespace: %v", err)
	}
	if err := rt.meta.UpdateTenantStatus(ctx, rt.tenantID, meta.TenantDeleting); err != nil {
		t.Fatal(err)
	}

	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusAccepted)
	}
	// The standalone cluster-deprovision path must never run for a shared
	// tenant, even when its placement row is already gone.
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls = %d, want 0", got)
	}
	hasJob, err := rt.meta.TenantDeleteJobExists(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if !hasJob {
		t.Fatal("tenant delete cleanup job was not enqueued")
	}
	placement, err := rt.meta.GetTenantPlacement(ctx, fsID)
	if err != nil {
		t.Fatalf("deleting placement after retry: %v", err)
	}
	if placement.Status != meta.PlacementStatusDeleting {
		t.Fatalf("placement status = %q, want %q", placement.Status, meta.PlacementStatusDeleting)
	}
}

// TestSharedTenantDeleteWithCleanupJobShortCircuits mirrors the standalone
// retry short-circuit: a deleting shared tenant with an existing cleanup job
// returns the deleting status immediately — without purging again and without
// touching the placement row.
func TestSharedTenantDeleteWithCleanupJobShortCircuits(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("EnsureFsID: %v", err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-delete",
		Host:                    "shared.example.com",
		Port:                    4000,
		User:                    "root",
		PasswordCipher:          []byte("cipher"),
		Name:                    "shared_db",
		MaxTenants:              100,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID:        fsID,
		DbID:        dbID,
		Placement:   meta.PlacementShared,
		SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatalf("UpsertTenantPlacement: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE tenant_placements SET status = ? WHERE fs_id = ?", meta.PlacementStatusDeleting, fsID); err != nil {
		t.Fatalf("mark placement deleting: %v", err)
	}
	if err := rt.meta.UpdateTenantStatus(ctx, rt.tenantID, meta.TenantDeleting); err != nil {
		t.Fatal(err)
	}
	if err := rt.meta.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID:    rt.tenantID,
		NamespaceID: rt.tenantID,
		Backend:     "local",
		Prefix:      rt.tenantID + "/",
	}); err != nil {
		t.Fatal(err)
	}

	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusAccepted)
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls = %d, want 0", got)
	}
	// The short-circuit must leave the placement row alone: no purge ran.
	if _, err := rt.meta.GetTenantPlacement(ctx, fsID); err != nil {
		t.Fatalf("placement after short-circuit: %v", err)
	}
}

// TestSharedTenantDeleteWithoutFsIDFails proves the shared delete path never
// allocates identity: a shared tenant without an fs_registry row fails the
// delete instead of minting a phantom fs_id (ResolveFsID, not EnsureFsID).
func TestSharedTenantDeleteWithoutFsIDFails(t *testing.T) {
	// The server's one-time startup backfill re-registers any tenant without
	// an fs_id, so the row deletion below must happen strictly after that
	// backfill has run — otherwise the test races it (the backfill recreates
	// the row and the delete returns 202). Watch the global logger for the
	// backfill's completion marker.
	core, recorded := observer.New(zap.InfoLevel)
	prev := logger.L()
	logger.Set(zap.New(core))
	t.Cleanup(func() { logger.Set(prev) })

	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	deadline := time.Now().Add(10 * time.Second)
	for recorded.FilterField(zap.String("event", "fs_registry_backfill_done")).Len() == 0 {
		if time.Now().After(deadline) {
			t.Fatal("startup fs_registry backfill did not complete in time")
		}
		time.Sleep(10 * time.Millisecond)
	}
	// InsertTenant allocated an fs_id at provision time; simulate a tenant
	// whose registry row was reaped out of band — the delete must fail, not
	// mint a new one.
	if _, err := rt.meta.DB().ExecContext(ctx, "DELETE FROM fs_registry WHERE tenant_id = ?", rt.tenantID); err != nil {
		t.Fatal(err)
	}

	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusInternalServerError {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusInternalServerError)
	}
	if _, err := rt.meta.ResolveFsID(ctx, rt.tenantID); err == nil {
		t.Fatal("delete allocated an fs_id for the tenant being deleted")
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls = %d, want 0", got)
	}
}

// TestSharedTenantDeleteFullPurgePath runs the complete owner delete on a
// shared tenant: drain, purge, placement removal, and final delete — the
// provider-routed dispatch never calls the cluster deprovisioner.
func TestSharedTenantDeleteFullPurgePath(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("EnsureFsID: %v", err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-delete",
		Host:                    "shared.example.com",
		Port:                    4000,
		User:                    "root",
		PasswordCipher:          []byte("cipher"),
		Name:                    "shared_db",
		MaxTenants:              100,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID:        fsID,
		DbID:        dbID,
		Placement:   meta.PlacementShared,
		SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatalf("UpsertTenantPlacement: %v", err)
	}

	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	// The purge needs a reachable shared DB, which "shared.example.com" is
	// not — the delete must fail there WITHOUT deprovisioning any cluster and
	// leave the tenant in deleting state for a later retry.
	if resp.StatusCode != http.StatusInternalServerError {
		t.Fatalf("status = %d, want %d (unreachable shared db)", resp.StatusCode, http.StatusInternalServerError)
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls = %d, want 0", got)
	}
	var status string
	if err := rt.meta.DB().QueryRow("SELECT status FROM tenants WHERE id = ?", rt.tenantID).Scan(&status); err != nil {
		t.Fatal(err)
	}
	if status != string(meta.TenantDeleting) {
		t.Fatalf("tenant status = %s, want %s", status, meta.TenantDeleting)
	}
	// The placement row must survive the failed purge so the retry can find
	// the shared DB again.
	if _, err := rt.meta.GetTenantPlacement(ctx, fsID); err != nil {
		t.Fatalf("placement after failed purge: %v", err)
	}
}

func TestAdminTenantCreateAllowsSharedPoolWithoutCloudClusterIdentity(t *testing.T) {
	rt := newQuotaRuntime(t, tenant.ProviderTiDBCloudNative)
	ctx := context.Background()
	rt.prov.iamIdentities = []*tenant.TiDBCloudAPIKeyIdentity{{
		OrganizationID: "org-shared-1", Role: tenant.TiDBCloudRoleOrgOwner,
	}}
	if _, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-1",
		Host:                    "shared.example.com",
		Port:                    4000,
		User:                    "root",
		PasswordCipher:          []byte("cipher"),
		Name:                    "shared_db",
		MaxTenants:              100,
	}); err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}

	body, err := json.Marshal(map[string]string{"public_key": "pk", "private_key": "sk"})
	if err != nil {
		t.Fatal(err)
	}
	ts := httptest.NewServer(rt.server)
	defer ts.Close()
	req, err := http.NewRequest(http.MethodPost, ts.URL+"/v1/admin/tenants", bytes.NewReader(body))
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		responseBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("status = %d, want %d: %s", resp.StatusCode, http.StatusAccepted, responseBody)
	}
	var tenantCount int
	if err := rt.meta.DB().QueryRow("SELECT COUNT(*) FROM tenants").Scan(&tenantCount); err != nil {
		t.Fatal(err)
	}
	if tenantCount != 2 {
		t.Fatalf("tenants = %d, want 2", tenantCount)
	}
	if got := rt.prov.listCalls.Load(); got != 0 {
		t.Fatalf("cluster list calls = %d, want 0", got)
	}
	if got := rt.prov.iamCalls.Load(); got != 1 {
		t.Fatalf("IAM calls = %d, want 1", got)
	}
}

func TestAdminTenantCreatePersistsSharedQuotaAndIgnoresSpendingLimit(t *testing.T) {
	rt := newQuotaRuntime(t, tenant.ProviderTiDBCloudNative)
	ctx := context.Background()
	rt.prov.iamIdentities = []*tenant.TiDBCloudAPIKeyIdentity{{
		OrganizationID: "org-shared-quota", Role: tenant.TiDBCloudRoleProjectOwner,
	}}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-quota", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 100,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE db_pool SET cluster_id = ? WHERE db_id = ?", "cluster-shared-quota", dbID); err != nil {
		t.Fatal(err)
	}
	ts := httptest.NewServer(rt.server)
	t.Cleanup(ts.Close)
	resp := postJSON(t, ts.URL+"/v1/admin/tenants", map[string]any{
		"public_key": "pk", "private_key": "sk",
		"max_storage_size": int64(500), "max_file_size": int64(10), "max_file_count": int64(100),
		"tidbcloud_spending_limit": int64(5),
	}, "")
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("status = %d, want %d: %s", resp.StatusCode, http.StatusAccepted, body)
	}
	var out adminTenantCreateResponse
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		t.Fatal(err)
	}
	cfg, err := rt.meta.GetQuotaConfig(ctx, out.TenantID)
	if err != nil {
		t.Fatal(err)
	}
	if cfg.MaxStorageBytes != 500*quotaStorageSizeBytes {
		t.Fatalf("max storage bytes = %d, want %d", cfg.MaxStorageBytes, 500*quotaStorageSizeBytes)
	}
	if cfg.MaxFileSizeBytes != 10*quotaStorageSizeBytes {
		t.Fatalf("max file size bytes = %d, want %d", cfg.MaxFileSizeBytes, 10*quotaStorageSizeBytes)
	}
	if cfg.MaxFileCount != 100 {
		t.Fatalf("max file count = %d, want 100", cfg.MaxFileCount)
	}
	if cfg.TiDBCloudSpendingLimit == nil || *cfg.TiDBCloudSpendingLimit != DefaultManagedSharedDBSpendingLimit {
		t.Fatalf("shared compatibility spending limit = %#v, want %d", cfg.TiDBCloudSpendingLimit, DefaultManagedSharedDBSpendingLimit)
	}
}

func TestAdminTenantGetAuthorizesSharedProviderThroughPhysicalPool(t *testing.T) {
	rt := newQuotaRuntime(t, tenant.ProviderTiDBCloudNative)
	rt.prov.iamIdentities = []*tenant.TiDBCloudAPIKeyIdentity{{OrganizationID: "org-shared-admin", Role: tenant.TiDBCloudRoleOrgOwner}}
	ctx := context.Background()
	if err := rt.meta.UpdateTenantProvider(ctx, rt.tenantID, tenant.ProviderTiDBCloudNativeShared); err != nil {
		t.Fatalf("UpdateTenantProvider: %v", err)
	}
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("EnsureFsID: %v", err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-admin", ClusterID: "cluster-shared-admin",
		Host: "shared.example.com", Port: 4000, User: "root", PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 100,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE db_pool SET cluster_id = ? WHERE db_id = ?", "cluster-shared-admin", dbID); err != nil {
		t.Fatalf("set shared cluster_id: %v", err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatalf("UpsertTenantPlacement: %v", err)
	}
	metricReq := httptest.NewRequest(http.MethodGet, "/v1/admin/tenants/"+rt.tenantID, nil)
	metricReq = metricReq.WithContext(withRequestMetricState(metricReq.Context(), &requestMetricState{}))
	if _, ok := rt.server.authorizedAdminTenant(httptest.NewRecorder(), metricReq, rt.tenantID,
		"org-shared-admin"); !ok {
		t.Fatal("shared admin tenant authorization failed")
	}
	metricTenantID, _, metricProvider, metricOrgID := requestMetricScope(metricReq.Context())
	if metricTenantID != rt.tenantID || metricProvider != tenant.ProviderTiDBCloudNativeShared || metricOrgID != "org-shared-admin" {
		t.Fatalf("request metric scope = tenant %q provider %q org %q", metricTenantID, metricProvider, metricOrgID)
	}
	ts := httptest.NewServer(rt.server)
	defer ts.Close()

	req, err := http.NewRequest(http.MethodGet, ts.URL+"/v1/admin/tenants/"+rt.tenantID, nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("X-TiDBCloud-Public-Key", "pk")
	req.Header.Set("X-TiDBCloud-Private-Key", "sk")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusOK)
	}
	if got := rt.prov.iamCalls.Load(); got != 1 {
		t.Fatalf("IAM authorization calls = %d, want 1", got)
	}
}

func TestAdminTenantQuotaSetSharedUpdatesVirtualQuotaAndIgnoresSpendingLimit(t *testing.T) {
	rt := newQuotaRuntime(t, tenant.ProviderTiDBCloudNative)
	rt.prov.iamIdentities = []*tenant.TiDBCloudAPIKeyIdentity{{OrganizationID: "org-shared-admin", Role: tenant.TiDBCloudRoleProjectOwner}}
	ctx := context.Background()
	if err := rt.meta.UpdateTenantProvider(ctx, rt.tenantID, tenant.ProviderTiDBCloudNativeShared); err != nil {
		t.Fatalf("UpdateTenantProvider: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE tenants SET cluster_id = NULL WHERE id = ?", rt.tenantID); err != nil {
		t.Fatalf("clear tenant cluster_id: %v", err)
	}
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("EnsureFsID: %v", err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-admin", Host: "shared.example.com", Port: 4000,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 100,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE db_pool SET cluster_id = ? WHERE db_id = ?", "cluster-shared-admin", dbID); err != nil {
		t.Fatalf("set shared cluster_id: %v", err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatalf("UpsertTenantPlacement: %v", err)
	}
	if err := rt.meta.EnsureQuotaUsageRow(ctx, rt.tenantID); err != nil {
		t.Fatalf("EnsureQuotaUsageRow: %v", err)
	}
	shadowSpendingLimit := int64(9_000)
	if err := rt.meta.SetQuotaConfig(ctx, &meta.QuotaConfig{
		TenantID: rt.tenantID, MaxStorageBytes: meta.DefaultMaxStorageBytes(),
		MaxFileSizeBytes: meta.DefaultMaxFileSizeBytes(), MaxFileCount: 100000,
		TiDBCloudSpendingLimit: &shadowSpendingLimit,
	}); err != nil {
		t.Fatalf("SetQuotaConfig: %v", err)
	}

	ts := httptest.NewServer(rt.server)
	defer ts.Close()
	resp := postJSON(t, ts.URL+"/v1/admin/tenants/"+rt.tenantID+"/quota", map[string]any{
		"public_key": "pk", "private_key": "sk", "max_storage_size": int64(200),
		"tidbcloud_spending_limit": int64(5),
	}, "")
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusOK)
	}
	if got := rt.prov.markCalls.Load(); got != 0 {
		t.Fatalf("physical quota mark calls = %d, want 0", got)
	}
	if got := rt.prov.updateCalls.Load(); got != 0 {
		t.Fatalf("physical quota update calls = %d, want 0", got)
	}
	cfg, err := rt.meta.GetQuotaConfig(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("GetQuotaConfig: %v", err)
	}
	if cfg.MaxStorageBytes != 200*quotaStorageSizeBytes {
		t.Fatalf("MaxStorageBytes = %d, want %d", cfg.MaxStorageBytes, 200*quotaStorageSizeBytes)
	}
	if cfg.TiDBCloudSpendingLimit == nil || *cfg.TiDBCloudSpendingLimit != 9_000 {
		t.Fatalf("shared compatibility spending limit = %#v, want unchanged 9000", cfg.TiDBCloudSpendingLimit)
	}
}

func TestAdminTenantDeleteSharedNeverDeprovisionsPhysicalPool(t *testing.T) {
	rt := newQuotaRuntime(t, tenant.ProviderTiDBCloudNative)
	rt.prov.iamIdentities = []*tenant.TiDBCloudAPIKeyIdentity{{OrganizationID: "org-shared-admin-delete", Role: tenant.TiDBCloudRoleOrgOwner}}
	ctx := context.Background()
	if err := rt.meta.UpdateTenantProvider(ctx, rt.tenantID, tenant.ProviderTiDBCloudNativeShared); err != nil {
		t.Fatalf("UpdateTenantProvider: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE tenants SET cluster_id = NULL WHERE id = ?", rt.tenantID); err != nil {
		t.Fatalf("clear tenant cluster_id: %v", err)
	}
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("EnsureFsID: %v", err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-admin-delete", Host: "127.0.0.1", Port: 1,
		User: "root", PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 100,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE db_pool SET cluster_id = ? WHERE db_id = ?", "cluster-shared-admin-delete", dbID); err != nil {
		t.Fatalf("set shared cluster_id: %v", err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatalf("UpsertTenantPlacement: %v", err)
	}

	ts := httptest.NewServer(rt.server)
	defer ts.Close()
	req, err := http.NewRequest(http.MethodDelete, ts.URL+"/v1/admin/tenants/"+rt.tenantID,
		bytes.NewBufferString(`{"public_key":"pk","private_key":"sk"}`))
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusInternalServerError {
		t.Fatalf("status = %d, want %d for unreachable shared DB purge", resp.StatusCode, http.StatusInternalServerError)
	}
	if got := rt.prov.markCalls.Load(); got != 0 {
		t.Fatalf("physical quota mark calls = %d, want 0", got)
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("physical cluster deprovision calls = %d, want 0", got)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE tenant_placements SET status = ? WHERE fs_id = ?", meta.PlacementStatusDeleting, fsID); err != nil {
		t.Fatalf("mark placement deleting: %v", err)
	}
	if err := rt.meta.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID: rt.tenantID, NamespaceID: "namespace-delete-retry", Backend: "local", Prefix: rt.tenantID + "/",
	}); err != nil {
		t.Fatalf("EnqueueTenantDeleteJob: %v", err)
	}
	retryReq, err := http.NewRequest(http.MethodDelete, ts.URL+"/v1/admin/tenants/"+rt.tenantID,
		bytes.NewBufferString(`{"public_key":"pk","private_key":"sk"}`))
	if err != nil {
		t.Fatal(err)
	}
	retryReq.Header.Set("Content-Type", "application/json")
	retryResp, err := http.DefaultClient.Do(retryReq)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = retryResp.Body.Close() }()
	if retryResp.StatusCode != http.StatusAccepted {
		t.Fatalf("retry status = %d, want %d", retryResp.StatusCode, http.StatusAccepted)
	}
	var deleteResponse adminTenantDeleteResponse
	if err := json.NewDecoder(retryResp.Body).Decode(&deleteResponse); err != nil {
		t.Fatalf("decode shared admin delete response: %v", err)
	}
	if deleteResponse.TenantID != rt.tenantID || deleteResponse.Status != string(meta.TenantDeleting) {
		t.Fatalf("shared admin delete response = %+v, want tenant_id %s and status %s", deleteResponse, rt.tenantID, meta.TenantDeleting)
	}
}

// TestSharedTenantDeletePlacementRemovalFailureStaysRetryable injects a
// failure into the placement-removal transaction (the pool row disappears
// between purge and removal) and proves the delete fails closed: 500, tenant
// stays deleting, the placement row survives for the retry, and the owner
// key stays active — instead of orphaning the placement on a terminal
// tenant.
func TestSharedTenantDeletePlacementRemovalFailureStaysRetryable(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNativeShared, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	fsID, err := rt.meta.EnsureFsID(ctx, rt.tenantID)
	if err != nil {
		t.Fatalf("EnsureFsID: %v", err)
	}
	db := newTenantDeleteDBInfo(t)
	passCipher, err := rt.pool.Encrypt(ctx, []byte(db.DBPass))
	if err != nil {
		t.Fatal(err)
	}
	dbID, err := rt.meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-shared-delete", Host: db.DBHost, Port: db.DBPort, User: db.DBUser,
		PasswordCipher: passCipher, Name: db.DBName, MaxTenants: 100,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	if err := rt.meta.UpsertTenantPlacement(ctx, &meta.TenantPlacement{
		FsID: fsID, DbID: dbID, Placement: meta.PlacementShared, SchemaShape: meta.SchemaShapeShared,
	}); err != nil {
		t.Fatalf("UpsertTenantPlacement: %v", err)
	}
	// Pre-warm the shared handle with a database in which every shared-schema
	// table is absent: the purge in the delete below reuses the cached handle and
	// succeeds trivially, but the placement removal transaction cannot release
	// capacity on a missing pool row and must roll back. Restore the standalone
	// shape in cleanup so this shared-schema test cannot pollute later tests in
	// the package-wide database.
	sharedTables := []string{
		"journal_entry_subjects", "journal_entries", "journal_append_requests", "journal_labels", "journals",
		"vault_audit_log", "vault_grants", "vault_tokens", "vault_secret_fields", "vault_secrets", "vault_deks",
		"git_workspace_object_packs", "git_workspace_overlay", "git_workspace_git_state", "git_workspace_tree_nodes", "git_workspaces",
		"fs_layer_checkpoints", "fs_layer_events", "fs_layer_tags", "fs_layer_entries", "fs_layers",
		"fs_events", "semantic_tasks", "file_gc_tasks", "uploads", "file_tags", "file_nodes", "semantic", "contents", "inodes",
	}
	dropSharedTables := func(t *testing.T) error {
		t.Helper()
		for _, tbl := range sharedTables {
			if _, err := rt.meta.DB().ExecContext(ctx, "DROP TABLE IF EXISTS "+tbl); err != nil {
				return fmt.Errorf("drop %s: %w", tbl, err)
			}
		}
		return nil
	}
	t.Cleanup(func() {
		if err := rt.pool.InvalidateSharedDB(dbID); err != nil {
			t.Errorf("invalidate shared database: %v", err)
		}
		if err := dropSharedTables(t); err != nil {
			t.Errorf("cleanup shared tables: %v", err)
		}
		initServerTenantSchema(t, testDSN)
	})
	if err := dropSharedTables(t); err != nil {
		t.Fatalf("prepare missing shared tables: %v", err)
	}
	if err := rt.pool.PurgeSharedTenant(ctx, fsID, dbID); err != nil {
		t.Fatalf("pre-warm purge: %v", err)
	}
	if _, err := rt.meta.DB().ExecContext(ctx, "DELETE FROM db_pool WHERE db_id = ?", dbID); err != nil {
		t.Fatal(err)
	}

	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusInternalServerError {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusInternalServerError)
	}
	var status string
	if err := rt.meta.DB().QueryRow("SELECT status FROM tenants WHERE id = ?", rt.tenantID).Scan(&status); err != nil {
		t.Fatal(err)
	}
	if status != string(meta.TenantDeleting) {
		t.Fatalf("tenant status = %s, want %s", status, meta.TenantDeleting)
	}
	if _, err := rt.meta.GetTenantPlacement(ctx, fsID); err != nil {
		t.Fatalf("placement must survive the rolled-back removal: %v", err)
	}
	var activeKeys int
	if err := rt.meta.DB().QueryRow("SELECT COUNT(*) FROM tenant_api_keys WHERE tenant_id = ? AND status = ?", rt.tenantID, meta.APIKeyActive).Scan(&activeKeys); err != nil {
		t.Fatal(err)
	}
	if activeKeys != 1 {
		t.Fatalf("active api keys = %d, want 1 (owner can retry)", activeKeys)
	}
}

// TestFindSharedDBForProvisionRequiresCloudOrganization proves that shared
// placement requires an exact customer organization and never captures
// providers without TiDB Cloud IAM organization data.
func TestFindSharedDBForProvisionSkipsNonTiDBProviders(t *testing.T) {
	db := newTenantDeleteDBInfo(t)
	testtidb.ResetMetaDB(t, db.Meta.DB())
	ctx := context.Background()
	if _, err := db.Meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-exact", Host: "shared.example.com", Port: 4000, User: "root",
		PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 100,
	}); err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	s := &Server{meta: db.Meta}

	got, err := s.findSharedDBForProvision(ctx, tenant.ProviderDB9, provisionTenantOptions{})
	if err != nil {
		t.Fatalf("db9 lookup: %v", err)
	}
	if got != nil {
		t.Fatalf("db9 matched shared pool %v, want no match", got)
	}
	// tidb_zero has no IAM organization and therefore cannot match an exact
	// organization-bound shared pool.
	got, err = s.findSharedDBForProvision(ctx, tenant.ProviderTiDBZero, provisionTenantOptions{})
	if err != nil {
		t.Fatalf("tidb_zero lookup: %v", err)
	}
	if got != nil {
		t.Fatalf("tidb_zero matched shared pool %v, want no match", got)
	}
}

// TestProvisionTenantOnSharedDBCommitsAtomically covers the happy path of
// the atomic provision transition: placement, provider re-label, active
// status, capacity slot, and the owner key all land together.
func TestProvisionTenantOnSharedDBCommitsAtomically(t *testing.T) {
	db := newTenantDeleteDBInfo(t)
	testtidb.ResetMetaDB(t, db.Meta.DB())
	ctx := context.Background()
	dbID, err := db.Meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-provision-ok", Host: "shared.example.com", Port: 4000, User: "root",
		PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 10,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	sharedDB, err := db.Meta.GetSharedDB(ctx, dbID)
	if err != nil {
		t.Fatalf("GetSharedDB: %v", err)
	}
	tokenSecret := make([]byte, 32)
	if _, err := rand.Read(tokenSecret); err != nil {
		t.Fatal(err)
	}
	s := &Server{meta: db.Meta, pool: db.Pool, tokenSecret: tokenSecret}
	now := time.Now().UTC()
	tenantID := token.NewID()
	if err := db.Meta.InsertTenant(ctx, &meta.Tenant{
		ID: tenantID, Status: meta.TenantPending, Kind: meta.TenantKindLive,
		Provider: tenant.ProviderTiDBCloudNative, DBPasswordCipher: []byte{},
		SchemaVersion: 1, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatalf("InsertTenant: %v", err)
	}

	res, err := s.provisionTenantOnSharedDB(ctx, tenantID, sharedDB, tenant.ProviderTiDBCloudNative, "default", provisionTenantOptions{}, now)
	if err != nil {
		t.Fatalf("provisionTenantOnSharedDB: %v", err)
	}
	if res.Provider != tenant.ProviderTiDBCloudNativeShared || res.APIKey == "" {
		t.Fatalf("result = provider %q api_key %q", res.Provider, res.APIKey)
	}
	got, err := db.Meta.GetTenant(ctx, tenantID)
	if err != nil {
		t.Fatalf("GetTenant: %v", err)
	}
	if got.Provider != tenant.ProviderTiDBCloudNativeShared || got.Status != meta.TenantActive {
		t.Fatalf("tenant = provider %q status %q, want tidb_cloud_native_shared/active", got.Provider, got.Status)
	}
	quotaCfg, err := db.Meta.GetQuotaConfig(ctx, tenantID)
	if err != nil {
		t.Fatalf("GetQuotaConfig: %v", err)
	}
	if quotaCfg.TiDBCloudSpendingLimit == nil || *quotaCfg.TiDBCloudSpendingLimit != DefaultManagedSharedDBSpendingLimit {
		t.Fatalf("shared compatibility spending limit = %#v, want %d", quotaCfg.TiDBCloudSpendingLimit, DefaultManagedSharedDBSpendingLimit)
	}
	fsID, err := db.Meta.ResolveFsID(ctx, tenantID)
	if err != nil {
		t.Fatalf("ResolveFsID: %v", err)
	}
	p, err := db.Meta.GetTenantPlacement(ctx, fsID)
	if err != nil {
		t.Fatalf("GetTenantPlacement: %v", err)
	}
	if p.DbID != dbID || p.SchemaShape != meta.SchemaShapeShared {
		t.Fatalf("placement = %+v", p)
	}
	dbRow, err := db.Meta.GetSharedDB(ctx, dbID)
	if err != nil {
		t.Fatalf("GetSharedDB: %v", err)
	}
	if dbRow.TenantCount != 1 {
		t.Fatalf("TenantCount = %d, want 1", dbRow.TenantCount)
	}
	var activeKeys int
	if err := db.Meta.DB().QueryRow("SELECT COUNT(*) FROM tenant_api_keys WHERE tenant_id = ? AND status = ?", tenantID, meta.APIKeyActive).Scan(&activeKeys); err != nil {
		t.Fatal(err)
	}
	if activeKeys != 1 {
		t.Fatalf("active owner keys = %d, want 1", activeKeys)
	}
}

func TestProvisionTenantOnSharedDBIgnoresDedicatedSpendingLimit(t *testing.T) {
	for _, spendingLimit := range []int64{0, 10_000} {
		t.Run(fmt.Sprintf("spending_%d", spendingLimit), func(t *testing.T) {
			db := newTenantDeleteDBInfo(t)
			testtidb.ResetMetaDB(t, db.Meta.DB())
			ctx := context.Background()
			dbID, err := db.Meta.RegisterSharedDB(ctx, &meta.SharedDB{
				TiDBCloudOrganizationID: "org-provision-ignore-spending", Host: "shared.example.com", Port: 4000, User: "root",
				PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 10,
			})
			if err != nil {
				t.Fatalf("RegisterSharedDB: %v", err)
			}
			sharedDB, err := db.Meta.GetSharedDB(ctx, dbID)
			if err != nil {
				t.Fatalf("GetSharedDB: %v", err)
			}
			tokenSecret := make([]byte, 32)
			if _, err := rand.Read(tokenSecret); err != nil {
				t.Fatal(err)
			}
			s := &Server{meta: db.Meta, pool: db.Pool, tokenSecret: tokenSecret}
			now := time.Now().UTC()
			tenantID := token.NewID()
			if err := db.Meta.InsertTenant(ctx, &meta.Tenant{
				ID: tenantID, Status: meta.TenantPending, Kind: meta.TenantKindLive,
				Provider: tenant.ProviderTiDBCloudNative, DBPasswordCipher: []byte{},
				SchemaVersion: 1, CreatedAt: now, UpdatedAt: now,
			}); err != nil {
				t.Fatalf("InsertTenant: %v", err)
			}

			res, err := s.provisionTenantOnSharedDB(ctx, tenantID, sharedDB, tenant.ProviderTiDBCloudNative, "default", provisionTenantOptions{
				Quota: &quotaRequest{quotaFields: quotaFields{TiDBCloudSpendingLimit: &spendingLimit}},
			}, now)
			if err != nil {
				t.Fatalf("provisionTenantOnSharedDB: %v", err)
			}
			if res == nil || res.Provider != tenant.ProviderTiDBCloudNativeShared {
				t.Fatalf("result = %+v, want shared tenant", res)
			}
			cfg, err := db.Meta.GetQuotaConfig(ctx, tenantID)
			if err != nil {
				t.Fatalf("GetQuotaConfig: %v", err)
			}
			if cfg.TiDBCloudSpendingLimit == nil || *cfg.TiDBCloudSpendingLimit != DefaultManagedSharedDBSpendingLimit {
				t.Fatalf("shared compatibility spending limit = %#v, want %d", cfg.TiDBCloudSpendingLimit, DefaultManagedSharedDBSpendingLimit)
			}
		})
	}
}

// TestProvisionTenantOnSharedDBFailureLeavesNoPartialState proves the atomic
// transition leaves nothing behind on failure: when any write of the
// provision transaction fails, no placement row, no reserved capacity, no
// owner key persists, and the tenant never becomes active — there is no
// best-effort compensation window that could orphan an active shared tenant.
func TestProvisionTenantOnSharedDBFailureLeavesNoPartialState(t *testing.T) {
	db := newTenantDeleteDBInfo(t)
	testtidb.ResetMetaDB(t, db.Meta.DB())
	ctx := context.Background()
	dbID, err := db.Meta.RegisterSharedDB(ctx, &meta.SharedDB{
		TiDBCloudOrganizationID: "org-provision-fail", Host: "shared.example.com", Port: 4000, User: "root",
		PasswordCipher: []byte("cipher"), Name: "shared_db", MaxTenants: 10,
	})
	if err != nil {
		t.Fatalf("RegisterSharedDB: %v", err)
	}
	sharedDB, err := db.Meta.GetSharedDB(ctx, dbID)
	if err != nil {
		t.Fatalf("GetSharedDB: %v", err)
	}
	tokenSecret := make([]byte, 32)
	if _, err := rand.Read(tokenSecret); err != nil {
		t.Fatal(err)
	}
	s := &Server{meta: db.Meta, pool: db.Pool, tokenSecret: tokenSecret}
	tenantID := token.NewID()
	// No tenant row is inserted, so the tenant activation inside the
	// transaction fails and rolls back the capacity reservation, the
	// placement, and the owner key insert.
	if _, err := s.provisionTenantOnSharedDB(ctx, tenantID, sharedDB, tenant.ProviderTiDBCloudNative, "default", provisionTenantOptions{}, time.Now().UTC()); err == nil {
		t.Fatal("provisionTenantOnSharedDB succeeded, want error")
	} else {
		var pe *provisionTenantError
		if !errors.As(err, &pe) || pe.status != http.StatusInternalServerError {
			t.Fatalf("error = %v, want provisionTenantError 500", err)
		}
	}
	fsID, err := db.Meta.ResolveFsID(ctx, tenantID)
	if err != nil {
		t.Fatalf("ResolveFsID: %v", err)
	}
	if _, err := db.Meta.GetTenantPlacement(ctx, fsID); !errors.Is(err, meta.ErrNotFound) {
		t.Fatalf("placement after failed provision = %v, want ErrNotFound", err)
	}
	got, err := db.Meta.GetSharedDB(ctx, dbID)
	if err != nil {
		t.Fatalf("GetSharedDB after failure: %v", err)
	}
	if got.TenantCount != 0 {
		t.Fatalf("TenantCount = %d, want 0 (nothing reserved)", got.TenantCount)
	}
	var keyCount int
	if err := db.Meta.DB().QueryRow("SELECT COUNT(*) FROM tenant_api_keys WHERE tenant_id = ?", tenantID).Scan(&keyCount); err != nil {
		t.Fatal(err)
	}
	if keyCount != 0 {
		t.Fatalf("owner keys after failed provision = %d, want 0", keyCount)
	}
}