package engine

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

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

func TestNormalizeInstanceEnumList(t *testing.T) {
	if got := normalizeInstanceEnumList([]string{"b", "a", "a", "c"}); got != "a,b,c" {
		t.Fatalf("got %q", got)
	}
	if got := normalizeInstanceEnumList([]string{"x"}); got != "x" {
		t.Fatalf("got %q", got)
	}
}

func TestNormalizeInstanceEnumFromCSV(t *testing.T) {
	if got := normalizeInstanceEnumFromCSV(" b , a "); got != "a,b" {
		t.Fatalf("got %q", got)
	}
}

func TestInstanceProjectionNeedsUpdate_ReplaceSemantics(t *testing.T) {
	// 与 Glue 一致：无需更新
	if instanceProjectionNeedsUpdate("a,b", []string{"a", "b"}) {
		t.Fatal("same set should not need update")
	}
	// S3 新增 instance
	if !instanceProjectionNeedsUpdate("a,b", []string{"a", "b", "c"}) {
		t.Fatal("new instance should need update")
	}
	// S3 已删除 instance（替换后变少）
	if !instanceProjectionNeedsUpdate("a,b,c", []string{"a", "c"}) {
		t.Fatal("removed instance should need update")
	}
	if !instanceProjectionNeedsUpdate("", nil) {
		t.Fatal("empty discover should repair an invalid empty enum")
	}
	if instanceProjectionNeedsUpdate(emptyInstanceProjectionValue, nil) {
		t.Fatal("empty marker should already represent an empty discovery")
	}
}

func projectionRepairConfig() *config.TableManagerConfig {
	return &config.TableManagerConfig{
		Enabled:    true,
		AutoCreate: true,
		TableTemplates: map[string]config.TableTemplate{
			"topsql_tidb":     {DDL: "CREATE TABLE {{.Table}} TBLPROPERTIES ('projection.instance.values'='{{.InstanceValues}}')"},
			"topsql_tikv":     {DDL: "CREATE TABLE {{.Table}} TBLPROPERTIES ('projection.instance.values'='{{.InstanceValues}}')"},
			"topsql_sql_meta": {DDL: "CREATE TABLE {{.Table}}"},
			"topsql_topru":    {DDL: "CREATE TABLE {{.Table}} TBLPROPERTIES ('projection.instance.values'='{{.InstanceValues}}')"},
			"topsql": {
				Type:      "view",
				DependsOn: []string{"topsql_tidb", "topsql_tikv", "topsql_sql_meta"},
				DDL:       "CREATE VIEW {{.Table}}",
			},
			"topru": {
				Type:      "view",
				DependsOn: []string{"topsql_topru", "topsql_sql_meta"},
				DDL:       "CREATE VIEW {{.Table}}",
			},
		},
	}
}

func TestInstanceProjectionTargetsCoverDirectTableAndTopRUView(t *testing.T) {
	tm := &TableManager{config: projectionRepairConfig()}

	direct := tm.instanceProjectionTargets(
		[]TableReference{{Database: "db", Table: "topsql_tidb_42"}},
		"s3://bucket/root/",
		"42",
	)
	require.Len(t, direct, 1)
	require.Equal(t, "topsql_tidb_42", direct[0].table)
	require.Contains(t, direct[0].dataSource, "component=tidb")

	topRU := tm.instanceProjectionTargets(
		[]TableReference{{Database: "db", Table: "topru_42"}},
		"s3://bucket/root/",
		"42",
	)
	require.Len(t, topRU, 1)
	require.Equal(t, "topsql_topru_42", topRU[0].table)
	require.Contains(t, topRU[0].dataSource, "component=topru")
}

func TestRepairInstanceProjectionExpandsViewAndUpdatesPhysicalTables(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{
		listPrefixes: []string{"root/instance=b/", "root/instance=a/", "root/instance=a/"},
		projections:  make(map[string]string),
	}
	tm := &TableManager{engine: fake, config: projectionRepairConfig()}

	repairs, err := tm.RepairInstanceProjection(
		context.Background(),
		[]TableReference{{Database: "db", Table: "topsql_42"}},
		"s3://bucket/root/",
		"us-west-2",
		"42",
	)
	require.NoError(t, err)
	require.Len(t, repairs, 2)
	require.Equal(t, []string{"a,b", "a,b"}, fake.updates)
	require.Equal(t, []string{"topsql_tidb", "topsql_tikv"}, []string{repairs[0].Table, repairs[1].Table})
	for _, repair := range repairs {
		require.Equal(t, 2, repair.InstanceCount)
		require.True(t, repair.Updated)
	}
}

func TestPeriodicInstanceProjectionSyncRepairsEmptyAndReplacesMarker(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{projections: make(map[string]string)}
	cfg := projectionRepairConfig()
	cfg.InstanceProjectionRefreshSeconds = 600
	tm := &TableManager{engine: fake, config: cfg}

	tm.maybeSyncInstanceProjection(
		context.Background(), "db", "topsql_tidb_42", "s3://bucket/root/", "us-west-2",
	)
	require.Equal(t, []string{emptyInstanceProjectionValue}, fake.updates)

	tm.instProjSync.Delete("instproj:db.topsql_tidb_42@us-west-2")
	fake.listPrefixes = []string{"root/instance=db-0/"}
	tm.maybeSyncInstanceProjection(
		context.Background(), "db", "topsql_tidb_42", "s3://bucket/root/", "us-west-2",
	)
	require.Equal(t, []string{emptyInstanceProjectionValue, "db-0"}, fake.updates)
}

func TestRepairInstanceProjectionUsesEmptyMarker(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{}
	tm := &TableManager{engine: fake, config: projectionRepairConfig()}

	repairs, err := tm.RepairInstanceProjection(
		context.Background(),
		[]TableReference{{Database: "db", Table: "topsql_tidb_42"}},
		"s3://bucket/root/",
		"us-west-2",
		"42",
	)
	require.NoError(t, err)
	require.Equal(t, []string{emptyInstanceProjectionValue}, fake.updates)
	require.Zero(t, repairs[0].InstanceCount)
}

func TestRepairInstanceProjectionDoesNotUpdateWhenDiscoveryFails(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{listErr: errors.New("access denied")}
	tm := &TableManager{engine: fake, config: projectionRepairConfig()}

	_, err := tm.RepairInstanceProjection(
		context.Background(),
		[]TableReference{{Database: "db", Table: "topsql_tidb_42"}},
		"s3://bucket/root/",
		"us-west-2",
		"42",
	)
	require.ErrorContains(t, err, "discover instances")
	require.Empty(t, fake.updates)
}

func TestCreateProjectionTableDoesNotRunDDLWhenDiscoveryFails(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{listErr: errors.New("access denied")}
	tm := &TableManager{engine: fake, config: projectionRepairConfig()}

	err := tm.createTable(context.Background(), "db", "topsql_tidb_42", "s3://bucket/root/", "us-west-2")
	require.ErrorContains(t, err, "discover instance projection values")
	require.Empty(t, fake.sql)
}

func TestCreateProjectionTableUsesEmptyMarker(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{}
	tm := &TableManager{engine: fake, config: projectionRepairConfig()}

	err := tm.createTable(context.Background(), "db", "topsql_tidb_42", "s3://bucket/root/", "us-west-2")
	require.NoError(t, err)
	require.Contains(t, fake.sql[0], emptyInstanceProjectionValue)
}

func TestRepairInstanceProjectionCollapsesConcurrentUpdates(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{listPrefixes: []string{"root/instance=a/"}}
	tm := &TableManager{engine: fake, config: projectionRepairConfig()}

	var wg sync.WaitGroup
	for range 5 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, _ = tm.RepairInstanceProjection(
				context.Background(),
				[]TableReference{{Database: "db", Table: "topsql_tidb_42"}},
				"s3://bucket/root/",
				"us-west-2",
				"42",
			)
		}()
	}
	wg.Wait()
	require.Len(t, fake.updates, 1)
}