package slowlog_explorer

import (
	"encoding/json"
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"time"
)

// Columns selected for slow log list queries.
var slowlogColumns = []string{
	"time", "digest", "plan_digest", "instance", "db",
	"user", "conn_id", "txn_start_ts",
	"query_time", "parse_time", "compile_time", "optimize_time",
	"rewrite_time", "wait_ts", "process_time", "wait_time",
	"backoff_time", "cop_time", "kv_total", "pd_total",
	"mem_max", "disk_max",
	"total_keys", "process_keys", "result_rows",
	"write_keys", "write_size", "prewrite_region",
	"request_count", "txn_retry",
	"cop_proc_max", "cop_proc_avg", "cop_proc_p90", "cop_proc_addr",
	"cop_wait_max", "cop_wait_avg", "cop_wait_p90", "cop_wait_addr",
	"exec_retry_count", "exec_retry_time",
	"preproc_subqueries", "preproc_subqueries_time",
	"rocksdb_delete_skipped_count", "rocksdb_key_skipped_count",
	"rocksdb_block_cache_hit_count", "rocksdb_block_read_count",
	"rocksdb_block_read_byte",
	"succ", "is_internal", "is_explicit_txn", "prepared",
	"plan_from_cache", "plan_from_binding",
	"resource_group", "request_unit_read", "request_unit_write",
	"time_queued_by_rc",
	"index_names", "query",
}

// Additional columns for detail view.
var detailExtraColumns = []string{
	"prewrite_time", "commit_time", "get_commit_ts_time", "commit_backoff_time",
	"resolve_lock_time", "local_latch_wait_time", "lockkeys_time",
	"backoff_types", "backoff_total", "write_sql_response_total",
	"wait_prewrite_binlog_time",
	"plan", "decoded_plan", "binary_plan", "prev_stmt", "warnings", "backoff_detail",
}

var selectClause = strings.Join(slowlogColumns, ", ")

type slowlogQueryDialect string

const (
	slowlogQueryDialectDefault    slowlogQueryDialect = ""
	slowlogQueryDialectMaxCompute slowlogQueryDialect = "maxcompute"
	slowlogQueryDialectBigQuery   slowlogQueryDialect = "bigquery"
	slowlogQueryDialectSynapse    slowlogQueryDialect = "synapse"
)

func slowlogQueryDialectForProvider(provider string) slowlogQueryDialect {
	switch strings.ToLower(strings.TrimSpace(provider)) {
	case "alicloud":
		return slowlogQueryDialectMaxCompute
	case "gcp", "google", "google-cloud":
		return slowlogQueryDialectBigQuery
	case "azure", "synapse", "mssql", "sqlserver", "sql_server", "azure_synapse":
		return slowlogQueryDialectSynapse
	default:
		return slowlogQueryDialectDefault
	}
}

func slowlogPercentileFunc(dialect slowlogQueryDialect) string {
	if dialect == slowlogQueryDialectSynapse {
		return "APPROX_PERCENTILE_CONT"
	}
	if dialect == slowlogQueryDialectMaxCompute {
		return "percentile_approx"
	}
	return "approx_percentile"
}

func slowlogPercentileIndex(metric string) (int, bool) {
	switch metric {
	case "p50":
		return 50, true
	case "p90":
		return 90, true
	case "p99":
		return 99, true
	default:
		return 0, false
	}
}

func slowlogNumericExpr(col string, dialect slowlogQueryDialect) string {
	switch dialect {
	case slowlogQueryDialectBigQuery:
		return fmt.Sprintf("PARSE_NUMERIC(NULLIF(CONCAT(%s, ''), ''))", col)
	case slowlogQueryDialectSynapse:
		return fmt.Sprintf("TRY_CAST(NULLIF(CAST(%s AS VARCHAR(64)), '') AS FLOAT)", col)
	case slowlogQueryDialectMaxCompute:
		// Data Proxy parses this logical SQL with TiDB's parser before rewriting
		// it for MaxCompute, so MaxCompute-only CAST targets such as STRING must
		// not appear here. Preserve empty-string handling for the columns stored
		// as STRING; canonical numeric columns can be cast directly.
		if maxComputeStringNumericColumns[col] {
			return fmt.Sprintf("CAST(NULLIF(%s, '') AS DOUBLE)", col)
		}
		return fmt.Sprintf("CAST(%s AS DOUBLE)", col)
	default:
		return fmt.Sprintf("CAST(%s AS DOUBLE)", col)
	}
}

// These numeric slow-log values are STRING in Data Proxy's canonical
// MaxCompute schema. Keep this list aligned with
// clinic_data_service/pkg/engine/schema_registry.go.
var maxComputeStringNumericColumns = map[string]bool{
	"exec_retry_count":              true,
	"preproc_subqueries":            true,
	"process_keys":                  true,
	"request_count":                 true,
	"rocksdb_block_cache_hit_count": true,
	"rocksdb_block_read_byte":       true,
	"rocksdb_block_read_count":      true,
	"rocksdb_delete_skipped_count":  true,
	"rocksdb_key_skipped_count":     true,
	"total_keys":                    true,
}

func slowlogZeroSafeNumericExpr(col string, dialect slowlogQueryDialect) string {
	return fmt.Sprintf("COALESCE(%s, 0)", slowlogNumericExpr(col, dialect))
}

func slowlogDurationExpr(col string, dialect slowlogQueryDialect) string {
	switch dialect {
	case slowlogQueryDialectBigQuery, slowlogQueryDialectMaxCompute, slowlogQueryDialectSynapse:
		return slowlogNumericExpr(col, dialect)
	default:
		return fmt.Sprintf("CAST(%s AS DOUBLE)", col)
	}
}

func slowlogZeroSafeDurationExpr(col string, dialect slowlogQueryDialect) string {
	return fmt.Sprintf("COALESCE(%s, 0)", slowlogDurationExpr(col, dialect))
}

func slowlogQueryTimeExpr(dialect slowlogQueryDialect) string {
	return slowlogDurationExpr(slowlogColumnRef("query_time", dialect), dialect)
}

func slowlogColumnRef(col string, dialect slowlogQueryDialect) string {
	if dialect == slowlogQueryDialectSynapse {
		return "[" + strings.ReplaceAll(col, "]", "]]") + "]"
	}
	return col
}

func slowlogSelectColumn(col string, dialect slowlogQueryDialect) string {
	if dialect == slowlogQueryDialectSynapse {
		return fmt.Sprintf("%s AS %s", slowlogColumnRef(col, dialect), slowlogColumnRef(col, dialect))
	}
	return col
}

func slowlogSelectClauseForDialect(cols []string, dialect slowlogQueryDialect) string {
	if dialect != slowlogQueryDialectSynapse {
		return strings.Join(cols, ", ")
	}
	selected := make([]string, 0, len(cols))
	for _, col := range cols {
		selected = append(selected, slowlogSelectColumn(col, dialect))
	}
	return strings.Join(selected, ", ")
}

func slowlogTableRef(_ slowlogQueryDialect) string {
	return "clinic_data_proxy.slow_query_logs"
}

func slowlogCountExpr(dialect slowlogQueryDialect) string {
	if dialect == slowlogQueryDialectSynapse {
		return "COUNT_BIG(*)"
	}
	return "COUNT(*)"
}

func slowlogLimitSuffix(limit int, dialect slowlogQueryDialect) string {
	if dialect == slowlogQueryDialectSynapse {
		return ""
	}
	return fmt.Sprintf(" LIMIT %d", limit)
}

func slowlogSelectPrefix(limit int, dialect slowlogQueryDialect) string {
	if dialect == slowlogQueryDialectSynapse {
		return fmt.Sprintf("SELECT TOP (%d)", limit)
	}
	return "SELECT"
}

func slowlogDataProxyOptionsForDialect(dialect slowlogQueryDialect) DataProxyOptions {
	if dialect == slowlogQueryDialectSynapse {
		return DataProxyOptions{
			Vendor:              "azure",
			SkipTableManagement: true,
		}
	}
	return DataProxyOptions{}
}

type topDigestMetricMeta struct {
	key       string
	column    string
	agg       string
	scaleToMS bool
}

var topDigestMetrics = []topDigestMetricMeta{
	{key: "total_time", column: "query_time", agg: "sum", scaleToMS: true},
	{key: "count", agg: "count"},
	{key: "avg_query_time", column: "query_time", agg: "avg", scaleToMS: true},
	{key: "sum_total_keys", column: "total_keys", agg: "sum"},
	{key: "avg_total_keys", column: "total_keys", agg: "avg"},
	{key: "sum_process_keys", column: "process_keys", agg: "sum"},
	{key: "avg_process_keys", column: "process_keys", agg: "avg"},
	{key: "sum_request_count", column: "request_count", agg: "sum"},
	{key: "avg_request_count", column: "request_count", agg: "avg"},
	{key: "sum_result_rows", column: "result_rows", agg: "sum"},
	{key: "avg_result_rows", column: "result_rows", agg: "avg"},
	{key: "sum_write_keys", column: "write_keys", agg: "sum"},
	{key: "avg_write_keys", column: "write_keys", agg: "avg"},
	{key: "sum_ru_read", column: "request_unit_read", agg: "sum"},
	{key: "avg_ru_read", column: "request_unit_read", agg: "avg"},
	{key: "sum_ru_write", column: "request_unit_write", agg: "sum"},
	{key: "avg_ru_write", column: "request_unit_write", agg: "avg"},
	{key: "sum_mem_max", column: "mem_max", agg: "sum"},
	{key: "avg_mem_max", column: "mem_max", agg: "avg"},
	{key: "sum_kv_total", column: "kv_total", agg: "sum", scaleToMS: true},
	{key: "avg_kv_total", column: "kv_total", agg: "avg", scaleToMS: true},
}

var topDigestMetricRegistry = func() map[string]topDigestMetricMeta {
	reg := make(map[string]topDigestMetricMeta, len(topDigestMetrics))
	for _, metric := range topDigestMetrics {
		reg[metric.key] = metric
	}
	return reg
}()

func normalizeTopDigestMetric(metric string) (string, bool) {
	metric = strings.TrimSpace(metric)
	if metric == "" {
		return "total_time", true
	}
	if _, ok := topDigestMetricRegistry[metric]; !ok {
		return "", false
	}
	return metric, true
}

func topDigestDisplayMetrics(metric string) []string {
	switch metric {
	case "count":
		return []string{}
	case "sum_total_keys", "sum_process_keys":
		return []string{"sum_total_keys", "sum_process_keys"}
	case "avg_total_keys", "avg_process_keys":
		return []string{"avg_total_keys", "avg_process_keys"}
	case "sum_ru_read", "sum_ru_write":
		return []string{"sum_ru_read", "sum_ru_write"}
	case "avg_ru_read", "avg_ru_write":
		return []string{"avg_ru_read", "avg_ru_write"}
	default:
		return []string{metric}
	}
}

func topDigestMetricExpression(metric topDigestMetricMeta, dialect slowlogQueryDialect) string {
	if metric.agg == "count" {
		return slowlogCountExpr(dialect)
	}

	column := slowlogColumnRef(metric.column, dialect)
	col := slowlogZeroSafeNumericExpr(column, dialect)
	if metric.scaleToMS {
		col = slowlogZeroSafeDurationExpr(column, dialect)
	}
	var expr string
	switch metric.agg {
	case "avg":
		expr = fmt.Sprintf("AVG(%s)", col)
	default:
		expr = fmt.Sprintf("SUM(%s)", col)
	}
	if metric.scaleToMS {
		expr = fmt.Sprintf("(%s) * 1000", expr)
	}
	return fmt.Sprintf("ROUND(%s, 2)", expr)
}

func topDigestMetricAlias(metric string) string {
	switch metric {
	case "count":
		return "cnt"
	case "total_time":
		return "total_ms"
	case "avg_query_time":
		return "avg_ms"
	default:
		return metric
	}
}

// topDigestGroupBy describes the grouping dimension of a top-digest query. The
// dimension only changes the first SELECT column and the GROUP BY expression;
// every other column (cnt/avg/p99/total/sample_query/display metrics) stays
// identical so the handler's fixed column offsets keep working. selectExpr and
// groupExpr are functions so each dialect's quoting is encapsulated here rather
// than branching inside the SQL builder.
type topDigestGroupBy struct {
	key        string
	selectExpr func(slowlogQueryDialect) string // first SELECT column, incl. alias
	groupExpr  func(slowlogQueryDialect) string // GROUP BY expression (same column)
}

var topDigestGroupRegistry = map[string]topDigestGroupBy{
	"digest": {
		key: "digest",
		selectExpr: func(d slowlogQueryDialect) string {
			return fmt.Sprintf("%s AS digest", slowlogColumnRef("digest", d))
		},
		groupExpr: func(d slowlogQueryDialect) string {
			return slowlogColumnRef("digest", d)
		},
	},
	"db": {
		key: "db",
		// COALESCE folds NULL/empty db into a single group rendered as "".
		selectExpr: func(d slowlogQueryDialect) string {
			return fmt.Sprintf("COALESCE(%s, '') AS db", slowlogColumnRef("db", d))
		},
		groupExpr: func(d slowlogQueryDialect) string {
			return fmt.Sprintf("COALESCE(%s, '')", slowlogColumnRef("db", d))
		},
	},
}

// normalizeTopDigestGroupBy resolves the group_by query param; empty defaults to
// digest (preserving the original top-digests behavior).
func normalizeTopDigestGroupBy(g string) (topDigestGroupBy, bool) {
	g = strings.TrimSpace(g)
	if g == "" {
		return topDigestGroupRegistry["digest"], true
	}
	meta, ok := topDigestGroupRegistry[g]
	return meta, ok
}

// datePartitions generates date partition keys (YYYYMMDD) for the given time range.
func datePartitions(start, end time.Time) []string {
	var dates []string
	d := start.UTC().Truncate(24 * time.Hour)
	endD := end.UTC().Truncate(24 * time.Hour)
	for !d.After(endD) {
		dates = append(dates, d.Format("20060102"))
		d = d.Add(24 * time.Hour)
	}
	return dates
}

// buildSlowlogSQL builds the SQL query for fetching slow logs.
func buildSlowlogSQL(startTime, endTime time.Time, limit int, beforeTime *float64, extraConditions []string, orderBy string) string {
	return buildSlowlogSQLForDialect(startTime, endTime, limit, beforeTime, extraConditions, orderBy, slowlogQueryDialectDefault)
}

func buildSlowlogSQLForDialect(startTime, endTime time.Time, limit int, beforeTime *float64, extraConditions []string, orderBy string, dialect slowlogQueryDialect) string {
	return buildSlowlogSQLForDialectWithColumns(startTime, endTime, limit, beforeTime, extraConditions, orderBy, dialect, slowlogColumns)
}

func buildSlowlogSQLForDialectWithColumns(startTime, endTime time.Time, limit int, beforeTime *float64, extraConditions []string, orderBy string, dialect slowlogQueryDialect, columns []string) string {
	dates := datePartitions(startTime, endTime)
	dateIn := make([]string, len(dates))
	for i, d := range dates {
		dateIn[i] = fmt.Sprintf("'%s'", d)
	}

	dateCol := slowlogColumnRef("date", dialect)
	timeCol := slowlogColumnRef("time", dialect)
	conditions := []string{
		fmt.Sprintf("%s IN (%s)", dateCol, strings.Join(dateIn, ", ")),
		fmt.Sprintf("%s >= %f", timeCol, float64(startTime.Unix())+float64(startTime.Nanosecond())/1e9),
		fmt.Sprintf("%s <= %f", timeCol, float64(endTime.Unix())+float64(endTime.Nanosecond())/1e9),
	}

	if beforeTime != nil {
		conditions = append(conditions, fmt.Sprintf("%s < %f", timeCol, *beforeTime))
	}

	conditions = append(conditions, extraConditions...)

	if orderBy == "" {
		orderBy = slowlogColumnRef("time", dialect) + " DESC"
	}

	return fmt.Sprintf(
		"%s %s FROM %s WHERE %s ORDER BY %s%s",
		slowlogSelectPrefix(limit, dialect),
		slowlogSelectClauseForDialect(columns, dialect),
		slowlogTableRef(dialect),
		strings.Join(conditions, " AND "),
		orderBy,
		slowlogLimitSuffix(limit, dialect),
	)
}

// buildStatsSQL builds the SQL for aggregate statistics.
func buildStatsSQL(startTime, endTime time.Time, extraConditions []string) string {
	return buildStatsSQLForDialect(startTime, endTime, extraConditions, slowlogQueryDialectDefault)
}

func buildStatsSQLForDialect(startTime, endTime time.Time, extraConditions []string, dialect slowlogQueryDialect) string {
	where := buildWhereForDialect(startTime, endTime, extraConditions, dialect)
	queryTime := slowlogQueryTimeExpr(dialect)
	if dialect == slowlogQueryDialectBigQuery {
		return fmt.Sprintf(
			"SELECT COUNT(*) AS total_count, "+
				"APPROX_QUANTILES(%s, 100) AS percentiles, "+
				"MAX(%s) AS max_val "+
				"FROM %s WHERE %s",
			queryTime, queryTime, slowlogTableRef(dialect), where,
		)
	}
	if dialect == slowlogQueryDialectSynapse {
		percentile := slowlogPercentileFunc(dialect)
		return fmt.Sprintf(
			"SELECT %s AS total_count, "+
				"%s(0.5) WITHIN GROUP (ORDER BY %s) AS p50, "+
				"%s(0.9) WITHIN GROUP (ORDER BY %s) AS p90, "+
				"%s(0.99) WITHIN GROUP (ORDER BY %s) AS p99, "+
				"MAX(%s) AS max_val "+
				"FROM %s WHERE %s",
			slowlogCountExpr(dialect),
			percentile, queryTime,
			percentile, queryTime,
			percentile, queryTime,
			queryTime,
			slowlogTableRef(dialect),
			where,
		)
	}

	percentile := slowlogPercentileFunc(dialect)
	return fmt.Sprintf(
		"SELECT %s AS total_count, "+
			"%s(%s, 0.5) AS p50, "+
			"%s(%s, 0.9) AS p90, "+
			"%s(%s, 0.99) AS p99, "+
			"MAX(%s) AS max_val "+
			"FROM %s WHERE %s",
		slowlogCountExpr(dialect), percentile, queryTime, percentile, queryTime, percentile, queryTime, queryTime, slowlogTableRef(dialect), where,
	)
}

// buildTrendSQL builds the SQL for time-series trend data.
func buildTrendSQL(startTime, endTime time.Time, metric string, interval int, extraConditions []string) string {
	return buildTrendSQLForDialect(startTime, endTime, metric, interval, extraConditions, slowlogQueryDialectDefault)
}

func buildTrendSQLForDialect(startTime, endTime time.Time, metric string, interval int, extraConditions []string, dialect slowlogQueryDialect) string {
	where := buildWhereForDialect(startTime, endTime, extraConditions, dialect)
	timeCol := slowlogColumnRef("time", dialect)
	bucketExpr := fmt.Sprintf("FLOOR(%s / %d) * %d", timeCol, interval, interval)
	percentile := slowlogPercentileFunc(dialect)
	queryTime := slowlogQueryTimeExpr(dialect)

	var aggExpr string
	switch metric {
	case "p50":
		if dialect == slowlogQueryDialectBigQuery {
			aggExpr = fmt.Sprintf("APPROX_QUANTILES(%s, 100)", queryTime)
		} else if dialect == slowlogQueryDialectSynapse {
			aggExpr = fmt.Sprintf("ROUND(%s(0.5) WITHIN GROUP (ORDER BY %s) * 1000, 2)", percentile, queryTime)
		} else {
			aggExpr = fmt.Sprintf("ROUND(%s(%s, 0.5) * 1000, 2)", percentile, queryTime)
		}
	case "p90":
		if dialect == slowlogQueryDialectBigQuery {
			aggExpr = fmt.Sprintf("APPROX_QUANTILES(%s, 100)", queryTime)
		} else if dialect == slowlogQueryDialectSynapse {
			aggExpr = fmt.Sprintf("ROUND(%s(0.9) WITHIN GROUP (ORDER BY %s) * 1000, 2)", percentile, queryTime)
		} else {
			aggExpr = fmt.Sprintf("ROUND(%s(%s, 0.9) * 1000, 2)", percentile, queryTime)
		}
	case "p99":
		if dialect == slowlogQueryDialectBigQuery {
			aggExpr = fmt.Sprintf("APPROX_QUANTILES(%s, 100)", queryTime)
		} else if dialect == slowlogQueryDialectSynapse {
			aggExpr = fmt.Sprintf("ROUND(%s(0.99) WITHIN GROUP (ORDER BY %s) * 1000, 2)", percentile, queryTime)
		} else {
			aggExpr = fmt.Sprintf("ROUND(%s(%s, 0.99) * 1000, 2)", percentile, queryTime)
		}
	case "max":
		aggExpr = fmt.Sprintf("ROUND(MAX(%s) * 1000, 2)", queryTime)
	default: // "count"
		aggExpr = slowlogCountExpr(dialect)
	}

	return fmt.Sprintf(
		"SELECT %s AS bucket, %s AS val "+
			"FROM %s WHERE %s GROUP BY %s ORDER BY %s",
		bucketExpr, aggExpr, slowlogTableRef(dialect), where, bucketExpr, bucketExpr,
	)
}

// buildTopDigestsSQL builds the SQL for top digest analysis.
func buildTopDigestsSQL(startTime, endTime time.Time, limit int, extraConditions []string) string {
	return buildTopDigestsSQLForDialect(startTime, endTime, limit, extraConditions, slowlogQueryDialectDefault, "total_time")
}

func buildTopDigestsSQLForDialect(startTime, endTime time.Time, limit int, extraConditions []string, dialect slowlogQueryDialect, metric string) string {
	return buildTopDigestsSQLWithGroupBy(startTime, endTime, limit, extraConditions, dialect, metric, topDigestGroupRegistry["digest"])
}

// buildTopDigestsSQLWithGroupBy is the generalized builder: the grouping
// dimension (digest or db) only swaps the first SELECT column and GROUP BY; the
// rest of the SELECT list keeps a fixed shape that the handler reads by offset.
func buildTopDigestsSQLWithGroupBy(startTime, endTime time.Time, limit int, extraConditions []string, dialect slowlogQueryDialect, metric string, groupBy topDigestGroupBy) string {
	where := buildWhereForDialect(startTime, endTime, extraConditions, dialect)
	queryTime := slowlogQueryTimeExpr(dialect)
	metric, ok := normalizeTopDigestMetric(metric)
	if !ok {
		metric = "total_time"
	}
	queryCol := slowlogColumnRef("query", dialect)
	selects := []string{
		groupBy.selectExpr(dialect),
		fmt.Sprintf("%s AS cnt", slowlogCountExpr(dialect)),
		fmt.Sprintf("ROUND(AVG(%s) * 1000, 2) AS avg_ms", queryTime),
	}

	if dialect == slowlogQueryDialectBigQuery {
		selects = append(selects, fmt.Sprintf("APPROX_QUANTILES(%s, 100) AS p99_ms", queryTime))
	} else if dialect == slowlogQueryDialectSynapse {
		percentile := slowlogPercentileFunc(dialect)
		selects = append(selects, fmt.Sprintf("ROUND(%s(0.99) WITHIN GROUP (ORDER BY %s) * 1000, 2) AS p99_ms", percentile, queryTime))
	} else {
		percentile := slowlogPercentileFunc(dialect)
		selects = append(selects, fmt.Sprintf("ROUND(%s(%s, 0.99) * 1000, 2) AS p99_ms", percentile, queryTime))
	}

	sampleQueryExpr := "any_value(query) AS sample_query"
	if dialect == slowlogQueryDialectSynapse {
		sampleQueryExpr = fmt.Sprintf("MAX(CAST(%s AS NVARCHAR(4000))) AS sample_query", queryCol)
	}
	selects = append(selects,
		fmt.Sprintf("ROUND(SUM(%s) * 1000, 2) AS total_ms", queryTime),
		sampleQueryExpr,
	)

	selectedMetricAliases := make(map[string]bool)
	for _, key := range topDigestDisplayMetrics(metric) {
		alias := topDigestMetricAlias(key)
		if alias == "avg_ms" || alias == "total_ms" {
			continue
		}
		if selectedMetricAliases[alias] {
			continue
		}
		meta := topDigestMetricRegistry[key]
		selects = append(selects, fmt.Sprintf("%s AS %s", topDigestMetricExpression(meta, dialect), alias))
		selectedMetricAliases[alias] = true
	}

	orderAlias := topDigestMetricAlias(metric)
	if orderAlias != "avg_ms" && orderAlias != "total_ms" && orderAlias != "cnt" && !selectedMetricAliases[orderAlias] {
		meta := topDigestMetricRegistry[metric]
		selects = append(selects, fmt.Sprintf("%s AS %s", topDigestMetricExpression(meta, dialect), orderAlias))
	}

	return fmt.Sprintf(
		"%s %s "+
			"FROM %s WHERE %s "+
			"GROUP BY %s ORDER BY %s DESC%s",
		slowlogSelectPrefix(limit, dialect),
		strings.Join(selects, ", "),
		slowlogTableRef(dialect),
		where,
		groupBy.groupExpr(dialect),
		orderAlias,
		slowlogLimitSuffix(limit, dialect),
	)
}

// buildDigestDbBreakdownSQL builds the per-DB breakdown SQL for a single digest:
// the inherited filter clauses plus a digest = ? scope, grouped by db. digest
// must already be hex-validated by the caller. The caller's filterClauses slice
// is not mutated.
func buildDigestDbBreakdownSQL(startTime, endTime time.Time, limit int, dialect slowlogQueryDialect, metric, digest string, filterClauses []string) string {
	clauses := make([]string, len(filterClauses), len(filterClauses)+1)
	copy(clauses, filterClauses)
	clauses = append(clauses, fmt.Sprintf("%s = '%s'", slowlogColumnRef("digest", dialect), digest))
	return buildTopDigestsSQLWithGroupBy(startTime, endTime, limit, clauses, dialect, metric, topDigestGroupRegistry["db"])
}

// buildDetailSQL builds the SQL for fetching a single event detail.
func buildDetailSQL(timestamp float64, digest, connID string) string {
	return buildDetailSQLForDialect(timestamp, digest, connID, slowlogQueryDialectDefault)
}

func buildDetailSQLForDialect(timestamp float64, digest, connID string, dialect slowlogQueryDialect) string {
	t := time.Unix(int64(timestamp), int64((timestamp-float64(int64(timestamp)))*1e9))
	dateStr := t.UTC().Format("20060102")
	allCols := append(append([]string{}, slowlogColumns...), detailExtraColumns...)
	dateCol := slowlogColumnRef("date", dialect)
	timeCol := slowlogColumnRef("time", dialect)
	digestCol := slowlogColumnRef("digest", dialect)
	connIDCol := slowlogColumnRef("conn_id", dialect)

	return fmt.Sprintf(
		"%s %s FROM %s "+
			"WHERE %s = '%s' AND %s >= %f AND %s <= %f "+
			"AND %s = '%s' AND %s = '%s' "+
			"ORDER BY ABS(%s - %f)%s",
		slowlogSelectPrefix(1, dialect),
		slowlogSelectClauseForDialect(allCols, dialect),
		slowlogTableRef(dialect),
		dateCol,
		dateStr,
		timeCol,
		timestamp-1,
		timeCol,
		timestamp+1,
		digestCol,
		escapeString(digest),
		connIDCol,
		escapeString(connID),
		timeCol,
		timestamp,
		slowlogLimitSuffix(1, dialect),
	)
}

func buildWhere(startTime, endTime time.Time, extraConditions []string) string {
	return buildWhereForDialect(startTime, endTime, extraConditions, slowlogQueryDialectDefault)
}

func buildWhereForDialect(startTime, endTime time.Time, extraConditions []string, dialect slowlogQueryDialect) string {
	dates := datePartitions(startTime, endTime)
	dateIn := make([]string, len(dates))
	for i, d := range dates {
		dateIn[i] = fmt.Sprintf("'%s'", d)
	}

	dateCol := slowlogColumnRef("date", dialect)
	timeCol := slowlogColumnRef("time", dialect)
	conditions := []string{
		fmt.Sprintf("%s IN (%s)", dateCol, strings.Join(dateIn, ", ")),
		fmt.Sprintf("%s >= %f", timeCol, float64(startTime.Unix())+float64(startTime.Nanosecond())/1e9),
		fmt.Sprintf("%s <= %f", timeCol, float64(endTime.Unix())+float64(endTime.Nanosecond())/1e9),
	}
	conditions = append(conditions, extraConditions...)
	return strings.Join(conditions, " AND ")
}

// trendInterval calculates the aggregation interval based on time range.
func trendInterval(start, end time.Time) int {
	hours := end.Sub(start).Hours()
	switch {
	case hours <= 1:
		return 60
	case hours <= 3:
		return 300
	case hours <= 12:
		return 600
	default:
		return 900
	}
}

// parseFilters parses JSON filter conditions into SQL WHERE clauses.
func parseFilters(raw string) ([]string, error) {
	return parseFiltersForDialect(raw, slowlogQueryDialectDefault)
}

func parseFiltersForDialect(raw string, dialect slowlogQueryDialect) ([]string, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" || raw == "[]" {
		return nil, nil
	}

	var filters []struct {
		Field string      `json:"field"`
		Op    string      `json:"op"`
		Value interface{} `json:"value"`
	}
	decoder := json.NewDecoder(strings.NewReader(raw))
	decoder.UseNumber()
	if err := decoder.Decode(&filters); err != nil {
		return nil, fmt.Errorf("invalid filters JSON: %w", err)
	}

	var clauses []string
	for _, f := range filters {
		meta, ok := filterRegistry[f.Field]
		if !ok {
			return nil, fmt.Errorf("unknown filter field: %q", f.Field)
		}
		op := strings.TrimSpace(f.Op)
		if !meta.operators[op] {
			return nil, fmt.Errorf("operator %q not allowed for field %q", op, f.Field)
		}

		clause, err := buildFilterClauseForDialect(meta, op, f.Value, dialect)
		if err != nil {
			return nil, err
		}
		clauses = append(clauses, clause)
	}
	return clauses, nil
}

// Validation patterns
var (
	hexPattern        = regexp.MustCompile(`^[0-9a-fA-F]+$`)
	safeStringPattern = regexp.MustCompile(`^[a-zA-Z0-9_.\-: ]+$`)

	validOrderByColumns = map[string]bool{
		"time": true, "query_time": true, "process_time": true, "wait_time": true,
		"mem_max": true, "disk_max": true, "total_keys": true, "process_keys": true,
		"result_rows": true, "write_keys": true, "kv_total": true, "cop_proc_max": true,
		"request_count": true, "request_unit_read": true, "request_unit_write": true,
	}
	numericOrderByColumns = map[string]bool{
		"query_time": true, "process_time": true, "wait_time": true,
		"mem_max": true, "disk_max": true, "total_keys": true, "process_keys": true,
		"result_rows": true, "write_keys": true, "kv_total": true, "cop_proc_max": true,
		"request_count": true, "request_unit_read": true, "request_unit_write": true,
	}
	durationOrderByColumns = map[string]bool{
		"query_time": true, "process_time": true, "wait_time": true,
		"kv_total": true, "cop_proc_max": true,
	}
)

type filterFieldType string

const (
	filterTypeString   filterFieldType = "string"
	filterTypeHex      filterFieldType = "hex"
	filterTypeDuration filterFieldType = "duration_s"
	filterTypeCount    filterFieldType = "count"
	filterTypeBytes    filterFieldType = "bytes"
	filterTypeBoolean  filterFieldType = "boolean"
	filterTypeRU       filterFieldType = "ru"
	maxFilterLikeLen                   = 200
)

type filterFieldMeta struct {
	key       string
	colType   filterFieldType
	operators map[string]bool
}

func ops(values ...string) map[string]bool {
	m := make(map[string]bool, len(values))
	for _, v := range values {
		m[v] = true
	}
	return m
}

func buildFilterRegistry() map[string]filterFieldMeta {
	stringOps := ops("=", "!=", "LIKE", "NOT LIKE", "IN", "NOT IN")
	hexOps := ops("=", "!=")
	numOps := ops("=", "!=", ">", ">=", "<", "<=")
	boolOps := ops("=")

	reg := make(map[string]filterFieldMeta)
	register := func(key string, colType filterFieldType, operators map[string]bool) {
		reg[key] = filterFieldMeta{key: key, colType: colType, operators: operators}
	}

	for _, key := range []string{
		"db", "user", "instance", "resource_group",
		"conn_id", "txn_start_ts", "index_names",
	} {
		register(key, filterTypeString, stringOps)
	}
	register("query", filterTypeString, ops("LIKE", "NOT LIKE"))
	register("digest", filterTypeHex, hexOps)
	register("plan_digest", filterTypeHex, hexOps)

	for _, key := range []string{
		"query_time", "parse_time", "compile_time", "optimize_time", "rewrite_time",
		"wait_ts", "process_time", "wait_time", "backoff_time", "cop_time",
		"kv_total", "pd_total", "exec_retry_time", "preproc_subqueries_time",
		"time_queued_by_rc", "cop_proc_max", "cop_proc_avg", "cop_proc_p90",
		"cop_wait_max", "cop_wait_avg", "cop_wait_p90",
	} {
		register(key, filterTypeDuration, numOps)
	}

	for _, key := range []string{
		"total_keys", "process_keys", "result_rows", "write_keys", "prewrite_region",
		"txn_retry", "exec_retry_count", "request_count", "preproc_subqueries",
		"rocksdb_delete_skipped_count", "rocksdb_key_skipped_count",
		"rocksdb_block_cache_hit_count", "rocksdb_block_read_count",
	} {
		register(key, filterTypeCount, numOps)
	}

	for _, key := range []string{"mem_max", "disk_max", "write_size", "rocksdb_block_read_byte"} {
		register(key, filterTypeBytes, numOps)
	}

	register("request_unit_read", filterTypeRU, numOps)
	register("request_unit_write", filterTypeRU, numOps)

	for _, key := range []string{
		"succ", "is_internal", "is_explicit_txn", "prepared",
		"plan_from_cache", "plan_from_binding",
	} {
		register(key, filterTypeBoolean, boolOps)
	}

	return reg
}

var filterRegistry = buildFilterRegistry()

type orderBySpec struct {
	column    string
	direction string
	clause    string
}

// parseOrderBy validates and returns a safe ORDER BY clause.
func parseOrderBy(raw string) (string, error) {
	spec, err := parseOrderByForDialect(raw, slowlogQueryDialectDefault)
	if err != nil {
		return "", err
	}
	return spec.clause, nil
}

func parseOrderByForDialect(raw string, dialect slowlogQueryDialect) (orderBySpec, error) {
	if raw == "" {
		return orderBySpec{column: "time", direction: "DESC", clause: slowlogColumnRef("time", dialect) + " DESC"}, nil
	}
	parts := strings.Fields(raw)
	if len(parts) < 1 || len(parts) > 2 {
		return orderBySpec{}, fmt.Errorf("invalid order_by: %q", raw)
	}
	col := strings.ToLower(parts[0])
	if !validOrderByColumns[col] {
		return orderBySpec{}, fmt.Errorf("invalid order_by column: %s", parts[0])
	}
	dir := "DESC"
	if len(parts) == 2 {
		d := strings.ToUpper(parts[1])
		if d == "ASC" || d == "DESC" {
			dir = d
		} else {
			return orderBySpec{}, fmt.Errorf("invalid order_by direction: %s", parts[1])
		}
	}
	orderExpr := slowlogColumnRef(col, dialect)
	if numericOrderByColumns[col] {
		orderExpr = slowlogNumericExpr(slowlogColumnRef(col, dialect), dialect)
		if durationOrderByColumns[col] {
			orderExpr = slowlogDurationExpr(slowlogColumnRef(col, dialect), dialect)
		}
	}
	return orderBySpec{column: col, direction: dir, clause: orderExpr + " " + dir}, nil
}

func buildFilterClause(meta filterFieldMeta, op string, value interface{}) (string, error) {
	return buildFilterClauseForDialect(meta, op, value, slowlogQueryDialectDefault)
}

func buildFilterClauseForDialect(meta filterFieldMeta, op string, value interface{}, dialect slowlogQueryDialect) (string, error) {
	col := meta.key
	colRef := slowlogColumnRef(col, dialect)

	switch meta.colType {
	case filterTypeHex:
		v, err := filterString(value)
		if err != nil {
			return "", err
		}
		v = strings.ToLower(strings.TrimSpace(v))
		if !hexPattern.MatchString(v) {
			return "", fmt.Errorf("invalid hex value for %s: %q", col, v)
		}
		return fmt.Sprintf("%s %s '%s'", colRef, op, v), nil

	case filterTypeString:
		switch op {
		case "=", "!=":
			v, err := filterString(value)
			if err != nil {
				return "", err
			}
			if !safeStringPattern.MatchString(v) {
				return "", fmt.Errorf("invalid string value for %s: %q", col, v)
			}
			return fmt.Sprintf("%s %s '%s'", colRef, op, escapeString(v)), nil
		case "LIKE", "NOT LIKE":
			v, err := filterString(value)
			if err != nil {
				return "", err
			}
			return fmt.Sprintf("%s %s '%%%s%%'", colRef, op, escapeString(truncateString(v, maxFilterLikeLen))), nil
		case "IN", "NOT IN":
			arr, ok := value.([]interface{})
			if !ok || len(arr) == 0 {
				return "", fmt.Errorf("%s requires a non-empty array for %s", op, col)
			}
			vals := make([]string, 0, len(arr))
			for _, item := range arr {
				v, err := filterString(item)
				if err != nil {
					return "", err
				}
				if !safeStringPattern.MatchString(v) {
					return "", fmt.Errorf("invalid string value in %s for %s: %q", op, col, v)
				}
				vals = append(vals, fmt.Sprintf("'%s'", escapeString(v)))
			}
			return fmt.Sprintf("%s %s (%s)", colRef, op, strings.Join(vals, ", ")), nil
		}

	case filterTypeDuration:
		num, err := filterFloat(value)
		if err != nil {
			return "", fmt.Errorf("invalid number for %s: %w", col, err)
		}
		return fmt.Sprintf("%s %s %s", slowlogDurationExpr(colRef, dialect), op, formatFilterFloat(num/1000.0)), nil

	case filterTypeCount, filterTypeBytes, filterTypeRU:
		num, err := filterFloat(value)
		if err != nil {
			return "", fmt.Errorf("invalid number for %s: %w", col, err)
		}
		return fmt.Sprintf("%s %s %s", slowlogNumericExpr(colRef, dialect), op, formatFilterFloat(num)), nil

	case filterTypeBoolean:
		v, err := filterString(value)
		if err != nil {
			return "", err
		}
		b := strings.ToLower(v)
		if b != "true" && b != "false" {
			return "", fmt.Errorf("invalid boolean for %s: %v", col, value)
		}
		if dialect == slowlogQueryDialectSynapse {
			if b == "true" {
				return fmt.Sprintf("%s = 1", colRef), nil
			}
			return fmt.Sprintf("%s = 0", colRef), nil
		}
		return fmt.Sprintf("%s = %s", colRef, b), nil
	}

	return "", fmt.Errorf("unsupported filter: %s %s %v", col, op, value)
}

func filterString(value interface{}) (string, error) {
	if value == nil {
		return "", fmt.Errorf("missing filter value")
	}
	switch v := value.(type) {
	case string:
		return v, nil
	case json.Number:
		return v.String(), nil
	case bool:
		if v {
			return "true", nil
		}
		return "false", nil
	default:
		return fmt.Sprintf("%v", v), nil
	}
}

func filterFloat(value interface{}) (float64, error) {
	switch v := value.(type) {
	case json.Number:
		return v.Float64()
	case float64:
		return v, nil
	case float32:
		return float64(v), nil
	case int:
		return float64(v), nil
	case int64:
		return float64(v), nil
	case string:
		return strconv.ParseFloat(strings.TrimSpace(v), 64)
	default:
		return strconv.ParseFloat(fmt.Sprintf("%v", v), 64)
	}
}

func formatFilterFloat(v float64) string {
	return strconv.FormatFloat(v, 'f', -1, 64)
}

func truncateString(s string, maxLen int) string {
	if maxLen < 0 {
		return s
	}
	runes := []rune(s)
	if len(runes) <= maxLen {
		return s
	}
	return string(runes[:maxLen])
}

func escapeString(s string) string {
	s = strings.ReplaceAll(s, "'", "''")
	s = strings.ReplaceAll(s, "\\", "\\\\")
	return s
}