package engine

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"go.uber.org/zap"
	"golang.org/x/sync/singleflight"
)

// TableMetadata 表元数据
type TableMetadata struct {
	Exists      bool      // 表是否存在
	LastChecked time.Time // 最后检查时间
}

// PartitionMetadata 分区元数据
type PartitionMetadata struct {
	Loaded      bool      // 分区是否已加载
	LastChecked time.Time // 最后检查时间
}

type athenaTableManagerEngine interface {
	CheckDatabaseExists(context.Context, string, string) (bool, error)
	CheckTableExists(context.Context, string, string, string) (bool, error)
	DeleteGlueTable(context.Context, string, string, string) error
	Execute(context.Context, string, *dto.DataSourceInfo) (*dto.DataProxyQueryResult, error)
	ExecuteDDL(context.Context, string, *dto.DataSourceInfo) error
	GetGlueProjectionInstanceValues(context.Context, string, string, string) (string, error)
	GetGlueTableInfo(context.Context, string, string, string) (map[string]string, string, error)
	GetGlueTableParameters(context.Context, string, string, string) (map[string]string, error)
	ListS3Prefixes(context.Context, string, string, string) ([]string, error)
	UpdateGlueProjectionInstanceValues(context.Context, string, string, string, string) error
}

// TableManager 表管理器
type TableManager struct {
	engine         athenaTableManagerEngine // Athena 引擎
	cache          sync.Map                 // 表缓存：key = "database.table@region", value = *TableMetadata
	partitionCache sync.Map                 // 分区缓存：key = "database.table@region:partition_key", value = *PartitionMetadata
	instProjSync   sync.Map                 // instance 投影刷新：key = "instproj:database.table@region", value = time.Time（上次检查时间）
	schemaRepairs  sync.Map                 // key = "database.table@region", value = time.Time（最近一次成功修复）
	repairGroup    singleflight.Group
	config         *config.TableManagerConfig // 配置
	mu             sync.Mutex                 // 保护创建表的并发操作
}

const emptyInstanceProjectionValue = "__clinic_no_instance__"

// InstanceProjectionRepair summarizes a safe Glue parameter reconciliation
// without exposing instance names or storage paths to callers or logs.
type InstanceProjectionRepair struct {
	Table         string
	InstanceCount int
	Updated       bool
}

// NewTableManager 创建表管理器
func NewTableManager(engine *AthenaEngine, cfg *config.TableManagerConfig) *TableManager {
	return &TableManager{
		engine: engine,
		config: cfg,
	}
}

// EnsureTableExists 确保表存在，不存在则自动创建
func (tm *TableManager) EnsureTableExists(ctx context.Context, database, table, dataSource, region string) error {
	if !tm.config.Enabled || !tm.config.AutoCreate {
		return nil // 未启用，直接返回
	}

	// 1. 检查缓存
	cacheKey := fmt.Sprintf("table:%s.%s@%s", database, table, region)
	if metadata, ok := tm.getCachedMetadata(cacheKey); ok {
		if metadata.Exists {
			tm.maybeSyncInstanceProjection(ctx, database, table, dataSource, region)
			return nil // 表已存在且在缓存有效期内
		}
	}

	// 2. 先确保数据库存在
	if err := tm.ensureDatabaseExists(ctx, database, region); err != nil {
		return fmt.Errorf("failed to ensure database exists: %w", err)
	}

	// 3. 检查表是否存在
	exists, err := tm.engine.CheckTableExists(ctx, database, table, region)
	if err != nil {
		return fmt.Errorf("failed to check table existence: %w", err)
	}

	if exists {
		if err := tm.ensureExistingTableCompatible(ctx, database, table, dataSource, region); err != nil {
			return err
		}
		// 更新缓存
		tm.setCachedMetadata(cacheKey, &TableMetadata{
			Exists:      true,
			LastChecked: time.Now(),
		})
		tm.maybeSyncInstanceProjection(ctx, database, table, dataSource, region)
		return nil
	}

	// 3. 表不存在，需要创建
	// 获取锁防止并发创建
	tm.mu.Lock()
	defer tm.mu.Unlock()

	// 双重检查（可能其他goroutine已创建）
	exists, _ = tm.engine.CheckTableExists(ctx, database, table, region)
	if exists {
		if err := tm.ensureExistingTableCompatibleLocked(ctx, database, table, dataSource, region); err != nil {
			return err
		}
		tm.setCachedMetadata(cacheKey, &TableMetadata{
			Exists:      true,
			LastChecked: time.Now(),
		})
		tm.maybeSyncInstanceProjection(ctx, database, table, dataSource, region)
		return nil
	}

	// 4. 创建表
	if err := tm.createTable(ctx, database, table, dataSource, region); err != nil {
		// 检查是否是表已存在错误（可能是并发创建导致）
		if strings.Contains(err.Error(), "AlreadyExistsException") ||
			strings.Contains(err.Error(), "already exists") {
			// 忽略该错误，更新缓存
			tm.setCachedMetadata(cacheKey, &TableMetadata{
				Exists:      true,
				LastChecked: time.Now(),
			})
			return nil
		}
		return fmt.Errorf("failed to create table: %w", err)
	}
	log.Info(ctx, "table created successfully",
		zap.String("database", database),
		zap.String("table", table),
		zap.String("region", region),
	)

	// 5. 表创建成功，尝试自动加载分区（从 data_source 提取）。Delta 表依赖
	// transaction log，不走 Glue partition 注册路径。
	template, ok := tm.findTableTemplate(table)
	if ok && !template.IsDeltaTable() {
		partitions := ExtractPartitionFromDataSource(dataSource)
		if len(partitions) > 0 {
			location := tm.extractLocation(dataSource)
			// 加载分区（忽略错误，不影响表的可用性）
			_ = tm.LoadPartitions(ctx, database, table, partitions, location, region)
		}
	}

	// 6. 更新缓存
	tm.setCachedMetadata(cacheKey, &TableMetadata{
		Exists:      true,
		LastChecked: time.Now(),
	})

	return nil
}

func (tm *TableManager) ensureExistingTableCompatible(ctx context.Context, database, table, dataSource, region string) error {
	template, ok := tm.findTableTemplate(table)
	if !ok || !template.IsDeltaTable() {
		return nil
	}

	isDelta, err := tm.isGlueDeltaTable(ctx, database, table, region)
	if err != nil {
		return err
	}
	if isDelta {
		return nil
	}

	tm.mu.Lock()
	defer tm.mu.Unlock()
	return tm.ensureExistingTableCompatibleLocked(ctx, database, table, dataSource, region)
}

func (tm *TableManager) ensureExistingTableCompatibleLocked(ctx context.Context, database, table, dataSource, region string) error {
	template, ok := tm.findTableTemplate(table)
	if !ok || !template.IsDeltaTable() {
		return nil
	}
	isDelta, err := tm.isGlueDeltaTable(ctx, database, table, region)
	if err != nil {
		return err
	}
	if isDelta {
		return nil
	}

	log.Warn(ctx, "existing Athena table is not Delta, recreating Glue table definition",
		zap.String("database", database),
		zap.String("table", table),
		zap.String("region", region),
	)

	dropSQL := fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", database, table)
	if err := tm.engine.ExecuteDDL(ctx, dropSQL, &dto.DataSourceInfo{Region: region, Vendor: "aws"}); err != nil {
		return fmt.Errorf("drop incompatible table failed: %w", err)
	}
	if err := tm.createTable(ctx, database, table, dataSource, region); err != nil {
		return fmt.Errorf("recreate delta table failed: %w", err)
	}
	return nil
}

func (tm *TableManager) isGlueDeltaTable(ctx context.Context, database, table, region string) (bool, error) {
	params, err := tm.engine.GetGlueTableParameters(ctx, database, table, region)
	if err != nil {
		log.Warn(ctx, "failed to read Glue table parameters",
			zap.String("database", database),
			zap.String("table", table),
		)
		return false, fmt.Errorf("read Glue table parameters failed: %w", err)
	}
	return strings.EqualFold(params["table_type"], "DELTA") ||
		strings.EqualFold(params["spark.sql.sources.provider"], "delta"), nil
}

type deltaRepairTarget struct {
	database   string
	table      string
	dataSource string
}

// RepairDeltaSchemaMismatch rebuilds only Clinic-managed Delta table
// definitions referenced by a failed query. Concurrent repairs for the same
// table are collapsed, and a repair completed after failedAt is reused.
func (tm *TableManager) RepairDeltaSchemaMismatch(
	ctx context.Context,
	tableRefs []TableReference,
	dataSource, region, clusterID string,
	failedAt time.Time,
) ([]string, error) {
	if !tm.config.Enabled || !tm.config.AutoCreate {
		return nil, fmt.Errorf("Athena table management is disabled")
	}

	targets := tm.deltaRepairTargets(tableRefs, dataSource, clusterID)
	if len(targets) == 0 {
		return nil, fmt.Errorf("query does not reference a Clinic-managed Delta table")
	}

	repaired := make([]string, 0, len(targets))
	for _, target := range targets {
		key := fmt.Sprintf("%s.%s@%s", target.database, target.table, region)
		_, err, _ := tm.repairGroup.Do(key, func() (interface{}, error) {
			if lastRepair, ok := tm.schemaRepairs.Load(key); ok &&
				!lastRepair.(time.Time).Before(failedAt) {
				return nil, nil
			}
			if err := tm.rebuildDeltaTable(ctx, target, region); err != nil {
				return nil, err
			}
			tm.schemaRepairs.Store(key, time.Now())
			return nil, nil
		})
		if err != nil {
			return repaired, fmt.Errorf("repair %s: %w", key, err)
		}
		repaired = append(repaired, target.database+"."+target.table)
	}
	return repaired, nil
}

func (tm *TableManager) deltaRepairTargets(tableRefs []TableReference, dataSource, clusterID string) []deltaRepairTarget {
	var targets []deltaRepairTarget
	seen := make(map[string]struct{})
	addTarget := func(database, table, source string) {
		template, ok := tm.findTableTemplate(table)
		if !ok || !template.IsDeltaTable() {
			return
		}
		key := database + "." + table
		if _, ok := seen[key]; ok {
			return
		}
		seen[key] = struct{}{}
		targets = append(targets, deltaRepairTarget{
			database:   database,
			table:      table,
			dataSource: ResolveTopSQLDataSource(source, table),
		})
	}

	for _, ref := range tableRefs {
		template, ok := tm.findTableTemplate(ref.Table)
		if !ok {
			continue
		}
		if template.IsView() {
			for _, dependency := range template.DependsOn {
				addTarget(ref.Database, dependencyTableName(dependency, clusterID), dataSource)
			}
			continue
		}
		addTarget(ref.Database, ref.Table, dataSource)
	}
	return targets
}

func (tm *TableManager) rebuildDeltaTable(ctx context.Context, target deltaRepairTarget, region string) error {
	params, catalogLocation, err := tm.engine.GetGlueTableInfo(
		ctx, target.database, target.table, region,
	)
	if err != nil {
		return fmt.Errorf("read Glue table metadata: %w", err)
	}
	if !strings.EqualFold(params["table_type"], "DELTA") &&
		!strings.EqualFold(params["spark.sql.sources.provider"], "delta") {
		return fmt.Errorf("existing Glue table is not Delta")
	}

	expectedLocation := tm.extractLocation(target.dataSource)
	if strings.TrimRight(catalogLocation, "/") != strings.TrimRight(expectedLocation, "/") {
		return fmt.Errorf(
			"Glue table location %q does not match request data source %q",
			catalogLocation,
			expectedLocation,
		)
	}

	tm.mu.Lock()
	defer tm.mu.Unlock()

	tm.RefreshCache(target.database, target.table, region)
	if err := tm.engine.DeleteGlueTable(ctx, target.database, target.table, region); err != nil {
		return fmt.Errorf("delete stale Delta Glue table: %w", err)
	}
	if err := tm.createTable(ctx, target.database, target.table, target.dataSource, region); err != nil {
		return fmt.Errorf("recreate Delta table: %w", err)
	}

	tm.setCachedMetadata(
		fmt.Sprintf("table:%s.%s@%s", target.database, target.table, region),
		&TableMetadata{Exists: true, LastChecked: time.Now()},
	)
	return nil
}

// createTable 创建表
func (tm *TableManager) createTable(ctx context.Context, database, table, dataSource, region string) error {
	// 1. 从配置获取表模板（支持模式匹配）
	template, ok := tm.findTableTemplate(table)
	if !ok {
		return fmt.Errorf("no template found for table: %s", table)
	}

	// 2. 解析 data_source 获取 location
	// 从 data_source 提取 location（去掉可能的分区部分）
	location := tm.extractLocation(dataSource)

	// 3. 参数化替换 DDL
	params := map[string]string{
		"Database": database,
		"Table":    table,
		"Location": location,
	}

	// 如果 DDL 模板包含 {{.InstanceValues}}，通过 S3 ListObjects 发现 instance 列表
	if strings.Contains(template.DDL, "{{.InstanceValues}}") {
		instances, err := tm.discoverInstances(ctx, dataSource, region)
		if err != nil {
			return fmt.Errorf("discover instance projection values: %w", err)
		}
		// enum values 格式: "val1,val2,val3"（逗号分隔，无空格）
		params["InstanceValues"] = desiredInstanceProjection(instances)
		log.Info(ctx, "resolved instance enum values for DDL",
			zap.Int("count", len(instances)),
			zap.String("table", table),
		)
	}

	ddl := tm.renderDDL(template.DDL, params)

	// 4. 执行 CREATE TABLE
	err := tm.engine.ExecuteDDL(ctx, ddl, &dto.DataSourceInfo{
		Region: region,
		Vendor: "aws",
	})
	if err != nil {
		return fmt.Errorf("execute CREATE TABLE failed: %w", err)
	}

	return nil
}

// discoverInstances 通过 S3 ListObjects 发现 instance 分区值
// dataSource 格式: s3://bucket/prefix/component=tidb/
func (tm *TableManager) discoverInstances(ctx context.Context, dataSource, region string) ([]string, error) {
	// 解析 S3 URI
	bucket, prefix, err := parseS3URI(dataSource)
	if err != nil {
		return nil, fmt.Errorf("parse S3 URI failed: %w", err)
	}

	// 确保 prefix 以 / 结尾
	if !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}

	// ListObjects with delimiter to get instance= prefixes
	prefixes, err := tm.engine.ListS3Prefixes(ctx, bucket, prefix, region)
	if err != nil {
		return nil, fmt.Errorf("list S3 prefixes failed: %w", err)
	}

	// 从前缀中提取 instance 名称
	// 前缀格式: .../instance=xxx/
	var instances []string
	for _, p := range prefixes {
		idx := strings.LastIndex(p, "instance=")
		if idx == -1 {
			continue
		}
		val := p[idx+len("instance="):]
		val = strings.TrimRight(val, "/")
		if val != "" && val != "_delta_log" {
			instances = append(instances, val)
		}
	}

	return instances, nil
}

// normalizeInstanceEnumList 将 S3 发现的 instance 列表规范为 Glue projection.instance.values 格式（去重、排序、逗号拼接）
func normalizeInstanceEnumList(discovered []string) string {
	set := make(map[string]struct{})
	for _, v := range discovered {
		v = strings.TrimSpace(v)
		if v != "" {
			set[v] = struct{}{}
		}
	}
	out := make([]string, 0, len(set))
	for v := range set {
		out = append(out, v)
	}
	sort.Strings(out)
	return strings.Join(out, ",")
}

// normalizeInstanceEnumFromCSV 解析 Glue 中已有的逗号分隔枚举并规范化，便于与 S3 快照比较
func normalizeInstanceEnumFromCSV(csv string) string {
	var parts []string
	for _, part := range strings.Split(csv, ",") {
		v := strings.TrimSpace(part)
		if v != "" {
			parts = append(parts, v)
		}
	}
	return normalizeInstanceEnumList(parts)
}

func instanceProjectionNeedsUpdate(currentCSV string, discovered []string) bool {
	desired := desiredInstanceProjection(discovered)
	currentNorm := normalizeInstanceEnumFromCSV(currentCSV)
	return desired != currentNorm
}

func desiredInstanceProjection(discovered []string) string {
	desired := normalizeInstanceEnumList(discovered)
	if desired == "" {
		return emptyInstanceProjectionValue
	}
	return desired
}

// maybeSyncInstanceProjection 表已存在时，按配置间隔用 S3 前缀刷新 Glue 中 projection.instance.values（仅 DDL 含 {{.InstanceValues}} 的模板）
func (tm *TableManager) maybeSyncInstanceProjection(ctx context.Context, database, table, dataSource, region string) {
	if tm.config.InstanceProjectionRefreshSeconds <= 0 {
		return
	}
	interval := time.Duration(tm.config.InstanceProjectionRefreshSeconds) * time.Second
	syncKey := fmt.Sprintf("instproj:%s.%s@%s", database, table, region)
	if t, ok := tm.instProjSync.Load(syncKey); ok {
		if time.Since(t.(time.Time)) < interval {
			return
		}
	}

	template, ok := tm.findTableTemplate(table)
	if !ok || !strings.Contains(template.DDL, "{{.InstanceValues}}") {
		tm.instProjSync.Store(syncKey, time.Now())
		return
	}

	discovered, err := tm.discoverInstances(ctx, dataSource, region)
	if err != nil {
		log.Warn(ctx, "instance projection sync: discoverInstances failed",
			zap.String("table", table))
		tm.instProjSync.Store(syncKey, time.Now())
		return
	}
	current, err := tm.engine.GetGlueProjectionInstanceValues(ctx, database, table, region)
	if err != nil {
		log.Warn(ctx, "instance projection sync: glue GetTable failed",
			zap.String("table", table))
		tm.instProjSync.Store(syncKey, time.Now())
		return
	}

	if !instanceProjectionNeedsUpdate(current, discovered) {
		tm.instProjSync.Store(syncKey, time.Now())
		return
	}

	// 以当前 S3 前缀下的 instance 列表为准整体替换（已下线的 instance 会从枚举中移除）
	desired := desiredInstanceProjection(discovered)
	err = tm.engine.UpdateGlueProjectionInstanceValues(ctx, database, table, region, desired)
	if err != nil {
		log.Warn(ctx, "instance projection sync: Glue UpdateTable failed",
			zap.String("database", database), zap.String("table", table))
	} else {
		instCount := 0
		if desired != "" {
			instCount = strings.Count(desired, ",") + 1
		}
		log.Info(ctx, "instance projection sync: replaced projection.instance.values from S3",
			zap.String("database", database), zap.String("table", table),
			zap.Int("instance_count", instCount))
	}
	tm.instProjSync.Store(syncKey, time.Now())
}

// RepairInstanceProjection reconciles only Clinic-managed tables whose DDL
// contains the instance enum placeholder. It never deletes or recreates Glue
// tables. Concurrent repairs for the same physical table are collapsed.
func (tm *TableManager) RepairInstanceProjection(
	ctx context.Context,
	tableRefs []TableReference,
	dataSource, region, clusterID string,
) ([]InstanceProjectionRepair, error) {
	if !tm.config.Enabled || !tm.config.AutoCreate {
		return nil, fmt.Errorf("Athena table management is disabled")
	}

	targets := tm.instanceProjectionTargets(tableRefs, dataSource, clusterID)
	if len(targets) == 0 {
		return nil, fmt.Errorf("query does not reference a Clinic-managed instance projection table")
	}

	repairs := make([]InstanceProjectionRepair, 0, len(targets))
	for _, target := range targets {
		key := fmt.Sprintf("instance-projection:%s.%s@%s", target.database, target.table, region)
		value, err, _ := tm.repairGroup.Do(key, func() (interface{}, error) {
			return tm.repairInstanceProjectionTable(ctx, target, region)
		})
		if err != nil {
			return repairs, fmt.Errorf("repair instance projection table: %w", err)
		}
		repairs = append(repairs, value.(InstanceProjectionRepair))
	}
	return repairs, nil
}

func (tm *TableManager) instanceProjectionTargets(
	tableRefs []TableReference,
	dataSource, clusterID string,
) []deltaRepairTarget {
	var targets []deltaRepairTarget
	seen := make(map[string]struct{})
	addTarget := func(database, table string) {
		template, ok := tm.findTableTemplate(table)
		if !ok || !strings.Contains(template.DDL, "{{.InstanceValues}}") {
			return
		}
		key := database + "." + table
		if _, ok := seen[key]; ok {
			return
		}
		seen[key] = struct{}{}
		targets = append(targets, deltaRepairTarget{
			database:   database,
			table:      table,
			dataSource: ResolveTopSQLDataSource(dataSource, table),
		})
	}

	for _, ref := range tableRefs {
		template, ok := tm.findTableTemplate(ref.Table)
		if !ok {
			continue
		}
		if template.IsView() {
			for _, dependency := range template.DependsOn {
				addTarget(ref.Database, dependencyTableName(dependency, clusterID))
			}
			continue
		}
		addTarget(ref.Database, ref.Table)
	}
	return targets
}

func (tm *TableManager) repairInstanceProjectionTable(
	ctx context.Context,
	target deltaRepairTarget,
	region string,
) (InstanceProjectionRepair, error) {
	instances, err := tm.discoverInstances(ctx, target.dataSource, region)
	if err != nil {
		return InstanceProjectionRepair{}, fmt.Errorf("discover instances: %w", err)
	}
	desired := desiredInstanceProjection(instances)
	current, err := tm.engine.GetGlueProjectionInstanceValues(ctx, target.database, target.table, region)
	if err != nil {
		return InstanceProjectionRepair{}, fmt.Errorf("read Glue instance projection: %w", err)
	}
	repair := InstanceProjectionRepair{
		Table:         logicalInstanceProjectionTable(tm, target.table),
		InstanceCount: len(strings.FieldsFunc(normalizeInstanceEnumList(instances), func(r rune) bool { return r == ',' })),
	}
	if normalizeInstanceEnumFromCSV(current) == desired {
		return repair, nil
	}
	if err := tm.engine.UpdateGlueProjectionInstanceValues(
		ctx, target.database, target.table, region, desired,
	); err != nil {
		return InstanceProjectionRepair{}, fmt.Errorf("update Glue instance projection: %w", err)
	}
	repair.Updated = true
	tm.instProjSync.Store(
		fmt.Sprintf("instproj:%s.%s@%s", target.database, target.table, region),
		time.Now(),
	)
	return repair, nil
}

func logicalInstanceProjectionTable(tm *TableManager, table string) string {
	if template, ok := tm.findTemplateNameForTable(table); ok {
		return template
	}
	return "instance_projection_table"
}

// renderDDL 渲染 DDL 模板
func (tm *TableManager) renderDDL(template string, params map[string]string) string {
	result := template
	for key, value := range params {
		placeholder := fmt.Sprintf("{{.%s}}", key)
		result = strings.ReplaceAll(result, placeholder, value)
	}
	return result
}

// extractLocation 从 data_source 提取 location
// 例如：s3://bucket/path/date=20251231/ -> s3://bucket/path/
func (tm *TableManager) extractLocation(dataSource string) string {
	// Location 就是 dataSource 本身（去掉末尾 /）
	// 注意：S3 路径中的 key=value 格式（如 org=xxx, cluster=xxx, type=topsql, component=tidb）
	// 是 S3 key 层级结构，不是 Athena 分区路径，不能去掉
	return strings.TrimRight(dataSource, "/")
}

// ensureDatabaseExists 确保数据库存在，不存在则自动创建
func (tm *TableManager) ensureDatabaseExists(ctx context.Context, database, region string) error {
	// 检查数据库缓存
	dbCacheKey := fmt.Sprintf("db:%s@%s", database, region)
	if metadata, ok := tm.getCachedMetadata(dbCacheKey); ok {
		if metadata.Exists {
			return nil // 数据库已存在
		}
	}

	exists, err := tm.engine.CheckDatabaseExists(ctx, database, region)
	if err != nil {
		return fmt.Errorf("failed to check database existence: %w", err)
	}
	if exists {
		tm.setCachedMetadata(dbCacheKey, &TableMetadata{
			Exists:      true,
			LastChecked: time.Now(),
		})
		return nil
	}

	// 创建数据库的 SQL
	createDBSQL := fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", database)

	// 执行创建数据库
	err = tm.engine.ExecuteDDL(ctx, createDBSQL, &dto.DataSourceInfo{
		Region: region,
		Vendor: "aws",
	})
	if err != nil {
		// 如果数据库已存在，忽略错误
		if strings.Contains(err.Error(), "AlreadyExistsException") ||
			strings.Contains(err.Error(), "already exists") {
			tm.setCachedMetadata(dbCacheKey, &TableMetadata{
				Exists:      true,
				LastChecked: time.Now(),
			})
			return nil
		}
		return fmt.Errorf("failed to create database: %w", err)
	}

	// 更新数据库缓存
	tm.setCachedMetadata(dbCacheKey, &TableMetadata{
		Exists:      true,
		LastChecked: time.Now(),
	})

	return nil
}

// getCachedMetadata 从缓存获取元数据
func (tm *TableManager) getCachedMetadata(key string) (*TableMetadata, bool) {
	if value, ok := tm.cache.Load(key); ok {
		metadata := value.(*TableMetadata)
		// 检查是否过期
		// TTL=0 表示永不过期
		if tm.config.CacheTTLSeconds == 0 {
			return metadata, true
		}
		ttl := time.Duration(tm.config.CacheTTLSeconds) * time.Second
		if time.Since(metadata.LastChecked) < ttl {
			return metadata, true
		}
	}
	return nil, false
}

// setCachedMetadata 设置缓存元数据
func (tm *TableManager) setCachedMetadata(key string, metadata *TableMetadata) {
	tm.cache.Store(key, metadata)
}

// RefreshCache 刷新指定表的缓存
func (tm *TableManager) RefreshCache(database, table, region string) {
	cacheKey := fmt.Sprintf("table:%s.%s@%s", database, table, region)
	tm.cache.Delete(cacheKey)
}

// LoadPartitions 加载分区（可选功能）- 带缓存优化
func (tm *TableManager) LoadPartitions(ctx context.Context, database, table string, partitions map[string]string, location, region string) error {
	if len(partitions) == 0 {
		return nil
	}

	// 1. 检查分区缓存
	partitionKey := tm.buildPartitionCacheKey(database, table, region, partitions)
	if metadata, ok := tm.getCachedPartitionMetadata(partitionKey); ok {
		if metadata.Loaded {
			// 分区已加载，跳过
			return nil
		}
	}

	// 2. 构建 ALTER TABLE ADD PARTITION 语句
	var partitionClauses []string
	var partitionLocations []string

	for key, value := range partitions {
		partitionClauses = append(partitionClauses, fmt.Sprintf("%s='%s'", key, value))
		partitionLocations = append(partitionLocations, fmt.Sprintf("%s=%s", key, value))
	}

	partitionClause := strings.Join(partitionClauses, ", ")
	partitionLocation := location + strings.Join(partitionLocations, "/") + "/"

	// 3. 使用反引号以支持包含特殊字符（如连字符）的数据库名
	sql := fmt.Sprintf(
		"ALTER TABLE `%s`.`%s` ADD IF NOT EXISTS PARTITION (%s) LOCATION '%s'",
		database, table, partitionClause, partitionLocation,
	)

	// 4. 执行 ALTER TABLE
	err := tm.engine.ExecuteDDL(ctx, sql, &dto.DataSourceInfo{
		Region: region,
		Vendor: "aws",
	})
	if err != nil {
		// 如果分区已存在，忽略错误并更新缓存
		if strings.Contains(err.Error(), "Partition already exists") ||
			strings.Contains(err.Error(), "AlreadyExistsException") {
			tm.setCachedPartitionMetadata(partitionKey, &PartitionMetadata{
				Loaded:      true,
				LastChecked: time.Now(),
			})
			return nil
		}
		return fmt.Errorf("load partition failed: %w", err)
	}

	// 5. 更新分区缓存
	tm.setCachedPartitionMetadata(partitionKey, &PartitionMetadata{
		Loaded:      true,
		LastChecked: time.Now(),
	})

	return nil
}

// buildPartitionCacheKey 构建分区缓存键
func (tm *TableManager) buildPartitionCacheKey(database, table, region string, partitions map[string]string) string {
	// 格式：partition:database.table@region:key1=value1,key2=value2
	var partitionParts []string
	for k, v := range partitions {
		partitionParts = append(partitionParts, fmt.Sprintf("%s=%s", k, v))
	}
	return fmt.Sprintf("partition:%s.%s@%s:%s", database, table, region, strings.Join(partitionParts, ","))
}

// getCachedPartitionMetadata 从缓存获取分区元数据
func (tm *TableManager) getCachedPartitionMetadata(key string) (*PartitionMetadata, bool) {
	if value, ok := tm.partitionCache.Load(key); ok {
		metadata := value.(*PartitionMetadata)
		// 检查是否过期（使用表缓存的 TTL）
		// TTL=0 表示永不过期
		if tm.config.CacheTTLSeconds == 0 {
			return metadata, true
		}
		ttl := time.Duration(tm.config.CacheTTLSeconds) * time.Second
		if time.Since(metadata.LastChecked) < ttl {
			return metadata, true
		}
	}
	return nil, false
}

// setCachedPartitionMetadata 设置分区缓存元数据
func (tm *TableManager) setCachedPartitionMetadata(key string, metadata *PartitionMetadata) {
	tm.partitionCache.Store(key, metadata)
}

// findTableTemplate 查找表模板（支持模式匹配）
// 例如：表名 slow_query_logs_123456 匹配模板 slow_query_logs
func (tm *TableManager) findTableTemplate(tableName string) (config.TableTemplate, bool) {
	// 1. 首先尝试精确匹配
	if template, ok := tm.config.TableTemplates[tableName]; ok {
		return template, true
	}

	// 2. 尝试最长前缀匹配：检查表名是否以某个模板名 + "_" 开头
	// slow_query_logs_c-kcbqpuhh -> slow_query_logs
	// 使用最长前缀匹配避免歧义（如 topsql_tidb_123 应匹配 topsql_tidb 而非 topsql）
	var bestTemplate config.TableTemplate
	bestPrefix := ""
	prefixFound := false
	for templateName, template := range tm.config.TableTemplates {
		prefix := templateName + "_"
		if strings.HasPrefix(tableName, prefix) && len(prefix) > len(bestPrefix) {
			bestTemplate = template
			bestPrefix = prefix
			prefixFound = true
		}
	}
	if prefixFound {
		return bestTemplate, true
	}

	// 3. 尝试模式匹配：去掉表名中的数字后缀（_123456）
	// slow_query_logs_123456 -> slow_query_logs
	if strings.Contains(tableName, "_") {
		parts := strings.Split(tableName, "_")
		// 检查最后一部分是否是纯数字
		lastPart := parts[len(parts)-1]
		isNumeric := true
		for _, c := range lastPart {
			if c < '0' || c > '9' {
				isNumeric = false
				break
			}
		}

		// 如果最后一部分是数字，去掉它后再尝试匹配
		if isNumeric && len(parts) > 1 {
			baseTableName := strings.Join(parts[:len(parts)-1], "_")
			if template, ok := tm.config.TableTemplates[baseTableName]; ok {
				return template, true
			}
		}
	}

	// 4. 未找到模板
	return config.TableTemplate{}, false
}

// findTemplateNameForTable 查找表名对应的模板名（与 findTableTemplate 逻辑一致，返回模板名而非模板对象）
func (tm *TableManager) findTemplateNameForTable(tableName string) (string, bool) {
	// 精确匹配
	if _, ok := tm.config.TableTemplates[tableName]; ok {
		return tableName, true
	}
	// 最长前缀匹配
	bestName := ""
	bestPrefix := ""
	for templateName := range tm.config.TableTemplates {
		prefix := templateName + "_"
		if strings.HasPrefix(tableName, prefix) && len(prefix) > len(bestPrefix) {
			bestName = templateName
			bestPrefix = prefix
		}
	}
	if bestName != "" {
		return bestName, true
	}
	// 数字后缀匹配
	if strings.Contains(tableName, "_") {
		parts := strings.Split(tableName, "_")
		lastPart := parts[len(parts)-1]
		isNumeric := true
		for _, c := range lastPart {
			if c < '0' || c > '9' {
				isNumeric = false
				break
			}
		}
		if isNumeric && len(parts) > 1 {
			baseTableName := strings.Join(parts[:len(parts)-1], "_")
			if _, ok := tm.config.TableTemplates[baseTableName]; ok {
				return baseTableName, true
			}
		}
	}
	return "", false
}

// IsViewTemplate 检查表名是否对应 VIEW 类型的模板
func (tm *TableManager) IsViewTemplate(tableName string) bool {
	template, ok := tm.findTableTemplate(tableName)
	return ok && template.IsView()
}

// EnsureViewExists 确保 VIEW 存在，如不存在则先创建依赖表再创建 VIEW
func (tm *TableManager) EnsureViewExists(ctx context.Context, database, viewName, dataSource, region, clusterId, clinicTenantId string) error {
	if !tm.config.Enabled || !tm.config.AutoCreate {
		return nil
	}

	// 1. 检查缓存
	cacheKey := fmt.Sprintf("table:%s.%s@%s", database, viewName, region)
	if metadata, ok := tm.getCachedMetadata(cacheKey); ok {
		if metadata.Exists {
			return nil
		}
	}

	// 2. 确保数据库存在
	if err := tm.ensureDatabaseExists(ctx, database, region); err != nil {
		return fmt.Errorf("failed to ensure database exists: %w", err)
	}

	// 3. 查找 VIEW 模板（在锁外执行）
	template, ok := tm.findTableTemplate(viewName)
	if !ok {
		return fmt.Errorf("no template found for view: %s", viewName)
	}
	if !template.IsView() {
		return fmt.Errorf("template for %s is not a view type", viewName)
	}
	templateName, _ := tm.findTemplateNameForTable(viewName)
	if templateName == "log_stats" && clinicTenantId == "" {
		return fmt.Errorf("clinicTenantId is required for log_stats view")
	}

	// 4. 确保所有依赖表存在（在锁外执行，避免死锁，EnsureTableExists 内部有自己的锁）
	if err := tm.ensureDependencyTables(ctx, database, template.DependsOn, dataSource, region, clusterId); err != nil {
		return err
	}

	// 5. 检查 VIEW 是否已存在（Athena SHOW TABLES 包含 VIEW）
	exists, err := tm.engine.CheckTableExists(ctx, database, viewName, region)
	if err != nil {
		return fmt.Errorf("failed to check view existence: %w", err)
	}
	if exists {
		tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
		return nil
	}

	// 6. 加锁防止并发创建（仅保护 VIEW 的创建，不包含依赖表创建）
	tm.mu.Lock()
	defer tm.mu.Unlock()

	// 双重检查
	exists, _ = tm.engine.CheckTableExists(ctx, database, viewName, region)
	if exists {
		tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
		return nil
	}

	// 7. 构建参数：Database, Table, 以及每个依赖表的实际名称
	params := map[string]string{
		"Database":       database,
		"Table":          viewName,
		"ClusterId":      clusterId,
		"ClinicTenantId": clinicTenantId,
	}
	for _, depName := range template.DependsOn {
		// 模板占位符格式: {{.DepTopsqlTidb}} → 将 topsql_tidb 转成 TopsqlTidb
		paramKey := "Dep" + snakeToPascal(depName)
		params[paramKey] = dependencyTableName(depName, clusterId)
	}

	// 8. 渲染并执行 DDL
	ddl := tm.renderDDL(template.DDL, params)
	err = tm.engine.ExecuteDDL(ctx, ddl, &dto.DataSourceInfo{
		Region: region,
		Vendor: "aws",
	})
	if err != nil {
		return fmt.Errorf("execute CREATE VIEW failed: %w", err)
	}

	log.Info(ctx, "view created successfully",
		zap.String("database", database),
		zap.String("view", viewName),
		zap.String("region", region),
	)

	// 9. 更新缓存
	tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
	return nil
}

func (tm *TableManager) ensureDependencyTables(ctx context.Context, database string, dependsOn []string, dataSource, region, clusterId string) error {
	if len(dependsOn) == 0 {
		return nil
	}

	startTime := time.Now()
	var wg sync.WaitGroup
	errCh := make(chan error, len(dependsOn))

	for _, depName := range dependsOn {
		depName := depName
		wg.Add(1)
		go func() {
			defer wg.Done()

			actualDepTable := dependencyTableName(depName, clusterId)
			depDataSource := ResolveTopSQLDataSource(dataSource, actualDepTable)
			depStart := time.Now()
			if err := tm.EnsureTableExists(ctx, database, actualDepTable, depDataSource, region); err != nil {
				log.Warn(ctx, "failed to ensure dependency table exists",
					zap.String("dependency", actualDepTable),
					zap.Duration("duration", time.Since(depStart)),
				)
				errCh <- fmt.Errorf("%s: %w", actualDepTable, err)
				return
			}
			log.Info(ctx, "dependency table ensured",
				zap.String("database", database),
				zap.String("table", actualDepTable),
				zap.String("region", region),
				zap.Duration("duration", time.Since(depStart)),
			)
		}()
	}

	wg.Wait()
	close(errCh)

	var depErrors []string
	for err := range errCh {
		depErrors = append(depErrors, err.Error())
	}
	if len(depErrors) > 0 {
		return fmt.Errorf("failed to ensure dependency tables: %s", strings.Join(depErrors, "; "))
	}

	log.Info(ctx, "dependency tables ensured",
		zap.String("database", database),
		zap.String("region", region),
		zap.Int("dependency_count", len(dependsOn)),
		zap.Duration("duration", time.Since(startTime)),
	)
	return nil
}

func dependencyTableName(depName, clusterId string) string {
	if depName == "log_stats_global" {
		return depName
	}
	return fmt.Sprintf("%s_%s", depName, clusterId)
}

// snakeToPascal 将 snake_case 转为 PascalCase
// 例如: topsql_tidb → TopsqlTidb, topsql_sql_meta → TopsqlSqlMeta
func snakeToPascal(s string) string {
	parts := strings.Split(s, "_")
	for i, p := range parts {
		if len(p) > 0 {
			parts[i] = strings.ToUpper(p[:1]) + p[1:]
		}
	}
	return strings.Join(parts, "")
}