package engine

import (
	"fmt"
	"regexp"
	"sort"
	"strings"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dialect"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

const (
	azureCTEPrefix = "az_"
)

var (
	azureDateInRE = regexp.MustCompile("(?is)(?:\\b[a-zA-Z_][a-zA-Z0-9_]*\\s*\\.\\s*)?[\"`\\[]?date[\"`\\]]?\\s+in\\s*\\(([^)]*)\\)")
	azureDateEqRE = regexp.MustCompile("(?is)(?:\\b[a-zA-Z_][a-zA-Z0-9_]*\\s*\\.\\s*)?[\"`\\[]?date[\"`\\]]?\\s*=\\s*'([^']+)'")
	quotedValueRE = regexp.MustCompile(`'([^']+)'`)
)

// BuildAzureSynapseSQL rewrites a logical Data Proxy SQL query to Synapse OPENROWSET SQL.
func BuildAzureSynapseSQL(sqlText, clusterID string, dataSource *dto.DataSourceInfo) (string, error) {
	if strings.TrimSpace(sqlText) == "" {
		return "", fmt.Errorf("sql is required")
	}
	if clusterID == "" {
		return "", fmt.Errorf("clusterId is required and cannot be empty")
	}
	if dataSource == nil {
		return "", fmt.Errorf("azure data source is required")
	}

	dates := ExtractAzureDateValues(sqlText)
	if len(dates) == 0 {
		return "", fmt.Errorf("SQL validation failed: missing Azure date partition filter. Please add WHERE date IN ('...')")
	}

	tableRefs := ExtractTableReferences(sqlText)
	if len(tableRefs) == 0 {
		return "", fmt.Errorf("SQL validation failed: no standard table reference found")
	}

	requestedTables := make(map[string]struct{})
	for _, ref := range tableRefs {
		table := strings.ToLower(strings.TrimSpace(ref.Table))
		if !IsStandardTable(table) {
			return "", fmt.Errorf("%s", UnsupportedStandardTableError(fmt.Sprintf("%s.%s", ref.Database, ref.Table)))
		}
		requestedTables[table] = struct{}{}
	}

	rootURI, err := ResolveAzureDeltaRootURI(dataSource)
	if err != nil {
		return "", err
	}

	neededPhysical := expandAzurePhysicalTables(requestedTables)
	tableNameMap := make(map[string]string, len(requestedTables))
	for table := range requestedTables {
		tableNameMap[table] = azureCTEName(table)
	}

	rewrittenUserSQL, err := dialect.RewriteForEngine(sqlText, "synapse", &dialect.RewriteOptions{
		TableNameMap: tableNameMap,
		DatabaseNameMap: map[string]string{
			"clinic_data_proxy": "",
			"clinic-data-proxy": "",
		},
	})
	if err != nil {
		return "", fmt.Errorf("Azure SQL rewrite failed: %w", err)
	}

	cteDefs := buildAzureCTEDefs(rootURI, neededPhysical, requestedTables, clusterID)
	return injectAzureCTEs(rewrittenUserSQL, cteDefs), nil
}

// ExtractAzureDateValues extracts literal date filters supported by Azure v1.
func ExtractAzureDateValues(sqlText string) []string {
	seen := make(map[string]struct{})
	var dates []string

	for _, match := range azureDateInRE.FindAllStringSubmatch(sqlText, -1) {
		if len(match) < 2 {
			continue
		}
		for _, valueMatch := range quotedValueRE.FindAllStringSubmatch(match[1], -1) {
			if len(valueMatch) < 2 {
				continue
			}
			value := strings.TrimSpace(valueMatch[1])
			if value == "" {
				continue
			}
			if _, ok := seen[value]; !ok {
				seen[value] = struct{}{}
				dates = append(dates, value)
			}
		}
	}

	for _, match := range azureDateEqRE.FindAllStringSubmatch(sqlText, -1) {
		if len(match) < 2 {
			continue
		}
		value := strings.TrimSpace(match[1])
		if value == "" {
			continue
		}
		if _, ok := seen[value]; !ok {
			seen[value] = struct{}{}
			dates = append(dates, value)
		}
	}

	sort.Strings(dates)
	return dates
}

// ResolveAzureDeltaRootURI returns an HTTPS dfs.core.windows.net URI for Synapse OPENROWSET.
func ResolveAzureDeltaRootURI(info *dto.DataSourceInfo) (string, error) {
	if info == nil {
		return "", fmt.Errorf("azure data source is required")
	}
	if info.CanonicalURI != "" {
		return ensureTrailingSlash(info.CanonicalURI), nil
	}

	account := info.Account
	container := info.Container
	if container == "" {
		container = info.Bucket
	}
	region := strings.ToLower(strings.TrimSpace(info.Region))

	if region != "" {
		if account == "" {
			account = defaultAzureStorageAccount(region)
		}
		if container == "" {
			container = defaultAzureStorageContainer(region)
		}
	}

	if account == "" || container == "" {
		return "", fmt.Errorf("azure data source requires storage account and container; got account=%q container=%q", account, container)
	}
	return ensureTrailingSlash(fmt.Sprintf("https://%s.dfs.core.windows.net/%s/%s", account, container, strings.TrimLeft(info.Path, "/"))), nil
}

func defaultAzureStorageAccount(region string) string {
	region = strings.ToLower(strings.TrimSpace(region))
	if region == "" {
		return ""
	}
	return "o11yprod" + region
}

func defaultAzureStorageContainer(region string) string {
	region = strings.ToLower(strings.TrimSpace(region))
	if region == "" {
		return ""
	}
	return "o11y-prod-shared-" + region
}

func expandAzurePhysicalTables(requested map[string]struct{}) map[string]struct{} {
	physical := make(map[string]struct{})
	for table := range requested {
		switch table {
		case "topsql":
			physical["topsql_tidb"] = struct{}{}
			physical["topsql_tikv"] = struct{}{}
			physical["topsql_sql_meta"] = struct{}{}
			physical["topsql_plan_meta"] = struct{}{}
		case "topru":
			physical["topsql_topru"] = struct{}{}
			physical["topsql_sql_meta"] = struct{}{}
			physical["topsql_plan_meta"] = struct{}{}
		default:
			if isAzurePhysicalTable(table) {
				physical[table] = struct{}{}
			}
		}
	}
	return physical
}

func buildAzureCTEDefs(rootURI string, physical map[string]struct{}, requested map[string]struct{}, clusterID string) []string {
	order := []string{
		"slow_query_logs",
		"statements",
		"topsql_tidb",
		"topsql_tikv",
		"topsql_sql_meta",
		"topsql_plan_meta",
		"topsql_topru",
	}

	var defs []string
	for _, table := range order {
		if _, ok := physical[table]; !ok {
			continue
		}
		tableRoot := rootURI
		if strings.HasPrefix(table, "topsql_") {
			tableRoot = ResolveTopSQLDataSource(rootURI, fmt.Sprintf("%s_%s", table, clusterID))
		}
		defs = append(defs, azureOpenRowsetCTE(azureCTEName(table), tableRoot))
	}

	if _, ok := requested["topsql"]; ok {
		defs = append(defs, azureTopSQLViewCTE())
	}
	if _, ok := requested["topru"]; ok {
		defs = append(defs, azureTopRUViewCTE())
	}
	return defs
}

func azureOpenRowsetCTE(name, uri string) string {
	return fmt.Sprintf("%s AS (\n  SELECT *\n  FROM OPENROWSET(\n    BULK '%s',\n    FORMAT = 'DELTA'\n  ) AS az_rows\n)", name, escapeSQLString(ensureTrailingSlash(uri)))
}

func azureTopSQLViewCTE() string {
	return `az_topsql AS (
  SELECT
    t."db",
    t."keyspace",
    t."plan_digest",
    t."region_id",
    t."sql_digest",
    t."table",
    t."table_id",
    t."tag_label",
    t."timestamps",
    t."topsql_cpu_time_ms",
    t."topsql_logical_read_bytes",
    t."topsql_logical_write_bytes",
    t."topsql_network_in_bytes",
    t."topsql_network_out_bytes",
    t."topsql_read_keys",
    t."topsql_stmt_duration_count",
    t."topsql_stmt_duration_sum_ns",
    t."topsql_stmt_exec_count",
    t."topsql_write_keys",
    t."instance",
    t."date",
    'tidb' AS "component",
    sm."normalized_sql",
    pm."normalized_plan",
    pm."encoded_normalized_plan"
  FROM az_topsql_tidb AS t
  LEFT JOIN az_topsql_sql_meta AS sm
    ON t."sql_digest" = sm."sql_digest" AND t."date" = sm."date"
  LEFT JOIN az_topsql_plan_meta AS pm
    ON t."plan_digest" = pm."plan_digest" AND t."date" = pm."date"
  UNION ALL
  SELECT
    t."db",
    t."keyspace",
    t."plan_digest",
    t."region_id",
    t."sql_digest",
    t."table",
    t."table_id",
    t."tag_label",
    t."timestamps",
    t."topsql_cpu_time_ms",
    t."topsql_logical_read_bytes",
    t."topsql_logical_write_bytes",
    t."topsql_network_in_bytes",
    t."topsql_network_out_bytes",
    t."topsql_read_keys",
    t."topsql_stmt_duration_count",
    t."topsql_stmt_duration_sum_ns",
    t."topsql_stmt_exec_count",
    t."topsql_write_keys",
    t."instance",
    t."date",
    'tikv' AS "component",
    sm."normalized_sql",
    pm."normalized_plan",
    pm."encoded_normalized_plan"
  FROM az_topsql_tikv AS t
  LEFT JOIN az_topsql_sql_meta AS sm
    ON t."sql_digest" = sm."sql_digest" AND t."date" = sm."date"
  LEFT JOIN az_topsql_plan_meta AS pm
    ON t."plan_digest" = pm."plan_digest" AND t."date" = pm."date"
)`
}

func azureTopRUViewCTE() string {
	return `az_topru AS (
  SELECT
    t."keyspace",
    t."plan_digest",
    t."sql_digest",
    t."timestamps",
    t."topru_exec_count",
    t."topru_exec_duration",
    t."topru_total_ru",
    t."user",
    t."instance",
    t."date",
    sm."normalized_sql",
    pm."normalized_plan",
    pm."encoded_normalized_plan"
  FROM az_topsql_topru AS t
  LEFT JOIN az_topsql_sql_meta AS sm
    ON t."sql_digest" = sm."sql_digest" AND t."date" = sm."date"
  LEFT JOIN az_topsql_plan_meta AS pm
    ON t."plan_digest" = pm."plan_digest" AND t."date" = pm."date"
)`
}

func injectAzureCTEs(sqlText string, cteDefs []string) string {
	if len(cteDefs) == 0 {
		return sqlText
	}
	defBlock := strings.Join(cteDefs, ",\n")
	trimmed := strings.TrimSpace(sqlText)
	if strings.HasPrefix(strings.ToLower(trimmed), "with ") {
		return "with " + defBlock + ",\n" + strings.TrimSpace(trimmed[len("with "):])
	}
	return "with " + defBlock + "\n" + trimmed
}

func azureCTEName(table string) string {
	return azureCTEPrefix + strings.ToLower(table)
}

func isAzurePhysicalTable(table string) bool {
	switch table {
	case "slow_query_logs", "statements", "topsql_tidb", "topsql_tikv", "topsql_plan_meta", "topsql_sql_meta", "topsql_topru":
		return true
	default:
		return false
	}
}

func ensureTrailingSlash(uri string) string {
	if strings.HasSuffix(uri, "/") {
		return uri
	}
	return uri + "/"
}

func escapeSQLString(value string) string {
	return strings.ReplaceAll(value, "'", "''")
}