package handlers

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

	"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/clustercontext"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

const canonicalTestBasePath = "/data-proxy/api/v1/orgs/123/clusters/456"

type fakeCanonicalDataProxyService struct {
	queryCalls  int
	queryReq    *dto.DataProxyQueryRequest
	queryResult *dto.DataProxyQueryResult
	queryErr    error

	batchCalls   int
	batchReqs    []*dto.DataProxyQueryRequest
	batchResults []*dto.DataProxyQueryResult
	batchErr     error

	schemaCalls   int
	schemaTables  []string
	schemaOptions []dto.DataProxyQueryOptions
	schemaFn      func(string, *dto.DataProxyQueryOptions) (*dto.SchemaResponse, error)

	vendorCalls int
	vendors     []string
}

func (f *fakeCanonicalDataProxyService) Query(_ context.Context, req *dto.DataProxyQueryRequest) (*dto.DataProxyQueryResult, error) {
	f.queryCalls++
	if req != nil {
		copyReq := *req
		f.queryReq = &copyReq
	}
	return f.queryResult, f.queryErr
}

func (f *fakeCanonicalDataProxyService) BatchQuery(_ context.Context, reqs []*dto.DataProxyQueryRequest) ([]*dto.DataProxyQueryResult, error) {
	f.batchCalls++
	f.batchReqs = make([]*dto.DataProxyQueryRequest, len(reqs))
	for i, req := range reqs {
		if req != nil {
			copyReq := *req
			f.batchReqs[i] = &copyReq
		}
	}
	return f.batchResults, f.batchErr
}

func (f *fakeCanonicalDataProxyService) GetSchemaForTable(
	_ context.Context,
	table string,
	options *dto.DataProxyQueryOptions,
) (*dto.SchemaResponse, error) {
	f.schemaCalls++
	f.schemaTables = append(f.schemaTables, table)
	if options != nil {
		f.schemaOptions = append(f.schemaOptions, *options)
	}
	if f.schemaFn != nil {
		return f.schemaFn(table, options)
	}
	return &dto.SchemaResponse{Table: table}, nil
}

func (f *fakeCanonicalDataProxyService) GetSupportedVendors() []string {
	f.vendorCalls++
	return f.vendors
}

type fakeCanonicalOwnershipReader struct {
	calls      int
	gotOrgID   uint64
	gotCluster string
	cluster    *clustercontext.Cluster
	err        error
}

func (f *fakeCanonicalOwnershipReader) GetByOrgAndClusterID(
	_ context.Context,
	orgID uint64,
	clusterID string,
) (*clustercontext.Cluster, error) {
	f.calls++
	f.gotOrgID = orgID
	f.gotCluster = clusterID
	return f.cluster, f.err
}

type fakeCanonicalTargetReader struct {
	calls      int
	gotCluster string
	target     *clinic_internal_api.ClusterDataURLs
	err        error
}

func (f *fakeCanonicalTargetReader) GetCanonicalClusterDataURLs(
	_ context.Context,
	clusterID string,
) (*clinic_internal_api.ClusterDataURLs, error) {
	f.calls++
	f.gotCluster = clusterID
	return f.target, f.err
}

func newCanonicalTestDependencies() (
	*fakeCanonicalDataProxyService,
	*fakeCanonicalOwnershipReader,
	*fakeCanonicalTargetReader,
) {
	service := &fakeCanonicalDataProxyService{
		queryResult: &dto.DataProxyQueryResult{
			Columns: []string{"value"},
			Rows:    [][]interface{}{{"ok"}},
		},
		vendors: []string{"aws", "gcp"},
	}
	ownership := &fakeCanonicalOwnershipReader{
		cluster: &clustercontext.Cluster{ClusterID: "456", OrgID: "123"},
	}
	target := &fakeCanonicalTargetReader{
		target: &clinic_internal_api.ClusterDataURLs{
			ClusterID:       "456",
			OrgID:           "123",
			Provider:        " AWS ",
			Region:          " us-west-2 ",
			ClinicTenantID:  "tenant-1",
			SlowQueryS3URL:  " s3://authoritative/slowlogs ",
			StatementsS3URL: " s3://authoritative/persisted_statements_summary ",
			TopSQLS3URL:     " s3://authoritative/topsql ",
			LogStatsS3URL:   " s3://authoritative/logstats ",
		},
	}
	return service, ownership, target
}

func newCanonicalTestRouter(
	service canonicalDataProxyService,
	ownership canonicalClusterOwnershipReader,
	target canonicalClusterTargetReader,
) *gin.Engine {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	NewCanonicalDataProxyHandler(service, ownership, target).RegisterRoutes(router)
	return router
}

func performCanonicalRequest(router http.Handler, method, path, body string) *httptest.ResponseRecorder {
	request := httptest.NewRequest(method, path, strings.NewReader(body))
	if body != "" {
		request.Header.Set("Content-Type", "application/json")
	}
	response := httptest.NewRecorder()
	router.ServeHTTP(response, request)
	return response
}

func TestCanonicalDataProxyHandlerRegistersRoutes(t *testing.T) {
	service, ownership, target := newCanonicalTestDependencies()
	router := newCanonicalTestRouter(service, ownership, target)

	want := map[string]bool{
		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("canonical routes not registered: %v", want)
	}
}

func TestCanonicalSQLQueryUsesAuthoritativeDataSources(t *testing.T) {
	tests := []struct {
		name       string
		sql        string
		dataSource string
	}{
		{
			name:       "slow query",
			sql:        "SELECT digest FROM clinic_data_proxy.slow_query_logs WHERE date = '20260721'",
			dataSource: "s3://authoritative/slowlogs",
		},
		{
			name:       "statements",
			sql:        "SELECT digest FROM clinic_data_proxy.statements WHERE date = '2026-07-21'",
			dataSource: "s3://authoritative/persisted_statements_summary",
		},
		{
			name:       "top sql",
			sql:        "SELECT sql_digest FROM clinic_data_proxy.topsql WHERE date = '2026-07-21'",
			dataSource: "s3://authoritative/topsql",
		},
		{
			name:       "log stats",
			sql:        "SELECT level FROM clinic_data_proxy.log_stats WHERE date = '20260721'",
			dataSource: "s3://authoritative/logstats",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			service, ownership, target := newCanonicalTestDependencies()
			router := newCanonicalTestRouter(service, ownership, target)
			body, err := json.Marshal(dto.CanonicalDataProxyQueryRequest{SQL: test.sql, Timeout: 60})
			if err != nil {
				t.Fatal(err)
			}

			response := performCanonicalRequest(router, http.MethodPost, canonicalTestBasePath+"/sql-queries", string(body))
			if response.Code != http.StatusOK {
				t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusOK, response.Body.String())
			}
			if ownership.calls != 1 || target.calls != 1 || service.queryCalls != 1 {
				t.Fatalf("calls: ownership=%d target=%d query=%d, want all 1", ownership.calls, target.calls, service.queryCalls)
			}
			if ownership.gotOrgID != 123 || ownership.gotCluster != "456" || target.gotCluster != "456" {
				t.Fatalf("resolved scope mismatch: ownership=(%d,%q) target=%q", ownership.gotOrgID, ownership.gotCluster, target.gotCluster)
			}
			if service.queryReq == nil {
				t.Fatal("query service did not receive a request")
			}
			if service.queryReq.ClusterId != "456" || service.queryReq.Timeout != time.Minute {
				t.Fatalf("request scope/timeout = (%q,%s), want (456,1m)", service.queryReq.ClusterId, service.queryReq.Timeout)
			}
			options := service.queryReq.Options
			if options.DataSource != test.dataSource || options.Vendor != "aws" || options.Region != "us-west-2" {
				t.Fatalf("authoritative routing options = %+v", options)
			}
			if options.ClusterId != "456" || options.ClinicTenantId != "tenant-1" {
				t.Fatalf("internal routing context = %+v", options)
			}
			if options.SkipSQLRewrite || options.SkipTableManagement {
				t.Fatalf("canonical request enabled caller-controlled skip flags: %+v", options)
			}
		})
	}
}

func TestCanonicalWarehouseRoutesRejectBYOCBeforeExecution(t *testing.T) {
	deployments := []struct {
		name         string
		deployType   string
		deployTypeV2 string
	}{
		{name: "logical cluster", deployType: "byoc", deployTypeV2: "byoc"},
		{name: "resource pool", deployType: "byoc-host", deployTypeV2: "byoc resource pool"},
	}
	requests := []struct {
		name   string
		method string
		path   string
		body   string
	}{
		{
			name:   "single query",
			method: http.MethodPost,
			path:   canonicalTestBasePath + "/sql-queries",
			body:   `{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"}`,
		},
		{
			name:   "batch query",
			method: http.MethodPost,
			path:   canonicalTestBasePath + "/sql-query-batches",
			body:   `[{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"}]`,
		},
		{
			name:   "table schema",
			method: http.MethodGet,
			path:   canonicalTestBasePath + "/table-schemas?tables=topsql",
		},
	}

	for _, deployment := range deployments {
		for _, request := range requests {
			t.Run(deployment.name+"/"+request.name, func(t *testing.T) {
				service, ownership, target := newCanonicalTestDependencies()
				target.target.DeployType = deployment.deployType
				target.target.DeployTypeV2 = deployment.deployTypeV2
				router := newCanonicalTestRouter(service, ownership, target)

				response := performCanonicalRequest(router, request.method, request.path, request.body)
				if response.Code != http.StatusBadRequest {
					t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
				}
				if ownership.calls != 1 || target.calls != 1 {
					t.Fatalf("metadata calls: ownership=%d target=%d, want both 1", ownership.calls, target.calls)
				}
				if service.queryCalls != 0 || service.batchCalls != 0 || service.schemaCalls != 0 {
					t.Fatalf(
						"BYOC request reached execution: query=%d batch=%d schema=%d",
						service.queryCalls,
						service.batchCalls,
						service.schemaCalls,
					)
				}
			})
		}
	}
}

func TestCanonicalSQLQueryUsesDefaultTimeout(t *testing.T) {
	if canonicalDefaultQueryTimeout != 2*time.Minute {
		t.Fatalf("canonical default timeout = %s, want 2m", canonicalDefaultQueryTimeout)
	}
	service, ownership, target := newCanonicalTestDependencies()
	router := newCanonicalTestRouter(service, ownership, target)
	body := `{"sql":"SELECT digest FROM clinic_data_proxy.slow_query_logs WHERE date = '20260721'"}`

	response := performCanonicalRequest(router, http.MethodPost, canonicalTestBasePath+"/sql-queries", body)
	if response.Code != http.StatusOK {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusOK, response.Body.String())
	}
	if service.queryReq == nil || service.queryReq.Timeout != canonicalDefaultQueryTimeout {
		t.Fatalf("timeout = %v, want %v", service.queryReq, canonicalDefaultQueryTimeout)
	}
}

func TestCanonicalSQLQueryAcceptsBoundedComplexQueries(t *testing.T) {
	queries := []string{
		"SELECT digest FROM clinic_data_proxy.slow_query_logs WHERE date = '20260721' " +
			"UNION ALL SELECT digest FROM clinic_data_proxy.slow_query_logs WHERE date = '20260720'",
		"WITH bounded AS (SELECT digest FROM clinic_data_proxy.slow_query_logs WHERE date = '20260721') " +
			"SELECT digest FROM bounded",
		"SELECT a.digest FROM clinic_data_proxy.slow_query_logs a " +
			"JOIN clinic_data_proxy.slow_query_logs b ON a.digest = b.digest " +
			"WHERE a.date = '20260721' AND b.date = '20260721'",
	}

	for _, sql := range queries {
		service, ownership, target := newCanonicalTestDependencies()
		router := newCanonicalTestRouter(service, ownership, target)
		body, err := json.Marshal(dto.CanonicalDataProxyQueryRequest{SQL: sql})
		if err != nil {
			t.Fatal(err)
		}

		response := performCanonicalRequest(router, http.MethodPost, canonicalTestBasePath+"/sql-queries", string(body))
		if response.Code != http.StatusOK {
			t.Fatalf("sql=%q status = %d, want %d; body=%s", sql, response.Code, http.StatusOK, response.Body.String())
		}
		if service.queryCalls != 1 {
			t.Fatalf("sql=%q query calls = %d, want 1", sql, service.queryCalls)
		}
	}
}

func TestCanonicalSQLQueryRejectsStrictInputsBeforeLookup(t *testing.T) {
	tests := []struct {
		name string
		path string
		body string
	}{
		{name: "invalid org", path: "/data-proxy/api/v1/orgs/0/clusters/456/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs"}`},
		{name: "invalid cluster", path: "/data-proxy/api/v1/orgs/123/clusters/not-numeric/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs"}`},
		{name: "caller cluster field", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs","clusterId":"999"}`},
		{name: "caller options field", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs","options":{"dataSource":"s3://caller"}}`},
		{name: "case variant field", path: canonicalTestBasePath + "/sql-queries", body: `{"SQL":"SELECT * FROM slow_query_logs"}`},
		{name: "duplicate field", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs","sql":"SELECT * FROM topsql"}`},
		{name: "null timeout", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs","timeout":null}`},
		{name: "trailing json", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs"}{}`},
		{name: "blank sql", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"  "}`},
		{name: "negative timeout", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs","timeout":-1}`},
		{name: "excessive timeout", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM slow_query_logs","timeout":1801}`},
		{name: "unsupported table", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM caller_table"}`},
		{name: "caller controlled database", path: canonicalTestBasePath + "/sql-queries", body: `{"sql":"SELECT * FROM other_db.slow_query_logs"}`},
		{
			name: "comment cannot spoof date partition",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"sql":"/* date IN ('20260721') */ ` +
				`SELECT * FROM clinic_data_proxy.slow_query_logs"}`,
		},
		{
			name: "string cannot spoof date partition",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"sql":"SELECT 'date IN (' AS marker ` +
				`FROM clinic_data_proxy.slow_query_logs"}`,
		},
		{
			name: "each union branch requires a date partition",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"sql":"SELECT digest FROM clinic_data_proxy.slow_query_logs ` +
				`WHERE date = '20260721' UNION ALL ` +
				`SELECT digest FROM clinic_data_proxy.slow_query_logs"}`,
		},
		{
			name: "outer date filter cannot bound an unfiltered cte scan",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"sql":"WITH unbounded AS (` +
				`SELECT digest, date FROM clinic_data_proxy.slow_query_logs` +
				`) SELECT digest FROM unbounded WHERE date = '20260721'"}`,
		},
		{
			name: "each joined scan requires its own date partition",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"sql":"SELECT a.digest FROM clinic_data_proxy.slow_query_logs a ` +
				`JOIN clinic_data_proxy.slow_query_logs b ON a.digest = b.digest ` +
				`WHERE a.date = '20260721'"}`,
		},
		{
			name: "or true cannot bypass date partition",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"sql":"SELECT * FROM clinic_data_proxy.slow_query_logs ` +
				`WHERE date = '20260721' OR 1 = 1"}`,
		},
		{
			name: "same name cte cannot hide an unfiltered base scan",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"sql":"WITH slow_query_logs AS (` +
				`SELECT * FROM slow_query_logs` +
				`) SELECT * FROM slow_query_logs WHERE date = '20260721'"}`,
		},
		{
			name: "mixed source families",
			path: canonicalTestBasePath + "/sql-queries",
			body: `{"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'"}`,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			service, ownership, target := newCanonicalTestDependencies()
			router := newCanonicalTestRouter(service, ownership, target)
			response := performCanonicalRequest(router, http.MethodPost, test.path, test.body)
			if response.Code != http.StatusBadRequest {
				t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
			}
			if ownership.calls != 0 || target.calls != 0 || service.queryCalls != 0 {
				t.Fatalf("invalid request reached dependencies: ownership=%d target=%d query=%d", ownership.calls, target.calls, service.queryCalls)
			}
		})
	}
}

func TestCanonicalSQLQueryRejectsOversizedBody(t *testing.T) {
	service, ownership, target := newCanonicalTestDependencies()
	router := newCanonicalTestRouter(service, ownership, target)
	body := `{"sql":"` + strings.Repeat("x", int(canonicalSingleQueryBodyBytes)) + `"}`

	response := performCanonicalRequest(router, http.MethodPost, canonicalTestBasePath+"/sql-queries", body)
	if response.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusRequestEntityTooLarge, response.Body.String())
	}
	if ownership.calls != 0 || target.calls != 0 || service.queryCalls != 0 {
		t.Fatalf("oversized request reached dependencies: ownership=%d target=%d query=%d", ownership.calls, target.calls, service.queryCalls)
	}
}

func TestCanonicalOwnershipAndTargetFailures(t *testing.T) {
	tests := []struct {
		name       string
		status     int
		configure  func(*fakeCanonicalOwnershipReader, *fakeCanonicalTargetReader)
		secretText string
	}{
		{
			name:   "cluster not found",
			status: http.StatusNotFound,
			configure: func(ownership *fakeCanonicalOwnershipReader, _ *fakeCanonicalTargetReader) {
				ownership.err = clustercontext.ErrNotFound
			},
		},
		{
			name:   "ownership unavailable",
			status: http.StatusServiceUnavailable,
			configure: func(ownership *fakeCanonicalOwnershipReader, _ *fakeCanonicalTargetReader) {
				ownership.err = clustercontext.ErrUnavailable
			},
		},
		{
			name:   "ownership timeout",
			status: http.StatusGatewayTimeout,
			configure: func(ownership *fakeCanonicalOwnershipReader, _ *fakeCanonicalTargetReader) {
				ownership.err = clustercontext.ErrTimeout
			},
		},
		{
			name:       "ownership details are hidden",
			status:     http.StatusInternalServerError,
			secretText: "internal-ownership-secret",
			configure: func(ownership *fakeCanonicalOwnershipReader, _ *fakeCanonicalTargetReader) {
				ownership.err = errors.New("internal-ownership-secret")
			},
		},
		{
			name:   "ownership identity mismatch",
			status: http.StatusBadGateway,
			configure: func(ownership *fakeCanonicalOwnershipReader, _ *fakeCanonicalTargetReader) {
				ownership.cluster = &clustercontext.Cluster{ClusterID: "456", OrgID: "999"}
			},
		},
		{
			name:   "target identity mismatch",
			status: http.StatusBadGateway,
			configure: func(_ *fakeCanonicalOwnershipReader, target *fakeCanonicalTargetReader) {
				target.target.OrgID = "999"
			},
		},
		{
			name:   "target timeout",
			status: http.StatusGatewayTimeout,
			configure: func(_ *fakeCanonicalOwnershipReader, target *fakeCanonicalTargetReader) {
				target.err = context.DeadlineExceeded
			},
		},
		{
			name:       "target details are hidden",
			status:     http.StatusServiceUnavailable,
			secretText: "internal-target-secret",
			configure: func(_ *fakeCanonicalOwnershipReader, target *fakeCanonicalTargetReader) {
				target.err = errors.New("internal-target-secret")
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			service, ownership, target := newCanonicalTestDependencies()
			test.configure(ownership, target)
			router := newCanonicalTestRouter(service, ownership, target)
			response := performCanonicalRequest(
				router,
				http.MethodPost,
				canonicalTestBasePath+"/sql-queries",
				`{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"}`,
			)
			if response.Code != test.status {
				t.Fatalf("status = %d, want %d; body=%s", response.Code, test.status, response.Body.String())
			}
			if service.queryCalls != 0 {
				t.Fatalf("query service called %d times after authorization failure", service.queryCalls)
			}
			if test.secretText != "" && strings.Contains(response.Body.String(), test.secretText) {
				t.Fatalf("response exposed upstream detail: %s", response.Body.String())
			}
		})
	}
}

func TestCanonicalExecutionFailuresUseStableErrors(t *testing.T) {
	tests := []struct {
		name   string
		result *dto.DataProxyQueryResult
		err    error
		status int
		secret string
	}{
		{name: "nil result", status: http.StatusBadGateway},
		{name: "timeout", err: context.DeadlineExceeded, status: http.StatusGatewayTimeout},
		{name: "internal detail hidden", err: errors.New("warehouse-secret"), status: http.StatusInternalServerError, secret: "warehouse-secret"},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			service, ownership, target := newCanonicalTestDependencies()
			service.queryResult = test.result
			service.queryErr = test.err
			router := newCanonicalTestRouter(service, ownership, target)
			response := performCanonicalRequest(
				router,
				http.MethodPost,
				canonicalTestBasePath+"/sql-queries",
				`{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"}`,
			)
			if response.Code != test.status {
				t.Fatalf("status = %d, want %d; body=%s", response.Code, test.status, response.Body.String())
			}
			if service.queryCalls != 1 || ownership.calls != 1 || target.calls != 1 {
				t.Fatalf("calls: ownership=%d target=%d query=%d, want all 1", ownership.calls, target.calls, service.queryCalls)
			}
			if test.secret != "" && strings.Contains(response.Body.String(), test.secret) {
				t.Fatalf("response exposed execution detail: %s", response.Body.String())
			}
		})
	}
}

func TestCanonicalBatchUsesOneLookupAndNoPartialExecution(t *testing.T) {
	service, ownership, target := newCanonicalTestDependencies()
	service.batchResults = []*dto.DataProxyQueryResult{{}, {}, {}}
	router := newCanonicalTestRouter(service, ownership, target)
	body := `[
		{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'","timeout":30},
		{"sql":"SELECT * FROM topsql WHERE date = '2026-07-21'"},
		{"sql":"SELECT * FROM log_stats WHERE date = '20260721'"}
	]`

	response := performCanonicalRequest(router, http.MethodPost, canonicalTestBasePath+"/sql-query-batches", body)
	if response.Code != http.StatusOK {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusOK, response.Body.String())
	}
	if ownership.calls != 1 || target.calls != 1 || service.batchCalls != 1 || service.queryCalls != 0 {
		t.Fatalf("calls: ownership=%d target=%d batch=%d query=%d", ownership.calls, target.calls, service.batchCalls, service.queryCalls)
	}
	if len(service.batchReqs) != 3 {
		t.Fatalf("batch request count = %d, want 3", len(service.batchReqs))
	}
	wantSources := []string{
		"s3://authoritative/slowlogs",
		"s3://authoritative/topsql",
		"s3://authoritative/logstats",
	}
	for i, req := range service.batchReqs {
		if req.ClusterId != "456" || req.Options.DataSource != wantSources[i] {
			t.Fatalf("batch item %d routing = %+v", i, req)
		}
	}
	var decoded struct {
		Results []json.RawMessage `json:"results"`
		Count   int               `json:"count"`
	}
	if err := json.Unmarshal(response.Body.Bytes(), &decoded); err != nil {
		t.Fatalf("decode response: %v", err)
	}
	if decoded.Count != 3 || len(decoded.Results) != 3 {
		t.Fatalf("batch response = %+v", decoded)
	}
}

func TestCanonicalBatchRejectsInvalidItemsBeforeLookup(t *testing.T) {
	tooMany := make([]dto.CanonicalDataProxyQueryRequest, canonicalMaxBatchQueries+1)
	for i := range tooMany {
		tooMany[i].SQL = "SELECT * FROM slow_query_logs"
	}
	tooManyBody, err := json.Marshal(tooMany)
	if err != nil {
		t.Fatal(err)
	}

	tests := []struct {
		name string
		body string
	}{
		{name: "empty batch", body: `[]`},
		{name: "null item", body: `[{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"},null]`},
		{name: "unknown item field", body: `[{"sql":"SELECT * FROM slow_query_logs","clusterId":"999"}]`},
		{name: "invalid later item", body: `[{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"},{"sql":"SELECT * FROM caller_table"}]`},
		{name: "statements unsupported", body: `[{"sql":"SELECT * FROM clinic_data_proxy.statements WHERE date = '2026-07-21'"}]`},
		{name: "too many items", body: string(tooManyBody)},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			service, ownership, target := newCanonicalTestDependencies()
			router := newCanonicalTestRouter(service, ownership, target)
			response := performCanonicalRequest(router, http.MethodPost, canonicalTestBasePath+"/sql-query-batches", test.body)
			if response.Code != http.StatusBadRequest {
				t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
			}
			if ownership.calls != 0 || target.calls != 0 || service.batchCalls != 0 || service.queryCalls != 0 {
				t.Fatalf("invalid batch reached dependencies: ownership=%d target=%d batch=%d query=%d", ownership.calls, target.calls, service.batchCalls, service.queryCalls)
			}
		})
	}
}

func TestCanonicalBatchRejectsOversizedBody(t *testing.T) {
	service, ownership, target := newCanonicalTestDependencies()
	router := newCanonicalTestRouter(service, ownership, target)
	body := `[{"sql":"` + strings.Repeat("x", int(canonicalBatchQueryBodyBytes)) + `"}]`

	response := performCanonicalRequest(router, http.MethodPost, canonicalTestBasePath+"/sql-query-batches", body)
	if response.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusRequestEntityTooLarge, response.Body.String())
	}
	if ownership.calls != 0 || target.calls != 0 || service.batchCalls != 0 {
		t.Fatalf("oversized batch reached dependencies: ownership=%d target=%d batch=%d", ownership.calls, target.calls, service.batchCalls)
	}
}

func TestCanonicalHTTPSurfaceRejectsUnsupportedInputsBeforeDependencies(t *testing.T) {
	validQueryBody := `{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"}`
	validBatchBody := `[{"sql":"SELECT * FROM slow_query_logs WHERE date = '20260721'"}]`
	tests := []struct {
		name   string
		method string
		path   string
		body   string
	}{
		{
			name:   "single query rejects query parameters",
			method: http.MethodPost,
			path:   canonicalTestBasePath + "/sql-queries?debug=true",
			body:   validQueryBody,
		},
		{
			name:   "batch rejects query parameters",
			method: http.MethodPost,
			path:   canonicalTestBasePath + "/sql-query-batches?debug=true",
			body:   validBatchBody,
		},
		{
			name:   "schema rejects request body",
			method: http.MethodGet,
			path:   canonicalTestBasePath + "/table-schemas?tables=topsql",
			body:   `{}`,
		},
		{
			name:   "schema rejects undocumented query parameter",
			method: http.MethodGet,
			path:   canonicalTestBasePath + "/table-schemas?tables=topsql&debug=true",
		},
		{
			name:   "capabilities rejects request body",
			method: http.MethodGet,
			path:   "/data-proxy/api/v1/capabilities",
			body:   `{}`,
		},
		{
			name:   "capabilities rejects query parameters",
			method: http.MethodGet,
			path:   "/data-proxy/api/v1/capabilities?debug=true",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			service, ownership, target := newCanonicalTestDependencies()
			router := newCanonicalTestRouter(service, ownership, target)
			response := performCanonicalRequest(router, test.method, test.path, test.body)
			if response.Code != http.StatusBadRequest {
				t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
			}
			if ownership.calls != 0 || target.calls != 0 || service.queryCalls != 0 ||
				service.batchCalls != 0 || service.schemaCalls != 0 || service.vendorCalls != 0 {
				t.Fatalf(
					"unsupported HTTP input reached dependencies: ownership=%d target=%d query=%d batch=%d schema=%d vendors=%d",
					ownership.calls,
					target.calls,
					service.queryCalls,
					service.batchCalls,
					service.schemaCalls,
					service.vendorCalls,
				)
			}
		})
	}
}

func TestCanonicalTableSchemasUseLogicalNamesAndAuthoritativeTarget(t *testing.T) {
	service, ownership, target := newCanonicalTestDependencies()
	service.schemaFn = func(table string, _ *dto.DataProxyQueryOptions) (*dto.SchemaResponse, error) {
		return &dto.SchemaResponse{Database: "warehouse", Table: table, DataSource: "engine-source"}, nil
	}
	router := newCanonicalTestRouter(service, ownership, target)
	response := performCanonicalRequest(
		router,
		http.MethodGet,
		canonicalTestBasePath+"/table-schemas?tables=topsql,topsql_sql_meta",
		"",
	)

	if response.Code != http.StatusOK {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusOK, response.Body.String())
	}
	if ownership.calls != 1 || target.calls != 1 || service.schemaCalls != 2 {
		t.Fatalf("calls: ownership=%d target=%d schema=%d", ownership.calls, target.calls, service.schemaCalls)
	}
	wantPhysical := []string{"topsql_456", "topsql_sql_meta_456"}
	for i, table := range wantPhysical {
		if service.schemaTables[i] != table {
			t.Fatalf("schema table %d = %q, want %q", i, service.schemaTables[i], table)
		}
		options := service.schemaOptions[i]
		if options.DataSource != "s3://authoritative/topsql" || options.Vendor != "aws" || options.Region != "us-west-2" {
			t.Fatalf("schema options %d = %+v", i, options)
		}
		if options.ClusterId != "456" || options.ClinicTenantId != "tenant-1" {
			t.Fatalf("schema internal context %d = %+v", i, options)
		}
	}
	var decoded dto.BatchSchemaResponse
	if err := json.Unmarshal(response.Body.Bytes(), &decoded); err != nil {
		t.Fatalf("decode response: %v", err)
	}
	if decoded.Count != 2 || len(decoded.Schemas) != 2 {
		t.Fatalf("schema response = %+v", decoded)
	}
	if decoded.Schemas[0].Table != "topsql" || decoded.Schemas[1].Table != "topsql_sql_meta" {
		t.Fatalf("response exposed physical table names: %+v", decoded.Schemas)
	}
}

func TestCanonicalTableSchemasRejectInvalidParametersBeforeLookup(t *testing.T) {
	tooManyTables := make([]string, canonicalMaxSchemaTables+1)
	for i := range tooManyTables {
		tooManyTables[i] = "topsql"
	}
	tests := []struct {
		name  string
		query string
	}{
		{name: "missing tables"},
		{name: "tables repeated", query: "?tables=topsql&tables=topsql_sql_meta"},
		{name: "unknown parameter", query: "?tables=topsql&clusterId=999"},
		{name: "empty table", query: "?tables="},
		{name: "unsupported table", query: "?tables=caller_table"},
		{name: "statements hidden", query: "?tables=statements"},
		{name: "mixed source families", query: "?tables=slow_query_logs,topsql"},
		{name: "too many tables", query: "?tables=" + strings.Join(tooManyTables, ",")},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			service, ownership, target := newCanonicalTestDependencies()
			router := newCanonicalTestRouter(service, ownership, target)
			response := performCanonicalRequest(router, http.MethodGet, canonicalTestBasePath+"/table-schemas"+test.query, "")
			if response.Code != http.StatusBadRequest {
				t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String())
			}
			if ownership.calls != 0 || target.calls != 0 || service.schemaCalls != 0 {
				t.Fatalf("invalid schema request reached dependencies: ownership=%d target=%d schema=%d", ownership.calls, target.calls, service.schemaCalls)
			}
		})
	}
}

func TestCanonicalCapabilities(t *testing.T) {
	service, ownership, target := newCanonicalTestDependencies()
	router := newCanonicalTestRouter(service, ownership, target)
	response := performCanonicalRequest(router, http.MethodGet, "/data-proxy/api/v1/capabilities", "")

	if response.Code != http.StatusOK {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusOK, response.Body.String())
	}
	var decoded dto.DataProxyCapabilitiesResponse
	if err := json.Unmarshal(response.Body.Bytes(), &decoded); err != nil {
		t.Fatalf("decode response: %v", err)
	}
	if decoded.Service != "Clinic Data Proxy" || decoded.Version != "1.0.0" {
		t.Fatalf("capabilities identity = %+v", decoded)
	}
	if len(decoded.SupportedVendors) != 2 || decoded.SupportedVendors[0] != "aws" || decoded.SupportedVendors[1] != "gcp" {
		t.Fatalf("supported vendors = %v", decoded.SupportedVendors)
	}
	if ownership.calls != 0 || target.calls != 0 || service.queryCalls != 0 || service.batchCalls != 0 || service.schemaCalls != 0 {
		t.Fatalf("capabilities touched cluster/query dependencies")
	}
	if service.vendorCalls != 1 {
		t.Fatalf("GetSupportedVendors calls = %d, want 1", service.vendorCalls)
	}
}

func TestCanonicalCapabilitiesUnavailableWithoutService(t *testing.T) {
	router := newCanonicalTestRouter(nil, nil, nil)
	response := performCanonicalRequest(router, http.MethodGet, "/data-proxy/api/v1/capabilities", "")
	if response.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusServiceUnavailable, response.Body.String())
	}
}