package sqls

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestStatementsSummaryHandler_Validation(t *testing.T) {
	gin.SetMode(gin.TestMode)

	okHeaders := map[string]string{
		"X-Provider":    "aws",
		"X-Region":      "us-west-2",
		"X-Org-Id":      "tenant-1",
		"X-Project-Id":  "project-1",
		"X-Cluster-Id":  "cluster-1",
		"X-Deploy-Type": "dedicated",
	}

	mockAdapter := &mockDuckDBAdapter{
		queryStatementsSummaryFunc: func(ctx context.Context, params *StatementsSummaryParams) ([]StatementsSummaryItem, error) {
			return []StatementsSummaryItem{
				{
					SQLDigest:    "digest-1",
					SQLText:      "select 1",
					TotalRU:      12.3,
					ExecCount:    3,
					DurationNs:   1000,
					TimestampSec: []int64{1, 2},
					ValueByTime:  []float64{5, 7.3},
				},
			}, nil
		},
	}
	svc := &Service{SQLsDuckDBAdapter: mockAdapter, SQLsQuery: mockAdapter}
	r := gin.New()
	r.GET("/statements/summary", svc.StatementsSummaryHandler)

	tests := []struct {
		name           string
		query          string
		headers        map[string]string
		expectedStatus int
	}{
		{
			name:           "missing begin_time",
			query:          "end_time=1678886400",
			headers:        okHeaders,
			expectedStatus: http.StatusBadRequest,
		},
		{
			name:           "invalid order_by",
			query:          "begin_time=1&end_time=2&order_by=bad",
			headers:        okHeaders,
			expectedStatus: http.StatusBadRequest,
		},
		{
			name:           "invalid dimension",
			query:          "begin_time=1&end_time=2&dimension=bad",
			headers:        okHeaders,
			expectedStatus: http.StatusBadRequest,
		},
		{
			name:           "v1 allows step=120",
			query:          "begin_time=1&end_time=2&step=120",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
		{
			name:           "invalid top",
			query:          "begin_time=1&end_time=2&top=0",
			headers:        okHeaders,
			expectedStatus: http.StatusBadRequest,
		},
		{
			name:           "success envelope",
			query:          "begin_time=1&end_time=2&top=5&order_by=exec_count&step=300",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
		{
			name:           "v2 coerces step=60 to 300",
			query:          "begin_time=1&end_time=600&step=60&data_source=v2",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
		{
			name:           "v2 coerces step=450 to 300",
			query:          "begin_time=1&end_time=600&step=450&data_source=v2",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
		{
			name:           "v2 accepts step=600",
			query:          "begin_time=1&end_time=600&step=600&data_source=v2",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			req := httptest.NewRequest(http.MethodGet, "/statements/summary?"+tt.query, nil)
			for k, v := range tt.headers {
				req.Header.Set(k, v)
			}
			w := httptest.NewRecorder()
			r.ServeHTTP(w, req)
			require.Equal(t, tt.expectedStatus, w.Code, w.Body.String())
			if tt.expectedStatus == http.StatusOK {
				var env StatementsSummaryEnvelope
				require.NoError(t, json.Unmarshal(w.Body.Bytes(), &env))
				require.Equal(t, "ok", env.Status)
				require.Len(t, env.Data, 1)
				require.Equal(t, "digest-1", env.Data[0].SQLDigest)
			}
		})
	}
}

func TestNormalizeStatementsV2SummaryWindow(t *testing.T) {
	begin, end, step, err := normalizeStatementsV2SummaryWindow(1786343949, 1786344546, 0)
	require.NoError(t, err)
	require.Equal(t, 300, step)
	require.Equal(t, 1786343700, begin) // floor to 300
	require.Equal(t, 1786344600, end)   // ceil to 300

	_, _, step, err = normalizeStatementsV2SummaryWindow(100, 600, 60)
	require.NoError(t, err)
	require.Equal(t, 300, step)

	_, _, step, err = normalizeStatementsV2SummaryWindow(100, 600, 450)
	require.NoError(t, err)
	require.Equal(t, 300, step)

	begin, end, step, err = normalizeStatementsV2SummaryWindow(300, 900, 600)
	require.NoError(t, err)
	require.Equal(t, 600, step)
	require.Equal(t, 300, begin)
	require.Equal(t, 900, end)
}

func TestBindSchemasAndStmtTypes_BracketSuffix(t *testing.T) {
	gin.SetMode(gin.TestMode)
	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest(http.MethodGet, "/?schemas[]=db1&schemas[]=db2&stmt_types[]=SELECT&stmt_types[]=Update", nil)

	schemas := []string{}
	stmtTypes := []string{}
	bindSchemasAndStmtTypes(c, &schemas, &stmtTypes)
	require.Equal(t, []string{"db1", "db2"}, schemas)
	require.Equal(t, []string{"SELECT", "Update"}, stmtTypes)
}

func TestBindSchemasAndStmtTypes_MergesPlainAndBracketKeys(t *testing.T) {
	gin.SetMode(gin.TestMode)
	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest(http.MethodGet, "/?schemas=db1&schemas[]=db2", nil)

	schemas := []string{"db1"}
	stmtTypes := []string{}
	bindSchemasAndStmtTypes(c, &schemas, &stmtTypes)
	require.Equal(t, []string{"db1", "db2"}, schemas)
}

func TestBuildStatementsSummaryBaseFilter(t *testing.T) {
	f := buildStatementsSummaryBaseFilter(100, 200, []string{"a'b"}, []string{"SELECT"}, "like%", "")
	require.Contains(t, f, "to_timestamp('200')")
	require.Contains(t, f, "to_timestamp('100')")
	require.Contains(t, f, "coalesce(SCHEMA_NAME, '') in ('a''b')")
	require.Contains(t, f, "coalesce(stmt_type, '') in ('SELECT')")
	require.Contains(t, f, "QUERY_SAMPLE_TEXT like '%")
	require.Contains(t, f, "digest like '%")
}

func TestBuildStatementsSummaryBaseFilter_V2SchemaColumn(t *testing.T) {
	f := buildStatementsSummaryBaseFilter(100, 200, []string{"db1"}, nil, "", "v2")
	require.Contains(t, f, "\"begin\" <= 200 and \"end\" >= 100")
	require.Contains(t, f, "coalesce(schema_name, '') in ('db1')")
	require.NotContains(t, f, "SCHEMA_NAME in")
}

func TestParseTopStatsDimension(t *testing.T) {
	tests := []struct {
		in      string
		want    string
		wantErr bool
	}{
		{"", "digest", false},
		{"digest", "digest", false},
		{"user", "user", false},
		{"sqltypes", "sqltypes", false},
		{"db", "db", false},
		{"DB", "db", false},
		{"table", "table", false},
		{"TABLE", "table", false},
		{"bad", "", true},
	}
	for _, tt := range tests {
		got, err := parseTopStatsDimension(tt.in)
		if tt.wantErr {
			require.Error(t, err)
			continue
		}
		require.NoError(t, err)
		require.Equal(t, tt.want, got)
	}
}

func TestStatementsDBNameColumn(t *testing.T) {
	require.Equal(t, "SCHEMA_NAME", statementsDBNameColumn(""))
	require.Equal(t, "SCHEMA_NAME", statementsDBNameColumn("deltalake"))
	require.Equal(t, "schema_name", statementsDBNameColumn("v2"))
	require.Equal(t, "schema_name", statementsDBNameColumn("V2"))
}

func TestStatementsTableNamesColumn(t *testing.T) {
	require.Equal(t, "table_names", statementsTableNamesColumn(""))
	require.Equal(t, "table_names", statementsTableNamesColumn("v2"))
}

func TestStatementsSampleUserColumn(t *testing.T) {
	require.Equal(t, "sample_user", statementsSampleUserColumn(""))
	require.Equal(t, "sample_user", statementsSampleUserColumn("deltalake"))
	require.Equal(t, "auth_users", statementsSampleUserColumn("v2"))
	require.Equal(t, "auth_users", statementsSampleUserColumn("V2"))
}

func TestStatementsTopStatsFromClause_UserV2(t *testing.T) {
	got := statementsTopStatsFromClause("stmt_tbl", "user", "v2")
	require.Equal(t, "stmt_tbl", got)
	require.Equal(t, "stmt_tbl", statementsTopStatsFromClause("stmt_tbl", "user", ""))
	require.Equal(t, "stmt_tbl", statementsTopStatsFromClause("stmt_tbl", "digest", "v2"))
}

func TestBuildStatementsSummaryBaseFilter_IgnoreEmptyValues(t *testing.T) {
	f := buildStatementsSummaryBaseFilter(100, 200, []string{""}, []string{"", "   "}, "", "")
	require.NotContains(t, f, "SCHEMA_NAME in")
	require.NotContains(t, f, "stmt_type in")
}

func TestAppendStatementsSampleUserFilter(t *testing.T) {
	base := "\"begin\" <= 200 and \"end\" >= 100"
	f := appendStatementsSampleUserFilter(base, "", "v2")
	require.Equal(t, base, f)

	f = appendStatementsSampleUserFilter(base, "user'a", "v2")
	require.Contains(t, f, "list_element(coalesce(json_keys(auth_users), []), 1)")
	require.Contains(t, f, "= 'user''a'")
	require.NotContains(t, f, "list_contains")

	f = appendStatementsSampleUserFilter(base, "root", "")
	require.Contains(t, f, "sample_user = 'root'")
}

func TestStatementsTopStatsHandler_Validation(t *testing.T) {
	gin.SetMode(gin.TestMode)

	okHeaders := map[string]string{
		"X-Provider":    "aws",
		"X-Region":      "us-west-2",
		"X-Org-Id":      "tenant-1",
		"X-Project-Id":  "project-1",
		"X-Cluster-Id":  "cluster-1",
		"X-Deploy-Type": "dedicated",
	}

	mockAdapter := &mockDuckDBAdapter{
		queryStatementsTopStatsFunc: func(ctx context.Context, params *StatementsTopStatsParams) ([]StatementsTopStatsItem, error) {
			if params.Dimension == "db" {
				return []StatementsTopStatsItem{
					{
						Key:        "mydb",
						Dimension:  "db",
						OrderBy:    params.OrderBy,
						Value:      5,
						ExecCount:  5,
						SumLatency: 500,
						DB:         "mydb",
					},
				}, nil
			}
			if params.Dimension == "table" {
				return []StatementsTopStatsItem{
					{
						Key:        "mydb.t1",
						Dimension:  "table",
						OrderBy:    params.OrderBy,
						Value:      3,
						ExecCount:  3,
						SumLatency: 300,
						Table:      "mydb.t1",
					},
				}, nil
			}
			return []StatementsTopStatsItem{
				{
					Key:        "user_a",
					Dimension:  "user",
					OrderBy:    "exec_count",
					Value:      10,
					ExecCount:  10,
					SumLatency: 1000,
					User:       "user_a",
				},
			}, nil
		},
	}
	svc := &Service{SQLsDuckDBAdapter: mockAdapter, SQLsQuery: mockAdapter}
	r := gin.New()
	r.GET("/statements/topstats", svc.StatementsTopStatsHandler)

	tests := []struct {
		name           string
		query          string
		headers        map[string]string
		expectedStatus int
	}{
		{
			name:           "missing begin_time",
			query:          "end_time=1678886400",
			headers:        okHeaders,
			expectedStatus: http.StatusBadRequest,
		},
		{
			name:           "invalid dimension",
			query:          "begin_time=1&end_time=2&dimension=bad",
			headers:        okHeaders,
			expectedStatus: http.StatusBadRequest,
		},
		{
			name:           "invalid order_by",
			query:          "begin_time=1&end_time=2&order_by=bad",
			headers:        okHeaders,
			expectedStatus: http.StatusBadRequest,
		},
		{
			name:           "success",
			query:          "begin_time=1&end_time=2&dimension=user&top=5",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
		{
			name:           "success db dimension",
			query:          "begin_time=1&end_time=2&dimension=db&top=5",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
		{
			name:           "success table dimension",
			query:          "begin_time=1&end_time=2&dimension=table&top=5",
			headers:        okHeaders,
			expectedStatus: http.StatusOK,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			req := httptest.NewRequest(http.MethodGet, "/statements/topstats?"+tt.query, nil)
			for k, v := range tt.headers {
				req.Header.Set(k, v)
			}
			w := httptest.NewRecorder()
			r.ServeHTTP(w, req)
			require.Equal(t, tt.expectedStatus, w.Code, w.Body.String())
			if tt.expectedStatus == http.StatusOK {
				var env StatementsTopStatsEnvelope
				require.NoError(t, json.Unmarshal(w.Body.Bytes(), &env))
				require.Equal(t, "ok", env.Status)
				require.Len(t, env.Data, 1)
				if tt.name == "success db dimension" {
					require.Equal(t, "db", env.Data[0].Dimension)
					require.Equal(t, "mydb", env.Data[0].DB)
				}
				if tt.name == "success table dimension" {
					require.Equal(t, "table", env.Data[0].Dimension)
					require.Equal(t, "mydb.t1", env.Data[0].Table)
				}
			}
		})
	}
}

func TestStatementsSummaryHandler_DimensionUser(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var captured *StatementsSummaryParams
	mockAdapter := &mockDuckDBAdapter{
		queryStatementsSummaryFunc: func(ctx context.Context, params *StatementsSummaryParams) ([]StatementsSummaryItem, error) {
			captured = params
			return []StatementsSummaryItem{
				{
					Dimension:    "user",
					User:         "u1",
					TotalRU:      9,
					OrderMetric:  9,
					TimestampSec: []int64{100},
					ValueByTime:  []float64{9},
				},
			}, nil
		},
	}
	svc := &Service{SQLsDuckDBAdapter: mockAdapter, SQLsQuery: mockAdapter}
	r := gin.New()
	r.GET("/statements/summary", svc.StatementsSummaryHandler)

	req := httptest.NewRequest(http.MethodGet, "/statements/summary?begin_time=1&end_time=2&dimension=user&data_source=v2", nil)
	req.Header.Set("X-Provider", "alicloud")
	req.Header.Set("X-Region", "ap-southeast-1")
	req.Header.Set("X-Org-Id", "tenant-1")
	req.Header.Set("X-Project-Id", "project-1")
	req.Header.Set("X-Cluster-Id", "cluster-1")
	req.Header.Set("X-Deploy-Type", "dedicated")

	w := httptest.NewRecorder()
	r.ServeHTTP(w, req)
	require.Equal(t, http.StatusOK, w.Code)
	require.NotNil(t, captured)
	require.Equal(t, "user", captured.Dimension)
	require.Equal(t, "v2", captured.DataSource)
	require.Equal(t, int64(0), captured.StartUnix)
	require.Equal(t, int64(300), captured.EndUnix)
	require.Equal(t, 300, captured.Step)
	require.Contains(t, captured.Filter, "\"begin\" <= 300 and \"end\" >= 0")

	var env StatementsSummaryEnvelope
	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &env))
	require.Equal(t, "user", env.Data[0].Dimension)
	require.Equal(t, "u1", env.Data[0].User)
}

func TestBuildStatementsSummaryQuery_UserV2(t *testing.T) {
	params := &StatementsSummaryParams{
		Filter:     "\"begin\" <= 200 and \"end\" >= 100",
		Top:        5,
		OrderBy:    "sum_ru",
		Step:       60,
		Dimension:  "user",
		DataSource: "v2",
	}
	q, err := buildStatementsSummaryQuery("stmt_alias", "1=1 AND "+params.Filter, params)
	require.NoError(t, err)
	require.Contains(t, q, "list_element(coalesce(json_keys(stmt_alias.auth_users), []), 1)")
	require.Contains(t, q, "filtered AS")
	require.NotContains(t, q, "CROSS JOIN")
	require.NotContains(t, q, "unnest")
	require.Contains(t, q, "entity_key")
	require.Contains(t, q, "tagged AS")
	require.Contains(t, q, statementsSummaryOthersEntityKey)
	require.NotContains(t, q, "entity_key <> ''")
	require.NotContains(t, q, "top_digests")
}

func TestBuildStatementsSummaryQuery_DigestDefault(t *testing.T) {
	params := &StatementsSummaryParams{
		Filter:     "SUMMARY_BEGIN_TIME <= to_timestamp('200')",
		Top:        10,
		OrderBy:    "exec_count",
		Step:       120,
		DataSource: "",
	}
	q, err := buildStatementsSummaryQuery("stmt_alias", "1=1 AND "+params.Filter, params)
	require.NoError(t, err)
	require.Contains(t, q, "coalesce(stmt_alias.digest, '') AS entity_key")
	require.Contains(t, q, "top_entities AS")
	require.Contains(t, q, "tagged AS")
	require.Contains(t, q, statementsSummaryOthersEntityKey)
	require.NotContains(t, q, "entity_key <> ''")
	require.NotContains(t, q, "digest IS NOT NULL")
}

func TestBuildStatementsSummaryQuery_DbIncludesEmptyDimensionValue(t *testing.T) {
	params := &StatementsSummaryParams{
		Filter:     "\"begin\" <= 200 and \"end\" >= 100",
		Top:        5,
		OrderBy:    "exec_count",
		Step:       60,
		Dimension:  "db",
		DataSource: "v2",
	}
	q, err := buildStatementsSummaryQuery("stmt_alias", "1=1 AND "+params.Filter, params)
	require.NoError(t, err)
	require.Contains(t, q, "coalesce(stmt_alias.schema_name, '') AS entity_key")
	require.NotContains(t, q, "entity_key <> ''")
}

func TestAggregateStatementsSummaryRows_OthersAtEnd(t *testing.T) {
	now := time.Now().UTC()
	rows := []statementsSummaryBucketRow{
		{
			EntityKey:         strPtr("digest-a"),
			EntityLabel:       strPtr("SELECT 1"),
			TimeWindow:        &now,
			SeriesValue:       10,
			EntityOrderMetric: float64Ptr(10),
		},
		{
			EntityKey:         strPtr(statementsSummaryOthersEntityKey),
			EntityLabel:       strPtr("others"),
			TimeWindow:        &now,
			SeriesValue:       100,
			EntityOrderMetric: float64Ptr(100),
		},
		{
			EntityKey:         strPtr(""),
			EntityLabel:       strPtr(""),
			TimeWindow:        &now,
			SeriesValue:       5,
			EntityOrderMetric: float64Ptr(5),
		},
	}
	out := aggregateStatementsSummaryRows(rows, &StatementsSummaryParams{
		Dimension:  "digest",
		StartUnix:  0,
		EndUnix:    60,
		DataSource: sqlsDataSourceV2,
	})
	require.Len(t, out, 3)
	assert.Equal(t, "digest-a", out[0].SQLDigest)
	assert.Equal(t, statementsNAValue, out[1].SQLDigest)
	assert.False(t, statementsSummaryItemIsOthers(out[0]))
	assert.False(t, statementsSummaryItemIsOthers(out[1]))
	assert.True(t, statementsSummaryItemIsOthers(out[2]))
	assert.Equal(t, "others", out[2].SQLText)
	assert.True(t, out[2].IsOther)
}

func TestAggregateStatementsSummaryRows_EmptyBucketDistinctFromOthers(t *testing.T) {
	now := time.Now().UTC()
	rows := []statementsSummaryBucketRow{
		{
			EntityKey:         strPtr(""),
			EntityLabel:       strPtr(""),
			TimeWindow:        &now,
			SeriesValue:       50,
			EntityOrderMetric: float64Ptr(50),
		},
		{
			EntityKey:         strPtr(statementsSummaryOthersEntityKey),
			EntityLabel:       strPtr("others"),
			TimeWindow:        &now,
			SeriesValue:       30,
			EntityOrderMetric: float64Ptr(30),
		},
	}
	out := aggregateStatementsSummaryRows(rows, &StatementsSummaryParams{
		Dimension:  "digest",
		StartUnix:  0,
		EndUnix:    60,
		DataSource: sqlsDataSourceV2,
	})
	require.Len(t, out, 2)
	assert.Equal(t, statementsNAValue, out[0].SQLDigest)
	assert.False(t, out[0].IsOther)
	assert.Equal(t, statementsNAValue, out[0].SQLText)
	assert.True(t, out[1].IsOther)
	assert.Equal(t, "others", out[1].SQLText)
}

func strPtr(v string) *string       { return &v }
func float64Ptr(v float64) *float64 { return &v }