package handlers

import (
	"encoding/json"
	"net/http"
	"net/url"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/gin-gonic/gin"

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

func TestDataProxyLegacyRoutesRemainRegistered(t *testing.T) {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	(&DataProxyHandler{}).RegisterRoutes(router)

	want := map[string]bool{
		http.MethodPost + " /data-proxy/query":       true,
		http.MethodPost + " /data-proxy/batch-query": true,
		http.MethodGet + " /data-proxy/schema":       true,
		http.MethodGet + " /data-proxy/info":         true,
	}
	for _, route := range router.Routes() {
		delete(want, route.Method+" "+route.Path)
	}
	if len(want) != 0 {
		t.Fatalf("legacy routes not registered: %v", want)
	}
}

func TestDataProxyCloudQueryRouteUsesUnifiedPublicPath(t *testing.T) {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	(&DataProxyHandler{}).RegisterRoutes(router)

	const publicPath = "/api/v1/orgs/:orgId/data-proxy/clusters/:clusterId/query"
	registered := false
	for _, route := range router.Routes() {
		if route.Method == http.MethodPost && route.Path == publicPath {
			registered = true
		}
		if route.Method == http.MethodPost && route.Path == "/api/v1/orgs/:org_id/clusters/:cluster_id/data-proxy/query" {
			t.Fatalf("deprecated public query route is still registered")
		}
	}
	if !registered {
		t.Fatalf("public query route %q is not registered", publicPath)
	}
}

func TestLegacyAndCanonicalDataProxyRoutesCoexist(t *testing.T) {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	(&DataProxyHandler{}).RegisterRoutes(router)
	service, ownership, target := newCanonicalTestDependencies()
	NewCanonicalDataProxyHandler(service, ownership, target).RegisterRoutes(router)

	want := map[string]bool{
		http.MethodPost + " /data-proxy/query":                                                    true,
		http.MethodPost + " /data-proxy/batch-query":                                              true,
		http.MethodGet + " /data-proxy/schema":                                                    true,
		http.MethodGet + " /data-proxy/info":                                                      true,
		http.MethodPost + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/sql-queries":       true,
		http.MethodPost + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/sql-query-batches": true,
		http.MethodGet + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/table-schemas":      true,
		http.MethodGet + " /data-proxy/api/v1/capabilities":                                       true,
	}
	for _, route := range router.Routes() {
		delete(want, route.Method+" "+route.Path)
	}
	if len(want) != 0 {
		t.Fatalf("legacy and canonical routes did not coexist: %v", want)
	}
}

func TestLegacyInfoAndCanonicalCapabilitiesRemainEquivalent(t *testing.T) {
	dataProxyService, err := service.NewDataProxyService(&config.DataProxyConfig{})
	if err != nil {
		t.Fatalf("create Data Proxy service: %v", err)
	}
	router := gin.New()
	NewDataProxyHandler(dataProxyService, &config.DataProxyConfig{}).RegisterRoutes(router)
	NewCanonicalDataProxyHandler(dataProxyService, nil, nil).RegisterRoutes(router)

	legacy := performCanonicalRequest(router, http.MethodGet, "/data-proxy/info", "")
	canonical := performCanonicalRequest(router, http.MethodGet, "/data-proxy/api/v1/capabilities", "")
	if legacy.Code != http.StatusOK || canonical.Code != http.StatusOK {
		t.Fatalf("legacy status=%d canonical status=%d", legacy.Code, canonical.Code)
	}
	var legacyBody, canonicalBody map[string]any
	if err := json.Unmarshal(legacy.Body.Bytes(), &legacyBody); err != nil {
		t.Fatalf("decode legacy info: %v", err)
	}
	if err := json.Unmarshal(canonical.Body.Bytes(), &canonicalBody); err != nil {
		t.Fatalf("decode canonical capabilities: %v", err)
	}
	if !reflect.DeepEqual(legacyBody, canonicalBody) {
		t.Fatalf("legacy info=%v canonical capabilities=%v", legacyBody, canonicalBody)
	}
}

func TestLegacyQueryAliasRoutingStillReachesExistingService(t *testing.T) {
	dataProxyService, err := service.NewDataProxyService(&config.DataProxyConfig{})
	if err != nil {
		t.Fatalf("create Data Proxy service: %v", err)
	}
	router := gin.New()
	NewDataProxyHandler(dataProxyService, &config.DataProxyConfig{}).RegisterRoutes(router)
	body := `{
		"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'",
		"cluster_id":"456",
		"options":{"data_source":"gs://caller/path","vendor":"gcp","region":"us-central1"}
	}`

	response := performCanonicalRequest(router, http.MethodPost, "/data-proxy/query", body)
	if response.Code != http.StatusInternalServerError {
		t.Fatalf("status=%d, want 500; body=%s", response.Code, response.Body.String())
	}
	if !strings.Contains(response.Body.String(), "no engine available for vendor: gcp") {
		t.Fatalf("legacy alias routing did not reach the existing gcp service path: %s", response.Body.String())
	}
}

func TestLegacySchemaAliasesAndLocalModeRemainCompatible(t *testing.T) {
	previousEnv := config.AppConfig.Env
	config.AppConfig.Env = "local"
	defer func() { config.AppConfig.Env = previousEnv }()

	dataProxyService, err := service.NewDataProxyService(&config.DataProxyConfig{})
	if err != nil {
		t.Fatalf("create Data Proxy service: %v", err)
	}
	router := gin.New()
	NewDataProxyHandler(dataProxyService, &config.DataProxyConfig{}).RegisterRoutes(router)

	queries := []url.Values{
		{
			"tables":     {"slow_query_logs"},
			"clusterId":  {"456"},
			"dataSource": {"gs://caller/path"},
			"vendor":     {"gcp"},
			"region":     {"us-central1"},
		},
		{
			"tables":             {"slow_query_logs"},
			"clusterId":          {"456"},
			"options.dataSource": {"gs://caller/path"},
			"options.vendor":     {"gcp"},
			"options.region":     {"us-central1"},
		},
	}
	for _, query := range queries {
		response := performCanonicalRequest(router, http.MethodGet, "/data-proxy/schema?"+query.Encode(), "")
		if response.Code != http.StatusInternalServerError ||
			!strings.Contains(response.Body.String(), "no engine available for vendor: gcp") {
			t.Fatalf("legacy schema aliases did not reach gcp service path: status=%d body=%s", response.Code, response.Body.String())
		}
	}

	missingDataSource := performCanonicalRequest(
		router,
		http.MethodGet,
		"/data-proxy/schema?tables=slow_query_logs&clusterId=456",
		"",
	)
	if missingDataSource.Code != http.StatusBadRequest ||
		!strings.Contains(missingDataSource.Body.String(), "DATASOURCE_REQUIRED") {
		t.Fatalf("legacy local schema behavior changed: status=%d body=%s", missingDataSource.Code, missingDataSource.Body.String())
	}
}

func TestLegacyDataProxyRequestAliasesRemainCompatible(t *testing.T) {
	body := `{
		"sql":"SELECT * FROM slow_query_logs",
		"cluster_id":"456",
		"timeout":45,
		"options":{
			"region":"us-west-2",
			"vendor":"aws",
			"data_source":"s3://caller/slowlogs",
			"skip_table_management":true,
			"skip_sql_rewrite":true
		}
	}`
	var request dto.DataProxyQueryRequest
	if err := json.Unmarshal([]byte(body), &request); err != nil {
		t.Fatalf("unmarshal legacy request: %v", err)
	}

	if request.SQL != "SELECT * FROM slow_query_logs" || request.ClusterId != "456" {
		t.Fatalf("legacy request identity = %+v", request)
	}
	if request.Timeout != 45*time.Second {
		t.Fatalf("legacy timeout = %s, want 45s", request.Timeout)
	}
	if request.Options.Region != "us-west-2" || request.Options.Vendor != "aws" {
		t.Fatalf("legacy provider options = %+v", request.Options)
	}
	if request.Options.DataSource != "s3://caller/slowlogs" {
		t.Fatalf("legacy caller data source = %q", request.Options.DataSource)
	}
	if !request.Options.SkipTableManagement || !request.Options.SkipSQLRewrite {
		t.Fatalf("legacy skip options were not preserved: %+v", request.Options)
	}
}

func TestLegacyCamelCaseFieldsKeepPrecedenceOverAliases(t *testing.T) {
	body := `{
		"sql":"SELECT * FROM slow_query_logs",
		"clusterId":"456",
		"cluster_id":"999",
		"options":{
			"dataSource":"s3://caller/camel",
			"data_source":"s3://caller/snake"
		}
	}`
	var request dto.DataProxyQueryRequest
	if err := json.Unmarshal([]byte(body), &request); err != nil {
		t.Fatalf("unmarshal legacy request: %v", err)
	}
	if request.ClusterId != "456" {
		t.Fatalf("clusterId = %q, want camel-case value", request.ClusterId)
	}
	if request.Options.DataSource != "s3://caller/camel" {
		t.Fatalf("dataSource = %q, want camel-case value", request.Options.DataSource)
	}
}

func TestLegacyBatchRequestStillSupportsPerItemClusterAndRouting(t *testing.T) {
	body := `[
		{
			"sql":"SELECT * FROM slow_query_logs",
			"clusterId":"111",
			"options":{"dataSource":"s3://caller/one","vendor":"aws"}
		},
		{
			"sql":"SELECT * FROM topsql",
			"cluster_id":"222",
			"options":{"data_source":"gs://caller/two","vendor":"gcp"}
		}
	]`
	var requests []*dto.DataProxyQueryRequest
	if err := json.Unmarshal([]byte(body), &requests); err != nil {
		t.Fatalf("unmarshal legacy batch: %v", err)
	}
	if len(requests) != 2 {
		t.Fatalf("legacy batch length = %d, want 2", len(requests))
	}
	if requests[0].ClusterId != "111" || requests[0].Options.DataSource != "s3://caller/one" || requests[0].Options.Vendor != "aws" {
		t.Fatalf("legacy batch item 0 = %+v", requests[0])
	}
	if requests[1].ClusterId != "222" || requests[1].Options.DataSource != "gs://caller/two" || requests[1].Options.Vendor != "gcp" {
		t.Fatalf("legacy batch item 1 = %+v", requests[1])
	}
}