package query

import (
	"os"
	"path/filepath"
	"strings"
	"testing"

	"github.com/parquet-go/parquet-go"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// Fixtures copied from prod us-east-1-serverless metrics-status Delta parquet files
// (biztype=dev-devtier/applicationID=devtier-infra-tidb/date=20260624).
const metricsStatusFixtureDir = "testdata/metrics-status"

func metricsStatusParquetFixturePath(t *testing.T, name string) string {
	t.Helper()
	path := filepath.Join(metricsStatusFixtureDir, name)
	if _, err := os.Stat(path); err != nil {
		t.Skipf("metrics-status parquet fixture not found: %s", path)
	}
	abs, err := filepath.Abs(path)
	require.NoError(t, err)
	return abs
}

func readParquetFileColumns(t *testing.T, path string) []ColumnMeta {
	t.Helper()
	f, err := os.Open(path)
	require.NoError(t, err)
	defer f.Close()

	st, err := f.Stat()
	require.NoError(t, err)

	pf, err := parquet.OpenFile(f, st.Size())
	require.NoError(t, err)

	fields := pf.Schema().Fields()
	cols := make([]ColumnMeta, 0, len(fields))
	for _, field := range fields {
		cols = append(cols, ColumnMeta{
			Name: field.Name(),
			Type: parquetTypeToGlue(field.Type().String()),
		})
	}
	return cols
}

func parquetTypeToGlue(parquetType string) string {
	switch strings.ToUpper(parquetType) {
	case "INT32":
		return "int"
	case "INT64":
		return "bigint"
	case "BOOLEAN":
		return "boolean"
	case "FLOAT":
		return "float"
	case "DOUBLE":
		return "double"
	default:
		return "string"
	}
}

func columnMetaNames(cols []ColumnMeta) []string {
	names := make([]string, len(cols))
	for i, col := range cols {
		names[i] = col.Name
	}
	return names
}

func assertColumnsMatch(t *testing.T, fixture string, def TableDefinition) {
	t.Helper()
	path := metricsStatusParquetFixturePath(t, fixture)
	got := readParquetFileColumns(t, path)

	require.Equal(t, len(def.FileColumns), len(got), "file column count mismatch for %s", fixture)
	for i, want := range def.FileColumns {
		assert.Equal(t, want.Name, got[i].Name, "column name mismatch at index %d", i)
		assert.Equal(t, want.Type, got[i].Type, "column type mismatch for %q", want.Name)
	}

	partitionNames := columnMetaNames(def.PartitionColumns)
	fileNames := columnMetaNames(got)
	for _, part := range partitionNames {
		assert.NotContains(t, fileNames, part,
			"partition column %q must not appear in parquet file columns (Hive path only)", part)
	}
}

func TestMetricsStatusSourcesParquetSchema(t *testing.T) {
	assertColumnsMatch(t, "sources.parquet", metricsStatusSourcesTableDef())
}

func TestMetricsStatusByMetricParquetSchema(t *testing.T) {
	assertColumnsMatch(t, "series_count_by_metric_name.parquet", metricsStatusByMetricTableDef())
}

func TestMetricsStatusByLabelParquetSchema(t *testing.T) {
	assertColumnsMatch(t, "series_count_by_label_name.parquet", metricsStatusByLabelTableDef())
}

func TestMetricsStatusByLabelValueParquetSchema(t *testing.T) {
	assertColumnsMatch(t, "series_count_by_label_value_pair.parquet", metricsStatusByLabelValueTableDef())
}

func TestMetricsStatusPartitionColumns(t *testing.T) {
	want := metricsStatusPartitionColumns()
	for _, defFn := range []func() TableDefinition{
		metricsStatusSourcesTableDef,
		metricsStatusByMetricTableDef,
		metricsStatusByLabelTableDef,
		metricsStatusByLabelValueTableDef,
	} {
		def := defFn()
		require.Equal(t, want, def.PartitionColumns, "dataset=%s", def.Dataset)
	}
}