package slowquery

import (
	"context"
	"fmt"
	"strings"

	"github.com/pingcap/log"
	"go.uber.org/zap"
)

type slowlogDetailColumn struct {
	expr     string
	required []string
}

type slowlogAvailableField struct {
	name     string
	required []string
}

// slowlogDetailColumnSpecs lists slowlog detail SELECT expressions and their required
// storage columns. Columns whose required fields are absent are omitted from detail queries.
var slowlogDetailColumnSpecs = []slowlogDetailColumn{
	{expr: "id", required: []string{"id"}},
	{expr: "time", required: []string{"time"}},
	{expr: "digest", required: []string{"digest"}},
	{expr: "date", required: []string{"date"}},
	{expr: "instance", required: []string{"instance"}},
	{expr: "txn_start_ts", required: []string{"txn_start_ts"}},
	{expr: "user", required: []string{"user"}},
	{expr: "host", required: []string{"host"}},
	{expr: "conn_id", required: []string{"conn_id"}},
	{expr: "session_alias", required: []string{"session_alias"}},
	{expr: "exec_retry_count", required: []string{"exec_retry_count"}},
	{expr: "exec_retry_time", required: []string{"exec_retry_time"}},
	{expr: "query_time", required: []string{"query_time"}},
	{expr: "parse_time", required: []string{"parse_time"}},
	{expr: "compile_time", required: []string{"compile_time"}},
	{expr: "rewrite_time", required: []string{"rewrite_time"}},
	{expr: "preproc_subqueries", required: []string{"preproc_subqueries"}},
	{expr: "preproc_subqueries_time", required: []string{"preproc_subqueries_time"}},
	{expr: "optimize_time", required: []string{"optimize_time"}},
	{expr: "wait_ts", required: []string{"wait_ts"}},
	{expr: "prewrite_time", required: []string{"prewrite_time"}},
	{expr: "wait_prewrite_binlog_time", required: []string{"wait_prewrite_binlog_time"}},
	{expr: "commit_time", required: []string{"commit_time"}},
	{expr: "get_commit_ts_time", required: []string{"get_commit_ts_time"}},
	{expr: "commit_backoff_time", required: []string{"commit_backoff_time"}},
	{expr: "backoff_types", required: []string{"backoff_types"}},
	{expr: "resolve_lock_time", required: []string{"resolve_lock_time"}},
	{expr: "local_latch_wait_time", required: []string{"local_latch_wait_time"}},
	{expr: "write_keys", required: []string{"write_keys"}},
	{expr: "write_size", required: []string{"write_size"}},
	{expr: "prewrite_region", required: []string{"prewrite_region"}},
	{expr: "txn_retry", required: []string{"txn_retry"}},
	{expr: "cop_time", required: []string{"cop_time"}},
	{expr: "process_time", required: []string{"process_time"}},
	{expr: "wait_time", required: []string{"wait_time"}},
	{expr: "backoff_time", required: []string{"backoff_time"}},
	{expr: "lockkeys_time", required: []string{"lockkeys_time"}},
	{expr: "request_count", required: []string{"request_count"}},
	{expr: "total_keys", required: []string{"total_keys"}},
	{expr: "process_keys", required: []string{"process_keys"}},
	{expr: "rocksdb_delete_skipped_count", required: []string{"rocksdb_delete_skipped_count"}},
	{expr: "rocksdb_key_skipped_count", required: []string{"rocksdb_key_skipped_count"}},
	{expr: "rocksdb_block_cache_hit_count", required: []string{"rocksdb_block_cache_hit_count"}},
	{expr: "rocksdb_block_read_count", required: []string{"rocksdb_block_read_count"}},
	{expr: "rocksdb_block_read_byte", required: []string{"rocksdb_block_read_byte"}},
	{expr: "db", required: []string{"db"}},
	{expr: "index_names", required: []string{"index_names"}},
	{expr: "is_internal", required: []string{"is_internal"}},
	{expr: "stats", required: []string{"stats"}},
	{expr: "cop_proc_avg", required: []string{"cop_proc_avg"}},
	{expr: "cop_proc_p90", required: []string{"cop_proc_p90"}},
	{expr: "cop_proc_max", required: []string{"cop_proc_max"}},
	{expr: "cop_proc_addr", required: []string{"cop_proc_addr"}},
	{expr: "cop_wait_avg", required: []string{"cop_wait_avg"}},
	{expr: "cop_wait_p90", required: []string{"cop_wait_p90"}},
	{expr: "cop_wait_max", required: []string{"cop_wait_max"}},
	{expr: "cop_wait_addr", required: []string{"cop_wait_addr"}},
	{expr: "mem_max", required: []string{"mem_max"}},
	{expr: "disk_max", required: []string{"disk_max"}},
	{expr: "kv_total", required: []string{"kv_total"}},
	{expr: "pd_total", required: []string{"pd_total"}},
	{expr: "backoff_total", required: []string{"backoff_total"}},
	{expr: "write_sql_response_total", required: []string{"write_sql_response_total"}},
	{expr: "result_rows", required: []string{"result_rows"}},
	{expr: "warnings", required: []string{"warnings"}},
	{expr: "backoff_detail", required: []string{"backoff_detail"}},
	{expr: "prepared", required: []string{"prepared"}},
	{expr: "succ", required: []string{"succ"}},
	{expr: "is_explicit_txn", required: []string{"is_explicit_txn"}},
	{expr: "is_write_cache_table", required: []string{"is_write_cache_table"}},
	{expr: "plan_from_cache", required: []string{"plan_from_cache"}},
	{expr: "plan_from_binding", required: []string{"plan_from_binding"}},
	{expr: "has_more_results", required: []string{"has_more_results"}},
	{expr: "resource_group", required: []string{"resource_group"}},
	{expr: "request_unit_read", required: []string{"request_unit_read"}},
	{expr: "request_unit_write", required: []string{"request_unit_write"}},
	{expr: "time_queued_by_rc", required: []string{"time_queued_by_rc"}},
	{expr: "plan_digest", required: []string{"plan_digest"}},
	{expr: "prev_stmt", required: []string{"prev_stmt"}},
	{expr: "serverless_tenant_id", required: []string{"serverless_tenant_id"}},
	{expr: "serverless_project_id", required: []string{"serverless_project_id"}},
	{expr: "serverless_cluster_id", required: []string{"serverless_cluster_id"}},
	{expr: "worker_role", required: []string{"worker_role"}},
	{expr: "is_tiflash", required: []string{"is_tiflash"}},
	{expr: "tidb_cpu_time", required: []string{"tidb_cpu_time"}},
	{expr: "tikv_cpu_time", required: []string{"tikv_cpu_time"}},
	{expr: "ia_remote_read_segment_count", required: []string{"ia_remote_read_segment_count"}},
	{expr: "ia_remote_read_segment_size", required: []string{"ia_remote_read_segment_size"}},
	{expr: "ia_remote_read_segment_wait_time", required: []string{"ia_remote_read_segment_wait_time"}},
	{expr: "(COALESCE(request_unit_read, 0) + COALESCE(request_unit_write, 0)) AS ru", required: []string{"request_unit_read", "request_unit_write"}},
	{expr: "(COALESCE(request_unit_v2, 0)) AS ru_v2", required: []string{"request_unit_v2"}},
	{expr: "query", required: []string{"query"}},
	{expr: "binary_plan", required: []string{"binary_plan"}},
	{expr: "plan", required: []string{"plan"}},
	{expr: "read_pool_task_details", required: []string{"read_pool_task_details"}},
}

// slowlogAvailableFieldSpecs maps API field names used by /slow_query/list `fields`
// and /slow_query/available_fields to required Delta Lake storage columns.
var slowlogAvailableFieldSpecs = []slowlogAvailableField{
	{name: "digest", required: []string{"digest"}},
	{name: "query", required: []string{"query"}},
	{name: "instance", required: []string{"instance"}},
	{name: "db", required: []string{"db"}},
	{name: "connection_id", required: []string{"conn_id"}},
	{name: "success", required: []string{"succ"}},
	{name: "timestamp", required: []string{"time"}},
	{name: "query_time", required: []string{"query_time"}},
	{name: "parse_time", required: []string{"parse_time"}},
	{name: "compile_time", required: []string{"compile_time"}},
	{name: "rewrite_time", required: []string{"rewrite_time"}},
	{name: "preproc_subqueries_time", required: []string{"preproc_subqueries_time"}},
	{name: "optimize_time", required: []string{"optimize_time"}},
	{name: "wait_ts", required: []string{"wait_ts"}},
	{name: "cop_time", required: []string{"cop_time"}},
	{name: "lock_keys_time", required: []string{"lockkeys_time"}},
	{name: "write_sql_response_total", required: []string{"write_sql_response_total"}},
	{name: "exec_retry_time", required: []string{"exec_retry_time"}},
	{name: "memory_max", required: []string{"mem_max"}},
	{name: "disk_max", required: []string{"disk_max"}},
	{name: "txn_start_ts", required: []string{"txn_start_ts"}},
	{name: "prev_stmt", required: []string{"prev_stmt"}},
	{name: "plan", required: []string{"plan"}},
	{name: "read_pool_task_details", required: []string{"read_pool_task_details"}},
	{name: "is_internal", required: []string{"is_internal"}},
	{name: "index_names", required: []string{"index_names"}},
	{name: "stats", required: []string{"stats"}},
	{name: "backoff_types", required: []string{"backoff_types"}},
	{name: "prepared", required: []string{"prepared"}},
	{name: "plan_from_cache", required: []string{"plan_from_cache"}},
	{name: "plan_from_binding", required: []string{"plan_from_binding"}},
	{name: "user", required: []string{"user"}},
	{name: "host", required: []string{"host"}},
	{name: "process_time", required: []string{"process_time"}},
	{name: "wait_time", required: []string{"wait_time"}},
	{name: "backoff_time", required: []string{"backoff_time"}},
	{name: "get_commit_ts_time", required: []string{"get_commit_ts_time"}},
	{name: "local_latch_wait_time", required: []string{"local_latch_wait_time"}},
	{name: "resolve_lock_time", required: []string{"resolve_lock_time"}},
	{name: "prewrite_time", required: []string{"prewrite_time"}},
	{name: "wait_prewrite_binlog_time", required: []string{"wait_prewrite_binlog_time"}},
	{name: "commit_time", required: []string{"commit_time"}},
	{name: "commit_backoff_time", required: []string{"commit_backoff_time"}},
	{name: "cop_proc_avg", required: []string{"cop_proc_avg"}},
	{name: "cop_proc_p90", required: []string{"cop_proc_p90"}},
	{name: "cop_proc_max", required: []string{"cop_proc_max"}},
	{name: "cop_wait_avg", required: []string{"cop_wait_avg"}},
	{name: "cop_wait_p90", required: []string{"cop_wait_p90"}},
	{name: "cop_wait_max", required: []string{"cop_wait_max"}},
	{name: "write_keys", required: []string{"write_keys"}},
	{name: "write_size", required: []string{"write_size"}},
	{name: "prewrite_region", required: []string{"prewrite_region"}},
	{name: "txn_retry", required: []string{"txn_retry"}},
	{name: "request_count", required: []string{"request_count"}},
	{name: "process_keys", required: []string{"process_keys"}},
	{name: "total_keys", required: []string{"total_keys"}},
	{name: "cop_proc_addr", required: []string{"cop_proc_addr"}},
	{name: "cop_wait_addr", required: []string{"cop_wait_addr"}},
	{name: "rocksdb_delete_skipped_count", required: []string{"rocksdb_delete_skipped_count"}},
	{name: "rocksdb_key_skipped_count", required: []string{"rocksdb_key_skipped_count"}},
	{name: "rocksdb_block_cache_hit_count", required: []string{"rocksdb_block_cache_hit_count"}},
	{name: "rocksdb_block_read_count", required: []string{"rocksdb_block_read_count"}},
	{name: "rocksdb_block_read_byte", required: []string{"rocksdb_block_read_byte"}},
	{name: "ru", required: []string{"request_unit_read", "request_unit_write"}},
	{name: "ru_v2", required: []string{"request_unit_v2"}},
	{name: "time_queued_by_rc", required: []string{"time_queued_by_rc"}},
	{name: "resource_group", required: []string{"resource_group"}},
	{name: "tidb_cpu_time", required: []string{"tidb_cpu_time"}},
	{name: "tikv_cpu_time", required: []string{"tikv_cpu_time"}},
	{name: "ia_remote_read_segment_count", required: []string{"ia_remote_read_segment_count"}},
	{name: "ia_remote_read_segment_size", required: []string{"ia_remote_read_segment_size"}},
	{name: "ia_remote_read_segment_wait_time", required: []string{"ia_remote_read_segment_wait_time"}},
}

func allSlowlogAvailableFieldNames() []string {
	names := make([]string, 0, len(slowlogAvailableFieldSpecs))
	for _, spec := range slowlogAvailableFieldSpecs {
		names = append(names, spec.name)
	}
	return names
}

func buildAvailableFieldsFromSchema(columns map[string]struct{}) []string {
	fields := make([]string, 0, len(slowlogAvailableFieldSpecs))
	for _, spec := range slowlogAvailableFieldSpecs {
		if hasSlowlogColumns(columns, spec.required...) {
			fields = append(fields, spec.name)
		}
	}
	return fields
}

func slowlogColumnNameSet(columns []string) map[string]struct{} {
	set := make(map[string]struct{}, len(columns))
	for _, column := range columns {
		set[strings.ToLower(strings.TrimSpace(column))] = struct{}{}
	}
	return set
}

func hasSlowlogColumns(columns map[string]struct{}, names ...string) bool {
	for _, name := range names {
		if _, ok := columns[strings.ToLower(name)]; !ok {
			return false
		}
	}
	return true
}

func buildDetailSelectColumnsFromSchema(columns map[string]struct{}) string {
	parts := make([]string, 0, len(slowlogDetailColumnSpecs))
	for _, spec := range slowlogDetailColumnSpecs {
		if hasSlowlogColumns(columns, spec.required...) {
			parts = append(parts, spec.expr)
		}
	}
	return strings.Join(parts, ", ")
}

type columnInfo struct {
	ColumnName string `db:"column_name"`
}

func (s *DuckDBAdapter) getSlowlogTableColumns(ctx context.Context, tablePath, alias string) ([]string, error) {
	query := fmt.Sprintf(
		`SELECT column_name FROM information_schema.columns WHERE table_name = '%s'`,
		alias,
	)

	var columns []columnInfo
	if err := s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &columns, alias, tablePath, query); err != nil {
		log.Error("Failed to get slowlog table schema",
			zap.String("tablePath", tablePath),
			zap.String("query", query),
			zap.Error(err))
		return nil, fmt.Errorf("failed to get slowlog table schema: %w", err)
	}

	names := make([]string, 0, len(columns))
	for _, col := range columns {
		names = append(names, col.ColumnName)
	}
	return names, nil
}