package sqls

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/utils/duckdb"
)

const statementsSummaryParquetFixture = "part-00001-389cd95a-41f3-4e49-93af-4a5f55a156f2-c000.snappy.parquet"

func statementsSummaryParquetPath(t *testing.T) string {
	t.Helper()
	candidates := []string{
		filepath.Join("testdata", statementsSummaryParquetFixture),
		filepath.Join("..", "..", "..", "..", "test", statementsSummaryParquetFixture),
	}
	for _, p := range candidates {
		if _, err := os.Stat(p); err == nil {
			abs, err := filepath.Abs(p)
			require.NoError(t, err)
			return abs
		}
	}
	t.Skipf("parquet fixture not found (tried %v)", candidates)
	return ""
}

func newStatementsSummaryParquetAdapter(t *testing.T) (*DuckDBAdapter, string) {
	t.Helper()
	parquetPath := statementsSummaryParquetPath(t)

	ctx := context.Background()
	fetcher, err := duckdb.NewQueryDuckDBFetcher(ctx, "local", "local")
	if err != nil {
		t.Skipf("duckdb not available: %v", err)
	}
	t.Cleanup(func() {
		_ = fetcher.Close(ctx)
	})

	clusterID := "testcluster"
	alias := tableAlias(clusterID, STATEMENTS_TABLE_NAME)
	ddl := fmt.Sprintf(
		"CREATE OR REPLACE VIEW %s AS SELECT * FROM read_parquet('%s')",
		alias,
		strings.ReplaceAll(parquetPath, "'", "''"),
	)
	require.NoError(t, fetcher.ExecDDL(ctx, ddl))

	return &DuckDBAdapter{
		vendor: "local",
		region: "local",
		duckdb: fetcher,
	}, alias
}

func queryStatementsSummaryFromParquetView(
	t *testing.T,
	adapter *DuckDBAdapter,
	alias string,
	p *StatementsSummaryParams,
) []StatementsSummaryItem {
	t.Helper()
	where := "1=1 AND " + p.Filter
	query, err := buildStatementsSummaryQuery(alias, where, p)
	require.NoError(t, err)

	var rows []statementsSummaryBucketRow
	err = adapter.duckdb.SelectSlowlogInDLWithoutFilterPushdown(
		context.Background(),
		&rows,
		alias,
		"",
		query,
	)
	require.NoError(t, err)
	return aggregateStatementsSummaryRows(rows, p)
}

func logStatementsSummaryResponse(t *testing.T, label string, items []StatementsSummaryItem) {
	t.Helper()
	b, err := json.MarshalIndent(items, "", "  ")
	require.NoError(t, err)
	t.Logf("%s summary response:\n%s", label, string(b))
}

// TestStatementsSummary_Parquet_DigestVsUser runs the same v2 summary SQL as production
// against a fixed parquet fixture and prints the JSON response for digest and user dimensions.
func TestStatementsSummary_Parquet_DigestVsUser(t *testing.T) {
	adapter, alias := newStatementsSummaryParquetAdapter(t)

	const (
		beginUnix = int64(1780231149)
		endUnix   = int64(1780231333)
		step      = 60
	)

	baseFilter := buildStatementsSummaryBaseFilter(int(beginUnix), int(endUnix), nil, nil, "", sqlsDataSourceV2)
	base := &StatementsSummaryParams{
		Env:        "dev",
		Provider:   "local",
		Region:     "local",
		TenantID:   "testtenant",
		ClusterID:  "testcluster",
		Filter:     baseFilter,
		StartUnix:  beginUnix,
		EndUnix:    endUnix,
		Top:        3,
		OrderBy:    "exec_count",
		Step:       step,
		DataSource: sqlsDataSourceV2,
	}

	digestParams := *base
	digestParams.Dimension = "digest"
	digestData := queryStatementsSummaryFromParquetView(t, adapter, alias, &digestParams)
	logStatementsSummaryResponse(t, "dimension=digest", digestData)

	userParams := *base
	userParams.Dimension = "user"
	userData := queryStatementsSummaryFromParquetView(t, adapter, alias, &userParams)
	logStatementsSummaryResponse(t, "dimension=user", userData)

	require.NotEmpty(t, digestData, "digest summary should return at least one row")
	require.NotEmpty(t, userData, "user summary should return at least one row")

	digestItem := digestData[0]
	userItem := userData[0]
	require.Equal(t, "digest", digestItem.Dimension)
	require.Equal(t, "user", userItem.Dimension)
	require.NotEmpty(t, digestItem.SQLDigest)
	require.NotEmpty(t, userItem.User)

	require.GreaterOrEqual(t, len(digestItem.TimestampSec), 2,
		"digest top entity should have multiple time buckets in this fixture")
	require.Equal(t, len(digestItem.TimestampSec), len(digestItem.ValueByTime))
	require.Equal(t, len(userItem.TimestampSec), len(userItem.ValueByTime))
	require.Equal(t, len(digestItem.TimestampSec), len(userItem.TimestampSec),
		"user and digest should produce the same number of time buckets for the full fixture window")
}

// TestStatementsSummary_Parquet_NarrowWindow shows a single-bucket response when the query
// window only overlaps one time_bucket interval (similar to production single-timestamp cases).
func TestStatementsSummary_Parquet_NarrowWindow(t *testing.T) {
	adapter, alias := newStatementsSummaryParquetAdapter(t)

	const narrowBegin = int64(1780231149)
	const narrowEnd = int64(1780231200)

	filter := buildStatementsSummaryBaseFilter(int(narrowBegin), int(narrowEnd), nil, nil, "", sqlsDataSourceV2)
	params := &StatementsSummaryParams{
		Env:        "dev",
		Provider:   "local",
		Region:     "local",
		TenantID:   "testtenant",
		ClusterID:  "testcluster",
		Filter:     filter,
		StartUnix:  narrowBegin,
		EndUnix:    narrowEnd,
		Top:        1,
		OrderBy:    "exec_count",
		Step:       60,
		Dimension:  "user",
		DataSource: sqlsDataSourceV2,
	}

	data := queryStatementsSummaryFromParquetView(t, adapter, alias, params)
	logStatementsSummaryResponse(t, "dimension=user narrow window", data)
	require.Len(t, data, 1)
	require.Len(t, data[0].TimestampSec, 1,
		"a narrow window that fits one bucket should return one timestamp point")
}