package server

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

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"github.com/go-sql-driver/mysql"
	"github.com/mem9-ai/drive9/pkg/encrypt"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/tenant/local"
	"github.com/mem9-ai/drive9/pkg/tenant/token"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

type tenantDeleteRuntime struct {
	meta        *meta.Store
	pool        *tenant.Pool
	tokenSecret []byte
	tenantID    string
	apiKey      string
	prov        *fakeProvisioner
	server      *Server
}

func TestAppendLogCleanupClientUsesBoundNamespaceConfiguration(t *testing.T) {
	db := newTestDBInfo(t)
	testtidb.ResetMetaDB(t, db.Meta.DB())
	ctx := context.Background()
	const tenantID = "append-log-cleanup-tenant"
	ns := &meta.StorageNamespace{
		ID:                    "append-log-cleanup-ns",
		OwnerTenantID:         tenantID,
		Backend:               "s3",
		Bucket:                "normal-bucket",
		Prefix:                "tenant/append-log-cleanup-tenant/",
		AppendLogBindingState: meta.AppendLogBindingBound,
		AppendLogBucket:       "directory-a",
		AppendLogPrefix:       "append-root/",
		AppendLogRegion:       "us-east-1",
		State:                 meta.StorageNamespaceDeleting,
	}
	if err := db.Meta.UpsertStorageNamespace(ctx, ns); err != nil {
		t.Fatal(err)
	}
	job := &meta.TenantDeleteJob{TenantID: tenantID, NamespaceID: ns.ID, Backend: "s3"}
	newServer := func(cfg tenant.PoolConfig) *Server {
		pool := tenant.NewPool(cfg, nil)
		pool.SetMetaStore(db.Meta)
		t.Cleanup(pool.Close)
		return &Server{meta: db.Meta, pool: pool}
	}
	matching := newServer(tenant.PoolConfig{
		S3ExpressBucket: "directory-a", S3ExpressPrefix: "append-root/", S3Region: "us-east-1",
		S3AccessKeyID: "test-key", S3SecretAccessKey: "test-secret",
	})
	client, err := matching.appendLogCleanupClient(ctx, job)
	if err != nil {
		t.Fatal(err)
	}
	if client == nil {
		t.Fatal("matching bound cleanup client = nil")
	}
	mismatched := newServer(tenant.PoolConfig{
		S3ExpressBucket: "directory-b", S3ExpressPrefix: "append-root/", S3Region: "us-east-1",
	})
	if _, err := mismatched.appendLogCleanupClient(ctx, job); !errors.Is(err, meta.ErrAppendLogBindingMismatch) {
		t.Fatalf("mismatched cleanup client error = %v, want ErrAppendLogBindingMismatch", err)
	}
	if err := db.Meta.UpsertStorageNamespace(ctx, &meta.StorageNamespace{
		ID: "append-log-cleanup-unbound", OwnerTenantID: tenantID, Backend: "s3", Bucket: "normal-bucket", Prefix: "tenant/unbound/",
		AppendLogBindingState: meta.AppendLogBindingUnbound, State: meta.StorageNamespaceDeleting,
	}); err != nil {
		t.Fatal(err)
	}
	client, err = newServer(tenant.PoolConfig{}).appendLogCleanupClient(ctx, &meta.TenantDeleteJob{
		TenantID: tenantID, NamespaceID: "append-log-cleanup-unbound", Backend: "s3",
	})
	if err != nil {
		t.Fatalf("unbound cleanup client error = %v", err)
	}
	if client != nil {
		t.Fatalf("unbound cleanup client = %T, want nil", client)
	}
}

func TestTenantDeleteBoundAppendLogConfigMismatchRetriesWithoutFinalizing(t *testing.T) {
	db := newTestDBInfo(t)
	testtidb.ResetMetaDB(t, db.Meta.DB())
	ctx := context.Background()
	const tenantID = "append-log-delete-mismatch"
	const namespaceID = "append-log-delete-mismatch-ns"
	now := time.Now().UTC()
	if err := db.Meta.InsertTenant(ctx, &meta.Tenant{
		ID: tenantID, Status: meta.TenantDeleting, StorageNamespaceID: namespaceID,
		Provider: tenant.ProviderLocal, DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := db.Meta.UpsertStorageNamespace(ctx, &meta.StorageNamespace{
		ID: namespaceID, OwnerTenantID: tenantID, Backend: "s3", Bucket: "normal-bucket", Prefix: "tenant/mismatch/",
		AppendLogBindingState: meta.AppendLogBindingBound,
		AppendLogBucket:       "directory-a",
		AppendLogPrefix:       "append-root/",
		AppendLogRegion:       "us-east-1",
		State:                 meta.StorageNamespaceDeleting,
	}); err != nil {
		t.Fatal(err)
	}
	if err := db.Meta.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID: tenantID, NamespaceID: namespaceID, Backend: "s3", Bucket: "normal-bucket", Prefix: "tenant/mismatch/",
	}); err != nil {
		t.Fatal(err)
	}
	var normalS3Requests int
	normalS3 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		normalS3Requests++
		if strings.Contains(r.URL.Path, "directory-b") {
			t.Fatalf("mismatched Express target was contacted: %s", r.URL.String())
		}
		w.Header().Set("Content-Type", "application/xml")
		switch {
		case r.URL.Query().Get("list-type") == "2":
			_, _ = io.WriteString(w, `<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><IsTruncated>false</IsTruncated></ListBucketResult>`)
		case r.URL.Query().Has("uploads"):
			_, _ = io.WriteString(w, `<ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><IsTruncated>false</IsTruncated></ListMultipartUploadsResult>`)
		default:
			t.Fatalf("unexpected normal S3 request: %s", r.URL.String())
		}
	}))
	t.Cleanup(normalS3.Close)
	pool := tenant.NewPool(tenant.PoolConfig{
		S3Bucket: "normal-bucket", S3Region: "us-east-1", S3Endpoint: normalS3.URL, S3ForcePathStyle: true,
		S3ExpressBucket: "directory-b", S3ExpressPrefix: "append-root/",
		S3AccessKeyID: "test-key", S3SecretAccessKey: "test-secret",
	}, nil)
	pool.SetMetaStore(db.Meta)
	t.Cleanup(pool.Close)
	srv := &Server{meta: db.Meta, pool: pool}
	srv.processTenantDeleteJobs(ctx)

	var state, lastError string
	var attempts int
	var completedAt sql.NullTime
	if err := db.Meta.DB().QueryRowContext(ctx, `SELECT state, attempts, COALESCE(last_error, ''), completed_at
		FROM tenant_delete_jobs WHERE tenant_id = ?`, tenantID).Scan(&state, &attempts, &lastError, &completedAt); err != nil {
		t.Fatal(err)
	}
	if state != string(meta.TenantDeleteJobPending) || attempts != 1 || completedAt.Valid {
		t.Fatalf("delete job = state=%s attempts=%d completed=%v, want pending/1/null", state, attempts, completedAt.Valid)
	}
	if !strings.Contains(lastError, meta.ErrAppendLogBindingMismatch.Error()) {
		t.Fatalf("delete job last_error = %q, want binding mismatch", lastError)
	}
	gotTenant, err := db.Meta.GetTenant(ctx, tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if gotTenant.Status != meta.TenantDeleting {
		t.Fatalf("tenant status = %s, want deleting", gotTenant.Status)
	}
	gotNamespace, err := db.Meta.GetStorageNamespace(ctx, namespaceID)
	if err != nil {
		t.Fatal(err)
	}
	if gotNamespace.State != meta.StorageNamespaceDeleting {
		t.Fatalf("namespace state = %s, want deleting", gotNamespace.State)
	}
	if normalS3Requests == 0 {
		t.Fatal("normal S3 cleanup was not attempted before bound Express mismatch")
	}
}

func newTenantDeleteRuntime(t *testing.T, provider string, scopeKind meta.APIKeyScopeKind) *tenantDeleteRuntime {
	t.Helper()
	db := newTenantDeleteDBInfo(t)
	testtidb.ResetMetaDB(t, db.Meta.DB())
	testtidb.ResetDB(t, db.Meta.DB())
	t.Cleanup(func() {
		testtidb.ResetMetaDB(t, db.Meta.DB())
		testtidb.ResetDB(t, db.Meta.DB())
	})
	tenantID := token.NewID()
	tokenSecret := make([]byte, 32)
	if _, err := rand.Read(tokenSecret); err != nil {
		t.Fatal(err)
	}
	apiKey, err := token.IssueToken(tokenSecret, tenantID, 1)
	if err != nil {
		t.Fatal(err)
	}
	apiKeyCipher, err := db.Pool.Encrypt(context.Background(), []byte(apiKey))
	if err != nil {
		t.Fatal(err)
	}
	dbPassCipher, err := db.Pool.Encrypt(context.Background(), []byte(db.DBPass))
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	if err := db.Meta.InsertTenant(context.Background(), &meta.Tenant{
		ID:               tenantID,
		Status:           meta.TenantActive,
		Kind:             meta.TenantKindLive,
		DBHost:           db.DBHost,
		DBPort:           db.DBPort,
		DBUser:           db.DBUser,
		DBPasswordCipher: dbPassCipher,
		DBName:           db.DBName,
		DBTLS:            false,
		Provider:         provider,
		ClusterID:        "cluster-delete-1",
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}
	var expiresAt *time.Time
	if scopeKind == meta.APIKeyScopeKindFS {
		// fs_scoped keys are fail-closed without expires_at (storage guard).
		exp := now.Add(2 * time.Hour)
		expiresAt = &exp
	}
	if err := db.Meta.InsertAPIKey(context.Background(), &meta.APIKey{
		ID:            token.NewID(),
		TenantID:      tenantID,
		KeyName:       "default",
		JWTCiphertext: apiKeyCipher,
		JWTHash:       token.HashToken(apiKey),
		TokenVersion:  1,
		Status:        meta.APIKeyActive,
		ScopeKind:     scopeKind,
		IssuedAt:      now,
		ExpiresAt:     expiresAt,
		CreatedAt:     now,
		UpdatedAt:     now,
	}); err != nil {
		t.Fatal(err)
	}
	prov := &fakeProvisioner{
		provider:          provider,
		cluster:           &tenant.ClusterInfo{},
		defaultPublicKey:  "default-public",
		defaultPrivateKey: "default-private",
	}
	cfg := Config{Meta: db.Meta, Pool: db.Pool, Provisioner: prov, TokenSecret: tokenSecret}
	if provider == tenant.ProviderTiDBCloudStarterLegacy {
		cfg.Provisioner = nil
		cfg.LegacyStarterProvisioner = prov
	}
	server := NewWithConfig(cfg)
	t.Cleanup(server.Close)
	return &tenantDeleteRuntime{
		meta:        db.Meta,
		pool:        db.Pool,
		tokenSecret: tokenSecret,
		tenantID:    tenantID,
		apiKey:      apiKey,
		prov:        prov,
		server:      server,
	}
}

func newTenantDeleteDBInfo(t *testing.T) *testDBInfo {
	t.Helper()
	if testDSN == "" {
		t.Skip("no test database available")
	}
	dropTenantSchemaLLMUsage(t)
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	initServerTenantSchema(t, testDSN)

	parsed, err := mysql.ParseDSN(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	host, port := "127.0.0.1", 3306
	if parsed.Addr != "" {
		h, p, _ := strings.Cut(parsed.Addr, ":")
		if h != "" {
			host = h
		}
		if p != "" {
			_, _ = fmt.Sscanf(p, "%d", &port)
		}
	}
	master := make([]byte, 32)
	if _, err := rand.Read(master); err != nil {
		t.Fatal(err)
	}
	enc, err := encrypt.NewLocalAESEncryptor(master)
	if err != nil {
		t.Fatal(err)
	}
	pool := tenant.NewPool(tenant.PoolConfig{S3Dir: mustTempDir(t), PublicURL: "http://localhost", SkipTiDBSchemaCheck: true}, enc)
	pool.SetMetaStore(metaStore)
	t.Cleanup(pool.Close)
	return &testDBInfo{Meta: metaStore, Pool: pool, DBHost: host, DBPort: port, DBUser: parsed.User, DBPass: parsed.Passwd, DBName: parsed.DBName}
}

func dropTenantSchemaLLMUsage(t *testing.T) {
	t.Helper()
	db := testtidb.OpenDB(t, testDSN)
	var tableCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'llm_usage'`).Scan(&tableCount); err != nil {
		t.Fatalf("check llm_usage table: %v", err)
	}
	if tableCount == 0 {
		return
	}
	var tenantIDColumnCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'llm_usage' AND column_name = 'tenant_id'`).Scan(&tenantIDColumnCount); err != nil {
		t.Fatalf("check llm_usage tenant_id column: %v", err)
	}
	if tenantIDColumnCount > 0 {
		return
	}
	if _, err := db.Exec("DROP TABLE llm_usage"); err != nil {
		t.Fatalf("drop tenant-schema llm_usage: %v", err)
	}
}

func (rt *tenantDeleteRuntime) deleteTenant(t *testing.T, body any) *http.Response {
	t.Helper()
	var reader *bytes.Reader
	if body == nil {
		reader = bytes.NewReader(nil)
	} else {
		raw, err := json.Marshal(body)
		if err != nil {
			t.Fatal(err)
		}
		reader = bytes.NewReader(raw)
	}
	req := httptest.NewRequest(http.MethodDelete, "/v1/tenant", reader)
	req.Header.Set("Authorization", "Bearer "+rt.apiKey)
	req.Header.Set("Content-Type", "application/json")
	rec := httptest.NewRecorder()
	rt.server.ServeHTTP(rec, req)
	return rec.Result()
}

func TestTenantDeleteNativeRequiresOwnerAndTiDBCloudCredentials(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	resp := rt.deleteTenant(t, map[string]string{"public_key": "public-1"})
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusBadRequest)
	}
	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.TenantActive) {
		t.Fatalf("tenant status = %s, want %s", status, meta.TenantActive)
	}
}

func TestTenantDeleteRejectsScopedAPIKey(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindFS)
	resp := rt.deleteTenant(t, map[string]string{"public_key": "public-1", "private_key": "private-1"})
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusForbidden {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusForbidden)
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls = %d, want 0", got)
	}
}

func TestTenantDeleteNativeRejectsUnknownCredentialFields(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	req := httptest.NewRequest(http.MethodDelete, "/v1/tenant", strings.NewReader(`{"public_key":"public-1","private_key":"private-1","unknown_field":"x"}`))
	req.Header.Set("Authorization", "Bearer "+rt.apiKey)
	req.Header.Set("Content-Type", "application/json")
	rec := httptest.NewRecorder()
	rt.server.ServeHTTP(rec, req)
	resp := rec.Result()
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusBadRequest)
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls = %d, want 0", got)
	}
}

func TestTenantDeleteNativeSucceedsWithCredentials(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	resp := rt.deleteTenant(t, map[string]string{"public_key": "public-1", "private_key": "private-1"})
	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 != 1 {
		t.Fatalf("deprovision calls = %d, want 1", got)
	}
	if rt.prov.lastCredentialReq.PublicKey != "public-1" || rt.prov.lastCredentialReq.PrivateKey != "private-1" {
		t.Fatalf("credential request = %+v", rt.prov.lastCredentialReq)
	}
	if rt.prov.lastDeprovision == nil || rt.prov.lastDeprovision.ClusterID != "cluster-delete-1" {
		t.Fatalf("deprovision cluster = %+v", rt.prov.lastDeprovision)
	}
	assertTenantMetricsCleanupNotify(t, rt)
	assertTenantDeletedAndKeysRevoked(t, rt)
}

func TestTenantDeleteLocalDropsDatabase(t *testing.T) {
	if testDSN == "" {
		t.Skip("no test database available")
	}
	t.Setenv(local.EnvAdminDSN, testDSN)
	prov, err := local.NewProvisionerFromEnv()
	if err != nil {
		t.Fatalf("NewProvisionerFromEnv: %v", err)
	}
	ctx := context.Background()
	tenantID := token.NewID()
	info, err := prov.Provision(ctx, tenantID)
	if err != nil {
		msg := strings.ToLower(err.Error())
		if strings.Contains(msg, "denied") || strings.Contains(msg, "privilege") {
			t.Skipf("test tidb user cannot CREATE DATABASE: %v", err)
		}
		t.Fatalf("Provision: %v", err)
	}
	t.Cleanup(func() { _ = prov.Deprovision(ctx, info) })

	db := newTenantDeleteDBInfo(t)
	testtidb.ResetMetaDB(t, db.Meta.DB())
	t.Cleanup(func() { testtidb.ResetMetaDB(t, db.Meta.DB()) })
	tokenSecret := make([]byte, 32)
	if _, err := rand.Read(tokenSecret); err != nil {
		t.Fatal(err)
	}
	apiKey, err := token.IssueToken(tokenSecret, tenantID, 1)
	if err != nil {
		t.Fatal(err)
	}
	apiKeyCipher, err := db.Pool.Encrypt(ctx, []byte(apiKey))
	if err != nil {
		t.Fatal(err)
	}
	dbPassCipher, err := db.Pool.Encrypt(ctx, []byte(info.Password))
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	if err := db.Meta.InsertTenant(ctx, &meta.Tenant{
		ID:               tenantID,
		Status:           meta.TenantActive,
		Kind:             meta.TenantKindLive,
		DBHost:           info.Host,
		DBPort:           info.Port,
		DBUser:           info.Username,
		DBPasswordCipher: dbPassCipher,
		DBName:           info.DBName,
		DBTLS:            false,
		Provider:         tenant.ProviderLocal,
		ClusterID:        info.ClusterID,
		SchemaVersion:    1,
		CreatedAt:        now,
		UpdatedAt:        now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := db.Meta.InsertAPIKey(ctx, &meta.APIKey{
		ID:            token.NewID(),
		TenantID:      tenantID,
		KeyName:       "default",
		JWTCiphertext: apiKeyCipher,
		JWTHash:       token.HashToken(apiKey),
		TokenVersion:  1,
		Status:        meta.APIKeyActive,
		ScopeKind:     meta.APIKeyScopeKindOwner,
		IssuedAt:      now,
		CreatedAt:     now,
		UpdatedAt:     now,
	}); err != nil {
		t.Fatal(err)
	}
	server := NewWithConfig(Config{Meta: db.Meta, Pool: db.Pool, Provisioner: prov, TokenSecret: tokenSecret})
	t.Cleanup(server.Close)
	rt := &tenantDeleteRuntime{meta: db.Meta, pool: db.Pool, tokenSecret: tokenSecret, tenantID: tenantID, apiKey: apiKey, server: server}

	resp := rt.deleteTenant(t, nil)
	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)
	}
	admin := testtidb.OpenDB(t, testDSN)
	var n int
	if err := admin.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.schemata WHERE schema_name = ?`, info.DBName).Scan(&n); err != nil {
		t.Fatalf("check dropped database: %v", err)
	}
	if n != 0 {
		t.Fatalf("local tenant database %s still exists after delete", info.DBName)
	}
	assertTenantDeletedAndKeysRevoked(t, rt)
}

func assertTenantMetricsCleanupNotify(t *testing.T, rt *tenantDeleteRuntime) {
	t.Helper()
	if rt.server.notifyCoalescer != nil {
		rt.server.notifyCoalescer.flush(context.Background())
	}
	rows, err := rt.meta.ListTenantNotifySince(context.Background(), 0, 100)
	if err != nil {
		t.Fatalf("list tenant cleanup notifications: %v", err)
	}
	for _, row := range rows {
		if row.TenantID == rt.tenantID && row.WorkMask&WorkMetricsCleanup != 0 {
			return
		}
	}
	t.Fatalf("tenant %s has no metrics-cleanup outbox notification: %+v", rt.tenantID, rows)
}

func TestTenantDeleteNativeUsesServerDefaultCredentials(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	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 != 1 {
		t.Fatalf("deprovision calls = %d, want 1", got)
	}
	if rt.prov.lastCredentialReq.PublicKey != "default-public" || rt.prov.lastCredentialReq.PrivateKey != "default-private" {
		t.Fatalf("native delete should use server default credentials: %+v", rt.prov.lastCredentialReq)
	}
	assertTenantDeletedAndKeysRevoked(t, rt)
}

func TestTenantDeleteLegacyStarterUsesLegacyProvisioner(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudStarterLegacy, meta.APIKeyScopeKindOwner)
	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 != 1 {
		t.Fatalf("deprovision calls = %d, want 1", got)
	}
	if rt.prov.lastCredentialReq.PublicKey != "" || rt.prov.lastCredentialReq.PrivateKey != "" {
		t.Fatalf("legacy starter delete should not use credential request: %+v", rt.prov.lastCredentialReq)
	}
	if rt.prov.lastDeprovision == nil || rt.prov.lastDeprovision.ClusterID != "cluster-delete-1" {
		t.Fatalf("deprovision cluster = %+v", rt.prov.lastDeprovision)
	}
	if rt.prov.lastDeprovision.Provider != tenant.ProviderTiDBCloudStarterLegacy {
		t.Fatalf("deprovision provider = %q, want %q", rt.prov.lastDeprovision.Provider, tenant.ProviderTiDBCloudStarterLegacy)
	}
	assertTenantDeletedAndKeysRevoked(t, rt)
}

func TestTenantDeleteWaitsForBorrowedBackendBeforeClusterDelete(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	tenantMeta, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	_, release, err := rt.pool.Acquire(ctx, tenantMeta)
	if err != nil {
		t.Fatalf("acquire tenant backend: %v", err)
	}
	defer release()

	respCh := make(chan *http.Response, 1)
	go func() {
		respCh <- rt.deleteTenant(t, nil)
	}()
	waitTenantDeleteStatus(t, rt, meta.TenantDeleting)
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls before backend release = %d, want 0", got)
	}
	select {
	case resp := <-respCh:
		defer func() { _ = resp.Body.Close() }()
		t.Fatalf("delete response returned before backend release: status %d", resp.StatusCode)
	default:
	}

	release()
	resp := receiveTenantDeleteResponse(t, respCh)
	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 != 1 {
		t.Fatalf("deprovision calls after backend release = %d, want 1", got)
	}
	assertTenantDeletingAndKeysRevoked(t, rt)
	rt.server.processTenantDeleteJobs(ctx)
	assertTenantDeletedAndKeysRevoked(t, rt)
}

func TestTenantDeleteRejectsNonDeletedFork(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	tenantMeta, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if _, release, err := rt.pool.Acquire(ctx, tenantMeta); err != nil {
		t.Fatalf("acquire tenant backend: %v", err)
	} else {
		release()
	}
	tenantMeta, err = rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	if err := rt.meta.InsertTenant(ctx, &meta.Tenant{
		ID:                 token.NewID(),
		Status:             meta.TenantActive,
		Kind:               meta.TenantKindFork,
		ParentTenantID:     rt.tenantID,
		StorageNamespaceID: tenantMeta.StorageNamespaceID,
		DBHost:             tenantMeta.DBHost,
		DBPort:             tenantMeta.DBPort,
		DBUser:             tenantMeta.DBUser,
		DBPasswordCipher:   tenantMeta.DBPasswordCipher,
		DBName:             tenantMeta.DBName,
		DBTLS:              tenantMeta.DBTLS,
		Provider:           tenantMeta.Provider,
		ClusterID:          tenantMeta.ClusterID,
		BranchID:           "branch-a",
		SchemaVersion:      tenantMeta.SchemaVersion,
		CreatedAt:          now,
		UpdatedAt:          now,
	}); err != nil {
		t.Fatal(err)
	}

	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusConflict {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusConflict)
	}
	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.TenantActive) {
		t.Fatalf("tenant status = %s, want %s", status, meta.TenantActive)
	}
}

func TestTenantDeleteRemovesS3PrefixAsyncAfterClusterDelete(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	payload := deterministicObjectGCPayload(8*1024*1024, 0x72)

	tenantMeta, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	backendFS, release, err := rt.pool.Acquire(ctx, tenantMeta)
	if err != nil {
		t.Fatalf("acquire tenant backend: %v", err)
	}
	if _, err := backendFS.Write("/large.bin", payload, 0, filesystem.WriteFlagCreate); err != nil {
		release()
		t.Fatalf("write s3 file: %v", err)
	}
	storageRef := objectGCFileStorageRef(t, backendFS, "/large.bin")
	tenantMeta, err = rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		release()
		t.Fatal(err)
	}
	namespaceID := tenantMeta.StorageNamespaceID
	assertTenantDeleteS3ObjectBytes(t, rt, namespaceID, storageRef, payload)
	release()

	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 != 1 {
		t.Fatalf("deprovision calls = %d, want 1", got)
	}
	assertTenantDeletingAndKeysRevoked(t, rt)
	assertTenantDeleteS3ObjectBytes(t, rt, namespaceID, storageRef, payload)

	rt.server.processTenantDeleteJobs(ctx)
	assertTenantDeletedAndKeysRevoked(t, rt)

	assertTenantDeleteS3ObjectMissing(t, rt, namespaceID, storageRef)
}

func TestTenantDeleteAsyncCleanupAbortsActiveMultipartUploads(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	payload := deterministicObjectGCPayload(8*1024*1024, 0x29)

	tenantMeta, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	backendFS, release, err := rt.pool.Acquire(ctx, tenantMeta)
	if err != nil {
		t.Fatalf("acquire tenant backend: %v", err)
	}
	plan, err := backendFS.InitiateUpload(ctx, "/uploading.bin", int64(len(payload)))
	if err != nil {
		release()
		t.Fatalf("initiate upload: %v", err)
	}
	if len(plan.Parts) == 0 {
		release()
		t.Fatal("upload plan has no parts")
	}
	uploadBeforeDelete, err := backendFS.Store().GetUpload(ctx, plan.UploadID)
	if err != nil {
		release()
		t.Fatalf("get upload before delete: %v", err)
	}
	if _, err := backendFS.S3().(*s3client.LocalS3Client).UploadPart(ctx, uploadBeforeDelete.S3UploadID, plan.Parts[0].Number, bytes.NewReader(payload[:plan.Parts[0].Size])); err != nil {
		release()
		t.Fatalf("upload part: %v", err)
	}
	tenantMeta, err = rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		release()
		t.Fatal(err)
	}
	namespaceID := tenantMeta.StorageNamespaceID
	release()

	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 != 1 {
		t.Fatalf("deprovision calls = %d, want 1", got)
	}
	assertTenantDeletingAndKeysRevoked(t, rt)

	rt.server.processTenantDeleteJobs(ctx)
	assertTenantDeletedAndKeysRevoked(t, rt)

	s3c := tenantDeleteS3ForNamespace(t, rt, namespaceID)
	parts, err := s3c.ListParts(ctx, uploadBeforeDelete.S3Key, uploadBeforeDelete.S3UploadID)
	if err == nil {
		if len(parts) != 0 {
			t.Fatalf("multipart parts after delete = %d, want 0", len(parts))
		}
		t.Fatal("multipart upload still exists after tenant delete")
	}
	if !strings.Contains(err.Error(), "upload not found") {
		t.Fatalf("list parts after delete error = %v, want upload not found", err)
	}
}

func tenantDeleteS3ForNamespace(t *testing.T, rt *tenantDeleteRuntime, namespaceID string) s3client.S3Client {
	t.Helper()
	ns, err := rt.meta.GetStorageNamespace(context.Background(), namespaceID)
	if err != nil {
		t.Fatalf("get storage namespace: %v", err)
	}
	s3c, err := rt.pool.S3ForStorageNamespace(context.Background(), ns)
	if err != nil {
		t.Fatalf("s3 for storage namespace: %v", err)
	}
	return s3c
}

func assertTenantDeleteS3ObjectBytes(t *testing.T, rt *tenantDeleteRuntime, namespaceID, storageRef string, want []byte) {
	t.Helper()
	reader, err := tenantDeleteS3ForNamespace(t, rt, namespaceID).GetObject(context.Background(), storageRef)
	if err != nil {
		t.Fatalf("get s3 object %q: %v", storageRef, err)
	}
	defer func() { _ = reader.Close() }()
	got, err := io.ReadAll(reader)
	if err != nil {
		t.Fatalf("read s3 object %q: %v", storageRef, err)
	}
	if !bytes.Equal(got, want) {
		t.Fatalf("s3 object %q bytes mismatch: got %d bytes want %d bytes", storageRef, len(got), len(want))
	}
}

func assertTenantDeleteS3ObjectMissing(t *testing.T, rt *tenantDeleteRuntime, namespaceID, storageRef string) {
	t.Helper()
	reader, err := tenantDeleteS3ForNamespace(t, rt, namespaceID).GetObject(context.Background(), storageRef)
	if err == nil {
		_ = reader.Close()
		t.Fatalf("s3 object %q still exists after tenant delete", storageRef)
	}
}

func TestTenantDeleteTiDBZeroUnsupported(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBZero, meta.APIKeyScopeKindOwner)
	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusConflict {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusConflict)
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls = %d, want 0", got)
	}
}

func assertTenantDeletedAndKeysRevoked(t *testing.T, rt *tenantDeleteRuntime) {
	t.Helper()
	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.TenantDeleted) {
		t.Fatalf("tenant status = %s, want %s", status, meta.TenantDeleted)
	}
	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 != 0 {
		t.Fatalf("active api keys = %d, want 0", activeKeys)
	}
	var revokedKeys int
	if err := rt.meta.DB().QueryRow("SELECT COUNT(*) FROM tenant_api_keys WHERE tenant_id = ? AND status = ?", rt.tenantID, meta.APIKeyRevoked).Scan(&revokedKeys); err != nil {
		t.Fatal(err)
	}
	if revokedKeys != 1 {
		t.Fatalf("revoked api keys = %d, want 1", revokedKeys)
	}
}

func waitTenantDeleteStatus(t *testing.T, rt *tenantDeleteRuntime, want meta.TenantStatus) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		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(want) {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("tenant status did not become %s, last status %s", want, status)
		}
		time.Sleep(10 * time.Millisecond)
	}
}

func receiveTenantDeleteResponse(t *testing.T, ch <-chan *http.Response) *http.Response {
	t.Helper()
	select {
	case resp := <-ch:
		return resp
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for delete response")
		return nil
	}
}

func assertTenantDeletingAndKeysRevoked(t *testing.T, rt *tenantDeleteRuntime) {
	t.Helper()
	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)
	}
	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 != 0 {
		t.Fatalf("active api keys = %d, want 0", activeKeys)
	}
}

func TestTenantDeleteClusterFailureKeepsTenantDeletingForCredentialRetry(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	rt.prov.deprovisionErr = fmt.Errorf("upstream down")
	resp := rt.deleteTenant(t, map[string]string{"public_key": "public-1", "private_key": "private-1"})
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusBadGateway {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusBadGateway)
	}
	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)
	}
}

func TestTenantDeleteCanRetryDeletingTenant(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	tenantRow, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	if _, release, err := rt.pool.Acquire(ctx, tenantRow); err != nil {
		t.Fatalf("acquire tenant backend: %v", err)
	} else {
		release()
	}
	tenantRow, err = rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	updated, err := rt.meta.UpdateTenantStatusIf(ctx, rt.tenantID, tenantRow.Status, meta.TenantDeleting)
	if err != nil || !updated {
		t.Fatalf("seed interrupted delete: updated=%v err=%v", updated, err)
	}
	rt.server.processTenantDeleteJobs(ctx)
	waitTenantDeleteStatus(t, rt, meta.TenantDeleting)

	resp := rt.deleteTenant(t, map[string]string{"public_key": "public-1", "private_key": "private-1"})
	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 != 1 {
		t.Fatalf("deprovision calls = %d, want 1", got)
	}
	rt.server.processTenantDeleteJobs(ctx)
	assertTenantDeletedAndKeysRevoked(t, rt)
}

func TestTenantDeleteDoesNotFinalizeBeforeExternalClusterDeleteConfirmation(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	tenantRow, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	_, release, err := rt.pool.Acquire(ctx, tenantRow)
	if err != nil {
		t.Fatalf("acquire tenant backend: %v", err)
	}
	defer release()

	respCh := make(chan *http.Response, 1)
	go func() {
		respCh <- rt.deleteTenant(t, map[string]string{"public_key": "public-1", "private_key": "private-1"})
	}()
	waitTenantDeleteStatus(t, rt, meta.TenantDeleting)
	var jobCount int
	if err := rt.meta.DB().QueryRow(`SELECT COUNT(*) FROM tenant_delete_jobs WHERE tenant_id = ?`, rt.tenantID).Scan(&jobCount); err != nil {
		t.Fatal(err)
	}
	if jobCount != 0 {
		t.Fatalf("cleanup jobs before cluster delete = %d, want 0", jobCount)
	}
	rt.server.processTenantDeleteJobs(context.Background())
	waitTenantDeleteStatus(t, rt, meta.TenantDeleting)
	if got := rt.prov.deprovisionCalls.Load(); got != 0 {
		t.Fatalf("deprovision calls while backend is borrowed = %d, want 0", got)
	}

	release()
	resp := receiveTenantDeleteResponse(t, respCh)
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("delete status = %d, want %d", resp.StatusCode, http.StatusAccepted)
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 1 {
		t.Fatalf("deprovision calls after backend release = %d, want 1", got)
	}
	rt.server.processTenantDeleteJobs(ctx)
	assertTenantDeletedAndKeysRevoked(t, rt)
}

func TestProcessTenantDeleteJobSkipsTenantNotDeleting(t *testing.T) {
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	ctx := context.Background()
	now := time.Now().UTC()
	if err := metaStore.InsertTenant(ctx, &meta.Tenant{
		ID: "job-guard-tenant", Status: meta.TenantActive, Provider: tenant.ProviderTiDBCloudNative,
		StorageNamespaceID: "job-guard-ns", DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := metaStore.UpsertStorageNamespace(ctx, &meta.StorageNamespace{
		ID: "job-guard-ns", OwnerTenantID: "job-guard-tenant", Backend: "local",
		Bucket: "bucket", Prefix: "tenant/job-guard-tenant", State: meta.StorageNamespaceActive,
		CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := metaStore.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID: "job-guard-tenant", NamespaceID: "job-guard-ns", Backend: "local",
		Bucket: "bucket", Prefix: "tenant/job-guard-tenant",
	}); err != nil {
		t.Fatal(err)
	}

	srv := &Server{meta: metaStore}
	t.Cleanup(srv.Close)
	if err := srv.processTenantDeleteJob(ctx, meta.TenantDeleteJob{
		TenantID: "job-guard-tenant", NamespaceID: "job-guard-ns", Backend: "local",
		Bucket: "bucket", Prefix: "tenant/job-guard-tenant",
	}); err != nil {
		t.Fatal(err)
	}

	got, err := metaStore.GetTenant(ctx, "job-guard-tenant")
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != meta.TenantActive {
		t.Fatalf("tenant status = %s, want %s (job must not touch a non-deleting tenant)", got.Status, meta.TenantActive)
	}
	var state string
	if err := metaStore.DB().QueryRow(`SELECT state FROM tenant_delete_jobs WHERE tenant_id = ?`, "job-guard-tenant").Scan(&state); !errors.Is(err, sql.ErrNoRows) {
		t.Fatalf("cancelled job row error = %v, want sql.ErrNoRows", err)
	}
}

func TestTenantDeleteFailsFastWhenStorageNamespaceMissing(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	if _, err := rt.meta.DB().ExecContext(ctx, "UPDATE tenants SET storage_namespace_id = ? WHERE id = ?", "missing-namespace", rt.tenantID); err != nil {
		t.Fatal(err)
	}

	resp := rt.deleteTenant(t, map[string]string{"public_key": "public-1", "private_key": "private-1"})
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusInternalServerError {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusInternalServerError)
	}
	// The namespace lookup now happens before the deleting mark and the
	// cluster deprovision, so a missing namespace row fails the request
	// without any side effects and stays retryable once the data is fixed.
	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.TenantActive) {
		t.Fatalf("tenant status = %s, want %s", status, meta.TenantActive)
	}
	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", activeKeys)
	}
}

func TestTenantDeleteDeletingTenantWithCleanupJobDoesNotRepeatClusterDelete(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	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, map[string]string{"public_key": "public-1", "private_key": "private-1"})
	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)
	}
}

// TestTenantDeleteTerminatesSSEStreamAndDrains reproduces the production
// incident where an established /v1/events SSE connection pinned the tenant
// backend reference for its whole connection lifetime: the delete path
// waited out the full pool drain timeout (30s) and failed. The delete must
// instead terminate the stream, let its auth-deferred release drop the
// reference, and proceed immediately.
func TestTenantDeleteTerminatesSSEStreamAndDrains(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)

	// Establish an SSE stream through the full middleware chain: auth
	// Acquires the tenant backend and defers release until the handler
	// returns, so the stream pins one pool reference. The request context
	// is never canceled, simulating a client that stays connected.
	sseDone := make(chan struct{})
	go func() {
		defer close(sseDone)
		req := httptest.NewRequest(http.MethodGet, "/v1/events?since=0", nil)
		req.Header.Set("Authorization", "Bearer "+rt.apiKey)
		rec := httptest.NewRecorder()
		rt.server.ServeHTTP(rec, req)
	}()

	// Wait until the stream is established (subscribed to the tenant bus).
	deadline := time.Now().Add(5 * time.Second)
	for {
		if bus := rt.server.events.getIfExists(rt.tenantID); bus != nil && bus.HasListeners() {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("SSE stream did not subscribe in time")
		}
		time.Sleep(10 * time.Millisecond)
	}

	// The drain must terminate the stream and complete promptly — without the
	// bus close this delete would wait out the full 30s pool drain timeout.
	start := time.Now()
	resp := rt.deleteTenant(t, nil)
	defer func() { _ = resp.Body.Close() }()
	if elapsed := time.Since(start); elapsed > 5*time.Second {
		t.Fatalf("delete took %v, want < 5s (drain timeout regression)", elapsed)
	}
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusAccepted)
	}
	select {
	case <-sseDone:
	case <-time.After(5 * time.Second):
		t.Fatal("SSE stream did not terminate after tenant delete drain")
	}
	if got := rt.prov.deprovisionCalls.Load(); got != 1 {
		t.Fatalf("deprovision calls = %d, want 1", got)
	}
	assertTenantDeletingAndKeysRevoked(t, rt)
}

// TestSSEEventsRejectsDeletingTenantBeforeSubscribe covers the in-flight
// connect race: a /v1/events request that passed the auth status gate while
// the tenant was still active must not subscribe (and keep pinning the pool
// backend reference via auth's deferred release) once the tenant has flipped
// to deleting — otherwise a delete drain racing it waits out the pool drain
// timeout.
func TestSSEEventsRejectsDeletingTenantBeforeSubscribe(t *testing.T) {
	rt := newTenantDeleteRuntime(t, tenant.ProviderTiDBCloudNative, meta.APIKeyScopeKindOwner)
	ctx := context.Background()
	tenantMeta, err := rt.meta.GetTenant(ctx, rt.tenantID)
	if err != nil {
		t.Fatal(err)
	}
	// Simulate the in-flight connect: auth already acquired the backend...
	b, release, err := rt.pool.Acquire(ctx, tenantMeta)
	if err != nil {
		t.Fatalf("acquire tenant backend: %v", err)
	}
	defer release()
	// ...then the tenant flips to deleting before the request subscribes.
	if err := rt.meta.UpdateTenantStatus(ctx, rt.tenantID, meta.TenantDeleting); err != nil {
		t.Fatal(err)
	}

	req := httptest.NewRequest(http.MethodGet, "/v1/events", nil)
	scope := &TenantScope{TenantID: rt.tenantID, Backend: b}
	rec := httptest.NewRecorder()
	rt.server.handleEvents(rec, req.WithContext(withScope(req.Context(), scope)))
	if rec.Code != http.StatusForbidden {
		t.Fatalf("status = %d, want %d", rec.Code, http.StatusForbidden)
	}
	// The re-check must also tear down the bus it just resolved, so a late
	// subscriber cannot park on it.
	if bus := rt.server.events.getIfExists(rt.tenantID); bus != nil {
		t.Fatal("bus still present after deleting-tenant rejection")
	}
}