package query

import (
	"encoding/json"
	"fmt"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestValidateSelectSQL(t *testing.T) {
	assert.NoError(t, ValidateSelectSQL("SELECT 1"))
	assert.NoError(t, ValidateSelectSQL("SELECT a FROM t WHERE b = 1"))
	assert.NoError(t, ValidateSelectSQL("WITH x AS (SELECT 1) SELECT * FROM x"))
	assert.NoError(t, ValidateSelectSQL("SELECT date_add('day', -1, current_date)"))
	assert.Error(t, ValidateSelectSQL("DELETE FROM t"))
	assert.Error(t, ValidateSelectSQL("INSERT INTO t VALUES (1)"))
	assert.Error(t, RejectDangerousSQL("SELECT * FROM t; UNLOAD (SELECT 1) TO 's3://x'"))
}

func TestScopeValueJSON(t *testing.T) {
	var s Scope
	err := json.Unmarshal([]byte(`{
		"env":"prod",
		"components":["tidb","tikv"],
		"date":{"gte":"2026-07-17","lt":"2026-07-20"}
	}`), &s)
	require.NoError(t, err)
	v, ok := s.GetString("env")
	require.True(t, ok)
	assert.Equal(t, "prod", v)
	assert.Equal(t, []string{"tidb", "tikv"}, s["components"].in)
	require.NotNil(t, s["date"].rng)
	assert.Equal(t, "2026-07-17", s["date"].rng.Gte)
	assert.Equal(t, "2026-07-20", s["date"].rng.Lt)
}

func TestBuildEffectiveSQLLogsDataset(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT event_time, message FROM logs WHERE level = 'ERROR' ORDER BY event_time DESC LIMIT 100",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-west-2",
			Scope: Scope{
				"env":         ScopeEq("prod"),
				"deploy_type": ScopeEq("dedicated"),
				"date":        ScopeEq("2026-06-18"),
				"components":  ScopeIn("tidb", "tikv"),
			},
			TimeRange: TimeRangeUnix(100, 200),
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH logs AS ("))
	assert.Contains(t, sql, `"o11y-logs".general_logs_us_west_2`)
	assert.Contains(t, sql, `"env" = 'prod'`)
	assert.Contains(t, sql, `"deploytype" = 'dedicated'`)
	assert.Contains(t, sql, `"date" = '2026-06-18'`)
	assert.Contains(t, sql, `"component" IN ('tidb', 'tikv')`)
	assert.Contains(t, sql, `"timestamp" >= 100`)
	assert.Contains(t, sql, `"timestamp" < 200`)
	assert.NotContains(t, sql, "cluster =")
	assert.NotContains(t, sql, "tenant_id =")
	assert.Contains(t, sql, "SELECT event_time, message FROM logs WHERE level = 'ERROR'")
	assert.NotContains(t, sql, ") AS _user_q")
}

func TestBuildEffectiveSQLLogsDateRange(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT * FROM logs",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-west-2",
			Scope: Scope{
				"env":         ScopeEq("prod"),
				"deploy_type": ScopeEq("dedicated"),
				"date":        ScopeRng(ScopeRange{Gte: "2026-07-17", Lt: "2026-07-20"}),
				"components":  ScopeIn("tidb"),
			},
			TimeRange: TimeRangeUnix(1, 2),
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, `"date" >= '2026-07-17'`)
	assert.Contains(t, sql, `"date" < '2026-07-20'`)
}

func TestBuildEffectiveSQLLogsLegacyBeginEndDate(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT * FROM logs",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-west-2",
			Scope: Scope{
				"env":         ScopeEq("prod"),
				"deploy_type": ScopeEq("dedicated"),
				"begin_date":  ScopeEq("2026-07-17"),
				"end_date":    ScopeEq("2026-07-19"),
				"components":  ScopeIn("tidb"),
			},
			TimeRange: TimeRangeUnix(1, 2),
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, `"date" >= '2026-07-17'`)
	assert.Contains(t, sql, `"date" <= '2026-07-19'`)
}

func TestBuildEffectiveSQLLogStatsDataset(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT component, level, count, msg_digest FROM log_stats WHERE level = 'ERROR' LIMIT 100",
		RewriteOptions{
			Dataset: DatasetLogStats,
			Region:  "us-west-2",
			Scope: Scope{
				"date":    ScopeEq("2026-07-17"),
				"org":     ScopeEq("1372813089187041280"),
				"cluster": ScopeEq("10004313793950238553"),
			},
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH log_stats AS ("))
	assert.Contains(t, sql, `"o11y-logs".log_stats_us_west_2`)
	assert.Contains(t, sql, `"date" = '20260717'`)
	assert.Contains(t, sql, `"org" = '1372813089187041280'`)
	assert.Contains(t, sql, `"cluster" = '10004313793950238553'`)
	assert.NotContains(t, sql, "timestamp")
}

func TestBuildEffectiveSQLLogStatsTimeRangeHour(t *testing.T) {
	// 2026-07-17T02:00:00Z .. 2026-07-17T05:00:00Z → hour >= 2 AND hour < 5
	var tr TimeRange
	require.NoError(t, json.Unmarshal([]byte(`{
		"begin":"2026-07-17T02:00:00Z",
		"end":"2026-07-17T05:00:00Z"
	}`), &tr))
	sql, err := BuildEffectiveSQL(
		"SELECT hour, count FROM log_stats LIMIT 10",
		RewriteOptions{
			Dataset: DatasetLogStats,
			Region:  "us-west-2",
			Scope: Scope{
				"date": ScopeEq("2026-07-17"), "org": ScopeEq("o"), "cluster": ScopeEq("c"),
			},
			TimeRange: &tr,
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, `"hour" >= 2`)
	assert.Contains(t, sql, `"hour" < 5`)
}

func TestBuildEffectiveSQLLogStatsTimeRangeRejectsMultiDay(t *testing.T) {
	_, err := BuildEffectiveSQL(
		"SELECT * FROM log_stats",
		RewriteOptions{
			Dataset:   DatasetLogStats,
			Region:    "us-west-2",
			Scope:     Scope{"date": ScopeEq("20260717"), "org": ScopeEq("o"), "cluster": ScopeEq("c")},
			TimeRange: TimeRangeUnix(1784253600, 1784253600+36*3600), // > 1 day
		},
	)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "one UTC day")
}

func TestBuildEffectiveSQLLogsTimeRangeOptional(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT * FROM logs",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-west-2",
			Scope: Scope{
				"env": ScopeEq("prod"), "deploy_type": ScopeEq("dedicated"),
				"date": ScopeEq("2026-06-18"), "components": ScopeIn("tidb"),
			},
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, `"date" = '2026-06-18'`)
	assert.NotContains(t, sql, `"timestamp"`)
}

func TestBuildEffectiveSQLLogsTimeRangeDatetimeString(t *testing.T) {
	var tr TimeRange
	require.NoError(t, json.Unmarshal([]byte(`{
		"begin":"2026-07-17T02:00:00Z",
		"end":"2026-07-17T03:00:00Z"
	}`), &tr))
	sql, err := BuildEffectiveSQL(
		"SELECT * FROM logs",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-west-2",
			Scope: Scope{
				"env": ScopeEq("prod"), "deploy_type": ScopeEq("dedicated"),
				"date": ScopeEq("2026-07-17"), "components": ScopeIn("tidb"),
			},
			TimeRange: &tr,
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, `"timestamp" >= 1784253600`)
	assert.Contains(t, sql, `"timestamp" < 1784257200`)
}

func TestBuildEffectiveSQLSQLStatementTimeRangeOverlap(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT digest FROM sqlstatement LIMIT 1",
		RewriteOptions{
			Dataset: DatasetSQLStatement,
			Region:  "us-west-2",
			Scope: Scope{
				"org": ScopeEq("o1"), "cluster": ScopeEq("c1"), "date": ScopeEq("2026-07-17"),
			},
			TimeRange: TimeRangeUnix(100, 200),
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, `"begin" < 200`)
	assert.Contains(t, sql, `"end" >= 100`)
}

func TestBuildEffectiveSQLLogStatsMissingScopeFields(t *testing.T) {
	_, err := BuildEffectiveSQL(
		"SELECT * FROM log-stats WHERE level = 'ERROR'",
		RewriteOptions{
			Dataset: DatasetLogStats,
			Region:  "us-west-2",
			Scope: Scope{
				"date":    ScopeEq("2026-07-17"),
				"org":     ScopeEq(""),
				"cluster": ScopeEq(""),
			},
		},
	)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "missing required scope fields")
	assert.Contains(t, err.Error(), "org")
	assert.Contains(t, err.Error(), "cluster")
}

func TestBuildEffectiveSQLLogStatsWrongVirtualTableHint(t *testing.T) {
	_, err := BuildEffectiveSQL(
		"SELECT * FROM log-stats WHERE level = 'ERROR'",
		RewriteOptions{
			Dataset: DatasetLogStats,
			Region:  "us-west-2",
			Scope: Scope{
				"date":    ScopeEq("2026-07-17"),
				"org":     ScopeEq("o1"),
				"cluster": ScopeEq("c1"),
			},
		},
	)
	require.Error(t, err)
	assert.Contains(t, err.Error(), `virtual table "log_stats"`)
	assert.Contains(t, err.Error(), `dataset id`)
	assert.Contains(t, err.Error(), "log_stats")
}

func TestBuildEffectiveSQLMetricsStatusSourcesDataset(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		`SELECT biztype, tenantID, totalSeries, totalLabelValuePairs
FROM metrics_status_sources
ORDER BY totalSeries DESC
LIMIT 100`,
		RewriteOptions{
			Dataset: DatasetMetricsStatusSources,
			Region:  "us-east-1-serverless",
			Scope: Scope{
				"biztype":       ScopeEq("prod-devtier"),
				"applicationID": ScopeEq("10196284766413416804"),
				"date":          ScopeEq("2026-08-19"),
			},
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH metrics_status_sources AS ("))
	assert.Contains(t, sql, `"o11y-logs".metrics_status_sources_us_east_1_serverless`)
	assert.Contains(t, sql, `"biztype" = 'prod-devtier'`)
	assert.Contains(t, sql, `"applicationID" = '10196284766413416804'`)
	assert.Contains(t, sql, `"date" = '20260819'`)
}

func TestBuildEffectiveSQLTopsqlDataset(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT sql_digest, SUM(topsql_cpu_time_ms) AS cpu FROM topsql GROUP BY 1 ORDER BY cpu DESC LIMIT 20",
		RewriteOptions{
			Dataset: DatasetTopsql,
			Region:  "us-west-2",
			Scope: Scope{
				"org":       ScopeEq("1372813089187041280"),
				"cluster":   ScopeEq("10004313793950238553"),
				"component": ScopeEq("tidb"),
				"instance":  ScopeEq("10.0.0.1:10080"),
				"date":      ScopeEq("2026-07-17"),
			},
			TimeRange: TimeRangeUnix(100, 200),
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH topsql AS ("))
	assert.Contains(t, sql, `"o11y-logs".topsql_us_west_2`)
	assert.Contains(t, sql, `"org" = '1372813089187041280'`)
	assert.Contains(t, sql, `"cluster" = '10004313793950238553'`)
	assert.Contains(t, sql, `"component" = 'tidb'`)
	assert.Contains(t, sql, `"instance" = '10.0.0.1:10080'`)
	assert.Contains(t, sql, `"date" = '2026-07-17'`)
	assert.Contains(t, sql, `"timestamps" >= 100`)
	assert.Contains(t, sql, `"timestamps" < 200`)
}

func TestBuildEffectiveSQLTopsqlSQLMeta(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT sql_digest, normalized_sql FROM topsql_sql_meta WHERE sql_digest IN ('abc')",
		RewriteOptions{
			Dataset: DatasetTopsqlSQLMeta,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("o1"),
				"cluster": ScopeEq("c1"),
				"date":    ScopeEq("2026-07-17"),
			},
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH topsql_sql_meta AS ("))
	assert.Contains(t, sql, `"o11y-logs".topsql_sql_meta_us_west_2`)
	assert.Contains(t, sql, `"date" = '2026-07-17'`)
}

func TestBuildEffectiveSQLTopsqlPlanMeta(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT plan_digest, normalized_plan FROM topsql_plan_meta WHERE plan_digest IN ('p1')",
		RewriteOptions{
			Dataset: DatasetTopsqlPlanMeta,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("o1"),
				"cluster": ScopeEq("c1"),
			},
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH topsql_plan_meta AS ("))
	assert.Contains(t, sql, `"o11y-logs".topsql_plan_meta_us_west_2`)
}

func TestBuildEffectiveSQLTopru(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT sql_digest, SUM(topru_total_ru) AS ru FROM topru GROUP BY 1 ORDER BY ru DESC LIMIT 100",
		RewriteOptions{
			Dataset: DatasetTopru,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("o1"),
				"cluster": ScopeEq("c1"),
				"date":    ScopeEq("2026-07-17"),
			},
			TimeRange: TimeRangeUnix(100, 200),
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH topru AS ("))
	assert.Contains(t, sql, `"o11y-logs".topru_us_west_2`)
	assert.Contains(t, sql, `"timestamps" >= 100`)
}

func TestBuildEffectiveSQLSQLStatementDataset(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT digest, exec_count FROM sqlstatement WHERE \"begin\" <= 200 AND \"end\" >= 100 LIMIT 50",
		RewriteOptions{
			Dataset: DatasetSQLStatement,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("o1"),
				"cluster": ScopeEq("c1"),
				"date":    ScopeEq("2026-07-17"),
			},
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH sqlstatement AS ("))
	assert.Contains(t, sql, `"o11y-logs".sqlstatement_us_west_2`)
	assert.Contains(t, sql, `"org" = 'o1'`)
	assert.Contains(t, sql, `"date" = '2026-07-17'`)
	assert.NotContains(t, sql, "timestamps")
}

func TestBuildEffectiveSQLSlowlogsDataset(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT digest, query_time, query FROM slowlogs WHERE query_time > 1 ORDER BY query_time DESC LIMIT 100",
		RewriteOptions{
			Dataset: DatasetSlowlogs,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("1372813089188151296"),
				"cluster": ScopeEq("019b23f6-ffc9-731d-be49-eb8f9e651855"),
				"date":    ScopeEq("2026-07-27"),
			},
			TimeRange: TimeRangeUnix(100, 200),
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH slowlogs AS ("))
	assert.Contains(t, sql, `"o11y-logs".slowlogs_us_west_2`)
	assert.Contains(t, sql, `"org" = '1372813089188151296'`)
	assert.Contains(t, sql, `"cluster" = '019b23f6-ffc9-731d-be49-eb8f9e651855'`)
	assert.Contains(t, sql, `"date" = '20260727'`)
	assert.Contains(t, sql, `"time" >= 100`)
	assert.Contains(t, sql, `"time" < 200`)
}

func TestBuildEffectiveSQLSlowlogsClusterIN(t *testing.T) {
	u1 := "019b23f6-ffc9-731d-be49-eb8f9e651855"
	u2 := "34750647-d0ba-4e59-9ffe-cd719c1c6af6"
	sql, err := BuildEffectiveSQL(
		"SELECT digest FROM slowlogs LIMIT 10",
		RewriteOptions{
			Dataset: DatasetSlowlogs,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("1372813089188151296"),
				"cluster": ScopeIn(u1, u2),
				"date":    ScopeEq("20260727"),
			},
			TimeRange: TimeRangeUnix(100, 200),
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, fmt.Sprintf(`"cluster" IN ('%s', '%s')`, u1, u2))
	assert.Contains(t, sql, `"date" = '20260727'`)
}

func TestBuildEffectiveSQLSlowlogsRequiresDate(t *testing.T) {
	_, err := BuildEffectiveSQL(
		"SELECT digest FROM slowlogs LIMIT 1",
		RewriteOptions{
			Dataset: DatasetSlowlogs,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("o1"),
				"cluster": ScopeEq("c1"),
			},
		},
	)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "date")
}

func TestBuildEffectiveSQLSlowlogsDateNormalize(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT digest FROM slowlogs LIMIT 1",
		RewriteOptions{
			Dataset: DatasetSlowlogs,
			Region:  "us-west-2",
			Scope: Scope{
				"org":     ScopeEq("o1"),
				"cluster": ScopeEq("c1"),
				"date":    ScopeEq("2026-07-17"),
			},
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, `"date" = '20260717'`)
}

func TestBuildEffectiveSQLSlowlogAlias(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"SELECT digest FROM slowlogs LIMIT 1",
		RewriteOptions{
			Dataset: "slowlog",
			Region:  "us-west-2",
			Scope: Scope{
				"org": ScopeEq("o1"), "cluster": ScopeEq("c1"), "date": ScopeEq("20260717"),
			},
			TimeRange: TimeRangeUnix(1, 2),
		},
	)
	require.NoError(t, err)
	assert.Contains(t, sql, "WITH slowlogs AS (")
}

func TestBuildEffectiveSQLLogsPrependsExistingWith(t *testing.T) {
	sql, err := BuildEffectiveSQL(
		"WITH ranked AS (SELECT * FROM logs) SELECT * FROM ranked",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-east-1",
			Scope: Scope{
				"env":         ScopeEq("prod"),
				"deploy_type": ScopeEq("serverless"),
				"date":        ScopeEq("2026-07-16"),
				"components":  ScopeIn("tidb"),
			},
			TimeRange: TimeRangeUnix(1, 2),
		},
	)
	require.NoError(t, err)
	assert.True(t, stringsHasPrefixFold(sql, "WITH logs AS ("))
	assert.Contains(t, sql, "),\nranked AS (SELECT * FROM logs)")
}

func TestBuildEffectiveSQLRawPassthrough(t *testing.T) {
	in := "SELECT DISTINCT serverless_cluster_id FROM sqlstatement_org_x_all WHERE \"date\" >= '2026-07-15'"
	sql, err := BuildEffectiveSQL(in, RewriteOptions{Dataset: DatasetRaw})
	require.NoError(t, err)
	assert.Equal(t, in, sql)
}

func TestBuildEffectiveSQLRawRejectsScope(t *testing.T) {
	_, err := BuildEffectiveSQL("SELECT 1", RewriteOptions{
		Dataset: DatasetRaw,
		Scope:   Scope{"env": ScopeEq("prod")},
	})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "scope is not allowed")
}

func TestBuildEffectiveSQLLogsRequiresAlias(t *testing.T) {
	_, err := BuildEffectiveSQL(
		"SELECT * FROM \"o11y-logs\".general_logs_us_west_2",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-west-2",
			Scope: Scope{
				"env": ScopeEq("prod"), "deploy_type": ScopeEq("dedicated"), "date": ScopeEq("2026-06-18"),
				"components": ScopeIn("tidb"),
			},
			TimeRange: TimeRangeUnix(1, 2),
		},
	)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "virtual table")
}

func TestBuildEffectiveSQLLogsRequiresComponents(t *testing.T) {
	_, err := BuildEffectiveSQL(
		"SELECT * FROM logs",
		RewriteOptions{
			Dataset: DatasetLogs,
			Region:  "us-west-2",
			Scope: Scope{
				"env": ScopeEq("prod"), "deploy_type": ScopeEq("dedicated"), "date": ScopeEq("2026-06-18"),
				"components": ScopeIn("*"),
			},
			TimeRange: TimeRangeUnix(1, 2),
		},
	)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "component")
}

func TestNewEngineNotReady(t *testing.T) {
	eng, err := NewEngine(Config{Vendor: "gcp", Region: "us-central1"})
	require.NoError(t, err)
	assert.Equal(t, "bigquery", eng.Name())
	_, err = eng.Submit(t.Context(), &EngineSubmitRequest{SQL: "SELECT 1"})
	var nr *NotReadyError
	require.ErrorAs(t, err, &nr)
	assert.Equal(t, "gcp", nr.Provider)
}

func stringsHasPrefixFold(s, prefix string) bool {
	if len(s) < len(prefix) {
		return false
	}
	return equalFoldASCII(s[:len(prefix)], prefix)
}

func equalFoldASCII(a, b string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := 0; i < len(a); i++ {
		ca, cb := a[i], b[i]
		if ca >= 'A' && ca <= 'Z' {
			ca += 'a' - 'A'
		}
		if cb >= 'A' && cb <= 'Z' {
			cb += 'a' - 'A'
		}
		if ca != cb {
			return false
		}
	}
	return true
}