package query

import (
	"fmt"
	"strings"
)

const (
	logsDatabaseName         = "o11y-logs"
	generalLogsBase          = "general_logs"
	logStatsBase             = "log_stats"
	topsqlBase               = "topsql"
	topsqlSQLMetaBase        = "topsql_sql_meta"
	topsqlPlanMetaBase       = "topsql_plan_meta"
	topruBase                = "topru"
	sqlstatementBase         = "sqlstatement"
	slowlogsBase             = "slowlogs"
	logsDataPathSuffix       = "deltalake/global/logs/"
	logStatsDataPathSuffix   = "deltalake/global/log-stats/"
	orgClusterDataPathSuffix = "deltalake/"
)

// ColumnMeta describes a table column for API responses.
type ColumnMeta struct {
	Name        string `json:"name"`
	Type        string `json:"type"`
	Description string `json:"description,omitempty"`
}

// TableDefinition is a built-in OLAP table schema known to diagnosis-query.
type TableDefinition struct {
	ID                  string
	Dataset             string // dataset id for /olap/tables (logs | log-stats | topsql | …)
	Description         string
	FileColumns         []ColumnMeta
	PartitionColumns    []ColumnMeta
	RequiresScope       bool
	PartitionProjection map[string]string
}

// ResolvedTable is a region-specific Glue table registration.
type ResolvedTable struct {
	Definition     TableDefinition
	Database       string
	Name           string
	FullName       string
	Location       string
	LocationPrefix string
}

// CatalogStatus values for Glue registration.
const (
	CatalogStatusReady        = "READY"
	CatalogStatusMissing      = "MISSING"
	CatalogStatusNotSupported = "NOT_SUPPORTED"
)

// DataStatus values for underlying S3 objects.
const (
	DataStatusAvailable = "AVAILABLE"
	DataStatusEmpty     = "EMPTY"
	DataStatusUnknown   = "UNKNOWN"
)

func generalLogsTableDef() TableDefinition {
	return TableDefinition{
		ID:            generalLogsBase,
		Dataset:       DatasetLogs,
		Description:   "Offline general logs stored as hourly Parquet under deltalake/global/logs/",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "event_time", Type: "string", Description: "Log time (ISO 8601, subsecond when available)"},
			{Name: "timestamp", Type: "bigint", Description: "Unix seconds"},
			{Name: "host", Type: "string"},
			{Name: "pod", Type: "string"},
			{Name: "level", Type: "string"},
			{Name: "position", Type: "string"},
			{Name: "message", Type: "string"},
			{Name: "cluster", Type: "string"},
			{Name: "instance", Type: "string"},
			{Name: "tenant_id", Type: "string", Description: "Tenant / org ID (file column)"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "env", Type: "string"},
			{Name: "deploytype", Type: "string"},
			{Name: "date", Type: "string"},
			{Name: "component", Type: "string"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":           "true",
			"projection.env.type":          "enum",
			"projection.env.values":        "prod,staging,dev",
			"projection.deploytype.type":   "enum",
			"projection.deploytype.values": "dedicated,premium,serverless",
			"projection.date.type":         "date",
			"projection.date.range":        "2024-01-01,NOW",
			"projection.date.format":       "yyyy-MM-dd",
			"projection.component.type":    "injected",
			"classification":               "parquet",
			"EXTERNAL":                     "TRUE",
		},
	}
}

// logStatsTableDef matches Deltalake output from deltalake-ops-server log_stats.
// Path: deltalake/global/log-stats/date={YYYYMMDD}/org={org}/cluster={cluster}/
func logStatsTableDef() TableDefinition {
	return TableDefinition{
		ID:            logStatsBase,
		Dataset:       DatasetLogStats,
		Description:   "Offline log statistics (Clinic aggregates) under deltalake/global/log-stats/",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "hour", Type: "bigint", Description: "UTC hour of day (0-23)"},
			{Name: "project", Type: "string"},
			{Name: "level", Type: "string"},
			{Name: "file", Type: "string", Description: "Source file from position"},
			{Name: "line", Type: "string", Description: "Source line from position"},
			{Name: "count", Type: "string", Description: "Occurrence count (string in Parquet)"},
			{Name: "component", Type: "string"},
			{Name: "msg_digest", Type: "string"},
			{Name: "sample", Type: "string", Description: "Sample log line"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "date", Type: "string", Description: "UTC date as yyyyMMdd (e.g. 20260319)"},
			{Name: "org", Type: "string"},
			{Name: "cluster", Type: "string"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":      "true",
			"projection.date.type":    "integer",
			"projection.date.range":   "20240101,20991231",
			"projection.date.digits":  "8",
			"projection.org.type":     "injected",
			"projection.cluster.type": "injected",
			"classification":          "parquet",
			"EXTERNAL":                "TRUE",
		},
	}
}

// topsqlTableDef matches TopSQL instance Delta tables.
// Path: deltalake/org={org}/cluster={cluster}/type=topsql/component={component}/instance={instance}/date={yyyy-MM-dd}/
func topsqlTableDef() TableDefinition {
	return TableDefinition{
		ID:            topsqlBase,
		Dataset:       DatasetTopsql,
		Description:   "TopSQL instance metrics under deltalake/org=/cluster=/type=topsql/",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "sql_digest", Type: "string"},
			{Name: "plan_digest", Type: "string"},
			{Name: "timestamps", Type: "bigint", Description: "Unix seconds"},
			{Name: "topsql_cpu_time_ms", Type: "bigint"},
			{Name: "topsql_stmt_exec_count", Type: "bigint"},
			{Name: "topsql_stmt_duration_sum_ns", Type: "bigint"},
			{Name: "topsql_stmt_duration_count", Type: "bigint"},
			{Name: "topsql_read_keys", Type: "bigint"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "org", Type: "string"},
			{Name: "cluster", Type: "string"},
			{Name: "component", Type: "string", Description: "tidb / tikv"},
			{Name: "instance", Type: "string"},
			{Name: "date", Type: "string", Description: "UTC date yyyy-MM-dd"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":        "true",
			"projection.org.type":       "injected",
			"projection.cluster.type":   "injected",
			"projection.component.type": "injected",
			"projection.instance.type":  "injected",
			"projection.date.type":      "date",
			"projection.date.range":     "2024-01-01,NOW",
			"projection.date.format":    "yyyy-MM-dd",
			"classification":            "parquet",
			"EXTERNAL":                  "TRUE",
		},
	}
}

// topsqlSQLMetaTableDef matches topsql_sql_meta (normalized SQL text by digest).
// Path: deltalake/org=/cluster=/type=topsql/component=topsql_sql_meta (no instance).
func topsqlSQLMetaTableDef() TableDefinition {
	return TableDefinition{
		ID:            topsqlSQLMetaBase,
		Dataset:       DatasetTopsqlSQLMeta,
		Description:   "TopSQL SQL text meta under component=topsql_sql_meta",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "sql_digest", Type: "string"},
			{Name: "normalized_sql", Type: "string"},
			{Name: "date", Type: "string", Description: "UTC date yyyy-MM-dd (file column)"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "org", Type: "string"},
			{Name: "cluster", Type: "string"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":      "true",
			"projection.org.type":     "injected",
			"projection.cluster.type": "injected",
			"classification":          "parquet",
			"EXTERNAL":                "TRUE",
		},
	}
}

// topsqlPlanMetaTableDef matches topsql_plan_meta (normalized plan text by digest).
// Path: deltalake/org=/cluster=/type=topsql/component=topsql_plan_meta (no instance).
func topsqlPlanMetaTableDef() TableDefinition {
	return TableDefinition{
		ID:            topsqlPlanMetaBase,
		Dataset:       DatasetTopsqlPlanMeta,
		Description:   "TopSQL plan text meta under component=topsql_plan_meta",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "plan_digest", Type: "string"},
			{Name: "normalized_plan", Type: "string"},
			{Name: "encoded_normalized_plan", Type: "string", Description: "base64+snappy+plancodec when normalized_plan is empty"},
			{Name: "date", Type: "string", Description: "UTC date yyyy-MM-dd (file column)"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "org", Type: "string"},
			{Name: "cluster", Type: "string"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":      "true",
			"projection.org.type":     "injected",
			"projection.cluster.type": "injected",
			"classification":          "parquet",
			"EXTERNAL":                "TRUE",
		},
	}
}

// topruTableDef matches TopRU metrics (same topsql_data_deltalake layout as topsql).
// Path: deltalake/org=/cluster=/type=topsql/component=topru/instance=default/date=/
func topruTableDef() TableDefinition {
	return TableDefinition{
		ID:            topruBase,
		Dataset:       DatasetTopru,
		Description:   "TopRU metrics under component=topru/instance=default",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "sql_digest", Type: "string"},
			{Name: "plan_digest", Type: "string"},
			{Name: "timestamps", Type: "bigint", Description: "Unix seconds"},
			{Name: "topru_total_ru", Type: "double"},
			{Name: "topru_exec_count", Type: "bigint"},
			{Name: "topru_exec_duration", Type: "bigint", Description: "Duration sum in ns"},
			{Name: "user", Type: "string"},
			{Name: "keyspace", Type: "string"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "org", Type: "string"},
			{Name: "cluster", Type: "string"},
			{Name: "date", Type: "string", Description: "UTC date yyyy-MM-dd"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":      "true",
			"projection.org.type":     "injected",
			"projection.cluster.type": "injected",
			"projection.date.type":    "date",
			"projection.date.range":   "2024-01-01,NOW",
			"projection.date.format":  "yyyy-MM-dd",
			"classification":          "parquet",
			"EXTERNAL":                "TRUE",
		},
	}
}

// sqlstatementTableDef matches SQL statement summary Delta tables.
// Path: deltalake/org={org}/cluster={cluster}/type=sqlstatement/component=sqlstatement/date={yyyy-MM-dd}/
func sqlstatementTableDef() TableDefinition {
	return TableDefinition{
		ID:            sqlstatementBase,
		Dataset:       DatasetSQLStatement,
		Description:   "SQL statement summaries under deltalake/org=/cluster=/type=sqlstatement/",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "digest", Type: "string"},
			{Name: "schema_name", Type: "string"},
			{Name: "stmt_type", Type: "string"},
			{Name: "begin", Type: "bigint", Description: "Window begin (unix seconds)"},
			{Name: "end", Type: "bigint", Description: "Window end (unix seconds)"},
			{Name: "exec_count", Type: "bigint"},
			{Name: "sum_latency", Type: "bigint"},
			{Name: "avg_latency", Type: "bigint"},
			{Name: "max_latency", Type: "bigint"},
			{Name: "sample_user", Type: "string"},
			{Name: "table_names", Type: "string"},
			{Name: "plan_digest", Type: "string"},
			{Name: "sample_query", Type: "string"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "org", Type: "string"},
			{Name: "cluster", Type: "string"},
			{Name: "date", Type: "string", Description: "UTC date yyyy-MM-dd"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":      "true",
			"projection.org.type":     "injected",
			"projection.cluster.type": "injected",
			"projection.date.type":    "date",
			"projection.date.range":   "2024-01-01,NOW",
			"projection.date.format":  "yyyy-MM-dd",
			"classification":          "parquet",
			"EXTERNAL":                "TRUE",
		},
	}
}

// slowlogsTableDef matches the primary legacy Delta Lake layout used by slowquery.
// Path: deltalake/{org}/{cluster}/slowlogs/date={yyyyMMdd}/
// (org/cluster are bare path segments; date is a Hive key=value partition).
func slowlogsTableDef() TableDefinition {
	return TableDefinition{
		ID:            slowlogsBase,
		Dataset:       DatasetSlowlogs,
		Description:   "Slow query logs (legacy Delta): deltalake/{org}/{cluster}/slowlogs/date=",
		RequiresScope: true,
		FileColumns: []ColumnMeta{
			{Name: "id", Type: "bigint"},
			{Name: "time", Type: "double", Description: "Query timestamp (unix seconds, float)"},
			{Name: "digest", Type: "string"},
			{Name: "instance", Type: "string"},
			{Name: "user", Type: "string"},
			{Name: "host", Type: "string"},
			{Name: "db", Type: "string"},
			{Name: "query_time", Type: "double"},
			{Name: "parse_time", Type: "double"},
			{Name: "compile_time", Type: "double"},
			{Name: "process_time", Type: "double"},
			{Name: "mem_max", Type: "bigint"},
			{Name: "plan_digest", Type: "string"},
			{Name: "query", Type: "string"},
			{Name: "succ", Type: "boolean"},
			{Name: "is_internal", Type: "boolean"},
		},
		PartitionColumns: []ColumnMeta{
			{Name: "org", Type: "string", Description: "Tenant / org id (bare path segment)"},
			{Name: "cluster", Type: "string", Description: "O11y cluster UUID (bare path segment)"},
			{Name: "date", Type: "string", Description: "UTC date as yyyyMMdd (Hive path partition date=)"},
		},
		PartitionProjection: map[string]string{
			"projection.enabled":      "true",
			"projection.org.type":     "injected",
			"projection.cluster.type": "injected",
			"projection.date.type":    "integer",
			"projection.date.range":   "20240101,20991231",
			"projection.date.digits":  "8",
			"classification":          "parquet",
			"EXTERNAL":                "TRUE",
		},
	}
}

func resolveRegistryTables(cfg Config) []ResolvedTable {
	bucket := strings.TrimSpace(cfg.LogsBucket)
	if bucket == "" {
		bucket = defaultLogsBucket(cfg.LogsTableRegionKey())
	}
	key := cfg.LogsTableRegionKey()
	suffix := cfg.LogsTableSuffix

	tables := []ResolvedTable{
		resolveGeneralLogsTable(bucket, key, suffix),
		resolveLogStatsTable(bucket, key, suffix),
		resolveTopsqlTable(bucket, key, suffix),
		resolveTopsqlSQLMetaTable(bucket, key, suffix),
		resolveTopsqlPlanMetaTable(bucket, key, suffix),
		resolveTopruTable(bucket, key, suffix),
		resolveSQLStatementTable(bucket, key, suffix),
		resolveSlowlogsTable(bucket, key, suffix),
	}
	return append(tables, resolveMetricsStatusTables(bucket, key, suffix)...)
}

func resolveGeneralLogsTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, logsDataPathSuffix)
	tableName := generalLogsTableName(logicalRegion, suffix)
	template := fmt.Sprintf("s3://%s/%senv=${env}/deployType=${deploytype}/date=${date}/component=${component}/",
		bucket, logsDataPathSuffix)

	def := generalLogsTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func resolveLogStatsTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, logStatsDataPathSuffix)
	tableName := logStatsTableName(logicalRegion, suffix)
	template := fmt.Sprintf("s3://%s/%sdate=${date}/org=${org}/cluster=${cluster}/",
		bucket, logStatsDataPathSuffix)

	def := logStatsTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func resolveTopsqlTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, orgClusterDataPathSuffix)
	tableName := topsqlTableName(logicalRegion, suffix)
	template := fmt.Sprintf("s3://%s/%sorg=${org}/cluster=${cluster}/type=topsql/component=${component}/instance=${instance}/date=${date}/",
		bucket, orgClusterDataPathSuffix)

	def := topsqlTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func resolveTopsqlSQLMetaTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, orgClusterDataPathSuffix)
	tableName := topsqlSQLMetaTableName(logicalRegion, suffix)
	template := fmt.Sprintf("s3://%s/%sorg=${org}/cluster=${cluster}/type=topsql/component=topsql_sql_meta",
		bucket, orgClusterDataPathSuffix)

	def := topsqlSQLMetaTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func resolveTopsqlPlanMetaTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, orgClusterDataPathSuffix)
	tableName := topsqlPlanMetaTableName(logicalRegion, suffix)
	template := fmt.Sprintf("s3://%s/%sorg=${org}/cluster=${cluster}/type=topsql/component=topsql_plan_meta",
		bucket, orgClusterDataPathSuffix)

	def := topsqlPlanMetaTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func resolveTopruTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, orgClusterDataPathSuffix)
	tableName := topruTableName(logicalRegion, suffix)
	template := fmt.Sprintf("s3://%s/%sorg=${org}/cluster=${cluster}/type=topsql/component=topru/instance=default/date=${date}/",
		bucket, orgClusterDataPathSuffix)

	def := topruTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func resolveSQLStatementTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, orgClusterDataPathSuffix)
	tableName := sqlstatementTableName(logicalRegion, suffix)
	template := fmt.Sprintf("s3://%s/%sorg=${org}/cluster=${cluster}/type=sqlstatement/component=sqlstatement/date=${date}/",
		bucket, orgClusterDataPathSuffix)

	def := sqlstatementTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func resolveSlowlogsTable(bucket, logicalRegion, suffix string) ResolvedTable {
	location := fmt.Sprintf("s3://%s/%s", bucket, orgClusterDataPathSuffix)
	tableName := slowlogsTableName(logicalRegion, suffix)
	// Legacy Delta roots: deltalake/{org}/{cluster}/slowlogs/date={yyyyMMdd}/
	template := fmt.Sprintf("s3://%s/%s${org}/${cluster}/slowlogs/date=${date}", bucket, orgClusterDataPathSuffix)

	def := slowlogsTableDef()
	def.PartitionProjection = cloneMap(def.PartitionProjection)
	def.PartitionProjection["storage.location.template"] = template

	return ResolvedTable{
		Definition:     def,
		Database:       logsDatabaseName,
		Name:           tableName,
		FullName:       logsDatabaseName + "." + tableName,
		Location:       location,
		LocationPrefix: location,
	}
}

func generalLogsTableName(region, suffix string) string {
	return glueTableName(generalLogsBase, region, suffix)
}

func logStatsTableName(region, suffix string) string {
	return glueTableName(logStatsBase, region, suffix)
}

func topsqlTableName(region, suffix string) string {
	return glueTableName(topsqlBase, region, suffix)
}

func topsqlSQLMetaTableName(region, suffix string) string {
	return glueTableName(topsqlSQLMetaBase, region, suffix)
}

func topsqlPlanMetaTableName(region, suffix string) string {
	return glueTableName(topsqlPlanMetaBase, region, suffix)
}

func topruTableName(region, suffix string) string {
	return glueTableName(topruBase, region, suffix)
}

func sqlstatementTableName(region, suffix string) string {
	return glueTableName(sqlstatementBase, region, suffix)
}

func slowlogsTableName(region, suffix string) string {
	return glueTableName(slowlogsBase, region, suffix)
}

func glueTableName(base, region, suffix string) string {
	name := base + "_" + regionToGlueSuffix(region)
	suffix = strings.TrimSpace(suffix)
	if suffix != "" {
		name += "_" + strings.ReplaceAll(suffix, "-", "_")
	}
	return name
}

func regionToGlueSuffix(region string) string {
	return strings.ReplaceAll(strings.TrimSpace(region), "-", "_")
}

func defaultLogsBucket(region string) string {
	return fmt.Sprintf("o11y-prod-shared-%s", strings.TrimSpace(region))
}

func cloneMap(in map[string]string) map[string]string {
	out := make(map[string]string, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}

func findResolvedTable(tables []ResolvedTable, database, name string) (*ResolvedTable, bool) {
	database = strings.TrimSpace(database)
	name = strings.TrimSpace(name)
	for i := range tables {
		t := &tables[i]
		if strings.EqualFold(t.Database, database) && strings.EqualFold(t.Name, name) {
			return t, true
		}
	}
	return nil, false
}