package engine

import (
	"strings"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

// StandardTableSchema returns the logical schema for a supported standard table.
func StandardTableSchema(table string) (*dto.SchemaResponse, bool) {
	table = strings.ToLower(strings.TrimSpace(table))
	switch table {
	case "slow_query_logs":
		return &dto.SchemaResponse{
			Table:      table,
			Columns:    slowQueryColumns(),
			Partitions: []dto.PartitionDef{{Name: "date", Type: "STRING"}},
		}, true
	case "topsql_tidb", "topsql_tikv":
		return &dto.SchemaResponse{
			Table:      table,
			Columns:    topsqlMetricColumns(),
			Partitions: []dto.PartitionDef{{Name: "instance", Type: "STRING"}, {Name: "date", Type: "STRING"}},
		}, true
	case "topsql_plan_meta":
		return &dto.SchemaResponse{
			Table:      table,
			Columns:    topsqlPlanMetaColumns(),
			Partitions: []dto.PartitionDef{{Name: "date", Type: "STRING"}},
		}, true
	case "topsql_sql_meta":
		return &dto.SchemaResponse{
			Table:      table,
			Columns:    topsqlSQLMetaColumns(),
			Partitions: []dto.PartitionDef{{Name: "date", Type: "STRING"}},
		}, true
	case "topsql_topru":
		return &dto.SchemaResponse{
			Table:      table,
			Columns:    topsqlTopRUColumns(),
			Partitions: []dto.PartitionDef{{Name: "instance", Type: "STRING"}, {Name: "date", Type: "STRING"}},
		}, true
	case "topsql":
		return &dto.SchemaResponse{
			Table:      table,
			Columns:    topsqlViewColumns(),
			Partitions: []dto.PartitionDef{{Name: "date", Type: "STRING"}},
			IsView:     true,
		}, true
	case "topru":
		return &dto.SchemaResponse{
			Table:      table,
			Columns:    topruViewColumns(),
			Partitions: []dto.PartitionDef{{Name: "date", Type: "STRING"}},
			IsView:     true,
		}, true
	case "log_stats":
		return &dto.SchemaResponse{
			Table: table,
			Columns: []dto.ColumnInfo{
				col("hour", "INT"),
				col("project", "STRING"),
				col("level", "STRING"),
				col("file", "STRING"),
				col("line", "INT"),
				col("count", "BIGINT"),
				col("msg_digest", "STRING"),
				col("sample", "STRING"),
				col("component", "STRING"),
			},
			Partitions: []dto.PartitionDef{
				{Name: "date", Type: "STRING"},
				{Name: "org", Type: "STRING"},
				{Name: "cluster", Type: "STRING"},
			},
			IsView: true,
		}, true
	default:
		return nil, false
	}
}

// LogicalStandardTableName strips the cluster suffix used by physical table names.
func LogicalStandardTableName(table, clusterID string) string {
	table = strings.ToLower(strings.TrimSpace(table))
	if clusterID != "" {
		suffix := "_" + clusterID
		if strings.HasSuffix(table, suffix) {
			table = strings.TrimSuffix(table, suffix)
		}
	}
	return table
}

func col(name, typ string) dto.ColumnInfo {
	return dto.ColumnInfo{Name: name, Type: typ}
}

func slowQueryColumns() []dto.ColumnInfo {
	return []dto.ColumnInfo{
		col("id", "BIGINT"),
		col("time", "DOUBLE"),
		col("digest", "STRING"),
		col("instance", "STRING"),
		col("txn_start_ts", "STRING"),
		col("user", "STRING"),
		col("host", "STRING"),
		col("conn_id", "STRING"),
		col("session_alias", "STRING"),
		col("exec_retry_count", "STRING"),
		col("exec_retry_time", "DOUBLE"),
		col("query_time", "DOUBLE"),
		col("parse_time", "DOUBLE"),
		col("compile_time", "DOUBLE"),
		col("rewrite_time", "DOUBLE"),
		col("preproc_subqueries", "STRING"),
		col("preproc_subqueries_time", "DOUBLE"),
		col("optimize_time", "DOUBLE"),
		col("wait_ts", "DOUBLE"),
		col("prewrite_time", "DOUBLE"),
		col("wait_prewrite_binlog_time", "DOUBLE"),
		col("commit_time", "DOUBLE"),
		col("get_commit_ts_time", "DOUBLE"),
		col("commit_backoff_time", "DOUBLE"),
		col("backoff_types", "STRING"),
		col("resolve_lock_time", "DOUBLE"),
		col("local_latch_wait_time", "DOUBLE"),
		col("write_keys", "BIGINT"),
		col("write_size", "BIGINT"),
		col("prewrite_region", "BIGINT"),
		col("txn_retry", "BIGINT"),
		col("cop_time", "DOUBLE"),
		col("process_time", "DOUBLE"),
		col("wait_time", "DOUBLE"),
		col("backoff_time", "DOUBLE"),
		col("lockkeys_time", "DOUBLE"),
		col("request_count", "STRING"),
		col("total_keys", "STRING"),
		col("process_keys", "STRING"),
		col("rocksdb_delete_skipped_count", "STRING"),
		col("rocksdb_key_skipped_count", "STRING"),
		col("rocksdb_block_cache_hit_count", "STRING"),
		col("rocksdb_block_read_count", "STRING"),
		col("rocksdb_block_read_byte", "STRING"),
		col("db", "STRING"),
		col("index_names", "STRING"),
		col("is_internal", "BOOLEAN"),
		col("stats", "STRING"),
		col("cop_proc_avg", "DOUBLE"),
		col("cop_proc_p90", "DOUBLE"),
		col("cop_proc_max", "DOUBLE"),
		col("cop_proc_addr", "STRING"),
		col("cop_wait_avg", "DOUBLE"),
		col("cop_wait_p90", "DOUBLE"),
		col("cop_wait_max", "DOUBLE"),
		col("cop_wait_addr", "STRING"),
		col("mem_max", "BIGINT"),
		col("disk_max", "BIGINT"),
		col("kv_total", "DOUBLE"),
		col("pd_total", "DOUBLE"),
		col("backoff_total", "DOUBLE"),
		col("write_sql_response_total", "DOUBLE"),
		col("result_rows", "BIGINT"),
		col("warnings", "STRING"),
		col("backoff_detail", "STRING"),
		col("prepared", "BOOLEAN"),
		col("succ", "BOOLEAN"),
		col("is_explicit_txn", "BOOLEAN"),
		col("is_write_cache_table", "BOOLEAN"),
		col("plan_from_cache", "BOOLEAN"),
		col("plan_from_binding", "BOOLEAN"),
		col("has_more_results", "BOOLEAN"),
		col("resource_group", "STRING"),
		col("request_unit_read", "DOUBLE"),
		col("request_unit_write", "DOUBLE"),
		col("time_queued_by_rc", "DOUBLE"),
		col("plan_digest", "STRING"),
		col("prev_stmt", "STRING"),
		col("serverless_tenant_id", "STRING"),
		col("serverless_project_id", "STRING"),
		col("serverless_cluster_id", "STRING"),
		col("worker_role", "STRING"),
		col("is_tiflash", "BOOLEAN"),
		col("query", "STRING"),
		col("binary_plan", "STRING"),
		col("plan", "STRING"),
		col("decoded_plan", "STRING"),
	}
}

func topsqlMetricColumns() []dto.ColumnInfo {
	return []dto.ColumnInfo{
		col("db", "STRING"),
		col("keyspace", "STRING"),
		col("plan_digest", "STRING"),
		col("region_id", "STRING"),
		col("sql_digest", "STRING"),
		col("table", "STRING"),
		col("table_id", "BIGINT"),
		col("tag_label", "STRING"),
		col("timestamps", "BIGINT"),
		col("topsql_cpu_time_ms", "INT"),
		col("topsql_logical_read_bytes", "BIGINT"),
		col("topsql_logical_write_bytes", "BIGINT"),
		col("topsql_network_in_bytes", "BIGINT"),
		col("topsql_network_out_bytes", "BIGINT"),
		col("topsql_read_keys", "BIGINT"),
		col("topsql_stmt_duration_count", "BIGINT"),
		col("topsql_stmt_duration_sum_ns", "BIGINT"),
		col("topsql_stmt_exec_count", "BIGINT"),
		col("topsql_write_keys", "BIGINT"),
	}
}

func topsqlPlanMetaColumns() []dto.ColumnInfo {
	return []dto.ColumnInfo{
		col("encoded_normalized_plan", "STRING"),
		col("normalized_plan", "STRING"),
		col("plan_digest", "STRING"),
	}
}

func topsqlSQLMetaColumns() []dto.ColumnInfo {
	return []dto.ColumnInfo{
		col("normalized_sql", "STRING"),
		col("sql_digest", "STRING"),
	}
}

func topsqlTopRUColumns() []dto.ColumnInfo {
	return []dto.ColumnInfo{
		col("keyspace", "STRING"),
		col("plan_digest", "STRING"),
		col("sql_digest", "STRING"),
		col("timestamps", "BIGINT"),
		col("topru_exec_count", "BIGINT"),
		col("topru_exec_duration", "BIGINT"),
		col("topru_total_ru", "DOUBLE"),
		col("user", "STRING"),
	}
}

func topsqlViewColumns() []dto.ColumnInfo {
	cols := append([]dto.ColumnInfo{}, topsqlMetricColumns()...)
	cols = append(cols,
		col("instance", "STRING"),
		col("date", "STRING"),
		col("component", "STRING"),
		col("normalized_sql", "STRING"),
		col("normalized_plan", "STRING"),
		col("encoded_normalized_plan", "STRING"),
	)
	return cols
}

func topruViewColumns() []dto.ColumnInfo {
	cols := append([]dto.ColumnInfo{}, topsqlTopRUColumns()...)
	cols = append(cols,
		col("instance", "STRING"),
		col("date", "STRING"),
		col("normalized_sql", "STRING"),
		col("normalized_plan", "STRING"),
		col("encoded_normalized_plan", "STRING"),
	)
	return cols
}