package handlers

import (
	"context"
	"encoding/json"
	stderrors "errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"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/dialect"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/engine"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/errors"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/observability"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/service"
)

const (
	canonicalSingleQueryBodyBytes = int64(1 << 20)
	canonicalBatchQueryBodyBytes  = int64(4 << 20)
	canonicalMaxSQLBytes          = 1 << 20
	canonicalMaxBatchQueries      = 100
	canonicalMaxTimeoutSeconds    = int64((30 * time.Minute) / time.Second)
	canonicalMaxSchemaTables      = 16
	canonicalMaxClusterIDBytes    = 255
	canonicalDefaultQueryTimeout  = defaultDataProxyQueryTimeout
)

var (
	errCanonicalTrailingJSON            = stderrors.New("request body contains trailing JSON data")
	errCanonicalDataSourceNotConfigured = stderrors.New("requested data source is not configured")
)

type canonicalDataProxyService interface {
	Query(context.Context, *dto.DataProxyQueryRequest) (*dto.DataProxyQueryResult, error)
	BatchQuery(context.Context, []*dto.DataProxyQueryRequest) ([]*dto.DataProxyQueryResult, error)
	GetSchemaForTable(context.Context, string, *dto.DataProxyQueryOptions) (*dto.SchemaResponse, error)
	GetSupportedVendors() []string
}

type canonicalClusterOwnershipReader interface {
	GetByOrgAndClusterID(context.Context, uint64, string) (*clustercontext.Cluster, error)
}

type canonicalClusterTargetReader interface {
	GetCanonicalClusterDataURLs(context.Context, string) (*clinic_internal_api.ClusterDataURLs, error)
}

// CanonicalDataProxyHandler is a strict, versioned HTTP adapter over the same
// DataProxyService used by the legacy endpoints. It owns only public request
// validation, org/cluster ownership checks, and server-authoritative routing.
type CanonicalDataProxyHandler struct {
	service         canonicalDataProxyService
	ownershipReader canonicalClusterOwnershipReader
	targetReader    canonicalClusterTargetReader
}

// NewCanonicalDataProxyHandler constructs the additive canonical Data Proxy
// adapter without changing the legacy handler or its caller-controlled options.
func NewCanonicalDataProxyHandler(
	service canonicalDataProxyService,
	ownershipReader canonicalClusterOwnershipReader,
	targetReader canonicalClusterTargetReader,
) *CanonicalDataProxyHandler {
	return &CanonicalDataProxyHandler{
		service:         service,
		ownershipReader: ownershipReader,
		targetReader:    targetReader,
	}
}

// CreateSQLQuery executes one server-routed warehouse query.
//
//	@Summary		Execute a cluster SQL query
//	@ID			canonical-create-sql-query
//	@Tags			data-proxy-sql
//	@Accept			json
//	@Produce		json
//	@Param			orgId		path		string							true	"TiDB Cloud/DBAAS org ID"
//	@Param			clusterId	path		string							true	"Numeric warehouse cluster ID"
//	@Param			request		body		dto.CanonicalDataProxyQueryRequest	true	"SQL query request"
//	@Success		200			{object}	dto.DataProxyQueryResult
//	@Failure		400			{object}	errors.APIError	"Invalid request; UNKNOWN_FIELD details are returned for unresolved query fields"
//	@Failure		401			{object}	errors.APIError
//	@Failure		403			{object}	errors.APIError
//	@Failure		404			{object}	errors.APIError
//	@Failure		413			{object}	errors.APIError
//	@Failure		500			{object}	errors.APIError
//	@Failure		502			{object}	errors.APIError
//	@Failure		503			{object}	errors.APIError
//	@Failure		504			{object}	errors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/sql-queries [post]
func (h *CanonicalDataProxyHandler) CreateSQLQuery(c *gin.Context) {
	start, rec := beginCanonicalDataProxyRequest(c, "canonical_sql_query")
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	orgID, clusterID, ok := parseCanonicalWarehouseScope(c, rec)
	if !ok {
		return
	}
	if !requireNoCanonicalQuery(c, rec) {
		return
	}

	var input dto.CanonicalDataProxyQueryRequest
	if err := decodeCanonicalJSON(c, &input, canonicalSingleQueryBodyBytes); err != nil {
		writeCanonicalDecodeError(c, rec, err)
		return
	}

	req, err := buildCanonicalQueryRequest(input, clusterID)
	if err != nil {
		rec.ErrCode = "INVALID_QUERY_REQUEST"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", err.Error(), nil)
		return
	}
	queryType, err := validateCanonicalQuery(req)
	if err != nil {
		rec.ErrCode = "INVALID_QUERY"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", err.Error(), nil)
		return
	}
	rec.QueryType = string(queryType)

	target, ok := h.authorizeAndResolveCluster(c, rec, orgID, clusterID)
	if !ok {
		return
	}
	if err := applyCanonicalQueryTarget(req, queryType, target); err != nil {
		h.writeCanonicalRoutingError(c, rec, queryType, target, err)
		return
	}

	h.executeCanonicalQuery(c, rec, req)
}

// CreateSQLQueryBatch executes a same-cluster batch through the existing batch
// query core after validating and resolving the target exactly once.
//
//	@Summary		Execute a same-cluster SQL query batch
//	@ID			canonical-create-sql-query-batch
//	@Tags			data-proxy-sql
//	@Accept			json
//	@Produce		json
//	@Param			orgId		path		string								true	"TiDB Cloud/DBAAS org ID"
//	@Param			clusterId	path		string								true	"Numeric warehouse cluster ID"
//	@Param			request		body		[]dto.CanonicalDataProxyQueryRequest	true	"SQL query request array"
//	@Success		200			{object}	dto.DataProxyBatchQueryResponse
//	@Failure		400			{object}	errors.APIError	"Invalid request; indexed UNKNOWN_FIELD details are returned for unresolved query fields"
//	@Failure		401			{object}	errors.APIError
//	@Failure		403			{object}	errors.APIError
//	@Failure		404			{object}	errors.APIError
//	@Failure		413			{object}	errors.APIError
//	@Failure		500			{object}	errors.APIError
//	@Failure		502			{object}	errors.APIError
//	@Failure		503			{object}	errors.APIError
//	@Failure		504			{object}	errors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/sql-query-batches [post]
func (h *CanonicalDataProxyHandler) CreateSQLQueryBatch(c *gin.Context) {
	start, rec := beginCanonicalDataProxyRequest(c, "canonical_sql_query_batch")
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	orgID, clusterID, ok := parseCanonicalWarehouseScope(c, rec)
	if !ok {
		return
	}
	if !requireNoCanonicalQuery(c, rec) {
		return
	}

	var input []*dto.CanonicalDataProxyQueryRequest
	if err := decodeCanonicalJSON(c, &input, canonicalBatchQueryBodyBytes); err != nil {
		writeCanonicalDecodeError(c, rec, err)
		return
	}
	if len(input) == 0 || len(input) > canonicalMaxBatchQueries {
		rec.ErrCode = "INVALID_BATCH_SIZE"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "query batch must contain between 1 and 100 items", nil)
		return
	}

	requests := make([]*dto.DataProxyQueryRequest, len(input))
	queryTypes := make([]dto.QueryType, len(input))
	for i, item := range input {
		if item == nil {
			rec.ErrCode = "INVALID_BATCH_ITEM"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", fmt.Sprintf("query batch item %d must not be null", i), nil)
			return
		}
		req, err := buildCanonicalQueryRequest(*item, clusterID)
		if err != nil {
			rec.ErrCode = "INVALID_BATCH_ITEM"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", fmt.Sprintf("query batch item %d is invalid", i), nil)
			return
		}
		queryType, err := validateCanonicalQuery(req)
		if err != nil {
			rec.ErrCode = "INVALID_BATCH_ITEM"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", fmt.Sprintf("query batch item %d is invalid", i), nil)
			return
		}
		if queryType == dto.QueryTypeStatements {
			rec.ErrCode = "STATEMENTS_BATCH_UNSUPPORTED"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "Statements batch queries are not supported", nil)
			return
		}
		requests[i] = req
		queryTypes[i] = queryType
	}
	rec.QueryType = fmt.Sprintf("batch:%d", len(requests))

	target, ok := h.authorizeAndResolveCluster(c, rec, orgID, clusterID)
	if !ok {
		return
	}
	for i := range requests {
		if err := applyCanonicalQueryTarget(requests[i], queryTypes[i], target); err != nil {
			h.writeCanonicalRoutingError(c, rec, queryTypes[i], target, err)
			return
		}
	}

	h.executeCanonicalBatchQuery(c, rec, requests)
}

// ListTableSchemas returns schemas for standard logical tables on one cluster.
//
//	@Summary		List cluster table schemas
//	@ID			canonical-list-table-schemas
//	@Tags			data-proxy-sql
//	@Produce		json
//	@Param			orgId		path		string	true	"TiDB Cloud/DBAAS org ID"
//	@Param			clusterId	path		string	true	"Numeric warehouse cluster ID"
//	@Param			tables		query		string	true	"Comma-separated standard logical table names"
//	@Success		200			{object}	dto.BatchSchemaResponse
//	@Failure		400			{object}	errors.APIError
//	@Failure		401			{object}	errors.APIError
//	@Failure		403			{object}	errors.APIError
//	@Failure		404			{object}	errors.APIError
//	@Failure		500			{object}	errors.APIError
//	@Failure		502			{object}	errors.APIError
//	@Failure		503			{object}	errors.APIError
//	@Failure		504			{object}	errors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/table-schemas [get]
func (h *CanonicalDataProxyHandler) ListTableSchemas(c *gin.Context) {
	start, rec := beginCanonicalDataProxyRequest(c, "canonical_table_schemas")
	rec.QueryType = "schema"
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	orgID, clusterID, ok := parseCanonicalWarehouseScope(c, rec)
	if !ok {
		return
	}
	if !requireNoCanonicalBody(c, rec) {
		return
	}
	tables, queryType, err := parseCanonicalSchemaTables(c)
	if err != nil {
		rec.ErrCode = "INVALID_SCHEMA_REQUEST"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", err.Error(), nil)
		return
	}

	target, ok := h.authorizeAndResolveCluster(c, rec, orgID, clusterID)
	if !ok {
		return
	}
	dataSource, err := canonicalDataSourceForQueryType(target, queryType)
	if err != nil {
		h.writeCanonicalRoutingError(c, rec, queryType, target, err)
		return
	}

	options := dto.DataProxyQueryOptions{
		DataSource:     dataSource,
		Region:         strings.TrimSpace(target.Region),
		Vendor:         strings.ToLower(strings.TrimSpace(target.Provider)),
		ClusterId:      clusterID,
		ClinicTenantId: target.ClinicTenantID,
	}
	schemas := make([]dto.SchemaResponse, 0, len(tables))
	for _, logicalTable := range tables {
		physicalTable := fmt.Sprintf("%s_%s", logicalTable, clusterID)
		result, err := h.service.GetSchemaForTable(c.Request.Context(), physicalTable, &options)
		if err != nil {
			writeCanonicalExecutionError(c, rec, err, "SCHEMA_FETCH_FAILED", "failed to get table schema")
			return
		}
		if result == nil {
			rec.ErrCode = "INVALID_SCHEMA_RESPONSE"
			writeResourceError(c, http.StatusBadGateway, "BadGateway", "schema service returned an invalid response", nil)
			return
		}
		responseSchema := *result
		responseSchema.Table = logicalTable
		schemas = append(schemas, responseSchema)
	}

	c.JSON(http.StatusOK, dto.BatchSchemaResponse{Schemas: schemas, Count: len(schemas)})
}

// Capabilities returns service-level Data Proxy capabilities.
//
//	@Summary		Get Data Proxy capabilities
//	@ID			canonical-get-capabilities
//	@Tags			data-proxy
//	@Produce		json
//	@Success		200	{object}	dto.DataProxyCapabilitiesResponse
//	@Failure		400	{object}	errors.APIError
//	@Failure		401	{object}	errors.APIError
//	@Failure		403	{object}	errors.APIError
//	@Failure		503	{object}	errors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/capabilities [get]
func (h *CanonicalDataProxyHandler) Capabilities(c *gin.Context) {
	start, rec := beginCanonicalDataProxyRequest(c, "canonical_capabilities")
	rec.QueryType = "info"
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	if !requireNoCanonicalQuery(c, rec) || !requireNoCanonicalBody(c, rec) {
		return
	}
	if h == nil || h.service == nil {
		rec.ErrCode = "DATA_PROXY_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "Data Proxy service is temporarily unavailable", nil)
		return
	}
	c.JSON(http.StatusOK, dto.DataProxyCapabilitiesResponse{
		Service:          "Clinic Data Proxy",
		Version:          "1.0.0",
		SupportedVendors: h.service.GetSupportedVendors(),
	})
}

// RegisterRoutes adds only canonical aliases. Legacy routes remain owned by
// DataProxyHandler and are intentionally neither redirected nor modified.
func (h *CanonicalDataProxyHandler) RegisterRoutes(r *gin.Engine) {
	group := r.Group("/data-proxy/api/v1")
	group.POST("/orgs/:orgId/clusters/:clusterId/sql-queries", h.CreateSQLQuery)
	group.POST("/orgs/:orgId/clusters/:clusterId/sql-query-batches", h.CreateSQLQueryBatch)
	group.GET("/orgs/:orgId/clusters/:clusterId/table-schemas", h.ListTableSchemas)
	group.GET("/capabilities", h.Capabilities)
}

func beginCanonicalDataProxyRequest(c *gin.Context, endpoint string) (time.Time, *observability.DataProxyRecord) {
	return time.Now(), &observability.DataProxyRecord{
		Endpoint: endpoint,
		Caller:   observability.CallerFromGin(c),
	}
}

func requireNoCanonicalQuery(c *gin.Context, rec *observability.DataProxyRecord) bool {
	if c.Request.URL.RawQuery == "" {
		return true
	}
	rec.ErrCode = "UNSUPPORTED_QUERY_PARAMETERS"
	writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "query parameters are not supported", nil)
	return false
}

func requireNoCanonicalBody(c *gin.Context, rec *observability.DataProxyRecord) bool {
	if c.Request.ContentLength == 0 && len(c.Request.TransferEncoding) == 0 {
		return true
	}
	rec.ErrCode = "UNSUPPORTED_REQUEST_BODY"
	writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "request body is not supported", nil)
	return false
}

func parseCanonicalWarehouseScope(c *gin.Context, rec *observability.DataProxyRecord) (uint64, string, bool) {
	orgID, ok := parseOrgID(c, rec)
	if !ok {
		return 0, "", false
	}
	clusterID := c.Param("clusterId")
	if len(clusterID) == 0 || len(clusterID) > canonicalMaxClusterIDBytes || validateClusterId(clusterID) != nil {
		rec.ErrCode = "INVALID_CLUSTER_ID"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "clusterId must contain between 1 and 255 digits", nil)
		return 0, "", false
	}
	return orgID, clusterID, true
}

func decodeCanonicalJSON(c *gin.Context, target any, maxBytes int64) error {
	if c.Request.Body == nil {
		return io.EOF
	}
	c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxBytes)
	decoder := json.NewDecoder(c.Request.Body)
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(target); err != nil {
		return err
	}
	var trailing any
	if err := decoder.Decode(&trailing); !stderrors.Is(err, io.EOF) {
		if err == nil {
			return errCanonicalTrailingJSON
		}
		return err
	}
	return nil
}

func writeCanonicalDecodeError(c *gin.Context, rec *observability.DataProxyRecord, err error) {
	var maxBytesError *http.MaxBytesError
	if stderrors.As(err, &maxBytesError) {
		rec.ErrCode = "PAYLOAD_TOO_LARGE"
		writeResourceError(c, http.StatusRequestEntityTooLarge, "PayloadTooLarge", "request body exceeds the allowed size", nil)
		return
	}
	rec.ErrCode = "INVALID_REQUEST"
	writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "request body must be one JSON value containing only documented fields", nil)
}

func buildCanonicalQueryRequest(input dto.CanonicalDataProxyQueryRequest, clusterID string) (*dto.DataProxyQueryRequest, error) {
	if strings.TrimSpace(input.SQL) == "" {
		return nil, stderrors.New("sql must not be empty")
	}
	if len(input.SQL) > canonicalMaxSQLBytes {
		return nil, fmt.Errorf("sql must not exceed %d bytes", canonicalMaxSQLBytes)
	}
	if input.Timeout < 0 || input.Timeout > canonicalMaxTimeoutSeconds {
		return nil, fmt.Errorf("timeout must be between 0 and %d seconds", canonicalMaxTimeoutSeconds)
	}
	timeout := canonicalDefaultQueryTimeout
	if input.Timeout > 0 {
		timeout = time.Duration(input.Timeout) * time.Second
	}
	return &dto.DataProxyQueryRequest{
		SQL:       input.SQL,
		ClusterId: clusterID,
		Timeout:   timeout,
	}, nil
}

func validateCanonicalQuery(req *dto.DataProxyQueryRequest) (dto.QueryType, error) {
	tableRefs := engine.ExtractTableReferences(req.SQL)
	if len(tableRefs) == 0 {
		return "", stderrors.New("sql must reference a supported logical table")
	}
	for _, ref := range tableRefs {
		if !engine.IsStandardTable(ref.Table) {
			return "", stderrors.New("sql references an unsupported table")
		}
		database := strings.TrimSpace(ref.Database)
		if database != "" && database != "clinic_data_proxy" && database != "clinic-data-proxy" {
			return "", stderrors.New("sql must use the canonical Data Proxy database")
		}
	}
	inferredType, err := engine.InferCompatibleQueryType(tableRefs)
	if err != nil {
		return "", stderrors.New("sql combines incompatible data-source families")
	}
	queryType := dto.QueryType(inferredType)
	if queryType == "" {
		return "", stderrors.New("sql must reference a supported logical table")
	}
	if queryType == dto.QueryTypeStatements {
		if err := engine.ValidateStatementsTableReferences(tableRefs, req.ClusterId); err != nil {
			return "", stderrors.New("statements must use the canonical logical table namespace")
		}
	}
	if err := dialect.ValidatePartitionFiltersForAllTableScans(req.SQL, []string{dto.DefaultPartitionColumn}); err != nil {
		return "", stderrors.New("sql must include a valid date partition filter")
	}
	return queryType, nil
}

func (h *CanonicalDataProxyHandler) authorizeAndResolveCluster(
	c *gin.Context,
	rec *observability.DataProxyRecord,
	orgID uint64,
	clusterID string,
) (*clinic_internal_api.ClusterDataURLs, bool) {
	if h == nil || h.service == nil || h.ownershipReader == nil || h.targetReader == nil {
		rec.ErrCode = "DATA_PROXY_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "Data Proxy service is temporarily unavailable", nil)
		return nil, false
	}

	cluster, err := h.ownershipReader.GetByOrgAndClusterID(c.Request.Context(), orgID, clusterID)
	if err != nil {
		writeCanonicalOwnershipError(c, rec, err)
		return nil, false
	}
	expectedOrgID := strconv.FormatUint(orgID, 10)
	if cluster == nil || cluster.ClusterID != clusterID || cluster.OrgID != expectedOrgID {
		rec.ErrCode = "OWNERSHIP_IDENTITY_MISMATCH"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "cluster ownership service returned an invalid response", nil)
		return nil, false
	}

	target, err := h.targetReader.GetCanonicalClusterDataURLs(c.Request.Context(), clusterID)
	if err != nil {
		writeCanonicalTargetError(c, rec, err)
		return nil, false
	}
	if target == nil || target.ClusterID != clusterID || target.OrgID != expectedOrgID {
		rec.ErrCode = "TARGET_IDENTITY_MISMATCH"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "cluster target service returned an invalid response", nil)
		return nil, false
	}
	rec.Provider = strings.ToLower(strings.TrimSpace(target.Provider))
	rec.Region = strings.TrimSpace(target.Region)
	rec.DeployTypeV2 = target.DeployTypeV2
	deployType := strings.ToLower(strings.TrimSpace(target.DeployType))
	deployTypeV2 := strings.ToLower(strings.TrimSpace(target.DeployTypeV2))
	if deployType == "byoc" || deployType == "byoc-host" ||
		deployTypeV2 == "byoc" || deployTypeV2 == "byoc resource pool" {
		rec.ErrCode = "UNSUPPORTED_DEPLOY_TYPE"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "Data Proxy warehouse queries are not supported for BYOC clusters", nil)
		return nil, false
	}
	return target, true
}

func writeCanonicalOwnershipError(c *gin.Context, rec *observability.DataProxyRecord, err error) {
	switch {
	case stderrors.Is(err, clustercontext.ErrNotFound):
		rec.ErrCode = "CLUSTER_NOT_FOUND"
		writeResourceError(c, http.StatusNotFound, "NotFound", "cluster not found", nil)
	case stderrors.Is(err, clustercontext.ErrUnavailable):
		rec.ErrCode = "METADATA_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "cluster metadata is temporarily unavailable", nil)
	case stderrors.Is(err, clustercontext.ErrTimeout):
		rec.ErrCode = "METADATA_TIMEOUT"
		writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "cluster metadata request timed out", nil)
	default:
		rec.ErrCode = "METADATA_QUERY_FAILED"
		writeResourceError(c, http.StatusInternalServerError, "InternalError", "failed to query cluster metadata", nil)
	}
}

func writeCanonicalTargetError(c *gin.Context, rec *observability.DataProxyRecord, err error) {
	switch {
	case stderrors.Is(err, context.Canceled):
		rec.ErrCode = "TARGET_REQUEST_CANCELED"
		writeResourceError(c, clientClosedStatus, "Canceled", "cluster target request was canceled", nil)
		return
	case stderrors.Is(err, context.DeadlineExceeded):
		rec.ErrCode = "TARGET_TIMEOUT"
		writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "cluster target request timed out", nil)
		return
	}

	var networkError net.Error
	if stderrors.As(err, &networkError) {
		if networkError.Timeout() {
			rec.ErrCode = "TARGET_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "cluster target request timed out", nil)
			return
		}
		rec.ErrCode = "TARGET_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "cluster target service is temporarily unavailable", nil)
		return
	}

	var upstreamError *clinic_internal_api.UpstreamError
	if stderrors.As(err, &upstreamError) {
		switch upstreamError.StatusCode {
		case http.StatusRequestTimeout, http.StatusGatewayTimeout:
			rec.ErrCode = "TARGET_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "cluster target request timed out", nil)
		case http.StatusNotFound, http.StatusTooManyRequests, http.StatusInternalServerError,
			http.StatusBadGateway, http.StatusServiceUnavailable:
			rec.ErrCode = "TARGET_UNAVAILABLE"
			writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "cluster target service is temporarily unavailable", nil)
		default:
			rec.ErrCode = "TARGET_BAD_RESPONSE"
			writeResourceError(c, http.StatusBadGateway, "BadGateway", "cluster target service returned an invalid response", nil)
		}
		return
	}

	rec.ErrCode = "TARGET_UNAVAILABLE"
	writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "cluster target service is temporarily unavailable", nil)
}

func canonicalDataSourceForQueryType(target *clinic_internal_api.ClusterDataURLs, queryType dto.QueryType) (string, error) {
	var dataSource string
	switch queryType {
	case dto.QueryTypeSlowQuery:
		dataSource = target.SlowQueryS3URL
	case dto.QueryTypeStatements:
		dataSource = target.StatementsS3URL
	case dto.QueryTypeTopSQL:
		dataSource = target.TopSQLS3URL
	case dto.QueryTypeLogStats:
		dataSource = target.LogStatsS3URL
	default:
		return "", stderrors.New("unsupported query type")
	}
	if strings.TrimSpace(dataSource) == "" {
		return "", errCanonicalDataSourceNotConfigured
	}
	return strings.TrimSpace(dataSource), nil
}

func applyCanonicalQueryTarget(
	req *dto.DataProxyQueryRequest,
	queryType dto.QueryType,
	target *clinic_internal_api.ClusterDataURLs,
) error {
	if queryType == dto.QueryTypeStatements && !statementsProviderSupported(target.Provider) {
		return fmt.Errorf("statements queries are not supported for provider %q", strings.TrimSpace(target.Provider))
	}
	dataSource, err := canonicalDataSourceForQueryType(target, queryType)
	if err != nil {
		return err
	}
	req.Options = dto.DataProxyQueryOptions{
		DataSource:     dataSource,
		Region:         strings.TrimSpace(target.Region),
		Vendor:         strings.ToLower(strings.TrimSpace(target.Provider)),
		ClusterId:      req.ClusterId,
		ClinicTenantId: target.ClinicTenantID,
	}
	return nil
}

func (h *CanonicalDataProxyHandler) writeCanonicalRoutingError(
	c *gin.Context,
	rec *observability.DataProxyRecord,
	queryType dto.QueryType,
	target *clinic_internal_api.ClusterDataURLs,
	err error,
) {
	if queryType == dto.QueryTypeStatements && target != nil && !statementsProviderSupported(target.Provider) {
		rec.ErrCode = "UNSUPPORTED_PROVIDER"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "Statements queries are not supported for this cluster provider", nil)
		return
	}
	if stderrors.Is(err, errCanonicalDataSourceNotConfigured) {
		rec.ErrCode = "DATASOURCE_NOT_CONFIGURED"
		writeResourceError(c, http.StatusBadRequest, "FailedPrecondition", "the requested data source is not configured for this cluster", nil)
		return
	}
	rec.ErrCode = "QUERY_ROUTING_FAILED"
	writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "the requested query type is not supported", nil)
}

func (h *CanonicalDataProxyHandler) executeCanonicalQuery(
	c *gin.Context,
	rec *observability.DataProxyRecord,
	req *dto.DataProxyQueryRequest,
) {
	result, err := h.service.Query(c.Request.Context(), req)
	if err != nil {
		writeCanonicalExecutionError(c, rec, err, "QUERY_FAILED", "query execution failed")
		return
	}
	if result == nil {
		rec.ErrCode = "INVALID_QUERY_RESPONSE"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "query service returned an invalid response", nil)
		return
	}
	rec.Engine = result.Metadata.Engine
	if rec.Provider == "" {
		rec.Provider = result.Metadata.Vendor
	}
	if rec.Region == "" {
		rec.Region = result.Metadata.Region
	}
	c.JSON(http.StatusOK, result)
}

func (h *CanonicalDataProxyHandler) executeCanonicalBatchQuery(
	c *gin.Context,
	rec *observability.DataProxyRecord,
	requests []*dto.DataProxyQueryRequest,
) {
	results, err := h.service.BatchQuery(c.Request.Context(), requests)
	if err != nil {
		writeCanonicalExecutionError(c, rec, err, "BATCH_QUERY_FAILED", "batch query execution failed")
		return
	}
	if len(results) > 0 && results[0] != nil {
		rec.Engine = results[0].Metadata.Engine
		if rec.Provider == "" {
			rec.Provider = results[0].Metadata.Vendor
		}
		if rec.Region == "" {
			rec.Region = results[0].Metadata.Region
		}
	}
	c.JSON(http.StatusOK, dto.DataProxyBatchQueryResponse{Results: results, Count: len(results)})
}

func writeCanonicalExecutionError(
	c *gin.Context,
	rec *observability.DataProxyRecord,
	err error,
	errCode string,
	message string,
) {
	switch {
	case isUnknownFieldError(err):
		rec.ErrCode = unknownFieldObservabilityCode(err)
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", unknownFieldMessage(err), unknownFieldDetailsForError(err))
	case stderrors.Is(err, context.Canceled):
		rec.ErrCode = errCode + "_CANCELED"
		writeResourceError(c, clientClosedStatus, "Canceled", "request was canceled", nil)
	case stderrors.Is(err, context.DeadlineExceeded):
		rec.ErrCode = errCode + "_TIMEOUT"
		writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "request timed out", nil)
	case isValidationError(err):
		rec.ErrCode = errCode + "_VALIDATION"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "query validation failed", nil)
	case isAuthenticationError(err):
		rec.ErrCode = errCode + "_CREDENTIALS"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "query service credentials are temporarily unavailable", nil)
	default:
		rec.ErrCode = errCode
		writeResourceError(c, http.StatusInternalServerError, "InternalError", message, nil)
	}
}

func unknownFieldObservabilityCode(err error) string {
	if _, ok := allBatchFailuresAreUnknownFields(err); ok {
		return "BATCH_QUERY_UNKNOWN_FIELD"
	}
	return "QUERY_UNKNOWN_FIELD"
}

const (
	unknownFieldReason        = "UNKNOWN_FIELD"
	unknownFieldSingleMessage = "query references fields that cannot be resolved in the target cluster schema"
	unknownFieldBatchMessage  = "one or more queries reference fields that cannot be resolved in the target cluster schema"
)

func isUnknownFieldError(err error) bool {
	_, ok := engine.AsUnknownFieldError(err)
	if ok {
		return true
	}
	_, ok = allBatchFailuresAreUnknownFields(err)
	return ok
}

func unknownFieldMessage(err error) string {
	if _, ok := allBatchFailuresAreUnknownFields(err); ok {
		return unknownFieldBatchMessage
	}
	return unknownFieldSingleMessage
}

func unknownFieldDetailsForError(err error) any {
	if queries, ok := allBatchFailuresAreUnknownFields(err); ok {
		return errors.UnknownFieldBatchDetails{Reason: unknownFieldReason, Retryable: false, Queries: queries}
	}
	unknown, _ := engine.AsUnknownFieldError(err)
	if unknown == nil {
		return nil
	}
	return errors.UnknownFieldDetails{Reason: unknownFieldReason, Fields: unknown.Fields, Retryable: false}
}

func allBatchFailuresAreUnknownFields(err error) ([]errors.UnknownFieldBatchQuery, bool) {
	var batchErr *service.BatchQueryError
	if !stderrors.As(err, &batchErr) || batchErr == nil || len(batchErr.Failures) == 0 {
		return nil, false
	}
	queries := make([]errors.UnknownFieldBatchQuery, 0, len(batchErr.Failures))
	for _, failure := range batchErr.Failures {
		unknown, ok := engine.AsUnknownFieldError(failure.Err)
		if !ok {
			return nil, false
		}
		queries = append(queries, errors.UnknownFieldBatchQuery{Index: failure.Index, Fields: unknown.Fields})
	}
	return queries, true
}

func parseCanonicalSchemaTables(c *gin.Context) ([]string, dto.QueryType, error) {
	query, err := url.ParseQuery(c.Request.URL.RawQuery)
	if err != nil {
		return nil, "", stderrors.New("query parameters are malformed")
	}
	for key := range query {
		if key != "tables" {
			return nil, "", fmt.Errorf("unsupported query parameter %q", key)
		}
	}
	tableValues, ok := query["tables"]
	if !ok || len(tableValues) != 1 {
		return nil, "", stderrors.New("tables must be provided exactly once")
	}
	rawTables := strings.Split(tableValues[0], ",")
	if len(rawTables) == 0 || len(rawTables) > canonicalMaxSchemaTables {
		return nil, "", fmt.Errorf("tables must contain between 1 and %d entries", canonicalMaxSchemaTables)
	}
	tables := make([]string, 0, len(rawTables))
	queryType := dto.QueryType("")
	for _, rawTable := range rawTables {
		table := strings.TrimSpace(rawTable)
		if table == "" || !engine.IsStandardTable(table) {
			return nil, "", stderrors.New("tables contains an unsupported logical table")
		}
		if table == "statements" {
			return nil, "", stderrors.New("statements schema is not exposed")
		}
		tableQueryType := dto.QueryType(engine.InferQueryType(table))
		if tableQueryType == "" {
			return nil, "", stderrors.New("tables contains an unsupported logical table")
		}
		if queryType == "" {
			queryType = tableQueryType
		} else if tableQueryType != queryType {
			return nil, "", stderrors.New("tables must use one data-source family")
		}
		tables = append(tables, table)
	}
	if queryType == "" {
		return nil, "", stderrors.New("tables must contain a supported logical table")
	}
	return tables, queryType, nil
}

// Keep the canonical OpenAPI error type reachable by the annotation parser.
var _ errors.APIError