package slowlog_explorer

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/api/common"
	apierr "github.com/pingcap-inc/clinic/pkg/api/error"
	cliniclog "github.com/pingcap-inc/clinic/pkg/log"
)

const (
	defaultLookbackHours = 3
	streamFirstBatch     = 30
	maxQueryHead         = 200
	sseHeartbeatInterval = 15 * time.Second
)

var requiredSlowlogListColumns = []string{"time", "digest", "conn_id"}

// checkClusterAccess verifies the user can read the cluster specified by :oid and :cid params.
func (s *Service) checkClusterAccess(c *gin.Context) (clusterID string, dialect slowlogQueryDialect, ok bool) {
	cid := c.Param("cid")
	_, _, cluster, canRead, err := common.CanReadCluster(c, s.rbac, cid)
	if err != nil {
		_ = c.Error(apierr.ErrNotFound.New("%s", err.Error()))
		return "", slowlogQueryDialectDefault, false
	}
	if !canRead {
		_ = c.Error(apierr.ErrForbidden.New("permission deny"))
		return "", slowlogQueryDialectDefault, false
	}
	provider := ""
	if cluster != nil {
		provider = cluster.Provider
	}
	return cid, slowlogQueryDialectForProvider(provider), true
}

// handleSlowlogs returns slow log rows via SSE streaming.
func (s *Service) handleSlowlogs(c *gin.Context) {
	var requestedFields *string
	if rawFields, fieldsProvided := c.GetQuery("fields"); fieldsProvided {
		requestedFields = &rawFields
	}
	logger := slowlogExplorerLogger(c).With(
		zap.Int("requested_field_count", slowlogRequestedFieldCount(requestedFields)),
	)

	clusterID, dialect, ok := s.checkClusterAccess(c)
	if !ok {
		logSlowlogListRejection(logger, "authorization", "cluster_access")
		return
	}
	logger = logger.With(
		zap.String("dialect", string(dialect)),
	)
	logger.Debug("slow log list access authorized",
		zap.String("stage", "authorization"),
		zap.String("status", "accepted"))

	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		logSlowlogListRejection(logger, "validation", "time_range")
		return
	}
	limit, ok := parseBoundedIntParam(c, "limit", 30, 1, 500)
	if !ok {
		logSlowlogListRejection(logger, "validation", "limit")
		return
	}
	logger = logger.With(zap.Int("limit", limit))

	var beforeTime *float64
	if bt := c.Query("before_time"); bt != "" {
		v, err := strconv.ParseFloat(bt, 64)
		if err != nil {
			logSlowlogListRejection(logger, "validation", "before_time")
			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid before_time"})
			return
		}
		beforeTime = &v
	}

	filterClauses, err := parseFiltersForDialect(c.Query("filters"), dialect)
	if err != nil {
		logSlowlogListRejection(logger, "validation", "filters")
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	orderSpec, err := parseOrderByForDialect(c.Query("order_by"), dialect)
	if err != nil {
		logSlowlogListRejection(logger, "validation", "order_by")
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}
	orderBy := orderSpec.clause
	selectedColumns, err := selectSlowlogColumns(requestedFields, dialect, orderSpec.column)
	if err != nil {
		logSlowlogListRejection(logger, "validation", "fields")
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}
	logger = logger.With(zap.Int("selected_field_count", len(selectedColumns)))
	logger.Info("slow log list request validated",
		zap.String("stage", "validation"),
		zap.String("status", "accepted"))

	// Set SSE headers
	c.Writer.Header().Set("Content-Type", "text/event-stream")
	c.Writer.Header().Set("Cache-Control", "no-cache")
	c.Writer.Header().Set("Connection", "keep-alive")
	c.Writer.Header().Set("X-Accel-Buffering", "no")
	writeSSEComment(c, "stream opened")

	ctx := c.Request.Context()
	dataProxyOptions := slowlogDataProxyOptionsForDialect(dialect)

	if limit <= streamFirstBatch {
		// Single query
		sql := buildSlowlogSQLForDialectWithColumns(startTime, endTime, limit+1, beforeTime, filterClauses, orderBy, dialect, selectedColumns)
		result, err := s.queryDataProxyWithSSEHeartbeat(c, clusterID, sql, 60, dataProxyOptions)
		logSlowlogUpstreamOutcome(logger, "single", result, err, ctx.Err(), false)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			writeSSEError(c, err.Error(), false)
			return
		}
		items := transformRows(result, clusterID)
		hasMore := len(items) > limit
		if hasMore {
			items = items[:limit]
		}
		nextCursor := extractLastTime(items)
		writeSSEBatch(c, items, 0, true)
		writeSSEDone(c, len(items), nextCursor, hasMore)
		logSlowlogListCompletion(logger, len(items), false)
		return
	}

	// Two-phase streaming: first batch (fast), then remainder
	sql1 := buildSlowlogSQLForDialectWithColumns(startTime, endTime, streamFirstBatch, beforeTime, filterClauses, orderBy, dialect, selectedColumns)
	result1, err := s.queryDataProxyWithSSEHeartbeat(c, clusterID, sql1, 60, dataProxyOptions)
	logSlowlogUpstreamOutcome(logger, "phase1", result1, err, ctx.Err(), false)
	if err != nil {
		if ctx.Err() != nil {
			return
		}
		writeSSEError(c, err.Error(), false)
		return
	}
	items1 := transformRows(result1, clusterID)
	if len(items1) < streamFirstBatch {
		writeSSEBatch(c, items1, 0, true)
		writeSSEDone(c, len(items1), extractLastTime(items1), false)
		logSlowlogListCompletion(logger, len(items1), false)
		return
	}
	writeSSEBatch(c, items1, 0, false)
	c.Writer.Flush()

	// Phase 2: for the default time-desc order, continue from the first batch
	// cursor instead of re-querying the first page.
	if orderSpec.column == "time" && orderSpec.direction == "DESC" {
		nextBeforeTime := extractLastTime(items1)
		remaining := limit - len(items1)
		sql2 := buildSlowlogSQLForDialectWithColumns(startTime, endTime, remaining+1, nextBeforeTime, filterClauses, orderBy, dialect, selectedColumns)
		result2, err := s.queryDataProxyWithSSEHeartbeat(c, clusterID, sql2, 120, dataProxyOptions)
		logSlowlogUpstreamOutcome(logger, "phase2", result2, err, ctx.Err(), true)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			writeSSEError(c, err.Error(), true)
			writeSSEDone(c, len(items1), extractLastTime(items1), true)
			logSlowlogListCompletion(logger, len(items1), true)
			return
		}
		items2 := transformRows(result2, clusterID)
		hasMore := len(items2) > remaining
		if hasMore {
			items2 = items2[:remaining]
		}
		writeSSEBatch(c, items2, 1, true)
		allItems := append(append([]map[string]interface{}{}, items1...), items2...)
		writeSSEDone(c, len(allItems), extractLastTime(allItems), hasMore)
		logSlowlogListCompletion(logger, len(allItems), false)
		return
	}

	// Phase 2 fallback for non-time sorts keeps the existing behavior. A full
	// keyset cursor for arbitrary sort columns requires carrying the sort value.
	sql2 := buildSlowlogSQLForDialectWithColumns(startTime, endTime, limit+1, beforeTime, filterClauses, orderBy, dialect, selectedColumns)
	result2, err := s.queryDataProxyWithSSEHeartbeat(c, clusterID, sql2, 120, dataProxyOptions)
	logSlowlogUpstreamOutcome(logger, "phase2", result2, err, ctx.Err(), true)
	if err != nil {
		if ctx.Err() != nil {
			return
		}
		writeSSEError(c, err.Error(), true)
		writeSSEDone(c, len(items1), extractLastTime(items1), true)
		logSlowlogListCompletion(logger, len(items1), true)
		return
	}
	allItems := transformRows(result2, clusterID)
	hasMore := len(allItems) > limit
	if hasMore {
		allItems = allItems[:limit]
	}
	var items2 []map[string]interface{}
	if len(allItems) > streamFirstBatch {
		items2 = allItems[streamFirstBatch:]
	}
	writeSSEBatch(c, items2, 1, true)
	writeSSEDone(c, len(allItems), extractLastTime(allItems), hasMore)
	logSlowlogListCompletion(logger, len(allItems), false)
}

// handleStats returns aggregate statistics.
func (s *Service) handleStats(c *gin.Context) {
	clusterID, dialect, ok := s.checkClusterAccess(c)
	if !ok {
		return
	}
	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		return
	}
	filterClauses, err := parseFiltersForDialect(c.Query("filters"), dialect)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	sql := buildStatsSQLForDialect(startTime, endTime, filterClauses, dialect)
	result, err := s.queryDataProxy(c.Request.Context(), clusterID, sql, 60, slowlogDataProxyOptionsForDialect(dialect))
	if err != nil {
		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
		return
	}

	if len(result.Rows) == 0 {
		c.JSON(http.StatusOK, gin.H{
			"total_count": 0, "p50_ms": 0, "p90_ms": 0, "p99_ms": 0, "max_ms": 0,
		})
		return
	}

	row := result.Rows[0]
	if dialect == slowlogQueryDialectBigQuery {
		percentiles := rawAt(row, 1)
		c.JSON(http.StatusOK, gin.H{
			"total_count": toIntOrZero(rawAt(row, 0)),
			"p50_ms":      toMsFloatOrZero(quantileAt(percentiles, 50)),
			"p90_ms":      toMsFloatOrZero(quantileAt(percentiles, 90)),
			"p99_ms":      toMsFloatOrZero(quantileAt(percentiles, 99)),
			"max_ms":      toMsFloatOrZero(rawAt(row, 2)),
		})
		return
	}
	c.JSON(http.StatusOK, gin.H{
		"total_count": toIntOrZero(rawAt(row, 0)),
		"p50_ms":      toMsFloatOrZero(rawAt(row, 1)),
		"p90_ms":      toMsFloatOrZero(rawAt(row, 2)),
		"p99_ms":      toMsFloatOrZero(rawAt(row, 3)),
		"max_ms":      toMsFloatOrZero(rawAt(row, 4)),
	})
}

// handleTrend returns time-series trend data.
func (s *Service) handleTrend(c *gin.Context) {
	clusterID, dialect, ok := s.checkClusterAccess(c)
	if !ok {
		return
	}
	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		return
	}
	metric := c.DefaultQuery("metric", "count")
	if !validTrendMetric(metric) {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid metric"})
		return
	}
	filterClauses, err := parseFiltersForDialect(c.Query("filters"), dialect)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	interval := trendInterval(startTime, endTime)
	sql := buildTrendSQLForDialect(startTime, endTime, metric, interval, filterClauses, dialect)
	result, err := s.queryDataProxy(c.Request.Context(), clusterID, sql, 60, slowlogDataProxyOptionsForDialect(dialect))
	if err != nil {
		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
		return
	}

	points := make([]map[string]interface{}, 0, len(result.Rows))
	for _, row := range result.Rows {
		t, err := toNumber(rawAt(row, 0))
		if err != nil {
			continue
		}
		rawValue := rawAt(row, 1)
		if percentileIndex, ok := slowlogPercentileIndex(metric); ok && dialect == slowlogQueryDialectBigQuery {
			rawValue = toMsFloat(quantileAt(rawValue, percentileIndex))
		}
		v, err := toNumber(rawValue)
		if err != nil {
			continue
		}
		var value interface{} = v
		if metric == "count" {
			value = int64(v)
		}
		points = append(points, map[string]interface{}{
			"t": int64(t),
			"v": value,
		})
	}

	c.JSON(http.StatusOK, gin.H{
		"metric":           metric,
		"interval_seconds": interval,
		"points":           points,
	})
}

// handleTopDigests returns top N slow query digests.
func (s *Service) handleTopDigests(c *gin.Context) {
	clusterID, dialect, ok := s.checkClusterAccess(c)
	if !ok {
		return
	}
	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		return
	}
	limit, ok := parseBoundedIntParam(c, "limit", 5, 1, 20)
	if !ok {
		return
	}
	metric, ok := normalizeTopDigestMetric(c.Query("metric"))
	if !ok {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid metric"})
		return
	}
	filterClauses, err := parseFiltersForDialect(c.Query("filters"), dialect)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	sql := buildTopDigestsSQLForDialect(startTime, endTime, limit, filterClauses, dialect, metric)
	result, err := s.queryDataProxy(c.Request.Context(), clusterID, sql, 60, slowlogDataProxyOptionsForDialect(dialect))
	if err != nil {
		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
		return
	}

	displayMetrics := topDigestDisplayMetrics(metric)
	items := make([]map[string]interface{}, 0, len(result.Rows))
	for _, row := range result.Rows {
		items = append(items, buildTopDigestItem(row, dialect, metric, displayMetrics, "digest", true))
	}

	c.JSON(http.StatusOK, gin.H{"items": items, "metric": metric})
}

// buildTopDigestItem converts one data-proxy row into a response item. The row
// has a fixed column shape (see buildTopDigestsSQLWithGroupBy): 0=group key,
// 1=cnt, 2=avg_ms, 3=p99(raw), 4=total_ms, 5=sample_query, 6+=display metrics.
// groupKeyField names the first column ("digest" or "db"); includeSampleQuery
// controls whether the sample_query column (always selected for offset padding)
// is exposed in the response.
func buildTopDigestItem(row []interface{}, dialect slowlogQueryDialect, metric string, displayMetrics []string, groupKeyField string, includeSampleQuery bool) map[string]interface{} {
	p99Ms := toFloatOrZero(rawAt(row, 3))
	if dialect == slowlogQueryDialectBigQuery {
		p99Ms = toMsFloatOrZero(quantileAt(rawAt(row, 3), 99))
	}
	metricValues := map[string]interface{}{}
	for i, key := range displayMetrics {
		value := rawAt(row, 6+i)
		if key == "total_time" {
			value = rawAt(row, 4)
		} else if key == "avg_query_time" {
			value = rawAt(row, 2)
		}
		metricValues[key] = toFloatOrZero(value)
	}
	sortValue := metricValues[metric]
	if metric == "count" {
		sortValue = toFloatOrZero(rawAt(row, 1))
	} else if metric == "total_time" {
		sortValue = toFloatOrZero(rawAt(row, 4))
	} else if metric == "avg_query_time" {
		sortValue = toFloatOrZero(rawAt(row, 2))
	}
	item := map[string]interface{}{
		groupKeyField:   rawStr(row, 0),
		"count":         toIntOrZero(rawAt(row, 1)),
		"avg_ms":        toFloatOrZero(rawAt(row, 2)),
		"p99_ms":        p99Ms,
		"total_ms":      toFloatOrZero(rawAt(row, 4)),
		"sort_metric":   metric,
		"sort_value":    sortValue,
		"metric_values": metricValues,
	}
	if includeSampleQuery {
		item["sample_query"] = truncateString(rawStr(row, 5), maxQueryHead)
	}
	return item
}

// parseDigestParam reads and validates the required digest query param for the
// DB breakdown: lower-cased to match the stored hex and validated as hex so it
// is injection-safe before being embedded into SQL.
func parseDigestParam(c *gin.Context) (string, bool) {
	digest := strings.ToLower(strings.TrimSpace(c.Query("digest")))
	if digest == "" || !hexPattern.MatchString(digest) {
		return "", false
	}
	return digest, true
}

// handleDigestDbBreakdown returns the per-DB breakdown for a single digest:
// same aggregation as top-digests but GROUP BY db, scoped to digest = ?.
func (s *Service) handleDigestDbBreakdown(c *gin.Context) {
	clusterID, dialect, ok := s.checkClusterAccess(c)
	if !ok {
		return
	}
	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		return
	}
	limit, ok := parseBoundedIntParam(c, "limit", 5, 1, 20)
	if !ok {
		return
	}
	metric, ok := normalizeTopDigestMetric(c.Query("metric"))
	if !ok {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid metric"})
		return
	}
	digest, ok := parseDigestParam(c)
	if !ok {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid or missing digest"})
		return
	}
	filterClauses, err := parseFiltersForDialect(c.Query("filters"), dialect)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	sql := buildDigestDbBreakdownSQL(startTime, endTime, limit, dialect, metric, digest, filterClauses)
	result, err := s.queryDataProxy(c.Request.Context(), clusterID, sql, 60, slowlogDataProxyOptionsForDialect(dialect))
	if err != nil {
		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
		return
	}

	displayMetrics := topDigestDisplayMetrics(metric)
	items := make([]map[string]interface{}, 0, len(result.Rows))
	for _, row := range result.Rows {
		items = append(items, buildTopDigestItem(row, dialect, metric, displayMetrics, "db", false))
	}

	c.JSON(http.StatusOK, gin.H{"items": items, "metric": metric})
}

// handleDetail returns a single slow log event detail.
func (s *Service) handleDetail(c *gin.Context) {
	clusterID, dialect, ok := s.checkClusterAccess(c)
	if !ok {
		return
	}
	digest, timestamp, connID, ok := parseDetailParams(c)
	if !ok {
		return
	}

	result, ok := s.queryDetailDataProxy(c, clusterID, digest, timestamp, connID, dialect)
	if !ok {
		return
	}

	record := transformDetail(result, clusterID)
	if record == nil {
		c.JSON(http.StatusNotFound, gin.H{"error": "event not found"})
		return
	}
	c.JSON(http.StatusOK, record)
}

// handleRawDetail returns the raw slow log event from Data Proxy for export.
func (s *Service) handleRawDetail(c *gin.Context) {
	clusterID, dialect, ok := s.checkClusterAccess(c)
	if !ok {
		return
	}
	digest, timestamp, connID, ok := parseDetailParams(c)
	if !ok {
		return
	}

	result, ok := s.queryDetailDataProxy(c, clusterID, digest, timestamp, connID, dialect)
	if !ok {
		return
	}

	record := rawRecord(result)
	if record == nil {
		c.JSON(http.StatusNotFound, gin.H{"error": "event not found"})
		return
	}
	c.JSON(http.StatusOK, record)
}

/* ------------------------------------------------------------------ */
/* SSE helpers                                                         */
/* ------------------------------------------------------------------ */

func slowlogExplorerLogger(c *gin.Context) *zap.Logger {
	return cliniclog.WithContextL(c.Request.Context()).With(
		zap.String("trace_id", slowlogExplorerTraceID(c)),
		zap.String("api", "slow_log_explorer"),
		zap.String("operation", "list_slowlogs"),
		zap.String("cluster_id", c.Param("cid")),
	)
}

func slowlogExplorerTraceID(c *gin.Context) string {
	spanContext := trace.SpanContextFromContext(c.Request.Context())
	if spanContext.HasTraceID() {
		return spanContext.TraceID().String()
	}
	return c.GetHeader("X-Trace-ID")
}

func logSlowlogListRejection(logger *zap.Logger, stage, validationTarget string) {
	logger.Warn("slow log list request rejected",
		zap.String("stage", stage),
		zap.String("validation_target", validationTarget),
		zap.String("status", "rejected"))
}

func logSlowlogUpstreamOutcome(logger *zap.Logger, phase string, response *DataProxyResponse, queryErr, requestErr error, partial bool) {
	fields := []zap.Field{
		zap.String("stage", "upstream"),
		zap.String("phase", phase),
		zap.Bool("partial", partial),
	}
	if requestErr != nil {
		logger.Info("slow log list upstream query canceled", append(fields,
			zap.String("status", "canceled"))...)
		return
	}
	if queryErr != nil {
		logger.Error("slow log list upstream query failed", append(fields,
			zap.String("status", "failed"),
			zap.String("error_type", fmt.Sprintf("%T", queryErr)))...)
		return
	}
	rowCount := 0
	if response != nil {
		rowCount = len(response.Rows)
	}
	logger.Info("slow log list upstream query completed", append(fields,
		zap.String("status", "success"),
		zap.Int("row_count", rowCount))...)
}

func logSlowlogListCompletion(logger *zap.Logger, rowCount int, partial bool) {
	status := "success"
	if partial {
		status = "partial"
	}
	logger.Info("slow log list request completed",
		zap.String("stage", "completion"),
		zap.String("status", status),
		zap.Int("row_count", rowCount),
		zap.Bool("partial", partial))
}

type slowlogQueryResult struct {
	response *DataProxyResponse
	err      error
}

func (s *Service) queryDataProxyWithSSEHeartbeat(c *gin.Context, clusterID, sql string, timeout int, options ...DataProxyOptions) (*DataProxyResponse, error) {
	return waitForSlowlogQuery(c, sseHeartbeatInterval, func(ctx context.Context) (*DataProxyResponse, error) {
		return s.queryDataProxy(ctx, clusterID, sql, timeout, options...)
	})
}

func waitForSlowlogQuery(c *gin.Context, heartbeatInterval time.Duration, query func(context.Context) (*DataProxyResponse, error)) (*DataProxyResponse, error) {
	ctx := c.Request.Context()
	resultCh := make(chan slowlogQueryResult, 1)
	go func() {
		response, err := query(ctx)
		resultCh <- slowlogQueryResult{response: response, err: err}
	}()

	ticker := time.NewTicker(heartbeatInterval)
	defer ticker.Stop()

	for {
		select {
		case result := <-resultCh:
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			return result.response, result.err
		case <-ticker.C:
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			writeSSEComment(c, "heartbeat")
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
}

func writeSSEComment(c *gin.Context, comment string) {
	_, _ = fmt.Fprintf(c.Writer, ": %s\n\n", comment)
	c.Writer.Flush()
}

func writeSSEBatch(c *gin.Context, items []map[string]interface{}, batchIndex int, isLast bool) {
	data, _ := json.Marshal(map[string]interface{}{
		"items":       publicItems(items),
		"batch_index": batchIndex,
		"is_last":     isLast,
	})
	_, _ = fmt.Fprintf(c.Writer, "event: batch\ndata: %s\n\n", data)
}

func writeSSEDone(c *gin.Context, totalRows int, nextCursor *float64, hasMore bool) {
	data, _ := json.Marshal(map[string]interface{}{
		"total_rows":  totalRows,
		"next_cursor": nextCursor,
		"has_more":    hasMore,
	})
	_, _ = fmt.Fprintf(c.Writer, "event: done\ndata: %s\n\n", data)
	c.Writer.Flush()
}

func writeSSEError(c *gin.Context, detail string, partial bool) {
	data, _ := json.Marshal(map[string]interface{}{
		"detail":  detail,
		"partial": partial,
	})
	_, _ = fmt.Fprintf(c.Writer, "event: error\ndata: %s\n\n", data)
	c.Writer.Flush()
}

/* ------------------------------------------------------------------ */
/* Utility helpers                                                     */
/* ------------------------------------------------------------------ */

func selectSlowlogColumns(rawFields *string, dialect slowlogQueryDialect, orderByColumn string) ([]string, error) {
	if dialect != slowlogQueryDialectMaxCompute || rawFields == nil {
		return slowlogColumns, nil
	}
	if strings.TrimSpace(*rawFields) == "" {
		return nil, fmt.Errorf("fields must not be empty")
	}

	allowed := make(map[string]struct{}, len(slowlogColumns))
	for _, column := range slowlogColumns {
		allowed[column] = struct{}{}
	}

	requested := make([]string, 0, len(slowlogColumns))
	seen := make(map[string]struct{}, len(slowlogColumns))
	for _, rawField := range strings.Split(*rawFields, ",") {
		field := strings.TrimSpace(rawField)
		if _, ok := allowed[field]; !ok {
			return nil, fmt.Errorf("invalid field %q", field)
		}
		if _, ok := seen[field]; ok {
			continue
		}
		seen[field] = struct{}{}
		requested = append(requested, field)
	}

	requiredColumns := append(append([]string{}, requiredSlowlogListColumns...), orderByColumn)
	for _, required := range requiredColumns {
		if _, ok := allowed[required]; !ok {
			return nil, fmt.Errorf("invalid required field %q", required)
		}
		if _, ok := seen[required]; ok {
			continue
		}
		seen[required] = struct{}{}
		requested = append(requested, required)
	}
	return requested, nil
}

func slowlogRequestedFieldCount(rawFields *string) int {
	if rawFields == nil || strings.TrimSpace(*rawFields) == "" {
		return 0
	}
	return len(strings.Split(*rawFields, ","))
}

func parseTimeRange(c *gin.Context) (time.Time, time.Time, bool) {
	now := time.Now()
	startTime := now.Add(-defaultLookbackHours * time.Hour)
	endTime := now

	if s := c.Query("start_time"); s != "" {
		t, err := parseTimeParam(s)
		if err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid start_time"})
			return time.Time{}, time.Time{}, false
		}
		startTime = t
	}
	if s := c.Query("end_time"); s != "" {
		t, err := parseTimeParam(s)
		if err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid end_time"})
			return time.Time{}, time.Time{}, false
		}
		endTime = t
	}

	if !startTime.Before(endTime) {
		c.JSON(http.StatusBadRequest, gin.H{"error": "start_time must be before end_time"})
		return time.Time{}, time.Time{}, false
	}

	return startTime, endTime, true
}

func parseTimeParam(raw string) (time.Time, error) {
	if t, err := time.Parse(time.RFC3339Nano, raw); err == nil {
		return t, nil
	}

	// Python datetime.fromisoformat accepts naive ISO strings and treats them as UTC.
	if hasTimezoneSuffix(raw) {
		return time.Time{}, fmt.Errorf("invalid RFC3339 time: %s", raw)
	}
	for _, layout := range []string{
		"2006-01-02T15:04:05.999999999",
		"2006-01-02T15:04:05",
		"2006-01-02 15:04:05.999999999",
		"2006-01-02 15:04:05",
	} {
		if t, err := time.ParseInLocation(layout, raw, time.UTC); err == nil {
			return t, nil
		}
	}
	return time.Time{}, fmt.Errorf("invalid ISO time: %s", raw)
}

func hasTimezoneSuffix(raw string) bool {
	if strings.HasSuffix(raw, "Z") || strings.HasSuffix(raw, "z") {
		return true
	}
	if len(raw) <= len("2006-01-02T15:04:05") {
		return false
	}
	tail := raw[len("2006-01-02T15:04:05"):]
	return strings.Contains(tail, "+") || strings.Contains(tail, "-")
}

func parseInt(s string, fallback int) int {
	v, err := strconv.Atoi(s)
	if err != nil {
		return fallback
	}
	return v
}

func parseBoundedIntParam(c *gin.Context, name string, fallback, min, max int) (int, bool) {
	raw := c.Query(name)
	if raw == "" {
		return fallback, true
	}
	v, err := strconv.Atoi(raw)
	if err != nil || v < min || v > max {
		c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("%s must be between %d and %d", name, min, max)})
		return 0, false
	}
	return v, true
}

func validTrendMetric(metric string) bool {
	switch metric {
	case "count", "p50", "p90", "p99", "max":
		return true
	default:
		return false
	}
}

func parseDetailParams(c *gin.Context) (string, float64, string, bool) {
	digest := c.Query("digest")
	timeStr := c.Query("time")
	connID := c.Query("conn_id")

	if digest == "" || timeStr == "" || connID == "" {
		c.JSON(http.StatusBadRequest, gin.H{"error": "digest, time, and conn_id are required"})
		return "", 0, "", false
	}
	if !hexPattern.MatchString(digest) {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid digest"})
		return "", 0, "", false
	}

	timestamp, err := strconv.ParseFloat(timeStr, 64)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid time parameter"})
		return "", 0, "", false
	}

	return digest, timestamp, connID, true
}

func (s *Service) queryDetailDataProxy(c *gin.Context, clusterID, digest string, timestamp float64, connID string, dialect slowlogQueryDialect) (*DataProxyResponse, bool) {
	sql := buildDetailSQLForDialect(timestamp, digest, connID, dialect)
	result, err := s.queryDataProxy(c.Request.Context(), clusterID, sql, 30, slowlogDataProxyOptionsForDialect(dialect))
	if err != nil {
		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
		return nil, false
	}

	if len(result.Rows) == 0 {
		c.JSON(http.StatusNotFound, gin.H{"error": "event not found"})
		return nil, false
	}
	return result, true
}