package topsqlv2

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

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

func TestBuildAvailableOrderByFromSchema(t *testing.T) {
	columns := columnNameSet([]string{
		"topsql_cpu_time_ms",
		"topsql_network_in_bytes",
		"topsql_network_out_bytes",
	})
	fields := buildAvailableOrderByFromSchema(columns)
	require.Equal(t, []string{"cpu", "network", "network_in", "network_out"}, fields)
	require.NotContains(t, fields, "logical_io")
	require.NotContains(t, fields, "rocksdb_block_read")
}

func TestBuildAvailableOrderByFromSchemaAllColumns(t *testing.T) {
	columns := columnNameSet([]string{
		"topsql_cpu_time_ms",
		"topsql_network_in_bytes",
		"topsql_network_out_bytes",
		"topsql_logical_read_bytes",
		"topsql_logical_write_bytes",
		"topsql_rocksdb_block_read_count",
	})
	fields := buildAvailableOrderByFromSchema(columns)
	require.Equal(t, []string{
		"cpu",
		"network",
		"network_in",
		"network_out",
		"logical_io",
		"logical_read",
		"logical_write",
		"rocksdb_block_read",
	}, fields)
}

func TestBuildTopSQLMetricSQLFragmentsOmitsMissingColumns(t *testing.T) {
	columns := columnNameSet([]string{
		"topsql_cpu_time_ms",
		"topsql_network_in_bytes",
		"topsql_rocksdb_block_read_count",
	})
	frags := buildTopSQLMetricSQLFragments(columns)
	joined := strings.Join(frags.RawSelectExprs, ",")
	require.Contains(t, joined, "network_in_bytes")
	require.NotContains(t, joined, "network_out_bytes")
	require.NotContains(t, joined, " AS network_bytes")
	require.NotContains(t, joined, "logical_read_bytes")
	require.Contains(t, joined, "rocksdb_block_read_count")
	require.Contains(t, frags.MetricSumSelect, "total_network_in_bytes")
	require.NotContains(t, frags.MetricSumSelect, "total_network_bytes")
	require.Contains(t, frags.MetricSumSelect, "total_rocksdb_block_read_count")
}

func TestTopsqlTableAlias(t *testing.T) {
	alias := topsqlTableAlias("cluster-1", "tikv", "10.0.0.1:20160")
	require.Equal(t, "topsql_tikv_cluster_1_10_0_0_1_20160", alias)
}

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

	t.Run("returns schema-filtered order_by fields", func(t *testing.T) {
		mockAdapter := &mockDuckDBAdapterForService{
			selectFunc: func(ctx context.Context, dest interface{}, tableAlias, table, query string) error {
				require.Contains(t, table, "component=tikv/instance=db-tikv-0")
				require.Contains(t, query, "DESCRIBE SELECT")
				if destPtr, ok := dest.(*[]topsqlColumnInfo); ok {
					*destPtr = []topsqlColumnInfo{
						{ColumnName: "topsql_cpu_time_ms"},
						{ColumnName: "topsql_logical_read_bytes"},
						{ColumnName: "topsql_logical_write_bytes"},
					}
				}
				return nil
			},
		}
		service := &Service{
			SQLsDuckDBAdapter: mockAdapter,
			SQLsQuery:         mockAdapter,
			store:             &mockStorageForService{},
		}

		router := gin.New()
		router.GET("/topsql/available_fields", service.availableFieldsHandler)
		req := httptest.NewRequest("GET", "/topsql/available_fields?instance=db-tikv-0&instance_type=tikv&data_source=deltalake", nil)
		req.Header.Set("X-Provider", "aws")
		req.Header.Set("X-Region", "us-west-2")
		req.Header.Set("X-Org-Id", "tenant-1")
		req.Header.Set("X-Cluster-Id", "cluster-1")
		req.Header.Set("X-Deploy-Type", "dedicated")

		w := httptest.NewRecorder()
		router.ServeHTTP(w, req)
		require.Equal(t, http.StatusOK, w.Code)

		var fields []string
		require.NoError(t, json.Unmarshal(w.Body.Bytes(), &fields))
		require.Equal(t, []string{"cpu", "logical_io", "logical_read", "logical_write"}, fields)
	})

	t.Run("requires deltalake data_source", func(t *testing.T) {
		service := &Service{
			SQLsDuckDBAdapter: &mockDuckDBAdapterForService{},
			store:             &mockStorageForService{},
		}
		router := gin.New()
		router.GET("/topsql/available_fields", service.availableFieldsHandler)
		req := httptest.NewRequest("GET", "/topsql/available_fields?instance=db-tidb-0&instance_type=tidb", nil)
		req.Header.Set("X-Provider", "aws")
		req.Header.Set("X-Region", "us-west-2")
		req.Header.Set("X-Org-Id", "tenant-1")
		req.Header.Set("X-Cluster-Id", "cluster-1")
		req.Header.Set("X-Deploy-Type", "dedicated")

		w := httptest.NewRecorder()
		router.ServeHTTP(w, req)
		require.Equal(t, http.StatusBadRequest, w.Code)
	})

	t.Run("schema error", func(t *testing.T) {
		mockAdapter := &mockDuckDBAdapterForService{
			selectFunc: func(ctx context.Context, dest interface{}, tableAlias, table, query string) error {
				return errors.New("schema missing")
			},
		}
		service := &Service{
			SQLsDuckDBAdapter: mockAdapter,
			store:             &mockStorageForService{},
		}
		router := gin.New()
		router.GET("/topsql/available_fields", service.availableFieldsHandler)
		req := httptest.NewRequest("GET", "/topsql/available_fields?instance=db-tidb-0&instance_type=tidb&data_source=deltalake", nil)
		req.Header.Set("X-Provider", "aws")
		req.Header.Set("X-Region", "us-west-2")
		req.Header.Set("X-Org-Id", "tenant-1")
		req.Header.Set("X-Cluster-Id", "cluster-1")
		req.Header.Set("X-Deploy-Type", "dedicated")

		w := httptest.NewRecorder()
		router.ServeHTTP(w, req)
		require.Equal(t, http.StatusInternalServerError, w.Code)
	})

	t.Run("empty schema is an error", func(t *testing.T) {
		mockAdapter := &mockDuckDBAdapterForService{
			selectFunc: func(ctx context.Context, dest interface{}, tableAlias, table, query string) error {
				// Simulate DuckDB returning success with zero columns (e.g. swallowed
				// "No files in log segment"), which previously produced [].
				return nil
			},
		}
		service := &Service{
			SQLsDuckDBAdapter: mockAdapter,
			store:             &mockStorageForService{},
		}
		router := gin.New()
		router.GET("/topsql/available_fields", service.availableFieldsHandler)
		req := httptest.NewRequest("GET", "/topsql/available_fields?instance=db-tidb-0&instance_type=tidb&data_source=deltalake", nil)
		req.Header.Set("X-Provider", "aws")
		req.Header.Set("X-Region", "us-west-2")
		req.Header.Set("X-Org-Id", "tenant-1")
		req.Header.Set("X-Cluster-Id", "cluster-1")
		req.Header.Set("X-Deploy-Type", "dedicated")

		w := httptest.NewRecorder()
		router.ServeHTTP(w, req)
		require.Equal(t, http.StatusInternalServerError, w.Code)
		require.Contains(t, w.Body.String(), "schema is empty")
	})
}