package topsqlv2

import (
	"context"
	"fmt"
	"strings"
	"unicode"

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

// topsqlOrderByFieldSpec maps an order_by / available_fields name to required
// Delta Lake storage columns under type=topsql/component={tidb|tikv}/instance=*.
type topsqlOrderByFieldSpec struct {
	name     string
	required []string
}

// Order of this list is the order returned by /topsql/available_fields.
var topsqlOrderByFieldSpecs = []topsqlOrderByFieldSpec{
	{name: "cpu", required: []string{"topsql_cpu_time_ms"}},
	{name: "network", required: []string{"topsql_network_in_bytes", "topsql_network_out_bytes"}},
	{name: "network_in", required: []string{"topsql_network_in_bytes"}},
	{name: "network_out", required: []string{"topsql_network_out_bytes"}},
	{name: "logical_io", required: []string{"topsql_logical_read_bytes", "topsql_logical_write_bytes"}},
	{name: "logical_read", required: []string{"topsql_logical_read_bytes"}},
	{name: "logical_write", required: []string{"topsql_logical_write_bytes"}},
	{name: "rocksdb_block_read", required: []string{"topsql_rocksdb_block_read_count"}},
}

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

func columnNameSet(names []string) map[string]struct{} {
	out := make(map[string]struct{}, len(names))
	for _, name := range names {
		out[strings.ToLower(strings.TrimSpace(name))] = struct{}{}
	}
	return out
}

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

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

func orderByRequiredColumns(orderBy string) ([]string, bool) {
	for _, spec := range topsqlOrderByFieldSpecs {
		if spec.name == orderBy {
			return spec.required, true
		}
	}
	return nil, false
}

func topsqlTableAlias(clusterID, instanceType, instance string) string {
	raw := fmt.Sprintf("topsql_%s_%s_%s", instanceType, clusterID, instance)
	var b strings.Builder
	b.Grow(len(raw))
	for _, r := range raw {
		if unicode.IsLetter(r) || unicode.IsDigit(r) {
			// Lowercase so ATTACH identifier and information_schema lookups stay consistent.
			b.WriteRune(unicode.ToLower(r))
			continue
		}
		b.WriteByte('_')
	}
	return b.String()
}

func (s *Service) getTopSQLTableColumns(ctx context.Context, tablePath, alias string) ([]string, error) {
	// Project only column_name: DuckDB DESCRIBE also returns column_type/null/key/...
	// and sqlx StructScan rejects unmapped result columns.
	names, describeErr := s.queryTopSQLTableColumns(ctx, tablePath, alias, fmt.Sprintf(
		`SELECT column_name FROM (DESCRIBE SELECT * FROM "%s")`, alias,
	))
	if describeErr == nil && len(names) > 0 {
		return names, nil
	}
	if describeErr != nil {
		log.Warn("DESCRIBE topsql schema failed, falling back to information_schema",
			zap.String("tablePath", tablePath),
			zap.String("alias", alias),
			zap.Error(describeErr))
	}

	// Fallback: Delta ATTACH registers the alias as both catalog and table name.
	// Match case-insensitively in case an older connection attached a mixed-case alias.
	names, err := s.queryTopSQLTableColumns(ctx, tablePath, alias, fmt.Sprintf(
		`SELECT column_name FROM information_schema.columns WHERE lower(table_name) = lower('%s') OR lower(table_catalog) = lower('%s')`,
		alias, alias,
	))
	if err != nil {
		if describeErr != nil {
			return nil, fmt.Errorf("failed to get topsql table schema: describe: %w; information_schema: %v", describeErr, err)
		}
		return nil, err
	}
	if len(names) == 0 {
		log.Warn("topsql table schema is empty",
			zap.String("tablePath", tablePath),
			zap.String("alias", alias))
		return nil, fmt.Errorf("topsql table schema is empty for alias %s (path %s)", alias, tablePath)
	}
	return names, nil
}

func (s *Service) queryTopSQLTableColumns(ctx context.Context, tablePath, alias, query string) ([]string, error) {
	var columns []topsqlColumnInfo
	if err := s.SQLsDuckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &columns, alias, tablePath, query); err != nil {
		log.Error("Failed to get topsql table schema",
			zap.String("tablePath", tablePath),
			zap.String("alias", alias),
			zap.String("query", query),
			zap.Error(err))
		return nil, fmt.Errorf("failed to get topsql table schema: %w", err)
	}

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

// topsqlMetricSQLFragments holds schema-filtered projection pieces for summary SQL.
type topsqlMetricSQLFragments struct {
	RawSelectExprs  []string // expressions after timestamps / group fields
	BucketSumExprs  []string // SUM(...) AS total_* in time_bucket_data
	MetricSumSelect string   // columns projected from time_bucket_data
	MetricSumAgg    string   // SUM(total_*) AS total_* for "other" union
}

func buildTopSQLMetricSQLFragments(columns map[string]struct{}) topsqlMetricSQLFragments {
	raw := []string{"topsql_cpu_time_ms"}
	bucket := []string{"SUM(topsql_cpu_time_ms) AS total_cpu_ms"}
	selectCols := []string{"total_cpu_ms"}
	aggCols := []string{"SUM(total_cpu_ms) AS total_cpu_ms"}

	add := func(rawExpr, bucketExpr, selectCol, aggExpr string) {
		raw = append(raw, rawExpr)
		bucket = append(bucket, bucketExpr)
		selectCols = append(selectCols, selectCol)
		aggCols = append(aggCols, aggExpr)
	}

	hasIn := hasColumns(columns, "topsql_network_in_bytes")
	hasOut := hasColumns(columns, "topsql_network_out_bytes")
	if hasIn {
		add(
			"COALESCE(CAST(topsql_network_in_bytes AS DOUBLE), 0) AS network_in_bytes",
			"SUM(network_in_bytes) AS total_network_in_bytes",
			"total_network_in_bytes",
			"SUM(total_network_in_bytes) AS total_network_in_bytes",
		)
	}
	if hasOut {
		add(
			"COALESCE(CAST(topsql_network_out_bytes AS DOUBLE), 0) AS network_out_bytes",
			"SUM(network_out_bytes) AS total_network_out_bytes",
			"total_network_out_bytes",
			"SUM(total_network_out_bytes) AS total_network_out_bytes",
		)
	}
	if hasIn && hasOut {
		add(
			"(COALESCE(CAST(topsql_network_in_bytes AS DOUBLE), 0) + COALESCE(CAST(topsql_network_out_bytes AS DOUBLE), 0)) AS network_bytes",
			"SUM(network_bytes) AS total_network_bytes",
			"total_network_bytes",
			"SUM(total_network_bytes) AS total_network_bytes",
		)
	}

	hasRead := hasColumns(columns, "topsql_logical_read_bytes")
	hasWrite := hasColumns(columns, "topsql_logical_write_bytes")
	if hasRead {
		add(
			"COALESCE(CAST(topsql_logical_read_bytes AS DOUBLE), 0) AS logical_read_bytes",
			"SUM(logical_read_bytes) AS total_logical_read_bytes",
			"total_logical_read_bytes",
			"SUM(total_logical_read_bytes) AS total_logical_read_bytes",
		)
	}
	if hasWrite {
		add(
			"COALESCE(CAST(topsql_logical_write_bytes AS DOUBLE), 0) AS logical_write_bytes",
			"SUM(logical_write_bytes) AS total_logical_write_bytes",
			"total_logical_write_bytes",
			"SUM(total_logical_write_bytes) AS total_logical_write_bytes",
		)
	}
	if hasRead && hasWrite {
		add(
			"(COALESCE(CAST(topsql_logical_read_bytes AS DOUBLE), 0) + COALESCE(CAST(topsql_logical_write_bytes AS DOUBLE), 0)) AS logical_io_bytes",
			"SUM(logical_io_bytes) AS total_logical_io_bytes",
			"total_logical_io_bytes",
			"SUM(total_logical_io_bytes) AS total_logical_io_bytes",
		)
	}

	if hasColumns(columns, "topsql_rocksdb_block_read_count") {
		add(
			"COALESCE(CAST(topsql_rocksdb_block_read_count AS DOUBLE), 0) AS rocksdb_block_read_count",
			"SUM(rocksdb_block_read_count) AS total_rocksdb_block_read_count",
			"total_rocksdb_block_read_count",
			"SUM(total_rocksdb_block_read_count) AS total_rocksdb_block_read_count",
		)
	}

	raw = append(raw,
		"topsql_stmt_exec_count",
		"topsql_stmt_duration_sum_ns",
		"topsql_stmt_duration_count",
	)
	bucket = append(bucket,
		"SUM(topsql_stmt_exec_count) AS total_stmt_exec_count",
		"SUM(topsql_stmt_duration_sum_ns) AS total_stmt_duration_sum_ns",
		"SUM(topsql_stmt_duration_count) AS total_stmt_duration_count",
	)

	return topsqlMetricSQLFragments{
		RawSelectExprs:  raw,
		BucketSumExprs:  bucket,
		MetricSumSelect: strings.Join(selectCols, ",\n\t\t"),
		MetricSumAgg:    strings.Join(aggCols, ",\n\t\t"),
	}
}