package engine

import (
	"strings"
	"testing"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

func TestExtractAzureDateValues(t *testing.T) {
	sql := `SELECT * FROM clinic_data_proxy.slow_query_logs WHERE "date" IN ('20260226', '20260227') OR t.date = '20260228'`
	got := ExtractAzureDateValues(sql)
	want := []string{"20260226", "20260227", "20260228"}
	if strings.Join(got, ",") != strings.Join(want, ",") {
		t.Fatalf("dates = %v, want %v", got, want)
	}
}

func TestBuildAzureSynapseSQLSlowQuery(t *testing.T) {
	ds, err := dto.ParseDataSource("https://o11yprodsoutheastasia.dfs.core.windows.net/o11y-prod-shared-southeastasia/deltalake/1372813089189381281/019c974a-66c4-7a02-abcb-f2d536b55747/slowlogs/")
	if err != nil {
		t.Fatal(err)
	}
	sql := `SELECT digest, count(*) AS cnt FROM clinic_data_proxy.slow_query_logs WHERE date IN ('20260226') GROUP BY digest LIMIT 3`

	got, err := BuildAzureSynapseSQL(sql, "1372813089189381281", ds)
	if err != nil {
		t.Fatalf("BuildAzureSynapseSQL() error = %v", err)
	}

	assertContains(t, got, "with az_slow_query_logs AS")
	assertContains(t, got, "OPENROWSET")
	assertContains(t, got, "FORMAT = 'DELTA'")
	assertContains(t, got, "BULK 'https://o11yprodsoutheastasia.dfs.core.windows.net/o11y-prod-shared-southeastasia/deltalake/1372813089189381281/019c974a-66c4-7a02-abcb-f2d536b55747/slowlogs/'")
	assertContains(t, got, `from "az_slow_query_logs"`)
	assertContains(t, got, "select top 3")
	if strings.Contains(got, "clinic_data_proxy") {
		t.Fatalf("planned SQL should not contain logical database: %s", got)
	}
}

func TestBuildAzureSynapseSQLStatements(t *testing.T) {
	ds, err := dto.ParseDataSource("https://o11yprodwestus3.dfs.core.windows.net/o11y-prod-shared-westus3/deltalake/1372813089189381281/019c974a-66c4-7a02-abcb-f2d536b55747/persisted_statements_summary/")
	if err != nil {
		t.Fatal(err)
	}
	sql := `SELECT digest, exec_count FROM clinic_data_proxy.statements WHERE date = '20260721' LIMIT 3`

	got, err := BuildAzureSynapseSQL(sql, "1372813089189381281", ds)
	if err != nil {
		t.Fatalf("BuildAzureSynapseSQL() error = %v", err)
	}

	assertContains(t, got, "with az_statements AS")
	assertContains(t, got, "FORMAT = 'DELTA'")
	assertContains(t, got, "BULK 'https://o11yprodwestus3.dfs.core.windows.net/o11y-prod-shared-westus3/deltalake/1372813089189381281/019c974a-66c4-7a02-abcb-f2d536b55747/persisted_statements_summary/'")
	assertContains(t, got, `from "az_statements"`)
	assertContains(t, got, "select top 3")
	if count := strings.Count(got, "OPENROWSET"); count != 1 {
		t.Fatalf("OPENROWSET count = %d, want 1:\n%s", count, got)
	}
	if strings.Contains(got, "slowlogs") || strings.Contains(got, "az_slow_query_logs") {
		t.Fatalf("statements plan should not include slow query data source:\n%s", got)
	}
}

func TestBuildAzureSynapseSQLStatementsRejectsMissingDateFilter(t *testing.T) {
	ds, err := dto.ParseDataSource("https://acct.dfs.core.windows.net/container/deltalake/persisted_statements_summary/")
	if err != nil {
		t.Fatal(err)
	}
	_, err = BuildAzureSynapseSQL(`SELECT count(*) FROM clinic_data_proxy.statements`, "123", ds)
	if err == nil || !strings.Contains(err.Error(), "missing Azure date partition filter") {
		t.Fatalf("error = %v, want missing date filter", err)
	}
}

func TestBuildAzureSynapseSQLRejectsMissingDateFilter(t *testing.T) {
	ds, err := dto.ParseDataSource("https://acct.dfs.core.windows.net/container/deltalake/slowlogs/")
	if err != nil {
		t.Fatal(err)
	}
	_, err = BuildAzureSynapseSQL(`SELECT count(*) FROM clinic_data_proxy.slow_query_logs`, "123", ds)
	if err == nil || !strings.Contains(err.Error(), "missing Azure date partition filter") {
		t.Fatalf("error = %v, want missing date filter", err)
	}
}

func TestBuildAzureSynapseSQLTopSQLView(t *testing.T) {
	ds, err := dto.ParseDataSource("https://acct.dfs.core.windows.net/container/deltalake/org/cluster/type=topsql/")
	if err != nil {
		t.Fatal(err)
	}
	sql := `SELECT sql_digest FROM clinic_data_proxy.topsql WHERE date IN ('2026-03-24') LIMIT 5`
	got, err := BuildAzureSynapseSQL(sql, "123", ds)
	if err != nil {
		t.Fatalf("BuildAzureSynapseSQL() error = %v", err)
	}
	assertContains(t, got, "az_topsql_tidb AS")
	assertContains(t, got, "az_topsql_tikv AS")
	assertContains(t, got, "az_topsql_sql_meta AS")
	assertContains(t, got, "az_topsql_plan_meta AS")
	assertContains(t, got, "az_topsql AS")
	assertContains(t, got, "/component=tidb/")
	assertContains(t, got, "/component=tikv/")
	assertContains(t, got, `from "az_topsql"`)
}

func TestResolveAzureDeltaRootURIShortABFSSRequiresRegion(t *testing.T) {
	info, err := dto.ParseDataSource("abfss://container/deltalake/table")
	if err != nil {
		t.Fatal(err)
	}
	_, err = ResolveAzureDeltaRootURI(info)
	if err == nil || !strings.Contains(err.Error(), "requires storage account") {
		t.Fatalf("error = %v, want missing storage account", err)
	}
}

func TestResolveAzureDeltaRootURIShortABFSSUsesRegionNamingRule(t *testing.T) {
	info, err := dto.ParseDataSource("abfss://o11y-prod-shared-japaneast/deltalake/1372813089209296719/019e1fd9-8917-75d3-a422-1be241ceda9c/slowlogs/")
	if err != nil {
		t.Fatal(err)
	}
	info.Region = "japaneast"

	got, err := ResolveAzureDeltaRootURI(info)
	if err != nil {
		t.Fatalf("ResolveAzureDeltaRootURI() error = %v", err)
	}
	want := "https://o11yprodjapaneast.dfs.core.windows.net/o11y-prod-shared-japaneast/deltalake/1372813089209296719/019e1fd9-8917-75d3-a422-1be241ceda9c/slowlogs/"
	if got != want {
		t.Fatalf("root = %q, want %q", got, want)
	}
}

func TestStandardTableSchemaSlowQuery(t *testing.T) {
	schema, ok := StandardTableSchema("slow_query_logs")
	if !ok {
		t.Fatal("slow_query_logs schema not found")
	}
	if len(schema.Columns) == 0 {
		t.Fatal("expected columns")
	}
	if len(schema.Partitions) != 1 || schema.Partitions[0].Name != "date" {
		t.Fatalf("partitions = %+v", schema.Partitions)
	}
}

func assertContains(t *testing.T, s, substr string) {
	t.Helper()
	if !strings.Contains(s, substr) {
		t.Fatalf("expected SQL to contain %q:\n%s", substr, s)
	}
}