package handlers

import (
	"context"
	"fmt"
	"net/http"
	"strings"
	"time"

	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	"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/engine"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/observability"
	"go.uber.org/zap"

	"github.com/gin-gonic/gin"
)

// DataProxyHandler Data Proxy HTTP 处理器
type DataProxyHandler struct {
	service      legacyDataProxyService
	config       *config.DataProxyConfig
	clinicClient *clinic_internal_api.Client
}

type legacyDataProxyService 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
	Health(context.Context) error
}

const (
	cloudDataProxyQueryPath      = "/api/v1/orgs/:orgId/data-proxy/clusters/:clusterId/query"
	defaultDataProxyQueryTimeout = 2 * time.Minute
)

// NewDataProxyHandler 创建处理器实例
func NewDataProxyHandler(svc legacyDataProxyService, cfg *config.DataProxyConfig) *DataProxyHandler {
	return &DataProxyHandler{
		service:      svc,
		config:       cfg,
		clinicClient: clinic_internal_api.NewClientWithTimeout(cfg.ClinicAPIBase, defaultDataProxyQueryTimeout),
	}
}

// Query handles the internal Data Proxy query endpoint.
// POST /data-proxy/query
func (h *DataProxyHandler) Query(c *gin.Context) {
	h.query(c, "")
}

// QueryByClusterPath handles the TiDB Cloud query endpoint. The path cluster ID
// is authoritative, so callers cannot select another cluster through the body.
func (h *DataProxyHandler) QueryByClusterPath(c *gin.Context) {
	h.query(c, c.Param("clusterId"))
}

func (h *DataProxyHandler) query(c *gin.Context, pathClusterID string) {
	start := time.Now()
	rec := &observability.DataProxyRecord{Endpoint: "query", Caller: observability.CallerFromGin(c)}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	var req dto.DataProxyQueryRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		rec.ErrCode = "INVALID_REQUEST"
		c.JSON(http.StatusBadRequest, gin.H{
			"error": fmt.Sprintf("invalid request: %v", err),
		})
		return
	}

	ctx := c.Request.Context()
	bodyClusterID := req.ClusterId
	if err := overrideClusterIDWithPath(&req, pathClusterID); err != nil {
		rec.ErrCode = "INVALID_PARAMETER"
		log.Warn(ctx, "Invalid cluster ID in request path", zap.String("clusterId", pathClusterID), zap.Error(err))
		c.JSON(http.StatusBadRequest, gin.H{
			"error": gin.H{
				"code":    "INVALID_PARAMETER",
				"message": "Invalid parameter: cluster_id must contain only digits",
				"details": gin.H{
					"field":  "cluster_id",
					"value":  pathClusterID,
					"reason": err.Error(),
				},
			},
		})
		return
	}
	if pathClusterID != "" && bodyClusterID != "" && bodyClusterID != pathClusterID {
		log.Info(ctx, "Overriding request clusterId with path cluster ID",
			zap.String("bodyClusterId", bodyClusterID),
			zap.String("pathClusterId", pathClusterID),
		)
	}

	// 1. 校验 clusterId 格式（如果提供了的话）
	if req.ClusterId != "" {
		if err := validateClusterId(req.ClusterId); err != nil {
			rec.ErrCode = "INVALID_PARAMETER"
			log.Warn(ctx, "Invalid clusterId format", zap.String("clusterId", req.ClusterId), zap.Error(err))
			c.JSON(http.StatusBadRequest, gin.H{
				"error": gin.H{
					"code":    "INVALID_PARAMETER",
					"message": "Invalid parameter: clusterId must contain only digits",
					"details": gin.H{
						"field":  "clusterId",
						"value":  req.ClusterId,
						"reason": err.Error(),
					},
				},
			})
			return
		}
	}

	// 2. 从 SQL 提取表引用
	tableRefs := engine.ExtractTableReferences(req.SQL)

	// 3. 验证：只支持标准表
	for _, ref := range tableRefs {
		if !engine.IsStandardTable(ref.Table) {
			rec.ErrCode = "UNSUPPORTED_TABLE"
			c.JSON(http.StatusBadRequest, gin.H{
				"error": engine.UnsupportedStandardTableError(fmt.Sprintf("%s.%s", ref.Database, ref.Table)),
			})
			return
		}
	}

	inferredType, err := engine.InferCompatibleQueryType(tableRefs)
	if err != nil {
		rec.ErrCode = "MIXED_QUERY_TYPES"
		c.JSON(http.StatusBadRequest, gin.H{
			"error": gin.H{
				"code":    "MIXED_QUERY_TYPES",
				"message": err.Error(),
			},
		})
		return
	}
	queryType := dto.QueryType(inferredType)
	rec.QueryType = string(queryType)
	log.Info(ctx, "Successfully extracted table references", zap.Any("tableRefs", tableRefs), zap.String("queryType", string(queryType)))
	if queryType == dto.QueryTypeStatements {
		if err := engine.ValidateStatementsTableReferences(tableRefs, req.ClusterId); err != nil {
			rec.ErrCode = "INVALID_STATEMENTS_NAMESPACE"
			c.JSON(http.StatusBadRequest, gin.H{
				"error": gin.H{
					"code":    "INVALID_STATEMENTS_NAMESPACE",
					"message": err.Error(),
				},
			})
			return
		}
		if req.ClusterId == "" {
			rec.ErrCode = "CLUSTER_ID_REQUIRED"
			c.JSON(http.StatusBadRequest, gin.H{
				"error": gin.H{
					"code":    "CLUSTER_ID_REQUIRED",
					"message": "clusterId is required for Statements queries",
				},
			})
			return
		}
	}

	// 4. 如果没有提供 dataSource，通过 clusterId 自动获取（使用推断的 queryType）。
	// LogStats 即使提供了 dataSource，也要获取 tenantId 用于 cluster-scoped view。
	// if !config.IsLocal() && (req.Options.DataSource == "" && req.ClusterId != "" && queryType != "") {
	if (req.Options.DataSource == "" || queryType == dto.QueryTypeLogStats || queryType == dto.QueryTypeStatements) && req.ClusterId != "" && queryType != "" {
		log.Info(ctx, "Attempting to auto-fill dataSource from clusterId",
			zap.String("clusterId", req.ClusterId),
			zap.String("inferredQueryType", string(queryType)),
		)

		//  Fetch cluster Loki URL and filters from clinic API
		clusterInfo, err := h.clinicClient.GetClusterDataURLs(ctx, req.ClusterId)
		if err != nil {
			rec.ErrCode = "CLUSTER_INFO_FETCH_FAILED"
			log.Error(ctx, "failed to fetch cluster data urls",
				zap.String("cluster_id", req.ClusterId),
				zap.Error(err))
			c.JSON(http.StatusInternalServerError, gin.H{
				"error": fmt.Sprintf("failed to fetch cluster info: %v", err),
			})
			return
		}
		rec.Provider = clusterInfo.Provider
		rec.DeployTypeV2 = clusterInfo.DeployTypeV2
		if queryType == dto.QueryTypeStatements && !statementsProviderSupported(clusterInfo.Provider) {
			rec.ErrCode = "UNSUPPORTED_PROVIDER"
			writeStatementsProviderUnsupported(c, clusterInfo.Provider)
			return
		}

		// if clusterURLs.LokiURL == "" {
		// 	log.Warn(ctx, "loki url not available for cluster", zap.String("cluster_id", req.ClusterId))
		// 	c.JSON(http.StatusNotFound, gin.H{
		// 		"error": "loki url not available for this cluster",
		// 	})
		// 	return
		// }

		/*
			// 调用内部 API 获取集群信息
			clusterInfo, err := h.fetchClusterO11yInfo(ctx, req.ClusterId)
			if err != nil {
				// 根据错误类型返回不同的 HTTP 状态码
				if apiErr, ok := err.(*ClusterAPIError); ok {
					log.Warn(ctx, "Cluster API error", zap.String("clusterId", req.ClusterId), zap.Int("statusCode", apiErr.StatusCode), zap.Error(err))
					c.JSON(apiErr.StatusCode, gin.H{
						"error": gin.H{
							"code":    "CLUSTER_NOT_FOUND",
							"message": fmt.Sprintf("Cluster '%s' not found, please check if the clusterId is correct", req.ClusterId),
							"details": gin.H{
								"field": "clusterId",
								"value": req.ClusterId,
							},
						},
					})
				} else {
					// 其他未知错误（网络超时、解析失败等）
					log.Error(ctx, "Failed to fetch cluster info", zap.String("clusterId", req.ClusterId), zap.Error(err))
					c.JSON(http.StatusInternalServerError, gin.H{
						"error": gin.H{
							"code":    "INTERNAL_ERROR",
							"message": "Service temporarily unavailable, please try again later",
						},
					})
				}
				return
			}
		*/
		log.Info(ctx, "Successfully fetched cluster info",
			zap.String("clusterId", req.ClusterId),
			zap.String("clusterName", clusterInfo.ClusterName),
			zap.String("region", clusterInfo.Region),
		)
		req.Options.ClinicTenantId = clusterInfo.ClinicTenantID

		// 根据推断的 queryType 填充 dataSource
		switch queryType {
		case dto.QueryTypeSlowQuery:
			if req.Options.DataSource == "" {
				req.Options.DataSource = clusterInfo.SlowQueryS3URL
			}
		case dto.QueryTypeStatements:
			// Statements is server-resolved only. Ignore caller-provided source,
			// vendor, and region so authorization cannot be bypassed.
			overrideStatementsDataSource(&req.Options, clusterInfo)
		case dto.QueryTypeTopSQL:
			if req.Options.DataSource == "" {
				req.Options.DataSource = clusterInfo.TopSQLS3URL
			}
		case dto.QueryTypeLogStats:
			if req.Options.DataSource == "" {
				req.Options.DataSource = clusterInfo.LogStatsS3URL
			}
		default:
			rec.ErrCode = "QUERY_TYPE_INFER_FAILED"
			log.Error(ctx, "Unable to infer query type from table name",
				zap.String("queryType", string(queryType)),
			)
			c.JSON(http.StatusBadRequest, gin.H{
				"error": "unable to infer query type from table name, please use standard table names",
			})
			return
		}

		// 验证 dataSource 是否为空
		if req.Options.DataSource == "" {
			rec.ErrCode = "DATASOURCE_NOT_CONFIGURED"
			message := fmt.Sprintf("Cluster '%s' does not have %s data source configured. Please contact administrator or specify dataSource manually.", req.ClusterId, queryType)
			hint := "The cluster's S3 URL is not configured in the system"
			if queryType == dto.QueryTypeStatements {
				message = fmt.Sprintf("Cluster '%s' does not have a persisted Statements data source configured.", req.ClusterId)
				hint = "Persistent Statements must be enabled on a supported Premium cluster"
			}
			log.Error(ctx, "DataSource is empty from cluster info",
				zap.String("clusterId", req.ClusterId),
				zap.String("queryType", string(queryType)),
			)
			c.JSON(http.StatusBadRequest, gin.H{
				"error": gin.H{
					"code":    "DATASOURCE_NOT_CONFIGURED",
					"message": message,
					"details": gin.H{
						"clusterId": req.ClusterId,
						"queryType": string(queryType),
						"hint":      hint,
					},
				},
			})
			return
		}

		// 如果 options 中没有指定 region，使用集群的 region
		if req.Options.Region == "" {
			req.Options.Region = clusterInfo.Region
		}
		if req.Options.Vendor == "" {
			req.Options.Vendor = clusterInfo.Provider
		}
		rec.Region = req.Options.Region

		log.Info(ctx, "Auto-filled dataSource from cluster info",
			zap.String("clusterId", req.ClusterId),
			zap.String("queryType", string(queryType)),
			zap.String("region", req.Options.Region),
		)
	}

	// // 5. 参数验证：dataSource 必填（或通过 clusterId 自动获取）
	// if req.DataSource == "" {
	// 	c.JSON(http.StatusBadRequest, gin.H{
	// 		"error": "dataSource is required (or provide clusterId with standard table)",
	// 	})
	// 	return
	// }

	// 设置默认超时
	if req.Timeout == 0 {
		req.Timeout = defaultDataProxyQueryTimeout
	}
	if rec.Region == "" {
		rec.Region = req.Options.Region
	}

	// 执行查询
	result, err := h.service.Query(ctx, &req)
	if err != nil {
		if isUnknownFieldError(err) {
			rec.ErrCode = "QUERY_UNKNOWN_FIELD"
			writeLegacyUnknownFieldError(c, err)
			return
		}
		// 判断是否是验证错误或认证错误，返回 400 Bad Request
		if isValidationError(err) || isAuthenticationError(err) {
			if isAuthenticationError(err) {
				rec.ErrCode = "AUTH_CREDENTIALS"
			} else {
				rec.ErrCode = "QUERY_VALIDATION_FAILED"
			}
			c.JSON(http.StatusBadRequest, gin.H{
				"error": fmt.Sprintf("query failed: %v", err),
			})
			return
		}
		rec.ErrCode = "ENGINE_QUERY_FAILED"
		// 其他错误返回 500
		c.JSON(http.StatusInternalServerError, gin.H{
			"error": fmt.Sprintf("query failed: %v", err),
		})
		return
	}

	if rec.Region == "" {
		rec.Region = result.Metadata.Region
	}
	if rec.Provider == "" {
		rec.Provider = result.Metadata.Vendor
	}
	rec.Engine = result.Metadata.Engine

	c.JSON(http.StatusOK, result)
}

// isValidationError 判断是否是验证错误
func isValidationError(err error) bool {
	if err == nil {
		return false
	}
	errMsg := strings.ToLower(err.Error())
	return strings.Contains(errMsg, "validation failed") ||
		strings.Contains(errMsg, "is required") ||
		strings.Contains(errMsg, "cannot be empty") ||
		strings.Contains(errMsg, "missing partition")
}

// isAuthenticationError 判断是否是认证错误
func isAuthenticationError(err error) bool {
	if err == nil {
		return false
	}
	errMsg := strings.ToLower(err.Error())
	return strings.Contains(errMsg, "aws credentials expired") ||
		strings.Contains(errMsg, "expiredtoken") ||
		strings.Contains(errMsg, "invalidclienttokenid") ||
		strings.Contains(errMsg, "securitytokenexpired") ||
		strings.Contains(errMsg, "unrecognizedclientexception") ||
		strings.Contains(errMsg, "security token included in the request is invalid") ||
		strings.Contains(errMsg, "azure service principal credential") ||
		strings.Contains(errMsg, "authentication failed") ||
		strings.Contains(errMsg, "login failed")
}

// BatchQuery HTTP 批量查询接口
// POST /dataproxy/batch-query
func (h *DataProxyHandler) BatchQuery(c *gin.Context) {
	start := time.Now()
	rec := &observability.DataProxyRecord{Endpoint: "batch_query", Caller: observability.CallerFromGin(c)}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	var requests []*dto.DataProxyQueryRequest
	if err := c.ShouldBindJSON(&requests); err != nil {
		rec.ErrCode = "INVALID_REQUEST"
		c.JSON(http.StatusBadRequest, gin.H{
			"error": fmt.Sprintf("invalid request: %v", err),
		})
		return
	}

	if len(requests) == 0 {
		rec.ErrCode = "EMPTY_BATCH"
		c.JSON(http.StatusBadRequest, gin.H{
			"error": "no queries provided",
		})
		return
	}

	rec.QueryType = fmt.Sprintf("batch:%d", len(requests))

	// 设置默认超时
	for _, req := range requests {
		if req == nil {
			rec.ErrCode = "INVALID_REQUEST"
			c.JSON(http.StatusBadRequest, gin.H{"error": "batch query request cannot be null"})
			return
		}
		for _, ref := range engine.ExtractTableReferences(req.SQL) {
			if engine.IsStatementsTableReference(ref.Table, req.ClusterId) {
				rec.ErrCode = "STATEMENTS_BATCH_UNSUPPORTED"
				c.JSON(http.StatusBadRequest, gin.H{
					"error": gin.H{
						"code":    "STATEMENTS_BATCH_UNSUPPORTED",
						"message": "Statements is supported only through POST /data-proxy/query",
					},
				})
				return
			}
		}
		if req.Timeout == 0 {
			req.Timeout = defaultDataProxyQueryTimeout
		}
	}

	if len(requests) > 0 && requests[0] != nil && requests[0].ClusterId != "" {
		if info, err := h.clinicClient.GetClusterDataURLs(c.Request.Context(), requests[0].ClusterId); err == nil {
			rec.DeployTypeV2 = info.DeployTypeV2
			if rec.Provider == "" {
				rec.Provider = info.Provider
			}
		}
	}

	// 执行批量查询
	results, err := h.service.BatchQuery(c.Request.Context(), requests)
	if err != nil {
		if isUnknownFieldError(err) {
			rec.ErrCode = "BATCH_QUERY_UNKNOWN_FIELD"
			writeLegacyUnknownFieldError(c, err)
			return
		}
		rec.ErrCode = "BATCH_QUERY_FAILED"
		c.JSON(http.StatusInternalServerError, gin.H{
			"error": fmt.Sprintf("batch query failed: %v", err),
		})
		return
	}
	if len(results) > 0 && results[0] != nil {
		rec.Engine = results[0].Metadata.Engine
		rec.Region = results[0].Metadata.Region
		rec.Provider = results[0].Metadata.Vendor
	}

	c.JSON(http.StatusOK, gin.H{
		"results": results,
		"count":   len(results),
	})
}

func writeLegacyUnknownFieldError(c *gin.Context, err error) {
	message := unknownFieldMessage(err)
	requestID := c.Writer.Header().Get("X-Trace-ID")
	c.JSON(http.StatusBadRequest, gin.H{
		"error":     message,
		"code":      "InvalidArgument",
		"requestId": requestID,
		"details":   unknownFieldDetailsForError(err),
	})
}

// Health 健康检查接口
// GET /dataproxy/health
func (h *DataProxyHandler) Health(c *gin.Context) {
	if err := h.service.Health(c.Request.Context()); err != nil {
		c.JSON(http.StatusServiceUnavailable, gin.H{
			"status": "unhealthy",
			"error":  err.Error(),
		})
		return
	}

	c.JSON(http.StatusOK, gin.H{
		"status": "healthy",
	})
}

// Info 服务信息接口
// GET /dataproxy/info
func (h *DataProxyHandler) Info(c *gin.Context) {
	start := time.Now()
	rec := &observability.DataProxyRecord{Endpoint: "info", Caller: observability.CallerFromGin(c), QueryType: "info"}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	vendors := h.service.GetSupportedVendors()

	c.JSON(http.StatusOK, gin.H{
		"service":           "Clinic Data Proxy",
		"version":           "1.0.0",
		"supported_vendors": vendors,
	})
}

// GetSchema 获取表结构接口
// GET /data-proxy/schema
func (h *DataProxyHandler) GetSchema(c *gin.Context) {
	start := time.Now()
	rec := &observability.DataProxyRecord{Endpoint: "schema", Caller: observability.CallerFromGin(c), QueryType: "schema"}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	ctx := c.Request.Context()

	// 手动解析查询参数（Gin 的 ShouldBindQuery 不支持嵌套结构）
	tablesParam := c.Query("tables")
	if tablesParam == "" {
		rec.ErrCode = "TABLES_REQUIRED"
		c.JSON(http.StatusBadRequest, gin.H{
			"error": "tables parameter is required (comma-separated list)",
		})
		return
	}

	// 解析表名列表（逗号分隔）
	tables := strings.Split(tablesParam, ",")
	for i := range tables {
		if !engine.IsStandardTable(tables[i]) {
			rec.ErrCode = "UNSUPPORTED_TABLE"
			c.JSON(http.StatusBadRequest, gin.H{
				"error": engine.UnsupportedStandardTableError(tables[i]),
			})
			return
		}
		if tables[i] == "statements" {
			rec.ErrCode = "STATEMENTS_SCHEMA_UNSUPPORTED"
			c.JSON(http.StatusBadRequest, gin.H{
				"error": gin.H{
					"code":    "STATEMENTS_SCHEMA_UNSUPPORTED",
					"message": "Statements schema is not exposed; query clinic_data_proxy.statements with POST /data-proxy/query",
				},
			})
			return
		}
	}

	// 解析其他参数
	clusterId := c.Query("clusterId")
	// 验证 clusterId 必填
	if clusterId == "" {
		rec.ErrCode = "CLUSTER_ID_REQUIRED"
		c.JSON(http.StatusBadRequest, gin.H{
			"error": "clusterId parameter is required",
		})
		return
	}

	// 校验 clusterId 格式
	if err := validateClusterId(clusterId); err != nil {
		rec.ErrCode = "INVALID_PARAMETER"
		log.Warn(ctx, "Invalid clusterId format", zap.String("clusterId", clusterId), zap.Error(err))
		c.JSON(http.StatusBadRequest, gin.H{
			"error": gin.H{
				"code":    "INVALID_PARAMETER",
				"message": "Invalid parameter: clusterId must contain only digits",
				"details": gin.H{
					"field":  "clusterId",
					"value":  clusterId,
					"reason": err.Error(),
				},
			},
		})
		return
	}
	dataSource := c.Query("dataSource") // 支持顶层参数
	if dataSource == "" {
		dataSource = c.Query("options.dataSource") // 支持嵌套参数
	}
	region := c.Query("region") // 支持顶层参数
	if region == "" {
		region = c.Query("options.region") // 支持嵌套参数
	}
	vendor := c.Query("vendor")
	if vendor == "" {
		vendor = c.Query("options.vendor")
	}

	// 构建请求对象
	req := dto.SchemaRequest{
		Tables:    tables,
		ClusterId: clusterId,
		Options: dto.DataProxyQueryOptions{
			DataSource: dataSource,
			Region:     region,
			Vendor:     vendor,
		},
	}

	schemaQueryType := ""
	for _, table := range req.Tables {
		if inferred := engine.InferQueryType(table); inferred != "" {
			schemaQueryType = inferred
			break
		}
	}

	// 如果没有提供 dataSource，通过 clusterId 自动获取（类似 Query 接口）
	// 只在非 local 环境才自动获取
	if !config.IsLocal() && ((req.Options.DataSource == "" || schemaQueryType == string(dto.QueryTypeLogStats)) && req.ClusterId != "") {
		log.Info(ctx, "Attempting to auto-fill dataSource from clusterId for schema request",
			zap.String("clusterId", req.ClusterId),
		)

		//  Fetch cluster Loki URL and filters from clinic API
		clusterInfo, err := h.clinicClient.GetClusterDataURLs(ctx, req.ClusterId)
		if err != nil {
			rec.ErrCode = "CLUSTER_INFO_FETCH_FAILED"
			log.Error(ctx, "failed to fetch cluster data urls",
				zap.String("cluster_id", req.ClusterId),
				zap.Error(err))
			c.JSON(http.StatusInternalServerError, gin.H{
				"error": fmt.Sprintf("failed to fetch cluster info: %v", err),
			})
			return
		}
		rec.Provider = clusterInfo.Provider
		rec.DeployTypeV2 = clusterInfo.DeployTypeV2

		// clusterInfo, err := h.fetchClusterO11yInfo(ctx, req.ClusterId)

		if err != nil {
			// 根据错误类型返回不同的 HTTP 状态码
			if apiErr, ok := err.(*ClusterAPIError); ok {
				log.Warn(ctx, "Cluster API error", zap.String("clusterId", req.ClusterId), zap.Int("statusCode", apiErr.StatusCode), zap.Error(err))
				c.JSON(apiErr.StatusCode, gin.H{
					"error": gin.H{
						"code":    "CLUSTER_NOT_FOUND",
						"message": fmt.Sprintf("Cluster '%s' not found, please check if the clusterId is correct", req.ClusterId),
						"details": gin.H{
							"field": "clusterId",
							"value": req.ClusterId,
						},
					},
				})
			} else {
				// 其他未知错误（网络超时、解析失败等）
				log.Error(ctx, "Failed to fetch cluster info", zap.String("clusterId", req.ClusterId), zap.Error(err))
				c.JSON(http.StatusInternalServerError, gin.H{
					"error": gin.H{
						"code":    "INTERNAL_ERROR",
						"message": "Service temporarily unavailable, please try again later",
					},
				})
			}
			return
		}

		log.Info(ctx, "Successfully fetched cluster info for schema request",
			zap.String("clusterId", req.ClusterId),
			zap.String("clusterName", clusterInfo.ClusterName),
			zap.String("region", clusterInfo.Region),
		)
		req.Options.ClinicTenantId = clusterInfo.ClinicTenantID

		// 根据表类型选择数据源：TopSQL 用 TopSQLS3URL，LogStats 用 LogStatsS3URL，其他用 SlowQueryS3URL
		if req.Options.DataSource == "" {
			switch dto.QueryType(schemaQueryType) {
			case dto.QueryTypeTopSQL:
				req.Options.DataSource = clusterInfo.TopSQLS3URL
			case dto.QueryTypeLogStats:
				req.Options.DataSource = clusterInfo.LogStatsS3URL
			default:
				req.Options.DataSource = clusterInfo.SlowQueryS3URL
			}
		}

		// 验证 dataSource 是否为空
		if req.Options.DataSource == "" {
			rec.ErrCode = "DATASOURCE_NOT_CONFIGURED"
			log.Error(ctx, "DataSource is empty from cluster info for schema request",
				zap.String("clusterId", req.ClusterId),
			)
			c.JSON(http.StatusBadRequest, gin.H{
				"error": gin.H{
					"code":    "DATASOURCE_NOT_CONFIGURED",
					"message": fmt.Sprintf("Cluster '%s' does not have slow query data source configured. Please contact administrator or specify dataSource manually.", req.ClusterId),
					"details": gin.H{
						"clusterId": req.ClusterId,
						"hint":      "The cluster's slow query S3 URL is not configured in the system",
					},
				},
			})
			return
		}

		// 如果 options 中没有指定 region，使用集群的 region
		if req.Options.Region == "" {
			req.Options.Region = clusterInfo.Region
		}
		if req.Options.Vendor == "" {
			req.Options.Vendor = clusterInfo.Provider
		}
		rec.Region = req.Options.Region

		log.Info(ctx, "Auto-filled dataSource from cluster info for schema request",
			zap.String("clusterId", req.ClusterId),
			zap.String("region", req.Options.Region),
		)
	}

	// 验证 dataSource 是否已设置
	if req.Options.DataSource == "" {
		rec.ErrCode = "DATASOURCE_REQUIRED"
		log.Error(ctx, "DataSource not provided and cannot be auto-filled",
			zap.String("clusterId", req.ClusterId),
		)
		c.JSON(http.StatusBadRequest, gin.H{
			"error": gin.H{
				"code":    "DATASOURCE_REQUIRED",
				"message": "dataSource is required (provide in options or via clusterId)",
				"details": gin.H{
					"hint": "Either provide dataSource in query parameters or ensure clusterId is valid and has configured S3 URL",
				},
			},
		})
		return
	}

	// 传递 clusterId 给 service 层（用于 VIEW 依赖表名生成）
	req.Options.ClusterId = req.ClusterId

	// 自动拼接完整表名 (table_{clusterId})
	actualTables := make([]string, len(req.Tables))
	tableNameMapping := make(map[string]string) // 映射 fullTable → originalTable
	for i, table := range req.Tables {
		// 拼接为完整表名：table_{clusterId}
		actualTables[i] = fmt.Sprintf("%s_%s", table, req.ClusterId)
		tableNameMapping[actualTables[i]] = table // 保存映射关系
		log.Info(ctx, "Auto-appended clusterId to table name",
			zap.String("originalTable", table),
			zap.String("fullTable", actualTables[i]),
		)
	}

	// 批量获取多个表的 schema
	var schemas []dto.SchemaResponse
	for _, table := range actualTables {
		// 调用 service 层获取单表 schema
		result, err := h.service.GetSchemaForTable(ctx, table, &req.Options)
		if err != nil {
			rec.ErrCode = "SCHEMA_FETCH_FAILED"
			log.Error(ctx, "Failed to get schema for table",
				zap.String("table", table),
				zap.Error(err),
			)
			c.JSON(http.StatusInternalServerError, gin.H{
				"error": fmt.Sprintf("get schema failed for table %s: %v", table, err),
			})
			return
		}

		// 将 Table 字段替换回原始表名
		if originalTable, ok := tableNameMapping[result.Table]; ok {
			result.Table = originalTable
		}

		schemas = append(schemas, *result)
	}

	// 返回批量结果
	response := dto.BatchSchemaResponse{
		Schemas: schemas,
		Count:   len(schemas),
	}

	if rec.Region == "" {
		rec.Region = req.Options.Region
	}

	c.JSON(http.StatusOK, response)
}

// getSchemaDatabaseFromDataSource 根据 dataSource URI 推断 vendor，返回对应的 database/dataset
func getSchemaDatabaseFromDataSource(dataSource string, cfg *config.DataProxyConfig) string {
	if strings.HasPrefix(dataSource, "gs://") {
		if cfg.GCP.BigQueryDataset != "" {
			return cfg.GCP.BigQueryDataset
		}
		return "clinic_data_proxy"
	}
	// 默认 AWS
	if cfg.AWS.AthenaDatabase != "" {
		return cfg.AWS.AthenaDatabase
	}
	return "clinic_data_proxy"
}

// RegisterRoutes 注册路由
func (h *DataProxyHandler) RegisterRoutes(r *gin.Engine) {
	group := r.Group("/data-proxy")
	{
		// 查询接口
		group.POST("/query", h.Query)
		group.POST("/batch-query", h.BatchQuery)

		// 表元数据接口
		group.GET("/schema", h.GetSchema)

		// 健康检查和信息接口
		group.GET("/info", h.Info)
	}

	r.POST(cloudDataProxyQueryPath, h.QueryByClusterPath)
}

func overrideClusterIDWithPath(req *dto.DataProxyQueryRequest, pathClusterID string) error {
	if pathClusterID == "" {
		return nil
	}
	if err := validateClusterId(pathClusterID); err != nil {
		return err
	}

	req.ClusterId = pathClusterID
	return nil
}

// ClusterAPIError 集群 API 错误（用于区分错误类型）
type ClusterAPIError struct {
	StatusCode int
	Message    string
}

func (e *ClusterAPIError) Error() string {
	return e.Message
}

func overrideStatementsDataSource(options *dto.DataProxyQueryOptions, clusterInfo *clinic_internal_api.ClusterDataURLs) {
	options.DataSource = clusterInfo.StatementsS3URL
	options.Region = clusterInfo.Region
	options.Vendor = strings.ToLower(strings.TrimSpace(clusterInfo.Provider))
	options.ClinicTenantId = clusterInfo.ClinicTenantID
	options.SkipSQLRewrite = false
	options.SkipTableManagement = false
}

func statementsProviderSupported(provider string) bool {
	switch strings.ToLower(strings.TrimSpace(provider)) {
	case "aws", "gcp", "azure":
		return true
	default:
		return false
	}
}

func writeStatementsProviderUnsupported(c *gin.Context, provider string) {
	c.JSON(http.StatusBadRequest, gin.H{
		"error": gin.H{
			"code":     "UNSUPPORTED_PROVIDER",
			"message":  fmt.Sprintf("Statements queries are not supported for provider '%s' yet", provider),
			"provider": provider,
		},
	})
}

// validateClusterId 校验 clusterId 格式
// clusterId 必须是纯数字字符串
func validateClusterId(clusterId string) error {
	if clusterId == "" {
		return nil // 空值由上层业务逻辑判断是否必填
	}

	// 检查是否只包含数字
	for _, c := range clusterId {
		if c < '0' || c > '9' {
			return fmt.Errorf("invalid clusterId format: must contain only digits, got '%s'", clusterId)
		}
	}

	return nil
}