package handlers

import (
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	"github.com/gin-gonic/gin"

	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

func TestOverrideStatementsDataSourceIgnoresCallerOverrides(t *testing.T) {
	options := dto.DataProxyQueryOptions{
		DataSource:          "s3://caller-controlled/path",
		Region:              "caller-region",
		Vendor:              "caller-vendor",
		ClinicTenantId:      "caller-tenant",
		SkipSQLRewrite:      true,
		SkipTableManagement: true,
	}
	clusterInfo := &clinic_internal_api.ClusterDataURLs{
		StatementsS3URL: "gs://o11y-prod-shared-us-central1/deltalake/org/o11y/persisted_statements_summary",
		Region:          "us-central1",
		Provider:        " GCP ",
		ClinicTenantID:  "12345",
	}

	overrideStatementsDataSource(&options, clusterInfo)

	if options.DataSource != clusterInfo.StatementsS3URL {
		t.Fatalf("data source = %q, want Clinic source %q", options.DataSource, clusterInfo.StatementsS3URL)
	}
	if options.Region != clusterInfo.Region {
		t.Fatalf("region = %q, want %q", options.Region, clusterInfo.Region)
	}
	if options.Vendor != "gcp" {
		t.Fatalf("vendor = %q, want %q", options.Vendor, "gcp")
	}
	if options.ClinicTenantId != clusterInfo.ClinicTenantID {
		t.Fatalf("tenant = %q, want %q", options.ClinicTenantId, clusterInfo.ClinicTenantID)
	}
	if options.SkipSQLRewrite || options.SkipTableManagement {
		t.Fatalf("Statements must force server-managed rewrite and table management: %+v", options)
	}
}

func TestStatementsProviderSupported(t *testing.T) {
	for _, provider := range []string{"aws", "gcp", "azure", " AWS "} {
		if !statementsProviderSupported(provider) {
			t.Fatalf("provider %q should support Statements queries", provider)
		}
	}
	for _, provider := range []string{"alicloud", "aliyun", "unknown", ""} {
		if statementsProviderSupported(provider) {
			t.Fatalf("provider %q should remain explicitly unsupported", provider)
		}
	}
}

func TestQueryRejectsMixedStatementsAndSlowQuery(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &DataProxyHandler{}
	router := gin.New()
	router.POST("/data-proxy/query", handler.Query)

	body := `{
		"clusterId":"123456",
		"sql":"SELECT s.digest FROM clinic_data_proxy.statements s JOIN clinic_data_proxy.slow_query_logs q ON s.digest = q.digest WHERE s.date = '20260721'"
	}`
	request := httptest.NewRequest(http.MethodPost, "/data-proxy/query", strings.NewReader(body))
	request.Header.Set("Content-Type", "application/json")
	response := httptest.NewRecorder()

	router.ServeHTTP(response, request)

	if response.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
	}
	if !strings.Contains(response.Body.String(), "MIXED_QUERY_TYPES") {
		t.Fatalf("body = %s, want MIXED_QUERY_TYPES", response.Body.String())
	}
}

func TestQueryRequiresClusterIDForStatements(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &DataProxyHandler{}
	router := gin.New()
	router.POST("/data-proxy/query", handler.Query)

	body := `{
		"sql":"SELECT digest FROM clinic_data_proxy.statements WHERE date = '2026-07-21'",
		"options":{"dataSource":"s3://caller-controlled/persisted_statements_summary"}
	}`
	request := httptest.NewRequest(http.MethodPost, "/data-proxy/query", strings.NewReader(body))
	request.Header.Set("Content-Type", "application/json")
	response := httptest.NewRecorder()

	router.ServeHTTP(response, request)

	if response.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
	}
	if !strings.Contains(response.Body.String(), "CLUSTER_ID_REQUIRED") {
		t.Fatalf("body = %s, want CLUSTER_ID_REQUIRED", response.Body.String())
	}
}

func TestQueryRejectsCallerControlledStatementsDatabase(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &DataProxyHandler{}
	router := gin.New()
	router.POST("/data-proxy/query", handler.Query)

	body := `{
		"clusterId":"123456",
		"sql":"SELECT digest FROM other_db.statements WHERE date = '2026-07-21'",
		"options":{"skipTableManagement":true,"skipSQLRewrite":true}
	}`
	request := httptest.NewRequest(http.MethodPost, "/data-proxy/query", strings.NewReader(body))
	request.Header.Set("Content-Type", "application/json")
	response := httptest.NewRecorder()

	router.ServeHTTP(response, request)

	if response.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
	}
	if !strings.Contains(response.Body.String(), "INVALID_STATEMENTS_NAMESPACE") {
		t.Fatalf("body = %s, want INVALID_STATEMENTS_NAMESPACE", response.Body.String())
	}
}

func TestGetSchemaRejectsStatements(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &DataProxyHandler{}
	router := gin.New()
	router.GET("/data-proxy/schema", handler.GetSchema)

	request := httptest.NewRequest(http.MethodGet, "/data-proxy/schema?tables=statements&clusterId=123456", nil)
	response := httptest.NewRecorder()
	router.ServeHTTP(response, request)

	if response.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
	}
	if !strings.Contains(response.Body.String(), "STATEMENTS_SCHEMA_UNSUPPORTED") {
		t.Fatalf("body = %s, want STATEMENTS_SCHEMA_UNSUPPORTED", response.Body.String())
	}
}

func TestBatchQueryRejectsPhysicalStatementsTable(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &DataProxyHandler{}
	router := gin.New()
	router.POST("/data-proxy/batch-query", handler.BatchQuery)

	body := `[{
		"clusterId":"123456",
		"sql":"SELECT digest FROM clinic_data_proxy.statements_123456_deadbeef WHERE date = '2026-07-21'",
		"options":{"dataSource":"s3://caller-controlled/path","skipTableManagement":true,"skipSQLRewrite":true}
	}]`
	request := httptest.NewRequest(http.MethodPost, "/data-proxy/batch-query", strings.NewReader(body))
	request.Header.Set("Content-Type", "application/json")
	response := httptest.NewRecorder()
	router.ServeHTTP(response, request)

	if response.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
	}
	if !strings.Contains(response.Body.String(), "STATEMENTS_BATCH_UNSUPPORTED") {
		t.Fatalf("body = %s, want STATEMENTS_BATCH_UNSUPPORTED", response.Body.String())
	}
}