package engine

import (
	"context"
	"errors"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
)

const (
	testMaxComputeProject    = "clinic_proxy_test_ap_northeast_1"
	testMaxComputeRegion     = "ap-northeast-1"
	testMaxComputeSlowTable  = "slow_query_logs_10109909546444159555"
	testMaxComputeDataSource = "oss://clinic-test/slowlogs"
)

type fakeMaxComputeTableManagerEngine struct {
	checkCalls   atomic.Int32
	executeCalls atomic.Int32

	checkFn   func(context.Context, string, string, string) (bool, error)
	executeFn func(context.Context, string, string) error

	statementsMu sync.Mutex
	statements   []string
}

func (f *fakeMaxComputeTableManagerEngine) CheckTableExists(
	ctx context.Context,
	project, table, region string,
) (bool, error) {
	f.checkCalls.Add(1)
	if f.checkFn != nil {
		return f.checkFn(ctx, project, table, region)
	}
	return true, nil
}

func (f *fakeMaxComputeTableManagerEngine) ExecuteStatement(
	ctx context.Context,
	statement, region string,
) error {
	f.executeCalls.Add(1)
	f.statementsMu.Lock()
	f.statements = append(f.statements, statement)
	f.statementsMu.Unlock()
	if f.executeFn != nil {
		return f.executeFn(ctx, statement, region)
	}
	return nil
}

func (f *fakeMaxComputeTableManagerEngine) recordedStatements() []string {
	f.statementsMu.Lock()
	defer f.statementsMu.Unlock()
	return append([]string(nil), f.statements...)
}

func newTestMaxComputeTableManager(
	engine maxComputeTableManagerEngine,
	cacheTTLSeconds int,
) *MaxComputeTableManager {
	return &MaxComputeTableManager{
		engine:       engine,
		prefetchSlot: make(chan struct{}, 1),
		config: &config.TableManagerConfig{
			Enabled:         true,
			AutoCreate:      true,
			CacheTTLSeconds: cacheTTLSeconds,
			TableTemplates: map[string]config.TableTemplate{
				"slow_query_logs": {
					DDL: "CREATE EXTERNAL TABLE IF NOT EXISTS `{{.Database}}`.`{{.Table}}` (time BIGINT) LOCATION '{{.Location}}'",
				},
				"topsql": {
					DDL: "CREATE EXTERNAL TABLE IF NOT EXISTS `{{.Database}}`.`{{.Table}}` (time BIGINT) LOCATION '{{.Location}}'",
				},
			},
		},
	}
}

func seedMaxComputeTableCache(tm *MaxComputeTableManager, table string, age time.Duration) *TableMetadata {
	metadata := &TableMetadata{Exists: true, LastChecked: time.Now().Add(-age)}
	tm.setCachedMetadata(
		maxComputeTableCacheKey(testMaxComputeProject, table, testMaxComputeRegion),
		metadata,
	)
	return metadata
}

func seedMaxComputePartitionCache(tm *MaxComputeTableManager, table string, age time.Duration) *PartitionMetadata {
	metadata := &PartitionMetadata{Loaded: true, LastChecked: time.Now().Add(-age)}
	tm.setCachedPartitionMetadata(
		maxComputePartitionCacheKey(testMaxComputeProject, table, testMaxComputeRegion),
		metadata,
	)
	return metadata
}

func waitForMaxComputeCondition(t *testing.T, condition func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for !condition() {
		if time.Now().After(deadline) {
			t.Fatal("timed out waiting for MaxCompute table-manager condition")
		}
		time.Sleep(time.Millisecond)
	}
}

func receiveMaxComputeTestValue[T any](t *testing.T, ch <-chan T) T {
	t.Helper()
	select {
	case value := <-ch:
		return value
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for MaxCompute table-manager test channel")
		var zero T
		return zero
	}
}

func waitForMaxComputePrefetchToFinish(t *testing.T, tm *MaxComputeTableManager, table string) {
	t.Helper()
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, table, testMaxComputeRegion)
	waitForMaxComputeCondition(t, func() bool {
		_, active := tm.activePrefetches.Load(cacheKey)
		return !active
	})
}

func TestMaxComputeSlowQueryFirstUseCollapsesConcurrentMaintenance(t *testing.T) {
	checkStarted := make(chan struct{})
	releaseCheck := make(chan struct{})
	var startOnce sync.Once
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			startOnce.Do(func() { close(checkStarted) })
			<-releaseCheck
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)

	const callers = 32
	errs := make(chan error, callers)
	for i := 0; i < callers; i++ {
		go func() {
			errs <- tm.EnsureTableExists(
				context.Background(),
				testMaxComputeProject,
				testMaxComputeSlowTable,
				testMaxComputeDataSource,
				testMaxComputeRegion,
			)
		}()
	}

	receiveMaxComputeTestValue(t, checkStarted)
	close(releaseCheck)
	for i := 0; i < callers; i++ {
		if err := receiveMaxComputeTestValue(t, errs); err != nil {
			t.Fatalf("EnsureTableExists() error = %v", err)
		}
	}

	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want 1", got)
	}
	if got := fake.executeCalls.Load(); got != 1 {
		t.Fatalf("ExecuteStatement calls = %d, want one MSCK", got)
	}
}

func TestMaxComputeSlowQueryMissingTableCreatesThenRepairs(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			return false, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)

	const callers = 32
	errs := make(chan error, callers)
	for i := 0; i < callers; i++ {
		go func() {
			errs <- tm.EnsureTableExists(
				context.Background(),
				testMaxComputeProject,
				testMaxComputeSlowTable,
				testMaxComputeDataSource,
				testMaxComputeRegion,
			)
		}()
	}
	for i := 0; i < callers; i++ {
		if err := receiveMaxComputeTestValue(t, errs); err != nil {
			t.Fatalf("EnsureTableExists() error = %v", err)
		}
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want one", got)
	}
	statements := fake.recordedStatements()
	if len(statements) != 2 || !strings.HasPrefix(statements[0], "CREATE EXTERNAL TABLE") ||
		!strings.HasPrefix(statements[1], "MSCK REPAIR TABLE") {
		t.Fatalf("statements = %#v, want CREATE followed by MSCK", statements)
	}
}

func TestMaxComputeSlowQueryCacheContracts(t *testing.T) {
	t.Run("fresh positive TTL", func(t *testing.T) {
		fake := &fakeMaxComputeTableManagerEngine{}
		tm := newTestMaxComputeTableManager(fake, 600)
		seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 569*time.Second)
		if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
			t.Fatalf("EnsureTableExists() error = %v", err)
		}
		if got := fake.checkCalls.Load(); got != 0 {
			t.Fatalf("CheckTableExists calls = %d, want fresh cache hit", got)
		}
	})

	t.Run("negative TTL keeps synchronous legacy behavior", func(t *testing.T) {
		fake := &fakeMaxComputeTableManagerEngine{}
		tm := newTestMaxComputeTableManager(fake, -1)
		seedMaxComputeTableCache(tm, testMaxComputeSlowTable, time.Second)
		if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
			t.Fatalf("EnsureTableExists() error = %v", err)
		}
		if got := fake.checkCalls.Load(); got != 1 {
			t.Fatalf("CheckTableExists calls = %d, want synchronous check", got)
		}
	})

	if got := slowQueryPrefetchThreshold(600 * time.Second); got != 570*time.Second {
		t.Fatalf("slowQueryPrefetchThreshold(600s) = %v, want 570s", got)
	}
}

func TestMaxComputeSlowQueryPrefetchDoesNotPublishFreshness(t *testing.T) {
	checkStarted := make(chan context.Context, 1)
	releaseCheck := make(chan struct{})
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(ctx context.Context, _, _, _ string) (bool, error) {
			checkStarted <- ctx
			<-releaseCheck
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	tableMetadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 571*time.Second)
	partitionMetadata := seedMaxComputePartitionCache(tm, testMaxComputeSlowTable, 571*time.Second)

	requestCtx, cancel := context.WithCancel(context.Background())
	cancel()
	if err := tm.EnsureTableExists(requestCtx, testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("prefetch-window EnsureTableExists() error = %v", err)
	}
	prefetchCtx := receiveMaxComputeTestValue(t, checkStarted)
	if err := prefetchCtx.Err(); err != nil {
		t.Fatalf("background prefetch context error = %v, want detached context", err)
	}
	close(releaseCheck)
	waitForMaxComputePrefetchToFinish(t, tm, testMaxComputeSlowTable)

	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	currentTable, _ := tm.getRawCachedMetadata(cacheKey)
	if currentTable != tableMetadata {
		t.Fatalf("table cache changed during prefetch: got %#v, want original %#v", currentTable, tableMetadata)
	}
	currentPartition, ok := tm.getCachedPartitionMetadata(
		maxComputePartitionCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion),
	)
	if !ok || currentPartition != partitionMetadata {
		t.Fatalf("partition cache changed during prefetch: got %#v, want original %#v", currentPartition, partitionMetadata)
	}
	if got := fake.executeCalls.Load(); got != 0 {
		t.Fatalf("ExecuteStatement calls = %d, want no background MSCK", got)
	}
	if _, ok := tm.loadSlowQueryTablePrefetch(cacheKey, tableMetadata.LastChecked.UnixNano()); !ok {
		t.Fatal("successful DESC result was not retained for the hard-TTL request")
	}
}

func TestMaxComputeSlowQueryConcurrentPrefetchStartsOneCheck(t *testing.T) {
	checkStarted := make(chan struct{})
	releaseCheck := make(chan struct{})
	var startOnce sync.Once
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			startOnce.Do(func() { close(checkStarted) })
			<-releaseCheck
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 571*time.Second)

	const callers = 128
	var wg sync.WaitGroup
	wg.Add(callers)
	for i := 0; i < callers; i++ {
		go func() {
			defer wg.Done()
			if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
				t.Errorf("EnsureTableExists() error = %v", err)
			}
		}()
	}
	wg.Wait()
	receiveMaxComputeTestValue(t, checkStarted)
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want one prefetch", got)
	}
	close(releaseCheck)
	waitForMaxComputePrefetchToFinish(t, tm, testMaxComputeSlowTable)
}

func TestMaxComputeSlowQueryPrefetchHasBoundedAdmission(t *testing.T) {
	const secondTable = "slow_query_logs_10109909546444159556"
	checkStarted := make(chan struct{})
	releaseCheck := make(chan struct{})
	var startOnce sync.Once
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			startOnce.Do(func() { close(checkStarted) })
			<-releaseCheck
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 571*time.Second)
	seedMaxComputeTableCache(tm, secondTable, 571*time.Second)

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("first prefetch error = %v", err)
	}
	receiveMaxComputeTestValue(t, checkStarted)
	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, secondTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("second prefetch-window call error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want one background operation per process", got)
	}

	close(releaseCheck)
	waitForMaxComputePrefetchToFinish(t, tm, testMaxComputeSlowTable)
	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, secondTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("second table retry error = %v", err)
	}
	waitForMaxComputeCondition(t, func() bool { return fake.checkCalls.Load() == 2 })
	waitForMaxComputePrefetchToFinish(t, tm, secondTable)
}

func TestMaxComputeSlowQueryHardTTLConsumesPrefetchAndKeepsPartitionPolicy(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	oldPartition := seedMaxComputePartitionCache(tm, testMaxComputeSlowTable, 601*time.Second)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	tm.prefetches.Store(cacheKey, &maxComputeTablePrefetch{
		cacheGeneration: metadata.LastChecked.UnixNano(),
		exists:          true,
		validUntil:      metadata.LastChecked.Add(630 * time.Second),
	})

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("hard-TTL EnsureTableExists() error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 0 {
		t.Fatalf("CheckTableExists calls = %d, want completed prefetch consumed", got)
	}
	if got := fake.executeCalls.Load(); got != 1 {
		t.Fatalf("ExecuteStatement calls = %d, want existing hard-TTL MSCK policy", got)
	}
	currentPartition, ok := tm.getCachedPartitionMetadata(
		maxComputePartitionCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion),
	)
	if !ok || !currentPartition.LastChecked.After(oldPartition.LastChecked) {
		t.Fatalf("partition metadata = %#v, want hard-TTL repair", currentPartition)
	}
}

func TestMaxComputeSlowQueryPrefetchRepairsPartitionDueByLegacyDescCompletion(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	seedMaxComputePartitionCache(tm, testMaxComputeSlowTable, 600*time.Second-10*time.Millisecond)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	tm.prefetches.Store(cacheKey, &maxComputeTablePrefetch{
		cacheGeneration: metadata.LastChecked.UnixNano(),
		exists:          true,
		checkDuration:   30 * time.Millisecond,
		validUntil:      time.Now().Add(30 * time.Millisecond),
	})

	startedAt := time.Now()
	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("hard-TTL EnsureTableExists() error = %v", err)
	}
	if elapsed := time.Since(startedAt); elapsed < 20*time.Millisecond {
		t.Fatalf("hard-TTL call completed in %v, want MSCK delayed to legacy DESC completion", elapsed)
	}
	currentTable, _ := tm.getRawCachedMetadata(cacheKey)
	if currentTable == metadata || !currentTable.LastChecked.After(metadata.LastChecked) {
		t.Fatalf("table cache = %#v, want prefetched freshness published", currentTable)
	}
	if got := fake.checkCalls.Load(); got != 0 {
		t.Fatalf("CheckTableExists calls = %d, want completed prefetch reused", got)
	}
	if got := fake.executeCalls.Load(); got != 1 {
		t.Fatalf("ExecuteStatement calls = %d, want counterfactual hard-path MSCK", got)
	}
}

func TestMaxComputeSlowQueryCanceledLeaderDoesNotPoisonMaintenanceFlight(t *testing.T) {
	checkStarted := make(chan context.Context, 1)
	releaseCheck := make(chan struct{})
	var releaseOnce sync.Once
	release := func() {
		releaseOnce.Do(func() { close(releaseCheck) })
	}
	defer release()
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(ctx context.Context, _, _, _ string) (bool, error) {
			checkStarted <- ctx
			<-releaseCheck
			if err := ctx.Err(); err != nil {
				return false, err
			}
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)

	canceledCtx, cancel := context.WithCancel(context.Background())
	cancel()
	leaderDone := make(chan error, 1)
	go func() {
		leaderDone <- tm.EnsureTableExists(
			canceledCtx,
			testMaxComputeProject,
			testMaxComputeSlowTable,
			testMaxComputeDataSource,
			testMaxComputeRegion,
		)
	}()

	leaderCtx := receiveMaxComputeTestValue(t, checkStarted)
	if err := leaderCtx.Err(); err != nil {
		t.Fatalf("maintenance leader context error = %v, want cancellation detached", err)
	}
	// The canceled caller is now deterministically leading the blocked check.
	// A healthy same-key caller should join without inheriting cancellation.
	healthyDone := make(chan error, 1)
	go func() {
		healthyDone <- tm.EnsureTableExists(
			context.Background(),
			testMaxComputeProject,
			testMaxComputeSlowTable,
			testMaxComputeDataSource,
			testMaxComputeRegion,
		)
	}()
	select {
	case err := <-healthyDone:
		t.Fatalf("healthy waiter returned before shared check completed: %v", err)
	case <-time.After(25 * time.Millisecond):
	}
	release()

	if err := receiveMaxComputeTestValue(t, leaderDone); err != nil {
		t.Fatalf("canceled maintenance leader error = %v, want legacy non-canceling behavior", err)
	}
	if err := receiveMaxComputeTestValue(t, healthyDone); err != nil {
		t.Fatalf("healthy maintenance waiter error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want one shared foreground check", got)
	}
	if got := fake.executeCalls.Load(); got != 1 {
		t.Fatalf("ExecuteStatement calls = %d, want one shared MSCK", got)
	}
}

func TestMaxComputeSlowQueryOldMaintenanceDoesNotDeleteNewGenerationPrefetch(t *testing.T) {
	executeStarted := make(chan struct{})
	releaseExecute := make(chan struct{})
	var startOnce sync.Once
	fake := &fakeMaxComputeTableManagerEngine{
		executeFn: func(context.Context, string, string) error {
			startOnce.Do(func() { close(executeStarted) })
			<-releaseExecute
			return nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	oldPrefetch := &maxComputeTablePrefetch{
		cacheGeneration: time.Now().Add(-601 * time.Second).UnixNano(),
		exists:          true,
		validUntil:      time.Now().Add(time.Minute),
	}
	tm.prefetches.Store(cacheKey, oldPrefetch)

	maintenanceDone := make(chan error, 1)
	go func() {
		maintenanceDone <- tm.completeSlowQueryTableMaintenance(
			context.Background(),
			testMaxComputeProject,
			testMaxComputeSlowTable,
			testMaxComputeDataSource,
			testMaxComputeRegion,
			oldPrefetch,
			time.Now(),
		)
	}()
	receiveMaxComputeTestValue(t, executeStarted)

	newPrefetch := &maxComputeTablePrefetch{
		cacheGeneration: time.Now().UnixNano(),
		exists:          true,
		validUntil:      time.Now().Add(time.Minute),
	}
	tm.prefetches.Store(cacheKey, newPrefetch)
	close(releaseExecute)
	if err := receiveMaxComputeTestValue(t, maintenanceDone); err != nil {
		t.Fatalf("old-generation maintenance error = %v", err)
	}
	got, ok := tm.prefetches.Load(cacheKey)
	if !ok || got != newPrefetch {
		t.Fatalf("prefetch after old maintenance = %#v, want newer generation %#v", got, newPrefetch)
	}
}

func TestMaxComputeSlowQueryPrefetchSkipsPartitionFreshPastLegacyDescCompletion(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	seedMaxComputePartitionCache(tm, testMaxComputeSlowTable, 500*time.Second)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	tm.prefetches.Store(cacheKey, &maxComputeTablePrefetch{
		cacheGeneration: metadata.LastChecked.UnixNano(),
		exists:          true,
		checkDuration:   200 * time.Millisecond,
		validUntil:      time.Now().Add(200 * time.Millisecond),
	})

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("hard-TTL EnsureTableExists() error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 0 {
		t.Fatalf("CheckTableExists calls = %d, want completed prefetch reused", got)
	}
	if got := fake.executeCalls.Load(); got != 0 {
		t.Fatalf("ExecuteStatement calls = %d, want counterfactually fresh partition skipped", got)
	}
	currentTable, _ := tm.getRawCachedMetadata(cacheKey)
	if currentTable == metadata || !currentTable.LastChecked.After(time.Now()) {
		t.Fatalf("table freshness = %#v, want counterfactual DESC completion timestamp", currentTable)
	}
}

func TestMaxComputeSlowQueryHardTTLJoinsActivePrefetch(t *testing.T) {
	checkStarted := make(chan struct{})
	releaseCheck := make(chan struct{})
	var startOnce sync.Once
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			startOnce.Do(func() { close(checkStarted) })
			<-releaseCheck
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	prefetchDone := make(chan error, 1)
	go func() {
		_, err := tm.checkSlowQueryTableExists(
			context.Background(),
			cacheKey,
			testMaxComputeProject,
			testMaxComputeSlowTable,
			testMaxComputeRegion,
			metadata.LastChecked.UnixNano(),
			time.Now().Add(30*time.Second),
		)
		prefetchDone <- err
	}()
	receiveMaxComputeTestValue(t, checkStarted)

	hardDone := make(chan error, 1)
	go func() {
		hardDone <- tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion)
	}()
	select {
	case err := <-hardDone:
		t.Fatalf("hard-TTL call returned before active prefetch completed: %v", err)
	case <-time.After(25 * time.Millisecond):
	}
	close(releaseCheck)
	if err := receiveMaxComputeTestValue(t, hardDone); err != nil {
		t.Fatalf("hard-TTL EnsureTableExists() error = %v", err)
	}
	if err := receiveMaxComputeTestValue(t, prefetchDone); err != nil {
		t.Fatalf("active prefetch error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want hard caller to join prefetch", got)
	}
}

func TestMaxComputeSlowQueryHardTTLRetriesWhenActivePrefetchFails(t *testing.T) {
	checkStarted := make(chan struct{})
	releaseCheck := make(chan struct{})
	var attempts atomic.Int32
	var releaseOnce sync.Once
	release := func() {
		releaseOnce.Do(func() { close(releaseCheck) })
	}
	defer release()

	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			if attempts.Add(1) == 1 {
				close(checkStarted)
				<-releaseCheck
				return false, errors.New("temporary background DESC failure")
			}
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	prefetchDone := make(chan error, 1)
	go func() {
		_, err := tm.checkSlowQueryTableExists(
			context.Background(),
			cacheKey,
			testMaxComputeProject,
			testMaxComputeSlowTable,
			testMaxComputeRegion,
			metadata.LastChecked.UnixNano(),
			time.Now().Add(30*time.Second),
		)
		prefetchDone <- err
	}()
	receiveMaxComputeTestValue(t, checkStarted)

	hardDone := make(chan error, 1)
	go func() {
		hardDone <- tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion)
	}()
	select {
	case err := <-hardDone:
		t.Fatalf("hard-TTL call returned before active prefetch completed: %v", err)
	case <-time.After(25 * time.Millisecond):
	}
	release()
	if err := receiveMaxComputeTestValue(t, hardDone); err != nil {
		t.Fatalf("hard-TTL EnsureTableExists() error = %v, want foreground retry", err)
	}
	if err := receiveMaxComputeTestValue(t, prefetchDone); err == nil {
		t.Fatal("active prefetch error = nil, want synthetic background failure")
	}
	if got := fake.checkCalls.Load(); got != 2 {
		t.Fatalf("CheckTableExists calls = %d, want failed active prefetch plus foreground retry", got)
	}
}

func TestMaxComputeSlowQueryHardTTLRechecksActiveNegativePrefetch(t *testing.T) {
	checkStarted := make(chan struct{})
	releaseCheck := make(chan struct{})
	var attempts atomic.Int32
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			if attempts.Add(1) == 1 {
				close(checkStarted)
				<-releaseCheck
				return false, nil
			}
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	prefetchDone := make(chan error, 1)
	go func() {
		_, err := tm.checkSlowQueryTableExists(
			context.Background(),
			cacheKey,
			testMaxComputeProject,
			testMaxComputeSlowTable,
			testMaxComputeRegion,
			metadata.LastChecked.UnixNano(),
			time.Now().Add(30*time.Second),
		)
		prefetchDone <- err
	}()
	receiveMaxComputeTestValue(t, checkStarted)

	hardDone := make(chan error, 1)
	go func() {
		hardDone <- tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion)
	}()
	select {
	case err := <-hardDone:
		t.Fatalf("hard-TTL call returned before negative prefetch completed: %v", err)
	case <-time.After(25 * time.Millisecond):
	}
	close(releaseCheck)
	if err := receiveMaxComputeTestValue(t, hardDone); err != nil {
		t.Fatalf("hard-TTL EnsureTableExists() error = %v", err)
	}
	if err := receiveMaxComputeTestValue(t, prefetchDone); err != nil {
		t.Fatalf("negative prefetch error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 2 {
		t.Fatalf("CheckTableExists calls = %d, want negative prefetch plus synchronous confirmation", got)
	}
	if statements := fake.recordedStatements(); len(statements) != 1 || !strings.HasPrefix(statements[0], "MSCK REPAIR TABLE") {
		t.Fatalf("statements = %#v, want confirmed existing table without CREATE", statements)
	}
}

func TestMaxComputeSlowQueryFailedPrefetchRetriesAtHardTTL(t *testing.T) {
	var failCheck atomic.Bool
	failCheck.Store(true)
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			if failCheck.Load() {
				return false, errors.New("temporary DESC failure")
			}
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	tm.startSlowQueryTablePrefetch(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion, metadata.LastChecked)
	waitForMaxComputePrefetchToFinish(t, tm, testMaxComputeSlowTable)
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("failed prefetch check calls = %d, want 1", got)
	}

	failCheck.Store(false)
	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("hard-TTL retry error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 2 {
		t.Fatalf("CheckTableExists calls = %d, want failed prefetch plus hard retry", got)
	}
}

func TestMaxComputeSlowQueryStalePrefetchIsNotConsumed(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	tm.prefetches.Store(cacheKey, &maxComputeTablePrefetch{
		cacheGeneration: metadata.LastChecked.Add(-time.Second).UnixNano(),
		exists:          false,
		validUntil:      time.Now().Add(time.Minute),
	})

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("EnsureTableExists() error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want stale prefetch ignored", got)
	}
	if statements := fake.recordedStatements(); len(statements) != 1 || !strings.HasPrefix(statements[0], "MSCK REPAIR TABLE") {
		t.Fatalf("statements = %#v, stale exists=false result must not create table", statements)
	}
}

func TestMaxComputeSlowQueryExpiredPrefetchIsNotConsumed(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 2*time.Hour)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	tm.prefetches.Store(cacheKey, &maxComputeTablePrefetch{
		cacheGeneration: metadata.LastChecked.UnixNano(),
		exists:          false,
		validUntil:      time.Now().Add(-time.Second),
	})

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("EnsureTableExists() error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want hours-old prefetch ignored", got)
	}
	if statements := fake.recordedStatements(); len(statements) != 1 || !strings.HasPrefix(statements[0], "MSCK REPAIR TABLE") {
		t.Fatalf("statements = %#v, expired exists=false result must not create table", statements)
	}
	if _, ok := tm.prefetches.Load(cacheKey); ok {
		t.Fatal("expired same-generation prefetch remained cached")
	}
}

func TestMaxComputeSlowQueryBackgroundDoesNotRetryExpiredResult(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)

	_, err := tm.checkSlowQueryTableExists(
		context.Background(),
		cacheKey,
		testMaxComputeProject,
		testMaxComputeSlowTable,
		testMaxComputeRegion,
		metadata.LastChecked.UnixNano(),
		time.Now().Add(-time.Second),
	)
	if !errors.Is(err, errMaxComputeTablePrefetchExpired) {
		t.Fatalf("expired background prefetch error = %v, want sentinel", err)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want no background fallback DESC", got)
	}
	if _, ok := tm.prefetches.Load(cacheKey); ok {
		t.Fatal("expired background result remained cached")
	}
}

func TestMaxComputeSlowQueryDelayedHardArrivalUsesArrivalBasedPartitionCheck(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 601*time.Second)
	seedMaxComputePartitionCache(tm, testMaxComputeSlowTable, 600*time.Second-25*time.Millisecond)
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion)
	validUntil := time.Now().Add(20 * time.Millisecond)
	tm.prefetches.Store(cacheKey, &maxComputeTablePrefetch{
		cacheGeneration: metadata.LastChecked.UnixNano(),
		exists:          true,
		checkDuration:   30 * time.Millisecond,
		validUntil:      validUntil,
	})

	startedAt := time.Now()
	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("hard-TTL EnsureTableExists() error = %v", err)
	}
	if elapsed := time.Since(startedAt); elapsed < 20*time.Millisecond {
		t.Fatalf("hard-TTL call completed in %v, want arrival-based counterfactual wait", elapsed)
	}
	if got := fake.checkCalls.Load(); got != 0 {
		t.Fatalf("CheckTableExists calls = %d, want valid prefetch consumed", got)
	}
	if got := fake.executeCalls.Load(); got != 1 {
		t.Fatalf("ExecuteStatement calls = %d, want arrival-based MSCK", got)
	}
	current, _ := tm.getRawCachedMetadata(cacheKey)
	if current.LastChecked.After(validUntil) {
		t.Fatalf("table freshness = %v, want capped at prefetch validity %v", current.LastChecked, validUntil)
	}
}

func TestMaxComputeSlowQueryPrefetchPanicDoesNotEscapeBackground(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			panic("synthetic DESC panic")
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	metadata := seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 571*time.Second)

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("prefetch-window EnsureTableExists() error = %v", err)
	}
	waitForMaxComputePrefetchToFinish(t, tm, testMaxComputeSlowTable)
	current, _ := tm.getRawCachedMetadata(
		maxComputeTableCacheKey(testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeRegion),
	)
	if current != metadata {
		t.Fatalf("table cache changed after recovered background panic: got %#v, want %#v", current, metadata)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want one recovered background attempt", got)
	}
}

func TestMaxComputeSlowQueryPartitionFailureRemainsFailOpen(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{
		executeFn: func(context.Context, string, string) error {
			return errors.New("temporary MSCK failure")
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("EnsureTableExists() error = %v, want partition failure to fail open", err)
	}
	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("warm EnsureTableExists() error = %v", err)
	}
	if got := fake.executeCalls.Load(); got != 1 {
		t.Fatalf("ExecuteStatement calls = %d, want existing fail-open cache behavior", got)
	}
}

func TestMaxComputeNonSlowQueryKeepsSynchronousTTLPath(t *testing.T) {
	const table = "topsql_10109909546444159555"
	checkStarted := make(chan struct{})
	releaseCheck := make(chan struct{})
	var startOnce sync.Once
	fake := &fakeMaxComputeTableManagerEngine{
		checkFn: func(context.Context, string, string, string) (bool, error) {
			startOnce.Do(func() { close(checkStarted) })
			<-releaseCheck
			return true, nil
		},
	}
	tm := newTestMaxComputeTableManager(fake, 600)
	seedMaxComputeTableCache(tm, table, 601*time.Second)

	done := make(chan error, 1)
	go func() {
		done <- tm.EnsureTableExists(context.Background(), testMaxComputeProject, table, testMaxComputeDataSource, testMaxComputeRegion)
	}()
	receiveMaxComputeTestValue(t, checkStarted)
	select {
	case err := <-done:
		t.Fatalf("non-slow table returned before synchronous check completed: %v", err)
	case <-time.After(25 * time.Millisecond):
	}
	close(releaseCheck)
	if err := receiveMaxComputeTestValue(t, done); err != nil {
		t.Fatalf("EnsureTableExists() error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 1 {
		t.Fatalf("CheckTableExists calls = %d, want one synchronous legacy check", got)
	}
	if got := fake.executeCalls.Load(); got != 1 {
		t.Fatalf("ExecuteStatement calls = %d, want one synchronous legacy MSCK", got)
	}
	cacheKey := maxComputeTableCacheKey(testMaxComputeProject, table, testMaxComputeRegion)
	if _, ok := tm.prefetches.Load(cacheKey); ok {
		t.Fatal("non-slow table populated prefetch result state")
	}
	if _, ok := tm.prefetchAttempts.Load(cacheKey); ok {
		t.Fatal("non-slow table populated prefetch admission state")
	}
	if _, ok := tm.activePrefetches.Load(cacheKey); ok {
		t.Fatal("non-slow table populated active prefetch state")
	}
	if got := len(tm.prefetchSlot); got != 0 {
		t.Fatalf("prefetch slot usage = %d, want non-slow table to leave it untouched", got)
	}
}

func TestMaxComputeSlowQueryZeroTTLRemainsIndefinite(t *testing.T) {
	fake := &fakeMaxComputeTableManagerEngine{}
	tm := newTestMaxComputeTableManager(fake, 0)
	seedMaxComputeTableCache(tm, testMaxComputeSlowTable, 24*time.Hour)

	if err := tm.EnsureTableExists(context.Background(), testMaxComputeProject, testMaxComputeSlowTable, testMaxComputeDataSource, testMaxComputeRegion); err != nil {
		t.Fatalf("EnsureTableExists() error = %v", err)
	}
	if got := fake.checkCalls.Load(); got != 0 {
		t.Fatalf("CheckTableExists calls = %d, want zero for indefinite cache", got)
	}
}

func TestPhysicalMaxComputeSlowQueryTableScope(t *testing.T) {
	tests := []struct {
		table string
		want  bool
	}{
		{table: "slow_query_logs_101", want: true},
		{table: "SLOW_QUERY_LOGS_101", want: true},
		{table: "slow_query_logs"},
		{table: "slow_query_logs_backup"},
		{table: "slow_query_logs_view_101"},
		{table: "topsql_101"},
	}
	for _, tt := range tests {
		if got := isPhysicalMaxComputeSlowQueryTable(tt.table); got != tt.want {
			t.Errorf("isPhysicalMaxComputeSlowQueryTable(%q) = %v, want %v", tt.table, got, tt.want)
		}
	}
}