package engine

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

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

const topSQLScopedSourceRefreshInterval = 30 * time.Minute

type bigQueryTableManagerEngine interface {
	CheckTableExists(context.Context, string, string, string) (bool, error)
	CreateExternalTable(context.Context, string, string, string) error
	EnsureTopSQLExternalTable(context.Context, string, string, string, string) error
	EnsureTopSQLView(context.Context, string, string, string, string) error
	EnsureDatasetExists(context.Context, string, string) error
	RefreshExternalTableSchema(context.Context, string, string, string) (bool, error)
}

// BigQueryTableManager BigQuery 表管理器
// 支持自动创建 dataset 和 Delta Lake 外部表，Schema 自动识别
type BigQueryTableManager struct {
	engine        bigQueryTableManagerEngine
	config        *config.TableManagerConfig
	cache         sync.Map // key = "table:dataset.table", value = *TableMetadata
	datasetCache  sync.Map // key = "dataset:datasetID", value = *TableMetadata（复用表示存在）
	schemaRefresh sync.Map // key = "schema:dataset.table", value = time.Time
	refreshGroup  singleflight.Group
	topSQLGroup   singleflight.Group
	mu            sync.Mutex
}

// NewBigQueryTableManager 创建 BigQuery 表管理器
func NewBigQueryTableManager(engine *BigQueryEngine, cfg *config.TableManagerConfig) *BigQueryTableManager {
	return &BigQueryTableManager{
		engine: engine,
		config: cfg,
	}
}

// EnsureTableExists 确保表存在，不存在则自动创建（Delta Lake，Schema 自动识别）
func (tm *BigQueryTableManager) EnsureTableExists(ctx context.Context, dataset, table, dataSource, region string) error {
	if !tm.config.Enabled || !tm.config.AutoCreate {
		return nil
	}

	cacheKey := fmt.Sprintf("table:%s.%s", dataset, table)
	if isBigQueryTopSQLView(table) || isBigQueryTopSQLPhysicalTable(table) {
		return tm.ensureBigQueryTopSQLObject(ctx, dataset, table, dataSource, region, cacheKey)
	}
	if metadata, ok := tm.getCachedMetadata(cacheKey); ok {
		if metadata.Exists {
			// A successful TopSQL creation is safe to cache. A fresh process never
			// trusts an existing BigQuery object without validating it below.
			tm.maybeRefreshSchema(ctx, dataset, table, dataSource)
			return nil
		}
	}
	exists, err := tm.engine.CheckTableExists(ctx, dataset, table, region)
	if err != nil {
		return fmt.Errorf("check table exists failed: %w", err)
	}

	if exists {
		tm.setCachedMetadata(cacheKey, &TableMetadata{
			Exists:      true,
			LastChecked: time.Now(),
		})
		tm.maybeRefreshSchema(ctx, dataset, table, dataSource)
		return nil
	}

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

	// 1. 确保 dataset 存在（每个 region 一个 dataset）
	if region != "" {
		dsCacheKey := fmt.Sprintf("dataset:%s", dataset)
		if _, ok := tm.datasetCache.Load(dsCacheKey); !ok {
			if err := tm.engine.EnsureDatasetExists(ctx, dataset, region); err != nil {
				return fmt.Errorf("ensure dataset exists failed: %w", err)
			}
			tm.datasetCache.Store(dsCacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
		}
	}

	gcsLocation := tm.extractGCSLocation(dataSource)
	if gcsLocation == "" {
		return fmt.Errorf("invalid GCS data source: %s", dataSource)
	}

	if err := tm.engine.CreateExternalTable(ctx, dataset, table, gcsLocation); err != nil {
		if strings.Contains(err.Error(), "Already Exists") || strings.Contains(err.Error(), "already exists") {
			tm.setCachedMetadata(cacheKey, &TableMetadata{
				Exists:      true,
				LastChecked: time.Now(),
			})
			return nil
		}
		return fmt.Errorf("create external table failed: %w", err)
	}

	log.Info(ctx, "BigQuery external table created",
		zap.String("dataset", dataset),
		zap.String("table", table),
	)

	tm.setCachedMetadata(cacheKey, &TableMetadata{
		Exists:      true,
		LastChecked: time.Now(),
	})
	if !isBigQueryTopSQLPhysicalTable(table) {
		tm.schemaRefresh.Store(schemaRefreshKey(dataset, table), time.Now())
	}

	return nil
}

func (tm *BigQueryTableManager) ensureBigQueryTopSQLObject(ctx context.Context, dataset, table, dataSource, region, cacheKey string) error {
	key := "topsql:" + dataset + "." + table + "@" + dataSource
	_, err, _ := tm.topSQLGroup.Do(key, func() (interface{}, error) {
		if region != "" {
			dsCacheKey := fmt.Sprintf("dataset:%s", dataset)
			if _, ok := tm.datasetCache.Load(dsCacheKey); !ok {
				if err := tm.engine.EnsureDatasetExists(ctx, dataset, region); err != nil {
					return nil, fmt.Errorf("ensure dataset exists failed: %w", err)
				}
				tm.datasetCache.Store(dsCacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
			}
		}
		root := tm.extractGCSLocation(dataSource)
		if root == "" {
			return nil, fmt.Errorf("invalid GCS data source")
		}
		if isBigQueryTopSQLView(table) {
			if err := tm.ensureBigQueryTopSQLView(ctx, dataset, table, root, region); err != nil {
				return nil, err
			}
		} else if err := tm.engine.EnsureTopSQLExternalTable(ctx, dataset, table, root, region); err != nil {
			return nil, fmt.Errorf("ensure TopSQL Parquet table: %w", err)
		}
		tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
		return nil, nil
	})
	return err
}

func (tm *BigQueryTableManager) ensureBigQueryTopSQLView(ctx context.Context, dataset, table, dataSource, region string) error {
	scope, ok := parseBigQueryTopSQLViewScope(table)
	if !ok {
		return fmt.Errorf("invalid TopSQL view name")
	}
	baseDataSource := topSQLBaseDataSource(dataSource)
	months := scope.months
	if len(months) == 0 {
		months = []string{""}
	}
	dependencies := make([]string, 0, len(months)*4)
	for _, month := range months {
		suffix := scope.clusterID
		if month != "" {
			suffix += "_m" + month
		}
		dependencies = append(dependencies,
			"topsql_tidb_"+suffix,
			"topsql_tikv_"+suffix,
			"topsql_sql_meta_"+suffix,
			"topsql_plan_meta_"+suffix,
		)
	}
	errs := make([]error, len(dependencies))
	var dependenciesWG sync.WaitGroup
	dependenciesWG.Add(len(dependencies))
	for index, dependency := range dependencies {
		go func() {
			defer dependenciesWG.Done()
			dependencyCacheKey := topSQLScopedDependencyCacheKey(dataset, dependency)
			if tm.topSQLScopedDependencyFresh(dependencyCacheKey, dependency) {
				return
			}
			root := ResolveTopSQLDataSource(baseDataSource, dependency)
			errs[index] = tm.engine.EnsureTopSQLExternalTable(ctx, dataset, dependency, root, region)
			if errs[index] == nil && topSQLTableMonth(dependency) != "" {
				tm.setCachedMetadata(dependencyCacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
			}
		}()
	}
	dependenciesWG.Wait()
	for index, err := range errs {
		if err != nil {
			return fmt.Errorf("ensure TopSQL dependency %s: %w", topSQLTableKind(dependencies[index]), err)
		}
	}
	if err := tm.engine.EnsureTopSQLView(ctx, dataset, table, baseDataSource, region); err != nil {
		return fmt.Errorf("ensure TopSQL view: %w", err)
	}
	return nil
}

func topSQLScopedDependencyCacheKey(dataset, table string) string {
	return "topsql-source:" + dataset + "." + table
}

func (tm *BigQueryTableManager) topSQLScopedDependencyFresh(cacheKey, table string) bool {
	if topSQLTableMonth(table) == "" {
		return false
	}
	value, ok := tm.cache.Load(cacheKey)
	if !ok {
		return false
	}
	metadata, ok := value.(*TableMetadata)
	return ok && metadata.Exists && time.Since(metadata.LastChecked) < topSQLScopedSourceRefreshInterval
}

func topSQLBaseDataSource(dataSource string) string {
	dataSource = strings.TrimRight(dataSource, "/")
	for _, suffix := range []string{"/component=tidb", "/component=tikv", "/component=topsql_sql_meta", "/component=topsql_plan_meta"} {
		if strings.HasSuffix(dataSource, suffix) {
			return strings.TrimSuffix(dataSource, suffix)
		}
	}
	return dataSource
}

func isBigQueryTopSQLView(table string) bool {
	_, ok := bigQueryTopSQLViewClusterID(table)
	return ok
}

func bigQueryTopSQLViewClusterID(table string) (string, bool) {
	scope, ok := parseBigQueryTopSQLViewScope(table)
	if !ok {
		return "", false
	}
	return scope.clusterID, true
}

type bigQueryTopSQLViewScope struct {
	clusterID string
	months    []string
}

func parseBigQueryTopSQLViewScope(table string) (bigQueryTopSQLViewScope, bool) {
	const prefix = "topsql_"
	if !strings.HasPrefix(table, prefix) {
		return bigQueryTopSQLViewScope{}, false
	}
	suffix := strings.TrimPrefix(table, prefix)
	clusterID, monthSuffix, scoped := strings.Cut(suffix, "_m")
	if clusterID == "" || !isAllDecimalDigits(clusterID) {
		return bigQueryTopSQLViewScope{}, false
	}
	if !scoped {
		return bigQueryTopSQLViewScope{clusterID: clusterID}, true
	}
	months := strings.Split(monthSuffix, "_")
	if len(months) == 0 {
		return bigQueryTopSQLViewScope{}, false
	}
	for index, month := range months {
		if !validTopSQLMonth(month) || (index > 0 && months[index-1] >= month) {
			return bigQueryTopSQLViewScope{}, false
		}
	}
	return bigQueryTopSQLViewScope{clusterID: clusterID, months: months}, true
}

func isBigQueryTopSQLPhysicalTable(table string) bool {
	return topSQLTableKind(table) != ""
}

func topSQLTableKind(table string) string {
	scope, ok := parseBigQueryTopSQLPhysicalScope(table)
	if !ok {
		return ""
	}
	return scope.kind
}

type bigQueryTopSQLPhysicalScope struct {
	kind      string
	clusterID string
	month     string
}

func parseBigQueryTopSQLPhysicalScope(table string) (bigQueryTopSQLPhysicalScope, bool) {
	for _, prefix := range []string{"topsql_tidb_", "topsql_tikv_", "topsql_sql_meta_", "topsql_plan_meta_"} {
		if !strings.HasPrefix(table, prefix) {
			continue
		}
		suffix := strings.TrimPrefix(table, prefix)
		clusterID, month, scoped := strings.Cut(suffix, "_m")
		if isAllDecimalDigits(clusterID) && (!scoped || validTopSQLMonth(month)) {
			return bigQueryTopSQLPhysicalScope{
				kind:      strings.TrimSuffix(prefix, "_"),
				clusterID: clusterID,
				month:     month,
			}, true
		}
	}
	return bigQueryTopSQLPhysicalScope{}, false
}

func topSQLTableMonth(table string) string {
	scope, ok := parseBigQueryTopSQLPhysicalScope(table)
	if !ok {
		return ""
	}
	return scope.month
}

func validTopSQLMonth(month string) bool {
	if len(month) != 6 || !isAllDecimalDigits(month) {
		return false
	}
	_, err := time.Parse("200601", month)
	return err == nil
}

func (tm *BigQueryTableManager) maybeRefreshSchema(ctx context.Context, dataset, table, dataSource string) {
	if isBigQueryTopSQLPhysicalTable(table) || isBigQueryTopSQLView(table) {
		return
	}
	if tm.config.SchemaRefreshSeconds <= 0 {
		return
	}
	gcsLocation := tm.extractGCSLocation(dataSource)
	if gcsLocation == "" {
		return
	}

	key := schemaRefreshKey(dataset, table)
	interval := time.Duration(tm.config.SchemaRefreshSeconds) * time.Second
	if refreshedAt, ok := tm.schemaRefresh.Load(key); ok &&
		time.Since(refreshedAt.(time.Time)) < interval {
		return
	}

	_, err, _ := tm.refreshGroup.Do(key, func() (interface{}, error) {
		if refreshedAt, ok := tm.schemaRefresh.Load(key); ok &&
			time.Since(refreshedAt.(time.Time)) < interval {
			return nil, nil
		}
		refreshed, err := tm.engine.RefreshExternalTableSchema(
			ctx, dataset, table, gcsLocation,
		)
		if err != nil {
			return nil, err
		}
		tm.schemaRefresh.Store(key, time.Now())
		if refreshed {
			observability.RecordDataProxySchemaRefresh("gcp", "success")
			log.Info(ctx, "BigQuery Delta schema refreshed",
				zap.String("dataset", dataset),
				zap.String("table", table),
			)
		} else {
			observability.RecordDataProxySchemaRefresh("gcp", "skipped")
		}
		return nil, nil
	})
	if err != nil {
		// Schema maintenance is fail-open: existing-field queries should remain
		// available while a later request retries the refresh.
		log.Warn(ctx, "BigQuery Delta schema refresh failed",
			zap.String("dataset", dataset),
			zap.String("table", table),
			zap.Error(err),
		)
		observability.RecordDataProxySchemaRefresh("gcp", "failure")
	}
}

func schemaRefreshKey(dataset, table string) string {
	return fmt.Sprintf("schema:%s.%s", dataset, table)
}

// extractGCSLocation 从 data_source 提取 GCS 基础路径
// 例如：gs://bucket/path/date=20251231/ -> gs://bucket/path/
func (tm *BigQueryTableManager) extractGCSLocation(dataSource string) string {
	dataSource = strings.TrimRight(dataSource, "/")

	if !strings.HasPrefix(dataSource, "gs://") {
		return ""
	}

	return dataSource
}

func (tm *BigQueryTableManager) getCachedMetadata(key string) (*TableMetadata, bool) {
	if value, ok := tm.cache.Load(key); ok {
		metadata := value.(*TableMetadata)
		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
}

func (tm *BigQueryTableManager) setCachedMetadata(key string, metadata *TableMetadata) {
	tm.cache.Store(key, metadata)
}