package engine

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/alibabacloud-go/darabonba-openapi/v2/client"
	mcclient "github.com/alibabacloud-go/maxcompute-20220104/v4/client"
	"github.com/alibabacloud-go/tea/tea"
	"github.com/aliyun/aliyun-odps-go-sdk/odps"
	"github.com/aliyun/aliyun-odps-go-sdk/odps/account"
	"github.com/aliyun/credentials-go/credentials"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

var (
	maxComputeUnknownFieldPattern   = regexp.MustCompile(`(?i)ODPS-0130071:\[[0-9]+,[0-9]+\]\s+Semantic analysis exception\s+-\s+column\s+([A-Za-z_][A-Za-z0-9_]*)\s+cannot be resolved\b`)
	maxComputeSlowQueryTablePattern = regexp.MustCompile(`(?i)^slow_query_logs(?:_[0-9]+)?$`)
)

// maxComputeEndpointFormat MaxCompute endpoint 固定格式
const maxComputeEndpointFormat = "https://service.%s.maxcompute.aliyun.com/api"

const defaultOIDCTokenPath = "/var/run/secrets/eks.amazonaws.com/serviceaccount/token"

const (
	maxComputeParquetNativeHint            = "odps.ext.parquet.native"
	maxComputeParquetPredicatePushdownHint = "odps.sql.parquet.use.predicate.pushdown"
)

// defaultOIDCRoleSessionSeconds OIDC 模式下 AssumeRoleWithOIDC 的会话时长（秒）。
// 必须明显短于 K8s ServiceAccount JWT 的典型 TTL（如 3600s），否则 credentials-go 仅按 STS 过期刷新，
// 可能在 JWT 已过期后才再次 AssumeRole，触发 AuthenticationFail.OIDCToken.Expired。
// 阿里云要求 900–角色最大会话时长之间；未配置时使用 1800（30 分钟）。
const defaultOIDCRoleSessionSeconds = 1800

// MaxComputeEngine 阿里云 MaxCompute (ODPS) 查询引擎
// 使用 aliyun-odps-go-sdk，SQL 方言与 MySQL 不同，需通过 dialect 转换
// 支持 AK/SK 或 OIDC（Pod token AssumeRoleWithOIDC）两种认证
type MaxComputeEngine struct {
	acc                                      account.Account // AliyunAccount 或 StsAccount
	projectPrefix                            string
	odpsPool                                 sync.Map // map[region]*odps.Odps
	mu                                       sync.RWMutex
	autoCreateProject                        bool
	projectProductType                       string
	sqlMeteringMax                           int // 单次 SQL 消费上限（GB），默认 1
	autoMvQuotaGb                            int // 自动物化视图配额（GB），默认 10
	enableSlowQueryParquetPredicatePushdown  bool
	slowQueryParquetPredicatePushdownRegions []string
	credProvider                             credentials.Credential
}

// MaxComputeEngineConfig MaxCompute 引擎配置
type MaxComputeEngineConfig struct {
	AccessKey     string
	SecretKey     string
	ProjectPrefix string
	// EnableSlowQueryParquetPredicatePushdown atomically enables the native
	// Parquet reader and row-group predicate pushdown for eligible slow-query
	// SELECT tasks only.
	EnableSlowQueryParquetPredicatePushdown bool
	// SlowQueryParquetPredicatePushdownRegions lists the regions eligible for
	// rollout. Empty disables the optimization so a rollout always
	// requires an explicit regional allowlist.
	SlowQueryParquetPredicatePushdownRegions []string
	// OIDC 无密钥（EKS 等）
	OIDCProviderArn string
	RoleArn         string
	OIDCTokenFile   string
	// OIDCRoleSessionSeconds AssumeRoleWithOIDC 的 DurationSeconds（秒），默认 defaultOIDCRoleSessionSeconds
	OIDCRoleSessionSeconds int
	// 自动创建 project（若不存在）
	AutoCreateProject  bool
	ProjectProductType string // payasyougo/subscription/dev，默认 payasyougo
	SqlMeteringMax     int    // 单次 SQL 消费上限（GB），默认 1
	AutoMvQuotaGb      int    // 自动物化视图配额（GB），默认 10
}

// NewMaxComputeEngine 创建 MaxCompute 引擎实例
// 认证方式：AK/SK 或 OIDC（oidc_provider_arn + role_arn + oidc_token_file）
func NewMaxComputeEngine(cfg *MaxComputeEngineConfig) (*MaxComputeEngine, error) {
	if cfg.ProjectPrefix == "" {
		return nil, fmt.Errorf("MaxCompute requires project_prefix (project = prefix_region)")
	}

	var acc account.Account
	var credProvider credentials.Credential
	if cfg.OIDCProviderArn != "" && cfg.RoleArn != "" {
		// OIDC 无密钥：使用 Pod service account token assume 到阿里云
		tokenFile := cfg.OIDCTokenFile
		if tokenFile == "" {
			tokenFile = defaultOIDCTokenPath
		}
		sessionSecs := cfg.OIDCRoleSessionSeconds
		if sessionSecs <= 0 {
			sessionSecs = defaultOIDCRoleSessionSeconds
		}
		buildOIDCCred := func() (credentials.Credential, error) {
			credConfig := new(credentials.Config).
				SetType("oidc_role_arn").
				SetOIDCProviderArn(cfg.OIDCProviderArn).
				SetRoleArn(cfg.RoleArn).
				SetOIDCTokenFilePath(tokenFile).
				SetRoleSessionName("clinic-data-proxy").
				SetRoleSessionExpiration(sessionSecs)
			return credentials.NewCredential(credConfig)
		}
		provider, err := buildOIDCCred()
		if err != nil {
			return nil, fmt.Errorf("MaxCompute OIDC credential init failed: %w", err)
		}
		credProvider = newOIDCCredentialWrap(provider, tokenFile, buildOIDCCred)
		acc = account.NewStsAccountWithCredential(credProvider)
	} else if cfg.AccessKey != "" && cfg.SecretKey != "" {
		acc = account.NewAliyunAccount(cfg.AccessKey, cfg.SecretKey)
		credConfig := new(credentials.Config).
			SetType("access_key").
			SetAccessKeyId(cfg.AccessKey).
			SetAccessKeySecret(cfg.SecretKey)
		provider, err := credentials.NewCredential(credConfig)
		if err != nil {
			return nil, fmt.Errorf("MaxCompute credential init failed: %w", err)
		}
		credProvider = provider
	} else {
		return nil, fmt.Errorf("MaxCompute requires either (access_key+secret_key) or (oidc_provider_arn+role_arn)")
	}

	productType := cfg.ProjectProductType
	if productType == "" {
		productType = "payasyougo"
	}
	// tenant_id 可选：未配置时将在首次自动创建 project 时通过 ListUsers 接口获取
	sqlMeteringMax := cfg.SqlMeteringMax
	if sqlMeteringMax <= 0 {
		sqlMeteringMax = 1
	}
	autoMvQuotaGb := cfg.AutoMvQuotaGb
	if autoMvQuotaGb <= 0 {
		autoMvQuotaGb = 10
	}

	return &MaxComputeEngine{
		acc:                                      acc,
		projectPrefix:                            cfg.ProjectPrefix,
		autoCreateProject:                        cfg.AutoCreateProject,
		projectProductType:                       productType,
		sqlMeteringMax:                           sqlMeteringMax,
		autoMvQuotaGb:                            autoMvQuotaGb,
		enableSlowQueryParquetPredicatePushdown:  cfg.EnableSlowQueryParquetPredicatePushdown,
		slowQueryParquetPredicatePushdownRegions: append([]string(nil), cfg.SlowQueryParquetPredicatePushdownRegions...),
		credProvider:                             credProvider,
	}, nil
}

func maxComputeRegionAllowed(region string, allowedRegions []string) bool {
	if len(allowedRegions) == 0 {
		return false
	}
	for _, allowed := range allowedRegions {
		if strings.EqualFold(strings.TrimSpace(allowed), region) {
			return true
		}
	}
	return false
}

func maxComputeSlowQueryParquetHints(
	enabled bool,
	allowedRegions []string,
	project string,
	sql string,
	dataSource *dto.DataSourceInfo,
) map[string]string {
	if !enabled || dataSource == nil ||
		!strings.EqualFold(dataSource.Vendor, "alicloud") ||
		!strings.EqualFold(dataSource.Scheme, "oss") ||
		!maxComputeRegionAllowed(dataSource.Region, allowedRegions) {
		return nil
	}
	// Avoid paying the AST parse cost for non-slow-query traffic such as TopSQL.
	// The AST and exact table-name checks below remain the source of truth.
	if !strings.Contains(strings.ToLower(sql), "slow_query_logs") {
		return nil
	}

	refs, err := extractTableReferencesAST(sql)
	if err != nil || len(refs) == 0 {
		return nil
	}
	for _, ref := range refs {
		if ref.Database != "" && !strings.EqualFold(ref.Database, project) {
			return nil
		}
		if !maxComputeSlowQueryTablePattern.MatchString(ref.Table) {
			return nil
		}
	}

	return map[string]string{
		maxComputeParquetNativeHint:            "true",
		maxComputeParquetPredicatePushdownHint: "true",
	}
}

func newMaxComputeQueryTask(sql string, hints map[string]string) odps.SQLTask {
	return odps.NewSqlTask("clinic_query", ensureTrailingSemicolon(sql), hints)
}

// ensureTrailingSemicolon MaxCompute 要求所有 SQL 以 ; 结尾
func ensureTrailingSemicolon(sql string) string {
	sql = strings.TrimSpace(sql)
	if sql == "" {
		return sql
	}
	if !strings.HasSuffix(sql, ";") {
		return sql + ";"
	}
	return sql
}

// GetDatabase 根据 region 返回 MaxCompute project 名（prefix_region 格式）
func (e *MaxComputeEngine) GetDatabase(region string) string {
	if region == "" {
		return e.projectPrefix
	}
	suffix := strings.ReplaceAll(region, "-", "_")
	return e.projectPrefix + "_" + suffix
}

// getOdpsForRegion 获取或创建指定 region 的 Odps 实例
func (e *MaxComputeEngine) getOdpsForRegion(region string) (*odps.Odps, error) {
	if o, ok := e.odpsPool.Load(region); ok {
		return o.(*odps.Odps), nil
	}

	e.mu.Lock()
	defer e.mu.Unlock()

	if o, ok := e.odpsPool.Load(region); ok {
		return o.(*odps.Odps), nil
	}

	endpoint := fmt.Sprintf(maxComputeEndpointFormat, region)
	project := e.GetDatabase(region)

	// 若开启自动创建且 project 不存在，则创建
	if e.autoCreateProject {
		if err := e.ensureProjectExists(region, project); err != nil {
			log.Printf("MaxCompute ensureProjectExists failed (region=%s project=%s): %v", region, project, err)
			return nil, fmt.Errorf("ensure project exists failed: %w", err)
		}
	}

	odpsIns := odps.NewOdps(e.acc, endpoint)
	odpsIns.SetDefaultProjectName(project)
	e.odpsPool.Store(region, odpsIns)
	return odpsIns, nil
}

// ensureProjectExists 检查 project 是否存在，不存在则创建
func (e *MaxComputeEngine) ensureProjectExists(region, project string) error {
	odpsIns := odps.NewOdps(e.acc, fmt.Sprintf(maxComputeEndpointFormat, region))
	odpsIns.SetDefaultProjectName(project)
	projects := odps.NewProjects(odpsIns)
	exists, err := projects.Exists(project)
	if err != nil && (!strings.Contains(err.Error(), "Project not found") &&
		!strings.Contains(err.Error(), "NoSuchObject")) {
		return fmt.Errorf("check project exists failed: %w", err)
	}
	if exists {
		return nil
	}

	mcClient, err := e.newMaxComputeOpenAPIClient(region)
	if err != nil {
		return fmt.Errorf("create MaxCompute OpenAPI client failed: %w", err)
	}

	// https://www.alibabacloud.com/help/zh/maxcompute/user-guide/api-maxcompute-2022-01-04-createproject
	body := map[string]interface{}{
		"name":         project,
		"comment":      "auto-created by clinic data proxy for " + region,
		"productType":  e.projectProductType,
		"defaultQuota": "os_PayAsYouGoQuota",
		"properties": map[string]interface{}{
			"typeSystem":     "2",
			"sqlMeteringMax": strconv.Itoa(e.sqlMeteringMax),
			"enableAutoMv":   true,
			"autoMvQuotaGb":  e.autoMvQuotaGb,
			"enableDr":       false,
		},
	}
	if e.projectProductType == "payasyougo" {
		body["defaultQuota"] = "os_PayAsYouGoQuota"
	}
	bodyJSON, err := json.Marshal(body)
	if err != nil {
		return fmt.Errorf("marshal create project body failed: %w", err)
	}
	bodyStr := string(bodyJSON)
	resp, err := mcClient.CreateProject(&mcclient.CreateProjectRequest{
		Body: tea.String(bodyStr),
	})
	if err != nil {
		return fmt.Errorf("CreateProject API failed: %w", err)
	}
	if resp.StatusCode != nil && *resp.StatusCode >= 400 {
		return fmt.Errorf("CreateProject API failed: status=%v", *resp.StatusCode)
	}
	if resp.Body == nil || resp.Body.Data == nil {
		return fmt.Errorf("CreateProject API returned empty response")
	}
	if resp.Body.Data == nil || *resp.Body.Data == "" {
		return fmt.Errorf("CreateProject API returned empty data")
	}
	if *resp.Body.Data != "success" {
		return fmt.Errorf("CreateProject API returned failed: %s", *resp.Body.Data)
	}

	return nil
}

// newMaxComputeOpenAPIClient 创建 MaxCompute 2022-01-04 OpenAPI 客户端（用于 CreateProject 等）
// 使用区域化 endpoint maxcompute.{region}.aliyuncs.com，与 OpenAPI 调试页一致
func (e *MaxComputeEngine) newMaxComputeOpenAPIClient(region string) (*mcclient.Client, error) {
	regionalHost := fmt.Sprintf("maxcompute.%s.aliyuncs.com", region)
	config := &client.Config{
		Credential: e.credProvider,
		RegionId:   tea.String(region),
		Endpoint:   tea.String(regionalHost),
	}
	return mcclient.NewClient(config)
}

// Name 返回引擎名称
func (e *MaxComputeEngine) Name() string {
	return "maxcompute"
}

// Execute 执行查询
func (e *MaxComputeEngine) Execute(ctx context.Context, sql string, dataSource *dto.DataSourceInfo) (*dto.DataProxyQueryResult, error) {
	startTime := time.Now()

	region := dataSource.Region
	if region == "" {
		return nil, fmt.Errorf("region is required in request parameters for MaxCompute")
	}

	odpsIns, err := e.getOdpsForRegion(region)
	if err != nil {
		return nil, err
	}

	project := odpsIns.DefaultProjectName()
	hints := maxComputeSlowQueryParquetHints(
		e.enableSlowQueryParquetPredicatePushdown,
		e.slowQueryParquetPredicatePushdownRegions,
		project,
		sql,
		dataSource,
	)
	sqlTask := newMaxComputeQueryTask(sql, hints)

	ins, err := sqlTask.Run(odpsIns, project)
	if err != nil {
		return nil, fmt.Errorf("query run failed: %w", err)
	}

	if err := ins.WaitForSuccess(); err != nil {
		if unknownErr := classifyMaxComputeUnknownField(err, ins.Id()); unknownErr != nil {
			return nil, unknownErr
		}
		return nil, fmt.Errorf("query execution failed: %w", err)
	}

	csvReader, err := sqlTask.GetSelectResultAsCsv(ins, true)
	if err != nil {
		return nil, fmt.Errorf("get result failed: %w", err)
	}

	result := &dto.DataProxyQueryResult{
		Columns: []string{},
		Rows:    make([][]interface{}, 0),
	}

	rowIndex := 0
	for {
		record, err := csvReader.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("read result row failed: %w", err)
		}

		if rowIndex == 0 && len(record) > 0 {
			// 第一行为列名（withHeader=true）
			result.Columns = record
			rowIndex++
			continue
		}

		row := make([]interface{}, len(record))
		for i, v := range record {
			row[i] = v
		}
		result.Rows = append(result.Rows, row)
		rowIndex++
	}

	result.Metadata = dto.DataProxyResultMetadata{
		QueryID:             ins.Id(),
		EngineExecutionTime: dto.FormattedDuration(ins.EndTime().Sub(ins.StartTime())),
		ExecutionTime:       dto.FormattedDuration(time.Since(startTime)),
		Engine:              e.Name(),
		RowCount:            int64(len(result.Rows)),
		Vendor:              dataSource.Vendor,
		Region:              region,
	}
	return result, nil
}

func classifyMaxComputeUnknownField(err error, queryID string) error {
	if err == nil {
		return nil
	}
	matches := maxComputeUnknownFieldPattern.FindAllStringSubmatch(err.Error(), -1)
	fields := make([]string, 0, len(matches))
	for _, match := range matches {
		if len(match) == 2 {
			fields = append(fields, match[1])
		}
	}
	return newUnknownFieldError("maxcompute", queryID, fields)
}

// ExecuteStatement 执行 DDL/DML 语句（不返回结果集，用于 CREATE TABLE、MSCK REPAIR 等）
func (e *MaxComputeEngine) ExecuteStatement(ctx context.Context, sql string, region string) error {
	odpsIns, err := e.getOdpsForRegion(region)
	if err != nil {
		return err
	}
	project := odpsIns.DefaultProjectName()
	sqlTask := odps.NewSqlTask("clinic_ddl", ensureTrailingSemicolon(sql), nil)
	ins, err := sqlTask.Run(odpsIns, project)
	if err != nil {
		return fmt.Errorf("execute statement failed: %w", err)
	}
	if err := ins.WaitForSuccess(); err != nil {
		return fmt.Errorf("execute statement failed: %w", err)
	}
	return nil
}

// CheckTableExists 检查表是否存在（project 由 region 对应的 Odps 决定）
func (e *MaxComputeEngine) CheckTableExists(ctx context.Context, project, table, region string) (bool, error) {
	odpsIns, err := e.getOdpsForRegion(region)
	if err != nil {
		return false, err
	}
	proj := odpsIns.DefaultProjectName()
	// DESC table 在表不存在时返回错误
	sql := ensureTrailingSemicolon(fmt.Sprintf("DESC `%s`", table))
	sqlTask := odps.NewSqlTask("check_table", sql, nil)
	ins, err := sqlTask.Run(odpsIns, proj)
	if err != nil {
		if strings.Contains(err.Error(), "not exist") || strings.Contains(err.Error(), "not found") ||
			strings.Contains(err.Error(), "ODPS-0130131") || strings.Contains(err.Error(), "does not exist") {
			return false, nil
		}
		return false, err
	}
	if err := ins.WaitForSuccess(); err != nil {
		if strings.Contains(err.Error(), "not exist") || strings.Contains(err.Error(), "not found") ||
			strings.Contains(err.Error(), "ODPS-0130161") || strings.Contains(err.Error(), "does not exist") {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

// GetTableSchema 获取表结构（使用 ODPS Table API）
func (e *MaxComputeEngine) GetTableSchema(ctx context.Context, project, table, region string) (*dto.SchemaResponse, error) {
	if region == "" {
		return nil, fmt.Errorf("region is required for GetTableSchema")
	}

	odpsIns, err := e.getOdpsForRegion(region)
	if err != nil {
		return nil, err
	}

	t := odps.NewTable(odpsIns, project, "", table)
	if err := t.Load(); err != nil {
		return nil, fmt.Errorf("load table metadata failed: %w", err)
	}

	schema := t.Schema()

	var columns []dto.ColumnInfo
	for _, col := range schema.Columns {
		colType := ""
		if col.Type != nil {
			colType = col.Type.Name()
		}
		columns = append(columns, dto.ColumnInfo{
			Name:    col.Name,
			Type:    colType,
			Comment: col.Comment,
		})
	}

	var partitions []dto.PartitionDef
	for _, col := range schema.PartitionColumns {
		colType := ""
		if col.Type != nil {
			colType = col.Type.Name()
		}
		partitions = append(partitions, dto.PartitionDef{
			Name: col.Name,
			Type: colType,
		})
	}

	return &dto.SchemaResponse{
		Database:   project,
		Table:      table,
		Columns:    columns,
		Partitions: partitions,
		IsView:     schema.IsVirtualView,
		Location:   schema.Location,
	}, nil
}

// Health 健康检查（使用 cn-hongkong 作为默认测试 region）
func (e *MaxComputeEngine) Health(ctx context.Context) error {
	// endpoint 按固定格式推导，任选一个常见 region 做连通性检查
	healthCheckRegion := "cn-hongkong"
	odpsIns, err := e.getOdpsForRegion(healthCheckRegion)
	if err != nil {
		return fmt.Errorf("MaxCompute health check failed for region %s: %w", healthCheckRegion, err)
	}
	sqlTask := odps.NewSqlTask("health_check", ensureTrailingSemicolon("SELECT 1"), nil)
	ins, err := sqlTask.Run(odpsIns, odpsIns.DefaultProjectName())
	if err != nil {
		return fmt.Errorf("MaxCompute health check failed: %w", err)
	}
	if err := ins.WaitForSuccess(); err != nil {
		return fmt.Errorf("MaxCompute health check failed: %w", err)
	}
	return nil
}