package engine

import (
	"context"
	"errors"
	"net/http"
	"net/url"
	"strings"
	"testing"
	"time"

	cloudauth "cloud.google.com/go/auth"
	"cloud.google.com/go/bigquery"
	"github.com/stretchr/testify/require"
	"google.golang.org/api/googleapi"
)

type fakeTopSQLSetupJob struct {
	id        string
	waitCalls int
}

func (j *fakeTopSQLSetupJob) ID() string { return j.id }

func (j *fakeTopSQLSetupJob) Wait(context.Context) (*bigquery.JobStatus, error) {
	j.waitCalls++
	return &bigquery.JobStatus{State: bigquery.Done}, nil
}

type fakeTopSQLSetupJobRunner struct {
	runJob         topSQLSetupJob
	runErr         error
	lookupJob      topSQLSetupJob
	runJobID       string
	runLocation    string
	lookupJobID    string
	lookupLocation string
}

func (r *fakeTopSQLSetupJobRunner) Run(_ context.Context, _, jobID, location string) (topSQLSetupJob, error) {
	r.runJobID = jobID
	r.runLocation = location
	return r.runJob, r.runErr
}

func (r *fakeTopSQLSetupJobRunner) Lookup(_ context.Context, jobID, location string) (topSQLSetupJob, error) {
	r.lookupJobID = jobID
	r.lookupLocation = location
	return r.lookupJob, nil
}

func TestDiscoverTopSQLExternalDefinitionIsolatesInstanceDeltaLogs(t *testing.T) {
	const root = "gs://bucket/root/component=tidb"
	lister := &fakeGCSObjectLister{prefixes: map[string][]string{
		"root/component=tidb/": {
			"root/component=tidb/instance=db-tidb-1.peer.svc:10080/",
			"root/component=tidb/_delta_log/",
			"root/component=tidb/instance=10.0.0.1:10080/",
			"root/component=tidb/instance=db-tidb-1.peer.svc:10080/",
		},
		"root/component=tidb/instance=10.0.0.1:10080/": {
			"root/component=tidb/instance=10.0.0.1:10080/_delta_log/",
			"root/component=tidb/instance=10.0.0.1:10080/date=2026-08-17/",
		},
		"root/component=tidb/instance=db-tidb-1.peer.svc:10080/": {
			"root/component=tidb/instance=db-tidb-1.peer.svc:10080/date=2026-08-17/",
		},
	}}
	engine := &BigQueryEngine{gcsObjectLister: lister}

	definition, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_tidb_42", root)
	require.NoError(t, err)
	require.Equal(t, []string{
		"gs://bucket/root/component=tidb/instance=10.0.0.1:10080/date=*",
		"gs://bucket/root/component=tidb/instance=db-tidb-1.peer.svc:10080/date=*",
	}, definition.sourceURIs)
	require.Equal(t, []string{"instance", "date"}, definition.partitionColumns)
	require.Equal(t, root, definition.partitionURIPrefix)
	require.Equal(t, 3, lister.prefixCalls)
	for _, uri := range definition.sourceURIs {
		require.Equal(t, 1, strings.Count(uri, "*"))
		require.NotContains(t, uri, "_delta_log")
	}
}

func TestDiscoverTopSQLExternalDefinitionSkipsInstancesWithoutDatePartitions(t *testing.T) {
	const root = "gs://bucket/root/component=tidb"
	engine := &BigQueryEngine{gcsObjectLister: &fakeGCSObjectLister{prefixes: map[string][]string{
		"root/component=tidb/": {
			"root/component=tidb/instance=empty/",
			"root/component=tidb/instance=ready/",
		},
		"root/component=tidb/instance=empty/": {
			"root/component=tidb/instance=empty/_delta_log/",
		},
		"root/component=tidb/instance=ready/": {
			"root/component=tidb/instance=ready/date=2026-08-17/",
		},
	}}}

	definition, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_tidb_42", root)
	require.NoError(t, err)
	require.Equal(t, []string{root + "/instance=ready/date=*"}, definition.sourceURIs)
	require.Equal(t, 1, definition.instanceCount)
	require.False(t, definition.empty)
}

func TestDiscoverTopSQLExternalDefinitionScopesInstancesAndURIsToMonth(t *testing.T) {
	const root = "gs://bucket/root/component=tidb"
	lister := &fakeGCSObjectLister{prefixes: map[string][]string{
		"root/component=tidb/": {
			"root/component=tidb/instance=july/",
			"root/component=tidb/instance=august/",
		},
		"root/component=tidb/instance=july/date=2026-07-": {
			"root/component=tidb/instance=july/date=2026-07-30/",
		},
		"root/component=tidb/instance=august/date=2026-07-": {},
	}}
	engine := &BigQueryEngine{gcsObjectLister: lister}

	definition, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_tidb_42_m202607", root)
	require.NoError(t, err)
	require.Equal(t, "202607", definition.month)
	require.Equal(t, []string{root + "/instance=july/date=2026-07-*"}, definition.sourceURIs)
	require.Equal(t, 1, definition.instanceCount)
	require.Equal(t, 3, lister.prefixCalls)
}

func TestDiscoverTopSQLMetaScopesURIToMonth(t *testing.T) {
	const root = "gs://bucket/root/component=topsql_sql_meta"
	engine := &BigQueryEngine{gcsObjectLister: &fakeGCSObjectLister{prefixes: map[string][]string{
		"root/component=topsql_sql_meta/date=2026-08-": {
			"root/component=topsql_sql_meta/date=2026-08-01/",
		},
	}}}

	definition, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_sql_meta_42_m202608", root)
	require.NoError(t, err)
	require.Equal(t, []string{root + "/date=2026-08-*"}, definition.sourceURIs)
}

func TestDiscoverTopSQLExternalDefinitionUsesEmptyInstanceSource(t *testing.T) {
	const root = "gs://bucket/root/component=tikv"
	engine := &BigQueryEngine{gcsObjectLister: &fakeGCSObjectLister{prefixes: map[string][]string{
		"root/component=tikv/": {},
	}}}

	definition, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_tikv_42", root)
	require.NoError(t, err)
	require.Equal(t, []string{root + "/instance=__clinic_no_instance__/date=*"}, definition.sourceURIs)
	require.True(t, definition.empty)
}

func TestDiscoverTopSQLExternalDefinitionRejectsUnsafeOrUnavailableSources(t *testing.T) {
	const root = "gs://bucket/root/component=tidb"
	tests := []struct {
		name     string
		prefixes []string
		err      error
		expected string
		source   string
		status   int
	}{
		{name: "unsafe instance", prefixes: []string{"root/component=tidb/instance=bad name/"}, expected: topSQLReasonInvalidInstanceName, source: bigQueryErrorSourceValidation},
		{name: "unexpected directory", prefixes: []string{"root/component=tidb/date=20260814/"}, expected: topSQLReasonInvalidInstanceLayout, source: bigQueryErrorSourceValidation},
		{name: "untyped list failure", err: errors.New("permission denied"), expected: topSQLReasonGCSListFailed, source: bigQueryErrorSourceGCS},
		{name: "API permission failure", err: &googleapi.Error{Code: 403, Errors: []googleapi.ErrorItem{{Reason: "forbidden"}}}, expected: "forbidden", source: bigQueryErrorSourceGoogleAPI, status: 403},
		{name: "ADC failure", err: &cloudauth.Error{Response: &http.Response{StatusCode: 401}}, expected: topSQLReasonCredentialError, source: bigQueryErrorSourceAuth, status: 401},
		{name: "network failure", err: &url.Error{Op: "GET", URL: "https://redacted.invalid", Err: context.DeadlineExceeded}, expected: topSQLReasonNetworkError, source: bigQueryErrorSourceNetwork},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			engine := &BigQueryEngine{gcsObjectLister: &fakeGCSObjectLister{
				prefixes:     map[string][]string{"root/component=tidb/": tt.prefixes},
				prefixErrors: map[string]error{"root/component=tidb/": tt.err},
			}}
			_, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_tidb_42", root)
			setupErr, ok := AsBigQueryTopSQLSetupError(err)
			require.True(t, ok)
			require.Equal(t, topSQLSetupOperationDiscover, setupErr.Operation)
			require.Equal(t, tt.expected, setupErr.Reason)
			require.Equal(t, tt.source, setupErr.ErrorSource)
			require.Equal(t, tt.status, setupErr.HTTPStatus)
			require.Zero(t, setupErr.SourceURICount)
		})
	}
}

func TestDiscoverTopSQLMetaClassifiesInvalidDateLayout(t *testing.T) {
	const root = "gs://bucket/root/component=topsql_sql_meta"
	engine := &BigQueryEngine{gcsObjectLister: &fakeGCSObjectLister{prefixes: map[string][]string{
		"root/component=topsql_sql_meta/": {"root/component=topsql_sql_meta/instance=unexpected/"},
	}}}

	_, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_sql_meta_42", root)
	setupErr, ok := AsBigQueryTopSQLSetupError(err)
	require.True(t, ok)
	require.Equal(t, topSQLReasonInvalidDateLayout, setupErr.Reason)
	require.Zero(t, setupErr.HTTPStatus)
}

func TestTopSQLMetaDefinitionAndDDLUseOnlyDatePartition(t *testing.T) {
	definition, 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=*"}, definition.sourceURIs)
	ddl := topSQLExternalTableDDL("project", "dataset", "topsql_sql_meta_42", definition)
	require.Contains(t, ddl, "WITH PARTITION COLUMNS (date STRING)")
	require.Contains(t, ddl, "format = 'PARQUET'")
	require.NotContains(t, ddl, "_delta_log")
	require.Equal(t, 1, strings.Count(ddl, "date=*"))
}

func TestDiscoverTopSQLMetaUsesReservedDateForEmptyComponent(t *testing.T) {
	const root = "gs://bucket/root/component=topsql_sql_meta"
	engine := &BigQueryEngine{gcsObjectLister: &fakeGCSObjectLister{prefixes: map[string][]string{
		"root/component=topsql_sql_meta/": {"root/component=topsql_sql_meta/_delta_log/"},
	}}}
	definition, err := engine.discoverTopSQLExternalDefinition(context.Background(), "topsql_sql_meta_42", root)
	require.NoError(t, err)
	require.Equal(t, []string{root + "/date=__clinic_no_date__/*"}, definition.sourceURIs)
	require.True(t, definition.empty)
}

func TestTopSQLEmptyViewProvidesSchemaWithoutReadingGCS(t *testing.T) {
	definition, err := topSQLExternalTableDefinition("topsql_sql_meta_42", "gs://bucket/root/component=topsql_sql_meta")
	require.NoError(t, err)
	query := topSQLEmptyViewQuery(definition)
	ddl := topSQLEmptyViewDDL("project", "dataset", "topsql_sql_meta_42", definition)

	require.Contains(t, query, "CAST(NULL AS STRING) AS `normalized_sql`")
	require.Contains(t, query, "CAST(NULL AS STRING) AS `sql_digest`")
	require.Contains(t, query, "CAST(NULL AS STRING) AS `date`")
	require.Contains(t, query, "WHERE FALSE")
	require.NotContains(t, query, "gs://")
	require.Equal(t, "CREATE VIEW `project.dataset.topsql_sql_meta_42` AS "+query, ddl)
	require.Equal(t, topSQLObjectStateManagedEmptyView, classifyTopSQLExternalObject(&bigquery.TableMetadata{
		Type:      bigquery.ViewTable,
		ViewQuery: query,
	}, definition))
}

func TestTopSQLViewSetupReplacesLegacyTableInSameJob(t *testing.T) {
	scope := bigQueryTopSQLViewScope{clusterID: "42"}
	query := topSQLViewSetupQuery("project", "dataset", "topsql_42", scope, true)
	require.Equal(t, 1, strings.Count(query, "DROP TABLE IF EXISTS `project.dataset.topsql_42`"))
	require.Equal(t, 1, strings.Count(query, "CREATE OR REPLACE VIEW `project.dataset.topsql_42`"))
	require.Less(t, strings.Index(query, "DROP TABLE IF EXISTS"), strings.Index(query, "CREATE OR REPLACE VIEW"))

	query = topSQLViewSetupQuery("project", "dataset", "topsql_42", scope, false)
	require.NotContains(t, query, "DROP TABLE")
}

func TestTopSQLScopedObjectNamesAndCrossMonthView(t *testing.T) {
	scope, ok := parseBigQueryTopSQLViewScope("topsql_42_m202607_202608")
	require.True(t, ok)
	require.Equal(t, "42", scope.clusterID)
	require.Equal(t, []string{"202607", "202608"}, scope.months)
	require.Equal(t, "topsql_tidb", topSQLTableKind("topsql_tidb_42_m202607"))
	require.Equal(t, "202607", topSQLTableMonth("topsql_tidb_42_m202607"))
	require.False(t, isBigQueryTopSQLView("topsql_42_m202613"))
	require.False(t, isBigQueryTopSQLPhysicalTable("topsql_tidb_42_m202607_202608"))

	ddl := topSQLViewDDL("project", "dataset", "topsql_42_m202607_202608", scope)
	require.Contains(t, ddl, "CREATE OR REPLACE VIEW `project.dataset.topsql_42_m202607_202608`")
	for _, dependency := range []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",
	} {
		require.Contains(t, ddl, dependency)
	}
	require.Equal(t, 3, strings.Count(ddl, " UNION ALL "))
}

func TestClassifyTopSQLExternalObjectAllowsOnlyManagedSources(t *testing.T) {
	definition, err := topSQLExternalTableDefinition("topsql_tidb_42", "gs://bucket/root/component=tidb")
	require.NoError(t, err)
	current := &bigquery.TableMetadata{Type: bigquery.ExternalTable, ExternalDataConfig: &bigquery.ExternalDataConfig{
		SourceFormat: bigquery.Parquet,
		SourceURIs: []string{
			"gs://bucket/root/component=tidb/instance=a/date=*",
			"gs://bucket/root/component=tidb/instance=b/date=*",
		},
	}}
	require.Equal(t, topSQLObjectStateCurrentParquet, classifyTopSQLExternalObject(current, definition))
	current.ExternalDataConfig.SourceURIs = append(current.ExternalDataConfig.SourceURIs, "gs://other/root/instance=c/date=*")
	require.Equal(t, topSQLObjectStateObjectConflict, classifyTopSQLExternalObject(current, definition))
}

func TestClassifyTopSQLExternalObjectRecognizesOnlyExpectedLegacyStates(t *testing.T) {
	definition, err := topSQLExternalTableDefinition("topsql_tidb_42", "gs://bucket/root/component=tidb")
	require.NoError(t, err)
	legacyDelta := &bigquery.TableMetadata{Type: bigquery.ExternalTable, ExternalDataConfig: &bigquery.ExternalDataConfig{
		SourceFormat: DeltaLakeFormat,
		SourceURIs:   []string{definition.root},
	}}
	require.Equal(t, topSQLObjectStateLegacyDelta, classifyTopSQLExternalObject(legacyDelta, definition))
	legacyBroad := &bigquery.TableMetadata{Type: bigquery.ExternalTable, ExternalDataConfig: &bigquery.ExternalDataConfig{
		SourceFormat: bigquery.Parquet,
		SourceURIs:   []string{definition.root + "/*"},
	}}
	require.Equal(t, topSQLObjectStateLegacyBroadParquet, classifyTopSQLExternalObject(legacyBroad, definition))
}

func TestTopSQLSourceURILimitsAndFingerprint(t *testing.T) {
	definition, err := topSQLExternalTableDefinition("topsql_tidb_42", "gs://bucket/root/component=tidb")
	require.NoError(t, err)
	definition.sourceURIs = make([]string, topSQLMaxSourceURIs+1)
	for index := range definition.sourceURIs {
		definition.sourceURIs[index] = "gs://bucket/root/component=tidb/instance=server/date=*"
	}
	require.Error(t, validateTopSQLSourceURIs(definition))
	require.Equal(
		t,
		topSQLSourceFingerprint([]string{"gs://bucket/a/instance=a/date=*", "gs://bucket/a/instance=b/date=*"}),
		topSQLSourceFingerprint([]string{"gs://bucket/a/instance=b/date=*", "gs://bucket/a/instance=a/date=*"}),
	)
	require.NotEqual(
		t,
		topSQLSourceFingerprint([]string{"gs://bucket/a/instance=a/date=*"}),
		topSQLSourceFingerprint([]string{"gs://bucket/a/instance=b/date=*"}),
	)
}

func TestTopSQLSetupJobIDIsStableOnlyWithinReconciliationWindow(t *testing.T) {
	now := time.Date(2026, 8, 17, 8, 23, 0, 0, time.UTC)
	first := topSQLSetupJobID("CREATE VIEW redacted", "etag-a", now)
	repeated := topSQLSetupJobID("CREATE VIEW redacted", "etag-a", now.Add(time.Minute))
	reconciled := topSQLSetupJobID("CREATE VIEW redacted", "etag-b", now.Add(time.Minute))
	nextWindow := topSQLSetupJobID("CREATE VIEW redacted", "etag-a", now.Add(topSQLSetupJobWindow))

	require.Equal(t, first, repeated)
	require.NotEqual(t, first, reconciled)
	require.NotEqual(t, first, nextWindow)
	require.Regexp(t, `^clinic_topsql_[0-9]+_[a-f0-9]{24}$`, first)
	require.NotContains(t, first, "etag")
	require.NotContains(t, first, "CREATE")
}

func TestRunTopSQLSetupQueryJoinsExistingCrossProcessJob(t *testing.T) {
	job := &fakeTopSQLSetupJob{id: "clinic_topsql_existing"}
	runner := &fakeTopSQLSetupJobRunner{
		runErr:    &googleapi.Error{Code: http.StatusConflict},
		lookupJob: job,
	}
	engine := &BigQueryEngine{
		topSQLSetupJobs: runner,
		topSQLNow: func() time.Time {
			return time.Date(2026, 8, 17, 8, 23, 0, 0, time.UTC)
		},
	}

	queryID, err := engine.runTopSQLSetupQuery(context.Background(), "CREATE VIEW redacted", "asia-northeast1", "etag-a")
	require.NoError(t, err)
	require.Equal(t, job.id, queryID)
	require.Equal(t, runner.runJobID, runner.lookupJobID)
	require.Equal(t, "asia-northeast1", runner.runLocation)
	require.Equal(t, runner.runLocation, runner.lookupLocation)
	require.Equal(t, 1, job.waitCalls)
}