package engine

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

	"github.com/aws/aws-sdk-go-v2/service/athena/types"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/stretchr/testify/require"
)

type fakeAthenaTableManagerEngine struct {
	mu           sync.Mutex
	params       map[string]string
	location     string
	sql          []string
	deleted      []string
	operations   []string
	deleteErr    error
	block        chan struct{}
	started      chan struct{}
	once         sync.Once
	listPrefixes []string
	listErr      error
	projection   string
	projections  map[string]string
	updates      []string
}

func (f *fakeAthenaTableManagerEngine) CheckDatabaseExists(context.Context, string, string) (bool, error) {
	return true, nil
}

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

func (f *fakeAthenaTableManagerEngine) DeleteGlueTable(_ context.Context, database, table, _ string) error {
	f.mu.Lock()
	f.deleted = append(f.deleted, database+"."+table)
	f.operations = append(f.operations, "delete:"+database+"."+table)
	f.mu.Unlock()
	if f.deleteErr != nil {
		return f.deleteErr
	}
	if f.block != nil {
		f.once.Do(func() { close(f.started) })
		<-f.block
	}
	return nil
}

func (f *fakeAthenaTableManagerEngine) Execute(
	_ context.Context,
	sql string,
	_ *dto.DataSourceInfo,
) (*dto.DataProxyQueryResult, error) {
	f.mu.Lock()
	f.sql = append(f.sql, sql)
	f.operations = append(f.operations, "execute:"+sql)
	f.mu.Unlock()
	return &dto.DataProxyQueryResult{}, nil
}

func (f *fakeAthenaTableManagerEngine) ExecuteDDL(
	_ context.Context,
	sql string,
	_ *dto.DataSourceInfo,
) error {
	f.mu.Lock()
	f.sql = append(f.sql, sql)
	f.operations = append(f.operations, "executeDDL:"+sql)
	f.mu.Unlock()
	return nil
}

func (f *fakeAthenaTableManagerEngine) GetGlueProjectionInstanceValues(_ context.Context, _, table, _ string) (string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.projections != nil {
		return f.projections[table], nil
	}
	return f.projection, nil
}

func (f *fakeAthenaTableManagerEngine) GetGlueTableInfo(context.Context, string, string, string) (map[string]string, string, error) {
	return f.params, f.location, nil
}

func (f *fakeAthenaTableManagerEngine) GetGlueTableParameters(context.Context, string, string, string) (map[string]string, error) {
	return f.params, nil
}

func (f *fakeAthenaTableManagerEngine) ListS3Prefixes(context.Context, string, string, string) ([]string, error) {
	return f.listPrefixes, f.listErr
}

func (f *fakeAthenaTableManagerEngine) UpdateGlueProjectionInstanceValues(_ context.Context, _, table, _ string, values string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.updates = append(f.updates, values)
	if f.projections != nil {
		f.projections[table] = values
	} else {
		f.projection = values
	}
	return nil
}

func deltaRepairConfig() *config.TableManagerConfig {
	return &config.TableManagerConfig{
		Enabled:    true,
		AutoCreate: true,
		TableTemplates: map[string]config.TableTemplate{
			"slow_query_logs": {
				DDL: "CREATE EXTERNAL TABLE `{{.Database}}`.`{{.Table}}` LOCATION '{{.Location}}' TBLPROPERTIES ('table_type'='DELTA')",
			},
			"log_stats_global": {
				DDL: "CREATE EXTERNAL TABLE `{{.Database}}`.`{{.Table}}` LOCATION '{{.Location}}' TBLPROPERTIES ('table_type'='DELTA')",
			},
			"log_stats": {
				Type:      "view",
				DependsOn: []string{"log_stats_global"},
				DDL:       "CREATE VIEW `{{.Database}}`.`{{.Table}}` AS SELECT * FROM `{{.Database}}`.`{{.DepLogStatsGlobal}}`",
			},
			"topsql_tidb": {
				DDL: "CREATE EXTERNAL TABLE `{{.Database}}`.`{{.Table}}` (id BIGINT) STORED AS PARQUET",
			},
		},
	}
}

func TestIsInvalidGlueSchema(t *testing.T) {
	err := &AthenaQueryError{
		QueryID: "query-1",
		State:   types.QueryExecutionStateFailed,
		Reason:  "INVALID_GLUE_SCHEMA: Delta Lake table schema is stale",
	}
	require.True(t, IsInvalidGlueSchema(err))
	require.Equal(t, "query-1", AthenaQueryID(err))
	require.False(t, IsInvalidGlueSchema(errors.New("INVALID_GLUE_SCHEMA: text only")))
	require.False(t, IsInvalidGlueSchema(&AthenaQueryError{Reason: "GENERIC_INTERNAL_ERROR"}))
}

func TestIsInvalidInstanceProjection(t *testing.T) {
	err := &AthenaQueryError{
		QueryID: "query-1",
		State:   types.QueryExecutionStateFailed,
		Reason:  "INVALID_TABLE_PROPERTY: bad enum (property: projection.instance.values)",
	}
	require.True(t, IsInvalidInstanceProjection(err))
	require.False(t, IsInvalidInstanceProjection(errors.New(err.Error())))
	require.False(t, IsInvalidInstanceProjection(&AthenaQueryError{
		Reason: "INVALID_TABLE_PROPERTY: another property",
	}))
	require.False(t, IsInvalidInstanceProjection(&AthenaQueryError{
		Reason: "projection.instance.values is empty",
	}))
}

func TestDeltaRepairTargetsExpandManagedViews(t *testing.T) {
	tm := &TableManager{config: deltaRepairConfig()}
	targets := tm.deltaRepairTargets([]TableReference{
		{Database: "db", Table: "slow_query_logs_42"},
		{Database: "db", Table: "log_stats_42"},
		{Database: "db", Table: "topsql_tidb_42"},
		{Database: "db", Table: "unknown"},
	}, "s3://bucket/root/", "42")

	require.Equal(t, []deltaRepairTarget{
		{database: "db", table: "slow_query_logs_42", dataSource: "s3://bucket/root/"},
		{database: "db", table: "log_stats_global", dataSource: "s3://bucket/root/"},
	}, targets)
}

func TestRepairDeltaSchemaMismatchDeletesGlueCreatesAndCaches(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{
		params:   map[string]string{"table_type": "DELTA"},
		location: "s3://bucket/root",
	}
	tm := &TableManager{engine: fake, config: deltaRepairConfig()}

	repaired, err := tm.RepairDeltaSchemaMismatch(
		context.Background(),
		[]TableReference{{Database: "db", Table: "slow_query_logs_42"}},
		"s3://bucket/root/",
		"us-west-2",
		"42",
		time.Now(),
	)
	require.NoError(t, err)
	require.Equal(t, []string{"db.slow_query_logs_42"}, repaired)
	require.Equal(t, []string{"db.slow_query_logs_42"}, fake.deleted)
	require.Len(t, fake.sql, 1)
	require.True(t, strings.HasPrefix(fake.sql[0], "CREATE EXTERNAL TABLE `db`.`slow_query_logs_42`"))
	require.Len(t, fake.operations, 2)
	require.Equal(t, "delete:db.slow_query_logs_42", fake.operations[0])
	require.True(t, strings.HasPrefix(fake.operations[1], "executeDDL:CREATE EXTERNAL TABLE `db`.`slow_query_logs_42`"))

	meta, ok := tm.getCachedMetadata("table:db.slow_query_logs_42@us-west-2")
	require.True(t, ok)
	require.True(t, meta.Exists)
}

func TestRepairDeltaSchemaMismatchStopsWhenGlueDeleteFails(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{
		params:    map[string]string{"table_type": "DELTA"},
		location:  "s3://bucket/root",
		deleteErr: errors.New("access denied"),
	}
	tm := &TableManager{engine: fake, config: deltaRepairConfig()}

	_, err := tm.RepairDeltaSchemaMismatch(
		context.Background(),
		[]TableReference{{Database: "db", Table: "slow_query_logs_42"}},
		"s3://bucket/root",
		"us-west-2",
		"42",
		time.Now(),
	)
	require.ErrorContains(t, err, "delete stale Delta Glue table")
	require.Equal(t, []string{"db.slow_query_logs_42"}, fake.deleted)
	require.Empty(t, fake.sql)
}

func TestRepairDeltaSchemaMismatchRejectsUnsafeTargets(t *testing.T) {
	tests := []struct {
		name     string
		params   map[string]string
		location string
	}{
		{
			name:     "not delta",
			params:   map[string]string{"classification": "parquet"},
			location: "s3://bucket/root",
		},
		{
			name:     "location mismatch",
			params:   map[string]string{"table_type": "DELTA"},
			location: "s3://other/root",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			fake := &fakeAthenaTableManagerEngine{params: tt.params, location: tt.location}
			tm := &TableManager{engine: fake, config: deltaRepairConfig()}
			_, err := tm.RepairDeltaSchemaMismatch(
				context.Background(),
				[]TableReference{{Database: "db", Table: "slow_query_logs_42"}},
				"s3://bucket/root",
				"us-west-2",
				"42",
				time.Now(),
			)
			require.Error(t, err)
			require.Empty(t, fake.deleted)
			require.Empty(t, fake.sql)
		})
	}
}

func TestRepairDeltaSchemaMismatchCollapsesConcurrentRepairs(t *testing.T) {
	fake := &fakeAthenaTableManagerEngine{
		params:   map[string]string{"table_type": "DELTA"},
		location: "s3://bucket/root",
		block:    make(chan struct{}),
		started:  make(chan struct{}),
	}
	tm := &TableManager{engine: fake, config: deltaRepairConfig()}
	failedAt := time.Now()
	run := func(done chan<- error) {
		_, err := tm.RepairDeltaSchemaMismatch(
			context.Background(),
			[]TableReference{{Database: "db", Table: "slow_query_logs_42"}},
			"s3://bucket/root",
			"us-west-2",
			"42",
			failedAt,
		)
		done <- err
	}

	done := make(chan error, 2)
	go run(done)
	<-fake.started
	go run(done)
	close(fake.block)
	require.NoError(t, <-done)
	require.NoError(t, <-done)
	require.Equal(t, []string{"db.slow_query_logs_42"}, fake.deleted)
	require.Len(t, fake.sql, 1)
}