package engine

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"net"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
	"github.com/microsoft/go-mssqldb"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

var synapseUnknownFieldPattern = regexp.MustCompile(`(?i)^Invalid column name '([A-Za-z_][A-Za-z0-9_]*)'\.?$`)

const (
	azureSQLScope             = "https://database.windows.net/.default"
	synapseConnectionAttempts = 3
)

// SynapseEngine executes Azure Delta queries through Synapse Serverless SQL.
type SynapseEngine struct {
	server       string
	port         int
	database     string
	authType     string
	username     string
	password     string
	queryTimeout time.Duration

	tokenCredential azcore.TokenCredential
	tokenMu         sync.Mutex
	accessToken     string
	tokenExpires    time.Time

	dbOnce sync.Once
	db     *sql.DB
	dbErr  error
}

// SynapseConfig is the engine-level Azure Synapse configuration.
type SynapseConfig struct {
	Server              string
	Port                int
	Database            string
	AuthType            string
	Username            string
	Password            string
	TenantID            string
	ClientID            string
	ClientSecret        string
	QueryTimeoutSeconds int
	MaxOpenConns        int
	MaxIdleConns        int
	ConnMaxLifetimeMin  int
}

// NewSynapseEngine creates a Synapse query engine.
func NewSynapseEngine(cfg *SynapseConfig) (*SynapseEngine, error) {
	if cfg == nil {
		return nil, fmt.Errorf("synapse config is required")
	}
	port := cfg.Port
	if port == 0 {
		port = 1433
	}
	authType := strings.ToLower(cfg.AuthType)
	if authType == "" {
		authType = "aad_service_principal"
	}

	queryTimeout := time.Duration(cfg.QueryTimeoutSeconds) * time.Second
	if queryTimeout == 0 {
		queryTimeout = 300 * time.Second
	}

	e := &SynapseEngine{
		server:       cfg.Server,
		port:         port,
		database:     cfg.Database,
		authType:     authType,
		username:     cfg.Username,
		password:     cfg.Password,
		queryTimeout: queryTimeout,
	}

	switch authType {
	case "aad_service_principal":
		cred, err := azidentity.NewClientSecretCredential(cfg.TenantID, cfg.ClientID, cfg.ClientSecret, nil)
		if err != nil {
			return nil, fmt.Errorf("create Azure service principal credential failed: %w", err)
		}
		e.tokenCredential = cred
	case "managed_identity":
		cred, err := azidentity.NewManagedIdentityCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("create Azure managed identity credential failed: %w", err)
		}
		e.tokenCredential = cred
	case "sql":
		if cfg.Username == "" || cfg.Password == "" {
			return nil, fmt.Errorf("synapse sql auth requires username and password")
		}
	default:
		return nil, fmt.Errorf("unsupported synapse auth type: %s", cfg.AuthType)
	}

	db, err := e.openDB()
	if err != nil {
		return nil, err
	}
	if cfg.MaxOpenConns > 0 {
		db.SetMaxOpenConns(cfg.MaxOpenConns)
	}
	if cfg.MaxIdleConns > 0 {
		db.SetMaxIdleConns(cfg.MaxIdleConns)
	}
	if cfg.ConnMaxLifetimeMin > 0 {
		db.SetConnMaxLifetime(time.Duration(cfg.ConnMaxLifetimeMin) * time.Minute)
	}
	e.db = db

	return e, nil
}

func (e *SynapseEngine) openDB() (*sql.DB, error) {
	connString := e.connString()

	// SQL username/password auth
	if e.authType == "sql" {
		db, err := sql.Open("sqlserver", connString)
		if err != nil {
			return nil, fmt.Errorf("open synapse sql connection failed: %w", err)
		}

		// Synapse serverless is not suitable for large connection pools.
		// Keep the pool small to avoid stale connections and TLS reset issues.
		db.SetMaxOpenConns(1)
		db.SetMaxIdleConns(0)

		// Short connection lifetime helps avoid server-side recycled connections.
		db.SetConnMaxLifetime(5 * time.Minute)
		db.SetConnMaxIdleTime(1 * time.Minute)

		return db, nil
	}

	tokenCtx, tokenCancel := context.WithTimeout(context.Background(), 30*time.Second)
	if _, err := e.getAccessToken(tokenCtx); err != nil {
		tokenCancel()
		return nil, err
	}
	tokenCancel()

	connector, err := mssql.NewAccessTokenConnector(connString, func() (string, error) {
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		return e.getAccessToken(ctx)
	})
	if err != nil {
		return nil, fmt.Errorf(
			"create synapse access token connector failed: %w",
			err,
		)
	}

	db := sql.OpenDB(connector)

	// Synapse Serverless best practice:
	// avoid large/long-lived pools.
	db.SetMaxOpenConns(3)
	db.SetMaxIdleConns(0)

	// Reduce stale connection probability.
	db.SetConnMaxLifetime(5 * time.Minute)
	db.SetConnMaxIdleTime(1 * time.Minute)

	return db, nil
}

func (e *SynapseEngine) connString() string {
	values := url.Values{}
	values.Set("database", e.database)
	values.Set("encrypt", "true")
	if e.authType == "sql" {
		values.Set("user id", e.username)
		values.Set("password", e.password)
	}

	connURL := url.URL{
		Scheme:   "sqlserver",
		Host:     net.JoinHostPort(e.server, strconv.Itoa(e.port)),
		RawQuery: values.Encode(),
	}
	return connURL.String()
}

func (e *SynapseEngine) getAccessToken(ctx context.Context) (string, error) {
	e.tokenMu.Lock()
	defer e.tokenMu.Unlock()

	if e.accessToken != "" && time.Until(e.tokenExpires) > 5*time.Minute {
		return e.accessToken, nil
	}

	token, err := e.tokenCredential.GetToken(ctx, policy.TokenRequestOptions{
		Scopes: []string{azureSQLScope},
	})
	if err != nil {
		return "", fmt.Errorf("get azure sql access token failed: %w", err)
	}
	e.accessToken = token.Token
	e.tokenExpires = token.ExpiresOn
	return e.accessToken, nil
}

func (e *SynapseEngine) databaseHandle() (*sql.DB, error) {
	if e.db != nil {
		return e.db, nil
	}
	e.dbOnce.Do(func() {
		e.db, e.dbErr = e.openDB()
	})
	return e.db, e.dbErr
}

// Name returns the engine name.
func (e *SynapseEngine) Name() string {
	return "synapse"
}

// GetDatabase returns the configured Synapse database.
func (e *SynapseEngine) GetDatabase(region string) string {
	return e.database
}

// Health checks Synapse connectivity.
func (e *SynapseEngine) Health(ctx context.Context) error {
	db, err := e.databaseHandle()
	if err != nil {
		return err
	}
	checkCtx, cancel := context.WithTimeout(ctx, minDuration(e.queryTimeout, 30*time.Second))
	defer cancel()
	if err := db.PingContext(checkCtx); err != nil {
		return fmt.Errorf("synapse health check failed: %w", err)
	}
	return nil
}

// HasDeltaLog reports whether dataSource points to an initialized Delta table.
// It uses the engine's Azure identity directly against Blob Storage and does not
// execute Synapse SQL, so query execution remains independent from availability
// checks.
func (e *SynapseEngine) HasDeltaLog(ctx context.Context, dataSource *dto.DataSourceInfo) (bool, error) {
	if e.authType == "sql" {
		return false, fmt.Errorf("azure Delta availability check is unsupported for Synapse SQL auth")
	}
	if e.tokenCredential == nil {
		return false, fmt.Errorf("azure token credential is not configured")
	}

	serviceURL, container, rootPrefix, err := azureDeltaBlobTarget(dataSource)
	if err != nil {
		return false, err
	}
	client, err := azblob.NewClient(serviceURL, e.tokenCredential, nil)
	if err != nil {
		return false, err
	}

	deltaLogPrefix := joinAzureBlobPrefix(rootPrefix, "_delta_log") + "/"
	hasDeltaLog, err := azureBlobPrefixHasRealObject(ctx, client, container, deltaLogPrefix)
	if err != nil {
		return false, err
	}
	if hasDeltaLog {
		return true, nil
	}

	rootListPrefix := rootPrefix
	if rootListPrefix != "" {
		rootListPrefix += "/"
	}
	hasRootObject, err := azureBlobPrefixHasRealObject(ctx, client, container, rootListPrefix)
	if err != nil {
		return false, err
	}
	return classifyAzureDeltaAvailability(dataSource.FullURI, false, hasRootObject)
}

func classifyAzureDeltaAvailability(location string, hasDeltaLog, hasRootObject bool) (bool, error) {
	if hasDeltaLog {
		return true, nil
	}
	if !hasRootObject {
		return false, nil
	}
	return false, fmt.Errorf("%w at %s", ErrInvalidDeltaSource, location)
}

func azureDeltaBlobTarget(dataSource *dto.DataSourceInfo) (serviceURL, container, prefix string, err error) {
	rootURI, err := ResolveAzureDeltaRootURI(dataSource)
	if err != nil {
		return "", "", "", err
	}
	resolved, err := dto.ParseDataSource(rootURI)
	if err != nil {
		return "", "", "", err
	}
	if resolved.Account == "" || resolved.Container == "" {
		return "", "", "", fmt.Errorf("azure Delta data source requires storage account and container")
	}

	return fmt.Sprintf("https://%s.blob.core.windows.net/", resolved.Account),
		resolved.Container,
		strings.Trim(resolved.Path, "/"),
		nil
}

func azureBlobPrefixHasRealObject(
	ctx context.Context,
	client *azblob.Client,
	container string,
	prefix string,
) (bool, error) {
	maxResults := int32(16)
	pager := client.NewListBlobsFlatPager(container, &azblob.ListBlobsFlatOptions{
		Prefix:     &prefix,
		MaxResults: &maxResults,
	})
	for pager.More() {
		page, err := pager.NextPage(ctx)
		if err != nil {
			return false, err
		}
		names := make([]string, 0, len(page.Segment.BlobItems))
		for _, item := range page.Segment.BlobItems {
			if item != nil && item.Name != nil {
				names = append(names, *item.Name)
			}
		}
		if azureBlobNamesContainRealObject(prefix, names) {
			return true, nil
		}
	}
	return false, nil
}

func azureBlobNamesContainRealObject(listPrefix string, names []string) bool {
	for _, name := range names {
		// Flat listings may include directory marker blobs. A marker either has
		// exactly the listed prefix as its name or ends with a slash.
		if name == listPrefix || strings.HasSuffix(name, "/") {
			continue
		}
		return true
	}
	return false
}

func joinAzureBlobPrefix(parent, child string) string {
	parent = strings.Trim(parent, "/")
	child = strings.Trim(child, "/")
	if parent == "" {
		return child
	}
	if child == "" {
		return parent
	}
	return parent + "/" + child
}

// Execute runs a Synapse SQL query and returns Data Proxy's unified response.
func (e *SynapseEngine) Execute(ctx context.Context, sqlText string, dataSource *dto.DataSourceInfo) (*dto.DataProxyQueryResult, error) {
	start := time.Now()
	db, err := e.databaseHandle()
	if err != nil {
		return nil, err
	}

	queryCtx := ctx
	var cancel context.CancelFunc
	if _, ok := ctx.Deadline(); !ok && e.queryTimeout > 0 {
		queryCtx, cancel = context.WithTimeout(ctx, e.queryTimeout)
		defer cancel()
	}

	var rows *sql.Rows
	var queryErr error
	for attempt := 1; attempt <= synapseConnectionAttempts; attempt++ {
		pingCtx, pingCancel := context.WithTimeout(queryCtx, minDuration(e.queryTimeout, 30*time.Second))
		pingErr := db.PingContext(pingCtx)
		pingCancel()
		if pingErr != nil {
			queryErr = fmt.Errorf("synapse ping failed before query: %w", pingErr)
			if !isSynapseTransientConnectionError(pingErr) || attempt == synapseConnectionAttempts {
				return nil, queryErr
			}
			if waitErr := sleepWithContext(queryCtx, time.Duration(attempt)*time.Second); waitErr != nil {
				return nil, queryErr
			}
			continue
		}

		rows, queryErr = db.QueryContext(queryCtx, sqlText)
		if queryErr == nil {
			break
		}
		if unknownErr := classifySynapseUnknownField(queryErr); unknownErr != nil {
			return nil, unknownErr
		}
		queryErr = fmt.Errorf("synapse query failed: %w", queryErr)
		if !isSynapseTransientConnectionError(queryErr) || attempt == synapseConnectionAttempts {
			return nil, queryErr
		}
		if waitErr := sleepWithContext(queryCtx, time.Duration(attempt)*time.Second); waitErr != nil {
			return nil, queryErr
		}
	}
	if rows == nil {
		return nil, queryErr
	}
	defer rows.Close()

	cols, err := rows.Columns()
	if err != nil {
		return nil, fmt.Errorf("read synapse result columns failed: %w", err)
	}

	result := &dto.DataProxyQueryResult{
		Columns: cols,
		Rows:    make([][]interface{}, 0),
		Metadata: dto.DataProxyResultMetadata{
			QueryID: fmt.Sprintf("synapse-%d", time.Now().UnixNano()),
			Engine:  e.Name(),
			Vendor:  "azure",
		},
	}
	if dataSource != nil {
		result.Metadata.Region = dataSource.Region
	}

	for rows.Next() {
		values := make([]interface{}, len(cols))
		scanTargets := make([]interface{}, len(cols))
		for i := range values {
			scanTargets[i] = &values[i]
		}
		if err := rows.Scan(scanTargets...); err != nil {
			return nil, fmt.Errorf("scan synapse row failed: %w", err)
		}
		for i, v := range values {
			if bytes, ok := v.([]byte); ok {
				values[i] = string(bytes)
			}
		}
		result.Rows = append(result.Rows, values)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate synapse results failed: %w", err)
	}

	elapsed := time.Since(start)
	result.Metadata.RowCount = int64(len(result.Rows))
	result.Metadata.ExecutionTime = dto.FormattedDuration(elapsed)
	result.Metadata.EngineExecutionTime = dto.FormattedDuration(elapsed)
	return result, nil
}

func classifySynapseUnknownField(err error) error {
	var sqlErr mssql.Error
	if !errors.As(err, &sqlErr) {
		return nil
	}
	items := sqlErr.All
	if len(items) == 0 {
		items = []mssql.Error{sqlErr}
	}
	var fields []string
	for _, item := range items {
		if item.Number != 207 {
			continue
		}
		match := synapseUnknownFieldPattern.FindStringSubmatch(strings.TrimSpace(item.Message))
		if len(match) == 2 {
			fields = append(fields, match[1])
		}
	}
	return newUnknownFieldError("synapse", "", fields)
}

// GetTableSchema is not used for Azure v1 because schema comes from the standard registry.
func (e *SynapseEngine) GetTableSchema(ctx context.Context, database, table, region string) (*dto.SchemaResponse, error) {
	return nil, fmt.Errorf("synapse live table schema is not supported; use standard schema registry")
}

func minDuration(a, b time.Duration) time.Duration {
	if a <= 0 {
		return b
	}
	if a < b {
		return a
	}
	return b
}

func isSynapseTransientConnectionError(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "tls handshake failed") ||
		strings.Contains(msg, "connection reset by peer") ||
		strings.Contains(msg, "operation timed out") ||
		strings.Contains(msg, "i/o timeout")
}

func sleepWithContext(ctx context.Context, d time.Duration) error {
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}