package config

import (
	"fmt"
	"net"
	"net/url"
	"os"
	"strconv"
	"strings"

	"gopkg.in/yaml.v3"
)

// DataProxyConfig Data Proxy 配置
type DataProxyConfig struct {
	ClinicAPIBase    string             `yaml:"clinic_api_base"`     // Clinic 内部 API 基础地址
	ClinicAPIPublic  string             `yaml:"clinic_api_public"`   // Clinic 公网 API 基础地址
	ClinicBaseURL    string             `yaml:"clinic_base_url"`     // Clinic 根域名，用于生成 slow query detail link
	TidbMgmtHTTPBase string             `yaml:"tidb_mgmt_http_base"` // TiDB management HTTP API 基础地址
	TiCDC            TiCDCConfig        `yaml:"ticdc"`
	AWS              AWSConfig          `yaml:"aws"`
	GCP              GCPConfig          `yaml:"gcp"`
	AliCloud         AliCloudConfig     `yaml:"alicloud"`
	Azure            AzureConfig        `yaml:"azure"`
	DuckDB           DuckDBConfig       `yaml:"duckdb"`
	MetricsQuery     MetricsQueryConfig `yaml:"metrics_query"` // Metrics 查询配置
	LokiProxy        LokiProxyConfig    `yaml:"loki_proxy"`    // Loki 代理配置
	AI               AIConfig           `yaml:"ai"`            // AI 分析配置
}

// TiCDCConfig contains the private gRPC endpoints used by the read-only TiCDC
// metadata APIs.
type TiCDCConfig struct {
	DataflowAddr   string `yaml:"dataflow_addr"`
	DataflowNGAddr string `yaml:"dataflow_ng_addr"`
	ServerlessAddr string `yaml:"serverless_addr"`
}

// MetricsQueryConfig Metrics 查询服务配置
type MetricsQueryConfig struct {
	Enabled        bool                 `yaml:"enabled"`         // 是否启用
	MetricsMapping MetricsMappingConfig `yaml:"metrics_mapping"` // 指标分类映射 (复用 bean.go 中的定义)
}

// AWSConfig AWS 配置
type AWSConfig struct {
	Enabled              bool   `yaml:"enabled"`
	Region               string `yaml:"region"`
	AthenaDatabase       string `yaml:"athena_database"`
	AthenaWorkgroup      string `yaml:"athena_workgroup"`
	AthenaOutputLocation string `yaml:"athena_output_location"`
	// 使用 AssumeRole 方式访问
	RoleArn      string             `yaml:"role_arn"`
	AccessKey    string             `yaml:"access_key"`
	SecretKey    string             `yaml:"secret_key"`
	SessionToken string             `yaml:"session_token"`
	TableManager TableManagerConfig `yaml:"table_manager"` // 表管理配置
}

// TableManagerConfig 表管理配置
type TableManagerConfig struct {
	Enabled         bool `yaml:"enabled"`           // 是否启用表管理
	AutoCreate      bool `yaml:"auto_create"`       // 是否自动创建表
	CacheTTLSeconds int  `yaml:"cache_ttl_seconds"` // 缓存有效期（秒）
	// SchemaRefreshSeconds controls periodic schema autodetection for existing
	// BigQuery Delta Lake external tables. Zero disables refresh.
	SchemaRefreshSeconds int `yaml:"schema_refresh_seconds"`
	// InstanceProjectionRefreshSeconds Athena 分区投影 projection.instance.values 与 S3 对齐的刷新间隔（秒）。
	// 仅对 DDL 含 {{.InstanceValues}} 的表生效。0 表示关闭（不访问 Glue/S3 做增量更新）。
	InstanceProjectionRefreshSeconds int                      `yaml:"instance_projection_refresh_seconds"`
	TableTemplates                   map[string]TableTemplate `yaml:"table_templates"` // 表模板定义
}

// TableTemplate 表模板
type TableTemplate struct {
	DDL       string   `yaml:"ddl"`                  // DDL 语句模板
	Type      string   `yaml:"type,omitempty"`       // "table" (默认) 或 "view"
	DependsOn []string `yaml:"depends_on,omitempty"` // view 依赖的表模板名称列表
}

// IsView 返回该模板是否为 VIEW 类型
func (t TableTemplate) IsView() bool {
	return strings.EqualFold(t.Type, "view")
}

// IsDeltaTable returns whether the Athena DDL registers a native Delta Lake table.
func (t TableTemplate) IsDeltaTable() bool {
	ddl := strings.ToLower(t.DDL)
	return strings.Contains(ddl, "table_type") &&
		strings.Contains(ddl, "delta")
}

// GCPConfig GCP 配置
type GCPConfig struct {
	Enabled             bool               `yaml:"enabled"`
	ProjectID           string             `yaml:"project_id"`
	BigQueryDataset     string             `yaml:"bigquery_dataset"`       // 数据集前缀，实际 dataset = prefix_region（如 clinic_data_proxy_dev_asia_east1）
	QueryScanBytesLimit string             `yaml:"query_scan_bytes_limit"` // 单次查询扫描字节数限制，默认 1GB
	TableManager        TableManagerConfig `yaml:"table_manager"`          // 表管理配置（Delta Lake 外部表）
}

// MaxComputeConfig MaxCompute 引擎配置
// endpoint 格式固定为 https://service.{region}.maxcompute.aliyun.com/api，无需 regions 映射
//
// 认证方式二选一：
//  1. AK/SK：配置 access_key 和 secret_key
//  2. OIDC 无密钥（EKS 等）：配置 oidc_provider_arn、role_arn、oidc_token_file，
//     使用 Pod 的 service account token 通过 AssumeRoleWithOIDC 获取临时凭证
type MaxComputeConfig struct {
	ProjectPrefix string `yaml:"project_prefix"` // 项目前缀，实际 project = prefix_region（如 clinic_data_proxy_dev_ap_southeast_1）
	AccessKey     string `yaml:"access_key"`     // 阿里云 AccessKey（与 OIDC 二选一）
	SecretKey     string `yaml:"secret_key"`     // 阿里云 SecretKey（与 OIDC 二选一）
	// EnableSlowQueryParquetPredicatePushdown atomically enables the MaxCompute
	// native Parquet reader and row-group predicate pushdown for eligible slow-query
	// SELECT tasks. It is disabled by default.
	EnableSlowQueryParquetPredicatePushdown bool `yaml:"enable_slow_query_parquet_predicate_pushdown"`
	// SlowQueryParquetPredicatePushdownRegions lists the MaxCompute regions
	// eligible for rollout. Empty keeps the optimization disabled so a
	// rollout always requires an explicit regional allowlist.
	SlowQueryParquetPredicatePushdownRegions []string `yaml:"slow_query_parquet_predicate_pushdown_regions"`
	// OIDC 无密钥访问（EKS 等 K8s 环境）
	// 需在阿里云 RAM 创建 OIDC IdP（issuer 指向 EKS 集群 OIDC）及信任该 IdP 的 RAM 角色
	OIDCProviderArn string `yaml:"oidc_provider_arn"` // 阿里云 RAM OIDC 身份提供方 ARN（如 acs:ram::123456789012:oidc-provider/eks-xxx）
	RoleArn         string `yaml:"role_arn"`          // 要扮演的 RAM 角色 ARN（如 acs:ram::123456789012:role/maxcompute-pod-role）
	OIDCTokenFile   string `yaml:"oidc_token_file"`   // Pod 内 token 路径；EKS 建议用 projected volume 挂载带集群 issuer 的 token
	// AssumeRoleWithOIDC 的会话时长（秒），须小于 K8s SA JWT 的 TTL，且满足阿里云 900s–角色上限；默认 1800
	OIDCRoleSessionSeconds int `yaml:"oidc_role_session_seconds"`
	// 自动创建 project（若不存在，需 odps:CreateProject 权限）
	AutoCreateProject  bool   `yaml:"auto_create_project"`  // 是否在 project 不存在时自动创建
	ProjectProductType string `yaml:"project_product_type"` // 创建 project 时的 productType：payasyougo/subscription/dev，默认 payasyougo
	TenantID           string `yaml:"tenant_id"`            // 租户 ID（可选；未配置时通过 ListUsers 接口自动获取）
	SqlMeteringMax     int    `yaml:"sql_metering_max"`     // 单次 SQL 消费上限（GB*复杂度），默认 1
	AutoMvQuotaGb      int    `yaml:"auto_mv_quota_gb"`     // 自动物化视图配额（GB），默认 10
}

// AliCloudConfig 阿里云配置
type AliCloudConfig struct {
	Enabled      bool               `yaml:"enabled"`
	TableManager TableManagerConfig `yaml:"table_manager"` // 表管理配置（OSS 外部表）
	MaxCompute   MaxComputeConfig   `yaml:"maxcompute"`    // MaxCompute 引擎配置
}

// AzureConfig Azure Synapse + ADLS/Blob 配置
type AzureConfig struct {
	Enabled bool               `yaml:"enabled"`
	Synapse AzureSynapseConfig `yaml:"synapse"`
}

// AzureSynapseConfig Synapse Serverless SQL 配置
type AzureSynapseConfig struct {
	Server              string `yaml:"server"`
	Port                int    `yaml:"port"`
	Database            string `yaml:"database"`
	AuthType            string `yaml:"auth_type"` // aad_service_principal | sql | managed_identity
	Username            string `yaml:"username"`
	Password            string `yaml:"password"`
	TenantID            string `yaml:"tenant_id"`
	ClientID            string `yaml:"client_id"`
	ClientSecret        string `yaml:"client_secret"`
	QueryTimeoutSeconds int    `yaml:"query_timeout_seconds"`
	MaxOpenConns        int    `yaml:"max_open_conns"`
	MaxIdleConns        int    `yaml:"max_idle_conns"`
	ConnMaxLifetimeMin  int    `yaml:"conn_max_lifetime_minutes"`
}

// DuckDBConfig DuckDB 配置
type DuckDBConfig struct {
	Enabled     bool   `yaml:"enabled"`
	S3AccessKey string `yaml:"s3_access_key"`
	S3SecretKey string `yaml:"s3_secret_key"`
}

// LokiProxyConfig Loki 代理配置
type LokiProxyConfig struct {
	Env                     string                     `yaml:"env"`
	MaxLimit                int                        `yaml:"max_limit"`                 // 查询结果的最大限制，默认 1000
	DefaultSystemDatasource string                     `yaml:"default_system_datasource"` // 管控面日志默认 datasource，用于 discovery 展示
	SystemEndpoints         []SystemLokiEndpointConfig `yaml:"system_endpoints"`          // 管控面 Loki endpoint 列表
}

// SystemLokiEndpointConfig maps a control-plane Loki datasource to its upstream URL.
type SystemLokiEndpointConfig struct {
	Datasource string `yaml:"datasource"`
	UID        string `yaml:"uid"`
	Provider   string `yaml:"provider"`
	Region     string `yaml:"region"`
	URL        string `yaml:"url"`
}

// LoadDataProxyConfig 从文件加载 DataProxy 配置
func LoadDataProxyConfig(filePath string) (*DataProxyConfig, error) {
	cfg, err := loadDataProxyConfig(filePath)
	if err != nil {
		return nil, err
	}

	// 验证配置
	if err := cfg.Validate(); err != nil {
		return nil, fmt.Errorf("invalid config: %w", err)
	}

	return cfg, nil
}

// LoadClinicAPIBase reads and validates only Clinic's internal service URL, so
// public resource routes remain independent from warehouse engine settings.
func LoadClinicAPIBase(filePath string) (string, error) {
	cfg, err := loadDataProxyConfig(filePath)
	if err != nil {
		return "", err
	}
	baseURL, err := url.Parse(cfg.ClinicAPIBase)
	if err != nil || (baseURL.Scheme != "http" && baseURL.Scheme != "https") || baseURL.Host == "" {
		return "", fmt.Errorf("clinic_api_base must be an absolute HTTP(S) URL")
	}
	if baseURL.User != nil || baseURL.RawQuery != "" || baseURL.Fragment != "" {
		return "", fmt.Errorf("clinic_api_base must not contain userinfo, query, or fragment")
	}
	return cfg.ClinicAPIBase, nil
}

// LoadTidbMgmtHTTPBase reads the optional user-authorized TiDB management HTTP
// base URL. It deliberately accepts HTTP for isolated tests and private
// non-production deployments, but disallows URL features that could redirect requests.
func LoadTidbMgmtHTTPBase(filePath string) (string, error) {
	cfg, err := loadDataProxyConfig(filePath)
	if err != nil {
		return "", err
	}
	if cfg.TidbMgmtHTTPBase == "" {
		return "", nil
	}
	baseURL, err := url.Parse(cfg.TidbMgmtHTTPBase)
	if err != nil || (baseURL.Scheme != "http" && baseURL.Scheme != "https") || baseURL.Host == "" {
		return "", fmt.Errorf("tidb_mgmt_http_base must be an absolute HTTP(S) URL")
	}
	if baseURL.User != nil || baseURL.RawQuery != "" || baseURL.Fragment != "" {
		return "", fmt.Errorf("tidb_mgmt_http_base must not contain userinfo, query, or fragment")
	}
	return cfg.TidbMgmtHTTPBase, nil
}

// LoadTiCDCConfig loads and validates both Dataflow endpoints. TiCDC routes
// are always registered, so missing endpoints are a startup configuration
// error rather than a partially available runtime mode.
func LoadTiCDCConfig(filePath string) (TiCDCConfig, error) {
	cfg, err := loadDataProxyConfig(filePath)
	if err != nil {
		return TiCDCConfig{}, err
	}
	for name, addr := range map[string]string{
		"ticdc.dataflow_addr":    cfg.TiCDC.DataflowAddr,
		"ticdc.dataflow_ng_addr": cfg.TiCDC.DataflowNGAddr,
		"ticdc.serverless_addr":  cfg.TiCDC.ServerlessAddr,
	} {
		addr = strings.TrimSpace(addr)
		host, port, splitErr := net.SplitHostPort(addr)
		portNumber, portErr := strconv.ParseUint(port, 10, 16)
		if splitErr != nil || host == "" || strings.Contains(host, "://") || portErr != nil || portNumber == 0 {
			return TiCDCConfig{}, fmt.Errorf("%s must be a host:port address", name)
		}
	}
	cfg.TiCDC.DataflowAddr = strings.TrimSpace(cfg.TiCDC.DataflowAddr)
	cfg.TiCDC.DataflowNGAddr = strings.TrimSpace(cfg.TiCDC.DataflowNGAddr)
	cfg.TiCDC.ServerlessAddr = strings.TrimSpace(cfg.TiCDC.ServerlessAddr)
	return cfg.TiCDC, nil
}

func loadDataProxyConfig(filePath string) (*DataProxyConfig, error) {
	// 先获取默认配置
	cfg := DefaultConfig()

	data, err := os.ReadFile(filePath)
	if err != nil {
		return nil, fmt.Errorf("failed to read config file: %w", err)
	}

	// 保护 Athena Partition Projection 占位符 ${date}, ${instance} 等，避免被 os.ExpandEnv 处理
	// 替换为临时占位符
	dataStr := string(data)
	dataStr = strings.ReplaceAll(dataStr, "${date}", "__PLACEHOLDER_DATE__")
	dataStr = strings.ReplaceAll(dataStr, "${instance}", "__PLACEHOLDER_INSTANCE__")

	// 替换环境变量
	dataStr = os.ExpandEnv(dataStr)

	// 恢复 Partition Projection 占位符
	dataStr = strings.ReplaceAll(dataStr, "__PLACEHOLDER_DATE__", "${date}")
	dataStr = strings.ReplaceAll(dataStr, "__PLACEHOLDER_INSTANCE__", "${instance}")
	if err := rejectMisplacedInstanceProjectionRefresh([]byte(dataStr)); err != nil {
		return nil, err
	}

	// 将配置文件内容合并到默认配置上
	if err := yaml.Unmarshal([]byte(dataStr), cfg); err != nil {
		return nil, fmt.Errorf("failed to unmarshal config: %w", err)
	}

	return cfg, nil
}

func rejectMisplacedInstanceProjectionRefresh(data []byte) error {
	var document yaml.Node
	if err := yaml.Unmarshal(data, &document); err != nil {
		return nil // the normal unmarshal below returns the authoritative syntax error
	}
	if len(document.Content) == 0 || len(document.Content[0].Content) == 0 {
		return nil
	}
	root := document.Content[0]
	if root.Kind == yaml.DocumentNode {
		root = root.Content[0]
	}
	if root.Kind != yaml.MappingNode {
		return nil
	}
	for i := 0; i+1 < len(root.Content); i += 2 {
		if root.Content[i].Value != "aws" {
			continue
		}
		awsNode := root.Content[i+1]
		for j := 0; j+1 < len(awsNode.Content); j += 2 {
			if awsNode.Content[j].Value == "instance_projection_refresh_seconds" {
				return fmt.Errorf(
					"aws.instance_projection_refresh_seconds is invalid; use aws.table_manager.instance_projection_refresh_seconds",
				)
			}
		}
	}
	return nil
}

// Validate 验证配置
func (c *DataProxyConfig) Validate() error {
	// 至少需要启用一个引擎
	if !c.AWS.Enabled && !c.GCP.Enabled && !c.AliCloud.Enabled && !c.Azure.Enabled && !c.DuckDB.Enabled {
		return fmt.Errorf("at least one engine must be enabled")
	}

	// 验证 AWS 配置
	if c.AWS.Enabled {
		// region 不再要求在配置文件中设置，可以从请求参数传递
		// 这样可以避免跨 Region 调用
		if c.AWS.AthenaDatabase == "" {
			return fmt.Errorf("aws.athena_database is required when AWS is enabled")
		}
		// athena_output_location 现在是可选的
		// 如果为空，将使用 Workgroup 的默认配置
	}

	// 验证 GCP 配置
	if c.GCP.Enabled {
		if c.GCP.ProjectID == "" {
			return fmt.Errorf("gcp.project_id is required when GCP is enabled")
		}
		if c.GCP.BigQueryDataset == "" {
			return fmt.Errorf("gcp.bigquery_dataset is required when GCP is enabled")
		}
	}

	// 验证阿里云配置
	if c.AliCloud.Enabled {
		if c.AliCloud.MaxCompute.ProjectPrefix == "" {
			return fmt.Errorf("alicloud.maxcompute.project_prefix is required when AliCloud is enabled (project = prefix_region)")
		}
		mc := &c.AliCloud.MaxCompute
		useOIDC := mc.OIDCProviderArn != "" && mc.RoleArn != ""
		useAKSK := mc.AccessKey != "" && mc.SecretKey != ""
		if !useOIDC && !useAKSK {
			return fmt.Errorf("alicloud.maxcompute: must configure either (access_key+secret_key) or (oidc_provider_arn+role_arn) for authentication")
		}
		if useOIDC && useAKSK {
			return fmt.Errorf("alicloud.maxcompute: cannot configure both AK/SK and OIDC; choose one authentication method")
		}
		if useOIDC && mc.OIDCRoleSessionSeconds != 0 && mc.OIDCRoleSessionSeconds < 900 {
			return fmt.Errorf("alicloud.maxcompute.oidc_role_session_seconds must be >= 900 (Aliyun AssumeRoleWithOIDC minimum) or omit for default 1800")
		}
		if mc.EnableSlowQueryParquetPredicatePushdown && len(mc.SlowQueryParquetPredicatePushdownRegions) == 0 {
			return fmt.Errorf("alicloud.maxcompute.slow_query_parquet_predicate_pushdown_regions is required when predicate pushdown is enabled")
		}
		for _, region := range mc.SlowQueryParquetPredicatePushdownRegions {
			if strings.TrimSpace(region) == "" {
				return fmt.Errorf("alicloud.maxcompute.slow_query_parquet_predicate_pushdown_regions must not contain an empty region")
			}
		}
		// tenant_id 可选：使用区域化 endpoint (maxcompute.{region}.aliyuncs.com) 时 CreateProject 无需 tenantId
	}

	if c.Azure.Enabled {
		if c.Azure.Synapse.Server == "" {
			return fmt.Errorf("azure.synapse.server is required when Azure is enabled")
		}
		if c.Azure.Synapse.Database == "" {
			return fmt.Errorf("azure.synapse.database is required when Azure is enabled")
		}
		authType := strings.ToLower(c.Azure.Synapse.AuthType)
		if authType == "" {
			authType = "aad_service_principal"
			c.Azure.Synapse.AuthType = authType
		}
		switch authType {
		case "aad_service_principal":
			if c.Azure.Synapse.TenantID == "" || c.Azure.Synapse.ClientID == "" || c.Azure.Synapse.ClientSecret == "" {
				return fmt.Errorf("azure.synapse tenant_id, client_id, and client_secret are required for aad_service_principal auth")
			}
		case "sql":
			if c.Azure.Synapse.Username == "" || c.Azure.Synapse.Password == "" {
				return fmt.Errorf("azure.synapse username and password are required for sql auth")
			}
		case "managed_identity":
			// Supported by config shape; engine can use DefaultAzureCredential.
		default:
			return fmt.Errorf("unsupported azure.synapse.auth_type: %s", c.Azure.Synapse.AuthType)
		}
		if c.Azure.Synapse.Port == 0 {
			c.Azure.Synapse.Port = 1433
		}
		if c.Azure.Synapse.QueryTimeoutSeconds == 0 {
			c.Azure.Synapse.QueryTimeoutSeconds = 300
		}
	}

	return nil
}

// DefaultConfig 返回默认配置
func DefaultConfig() *DataProxyConfig {
	return &DataProxyConfig{
		ClinicAPIBase:   "http://clinic-internal-api:80",
		ClinicAPIPublic: "",
		ClinicBaseURL:   "https://clinic.pingcap.com",
		AWS: AWSConfig{
			Enabled:              false,
			Region:               "us-east-1",
			AthenaDatabase:       "default",
			AthenaWorkgroup:      "primary",
			AthenaOutputLocation: "s3://my-bucket/athena-results/",
		},
		GCP: GCPConfig{
			Enabled:         false,
			ProjectID:       "",
			BigQueryDataset: "",
			TableManager: TableManagerConfig{
				Enabled:         true,
				AutoCreate:      true,
				CacheTTLSeconds: 0,
			},
		},
		AliCloud: AliCloudConfig{
			Enabled: false,
			TableManager: TableManagerConfig{
				Enabled:         true,
				AutoCreate:      true,
				CacheTTLSeconds: 0,
			},
			MaxCompute: MaxComputeConfig{},
		},
		Azure: AzureConfig{
			Enabled: false,
			Synapse: AzureSynapseConfig{
				Port:                1433,
				Database:            "master",
				AuthType:            "aad_service_principal",
				QueryTimeoutSeconds: 300,
				MaxOpenConns:        10,
				MaxIdleConns:        5,
				ConnMaxLifetimeMin:  30,
			},
		},
		DuckDB: DuckDBConfig{
			Enabled: false,
		},
		MetricsQuery: MetricsQueryConfig{
			Enabled: true,
			MetricsMapping: MetricsMappingConfig{
				DataPlaneMetrics: []string{
					"node_*",
					"kube_*",
					"container_*",
					"aws_ebs_csi_*",
					"namespace_*",
					"node_memory_MemTotal_bytes",
					"node_memory_MemAvailable_bytes",
					"node_memory_MemFree_bytes",
					"node_memory_Buffers_bytes",
					"node_memory_Cached_bytes",
					"node_memory_Slab_bytes",
					"node_cpu_seconds_total",
					"node_disk_io_time_seconds_total",
					"node_disk_read_bytes_total",
					"node_disk_written_bytes_total",
					"node_disk_read_time_seconds_total",
					"node_disk_reads_completed_total",
					"node_disk_write_time_seconds_total",
					"node_disk_writes_completed_total",
					"kube_pod_info",
					"aws_ebs_csi_exceeded_iops_seconds_total",
				},
			},
		},
	}
}