package engine

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	"cloud.google.com/go/bigquery"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/stretchr/testify/require"
)

type fakeBigQueryTableManagerEngine struct {
	mu               sync.Mutex
	refreshCalls     int
	refreshErr       error
	refreshed        bool
	block            chan struct{}
	started          chan struct{}
	once             sync.Once
	topSQLTables     []string
	topSQLSources    []string
	topSQLViews      []string
	topSQLBlock      chan struct{}
	topSQLAllStarted chan struct{}
	topSQLStartOnce  sync.Once
	topSQLActive     int
	topSQLMaxActive  int
	tableExists      bool
}

func (f *fakeBigQueryTableManagerEngine) CheckTableExists(context.Context, string, string, string) (bool, error) {
	return f.tableExists, nil
}

func (f *fakeBigQueryTableManagerEngine) CreateExternalTable(context.Context, string, string, string) error {
	return nil
}

func (f *fakeBigQueryTableManagerEngine) EnsureTopSQLExternalTable(_ context.Context, _ string, table, source, _ string) error {
	f.mu.Lock()
	f.topSQLTables = append(f.topSQLTables, table)
	f.topSQLSources = append(f.topSQLSources, source)
	f.topSQLActive++
	if f.topSQLActive > f.topSQLMaxActive {
		f.topSQLMaxActive = f.topSQLActive
	}
	if f.topSQLAllStarted != nil && f.topSQLActive == 4 {
		f.topSQLStartOnce.Do(func() { close(f.topSQLAllStarted) })
	}
	topSQLBlock := f.topSQLBlock
	f.mu.Unlock()
	if topSQLBlock != nil {
		<-topSQLBlock
	}
	f.mu.Lock()
	f.topSQLActive--
	f.mu.Unlock()
	if f.block != nil {
		f.once.Do(func() { close(f.started) })
		<-f.block
	}
	return nil
}

func TestBigQueryTopSQLViewEnsuresDependenciesConcurrentlyBeforeView(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{
		topSQLBlock:      make(chan struct{}),
		topSQLAllStarted: make(chan struct{}),
	}
	tm := &BigQueryTableManager{engine: fake, config: &config.TableManagerConfig{Enabled: true, AutoCreate: true}}
	done := make(chan error, 1)
	go func() {
		done <- tm.EnsureTableExists(context.Background(), "dataset", "topsql_42", "gs://bucket/root", "asia-northeast1")
	}()

	select {
	case <-fake.topSQLAllStarted:
	case <-time.After(time.Second):
		t.Fatal("TopSQL dependencies did not start concurrently")
	}
	fake.mu.Lock()
	require.Equal(t, 4, fake.topSQLMaxActive)
	require.Empty(t, fake.topSQLViews)
	fake.mu.Unlock()

	close(fake.topSQLBlock)
	require.NoError(t, <-done)
	fake.mu.Lock()
	require.Equal(t, []string{"topsql_42"}, fake.topSQLViews)
	fake.mu.Unlock()
}

func (f *fakeBigQueryTableManagerEngine) EnsureTopSQLView(_ context.Context, _ string, table, _, _ string) error {
	f.mu.Lock()
	f.topSQLViews = append(f.topSQLViews, table)
	f.mu.Unlock()
	return nil
}

func (f *fakeBigQueryTableManagerEngine) EnsureDatasetExists(context.Context, string, string) error {
	return nil
}

func (f *fakeBigQueryTableManagerEngine) RefreshExternalTableSchema(context.Context, string, string, string) (bool, error) {
	f.mu.Lock()
	f.refreshCalls++
	f.mu.Unlock()
	if f.block != nil {
		f.once.Do(func() { close(f.started) })
		<-f.block
	}
	return f.refreshed, f.refreshErr
}

func newCachedBigQueryManager(fake *fakeBigQueryTableManagerEngine, refreshSeconds int) *BigQueryTableManager {
	tm := &BigQueryTableManager{
		engine: fake,
		config: &config.TableManagerConfig{
			Enabled:              true,
			AutoCreate:           true,
			SchemaRefreshSeconds: refreshSeconds,
		},
	}
	tm.setCachedMetadata("table:dataset.table", &TableMetadata{
		Exists:      true,
		LastChecked: time.Now(),
	})
	return tm
}

func TestBigQuerySchemaRefreshDisabled(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{refreshed: true}
	tm := newCachedBigQueryManager(fake, 0)
	require.NoError(t, tm.EnsureTableExists(
		context.Background(), "dataset", "table", "gs://bucket/root/", "us",
	))
	require.Zero(t, fake.refreshCalls)
}

func TestBigQuerySchemaRefreshRespectsInterval(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{refreshed: true}
	tm := newCachedBigQueryManager(fake, 300)

	require.NoError(t, tm.EnsureTableExists(
		context.Background(), "dataset", "table", "gs://bucket/root/", "us",
	))
	require.NoError(t, tm.EnsureTableExists(
		context.Background(), "dataset", "table", "gs://bucket/root/", "us",
	))
	require.Equal(t, 1, fake.refreshCalls)
}

func TestBigQuerySchemaRefreshSkippedCheckRespectsInterval(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{refreshed: false}
	tm := newCachedBigQueryManager(fake, 300)

	require.NoError(t, tm.EnsureTableExists(
		context.Background(), "dataset", "table", "gs://bucket/root/", "us",
	))
	require.NoError(t, tm.EnsureTableExists(
		context.Background(), "dataset", "table", "gs://bucket/root/", "us",
	))
	require.Equal(t, 1, fake.refreshCalls)
}

func TestBigQuerySchemaRefreshFailureIsFailOpenAndRetried(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{
		refreshErr: errors.New("permission denied"),
	}
	tm := newCachedBigQueryManager(fake, 300)

	require.NoError(t, tm.EnsureTableExists(
		context.Background(), "dataset", "table", "gs://bucket/root", "us",
	))
	require.NoError(t, tm.EnsureTableExists(
		context.Background(), "dataset", "table", "gs://bucket/root", "us",
	))
	require.Equal(t, 2, fake.refreshCalls)
}

func TestBigQuerySchemaRefreshCollapsesConcurrentCalls(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{
		refreshed: true,
		block:     make(chan struct{}),
		started:   make(chan struct{}),
	}
	tm := newCachedBigQueryManager(fake, 300)
	done := make(chan error, 2)
	run := func() {
		done <- tm.EnsureTableExists(
			context.Background(), "dataset", "table", "gs://bucket/root", "us",
		)
	}
	go run()
	<-fake.started
	go run()
	close(fake.block)
	require.NoError(t, <-done)
	require.NoError(t, <-done)
	require.Equal(t, 1, fake.refreshCalls)
}

func TestSameGCSDeltaRoot(t *testing.T) {
	require.True(t, sameGCSDeltaRoot([]string{"gs://bucket/root"}, "gs://bucket/root/"))
	require.False(t, sameGCSDeltaRoot([]string{"gs://bucket/other"}, "gs://bucket/root"))
	require.False(t, sameGCSDeltaRoot(
		[]string{"gs://bucket/root", "gs://bucket/other"},
		"gs://bucket/root",
	))
}

func TestExternalDataConfigForSchemaRefreshPreservesSettings(t *testing.T) {
	current := &bigquery.ExternalDataConfig{
		SourceFormat: DeltaLakeFormat,
		SourceURIs:   []string{"gs://bucket/root"},
		Schema:       bigquery.Schema{{Name: "old_field", Type: bigquery.StringFieldType}},
		ConnectionID: "project.us.connection",
	}
	update := externalDataConfigForSchemaRefresh(current)

	require.True(t, update.AutoDetect)
	require.Nil(t, update.Schema)
	require.Equal(t, current.SourceFormat, update.SourceFormat)
	require.Equal(t, current.SourceURIs, update.SourceURIs)
	require.Equal(t, current.ConnectionID, update.ConnectionID)
	require.NotSame(t, current, update)
}

func TestBigQueryTopSQLViewEnsuresDependencies(t *testing.T) {
	for _, dataSource := range []string{
		"gs://bucket/root",
		"gs://bucket/root/",
		"gs://bucket/root/component=tidb",
		"gs://bucket/root/component=tidb/",
	} {
		t.Run(dataSource, func(t *testing.T) {
			fake := &fakeBigQueryTableManagerEngine{}
			tm := &BigQueryTableManager{engine: fake, config: &config.TableManagerConfig{Enabled: true, AutoCreate: true}}
			require.NoError(t, tm.EnsureTableExists(context.Background(), "dataset", "topsql_42", dataSource, "us"))
			require.ElementsMatch(t, []string{"topsql_tidb_42", "topsql_tikv_42", "topsql_sql_meta_42", "topsql_plan_meta_42"}, fake.topSQLTables)
			require.ElementsMatch(t, []string{
				"gs://bucket/root/component=tidb/",
				"gs://bucket/root/component=tikv/",
				"gs://bucket/root/component=topsql_sql_meta/",
				"gs://bucket/root/component=topsql_plan_meta/",
			}, fake.topSQLSources)
			require.Equal(t, []string{"topsql_42"}, fake.topSQLViews)
		})
	}
}

func TestBigQueryScopedTopSQLViewEnsuresEachMonthDependency(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{}
	tm := &BigQueryTableManager{engine: fake, config: &config.TableManagerConfig{Enabled: true, AutoCreate: true}}
	require.NoError(t, tm.EnsureTableExists(context.Background(), "dataset", "topsql_42_m202607_202608", "gs://bucket/root", "us"))
	require.ElementsMatch(t, []string{
		"topsql_tidb_42_m202607", "topsql_tikv_42_m202607", "topsql_sql_meta_42_m202607", "topsql_plan_meta_42_m202607",
		"topsql_tidb_42_m202608", "topsql_tikv_42_m202608", "topsql_sql_meta_42_m202608", "topsql_plan_meta_42_m202608",
	}, fake.topSQLTables)
	require.Equal(t, []string{"topsql_42_m202607_202608"}, fake.topSQLViews)
}

func TestBigQueryScopedTopSQLDependenciesRefreshAtMostEveryThirtyMinutes(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{}
	tm := &BigQueryTableManager{engine: fake, config: &config.TableManagerConfig{Enabled: true, AutoCreate: true}}
	const table = "topsql_42_m202607"
	require.NoError(t, tm.EnsureTableExists(context.Background(), "dataset", table, "gs://bucket/root", "us"))
	require.NoError(t, tm.EnsureTableExists(context.Background(), "dataset", table, "gs://bucket/root", "us"))
	require.Len(t, fake.topSQLTables, 4)
	require.Equal(t, []string{table, table}, fake.topSQLViews)

	for _, dependency := range []string{"topsql_tidb_42_m202607", "topsql_tikv_42_m202607", "topsql_sql_meta_42_m202607", "topsql_plan_meta_42_m202607"} {
		tm.setCachedMetadata(topSQLScopedDependencyCacheKey("dataset", dependency), &TableMetadata{
			Exists:      true,
			LastChecked: time.Now().Add(-topSQLScopedSourceRefreshInterval),
		})
	}
	require.NoError(t, tm.EnsureTableExists(context.Background(), "dataset", table, "gs://bucket/root", "us"))
	require.Len(t, fake.topSQLTables, 8)
}

func TestTopSQLBaseDataSourceNormalizesComponentSuffix(t *testing.T) {
	for _, dataSource := range []string{
		"gs://bucket/root/component=tidb",
		"gs://bucket/root/component=tidb/",
		"gs://bucket/root/component=tikv",
		"gs://bucket/root/component=topsql_sql_meta/",
		"gs://bucket/root/component=topsql_plan_meta",
	} {
		require.Equal(t, "gs://bucket/root", topSQLBaseDataSource(dataSource))
	}
	require.Equal(t, "gs://bucket/root", topSQLBaseDataSource("gs://bucket/root/"))
}

func TestBigQueryTopSQLViewRevalidatesEachRequest(t *testing.T) {
	fake := &fakeBigQueryTableManagerEngine{}
	tm := &BigQueryTableManager{engine: fake, config: &config.TableManagerConfig{Enabled: true, AutoCreate: true}}
	require.NoError(t, tm.EnsureTableExists(context.Background(), "dataset", "topsql_42", "gs://bucket/root", "us"))
	require.NoError(t, tm.EnsureTableExists(context.Background(), "dataset", "topsql_42", "gs://bucket/root", "us"))
	require.Len(t, fake.topSQLTables, 8)
	require.Equal(t, []string{"topsql_42", "topsql_42"}, fake.topSQLViews)
}

func TestTopSQLExternalTableDefinition(t *testing.T) {
	metric, err := topSQLExternalTableDefinition("topsql_tidb_42", "gs://bucket/root/component=tidb/")
	require.NoError(t, err)
	require.Equal(t, []string{"gs://bucket/root/component=tidb/instance=__clinic_no_instance__/date=*"}, metric.sourceURIs)
	require.Equal(t, []string{"instance", "date"}, metric.partitionColumns)
	require.Equal(t, "gs://bucket/root/component=tidb", metric.partitionURIPrefix)

	meta, err := topSQLExternalTableDefinition("topsql_sql_meta_42", "gs://bucket/root/component=topsql_sql_meta/")
	require.NoError(t, err)
	require.Equal(t, []string{"gs://bucket/root/component=topsql_sql_meta/date=*"}, meta.sourceURIs)
	require.Equal(t, []string{"date"}, meta.partitionColumns)
	require.Len(t, meta.schema, 2)
}

func TestTopSQLViewDDLUsesCanonicalFields(t *testing.T) {
	ddl := topSQLViewDDL("project", "dataset", "topsql_42", bigQueryTopSQLViewScope{clusterID: "42"})
	require.Contains(t, ddl, "topsql_cpu_time_ms")
	require.Contains(t, ddl, "timestamps")
	require.NotContains(t, ddl, " AS time")
	require.NotContains(t, ddl, " AS cpu_time_ms")
}