package streaming

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"database/sql"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"sync/atomic"

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/tidbcloud/service-proxy/pkg/checker/core"
)

var registerDefaultTLSOnce sync.Once

func init() {
	core.DefaultRegistry.Register("mysql", func(req *core.CheckRequest) (core.Checker, error) {
		if req.MySQL == nil {
			return nil, errors.New("mysql params required")
		}
		return NewMySQLChecker(req.MySQL), nil
	})

	// Register default TLS config to match clusterproxy behavior (only once)
	registerDefaultTLSOnce.Do(func() {
		tlsConf := &tls.Config{MinVersion: tls.VersionTLS12}
		if err := mysql.RegisterTLSConfig(mysqlTLSConfigName, tlsConf); err != nil {
			log.Error("failed to register default mysql TLS config", zap.Error(err))
		}
	})
}

var tlsConfigCounter int64

const (
	// keep a single, well-known TLS config name to match the "original" way
	mysqlTLSConfigName = "custom"
)

// MySQLChecker checks MySQL/TiDB connectivity.
type MySQLChecker struct {
	params *core.MySQLParams
}

// NewMySQLChecker creates a new MySQL checker.
func NewMySQLChecker(params *core.MySQLParams) *MySQLChecker {
	return &MySQLChecker{params: params}
}

// Type returns the checker type.
func (c *MySQLChecker) Type() string {
	return "mysql"
}

// Check performs MySQL connectivity check.
func (c *MySQLChecker) Check(ctx context.Context) error {
	if c.params.User == "" || c.params.Password == "" || c.params.Host == "" {
		return fmt.Errorf("user, password and host are required fields")
	}
	db, cleanup, err := c.openMySQLDB()
	if err != nil {
		return err
	}
	defer cleanup()

	if err := db.PingContext(ctx); err != nil {
		return fmt.Errorf("unable to connect to mysql: %w", err)
	}
	return nil
}

func (c *MySQLChecker) openMySQLDB() (*sql.DB, func(), error) {
	if c.params.User == "" || c.params.Password == "" || c.params.Host == "" || c.params.Port <= 0 {
		return nil, func() {}, fmt.Errorf("user, password, host, and port are required fields")
	}

	dsn := fmt.Sprintf("%s:%s@(%s:%d)/", c.params.User, c.params.Password, c.params.Host, c.params.Port)

	tlsCleanup := func() {}
	// Configure TLS if enabled
	if c.params.TLS != nil {
		// Create custom TLS config with minimum TLS 1.2
		tlsConf := &tls.Config{MinVersion: tls.VersionTLS12}
		hasCA := len(c.params.TLS.CABytes) > 0
		hasCert := len(c.params.TLS.CertBytes) > 0
		hasKey := len(c.params.TLS.KeyBytes) > 0
		if hasCA || hasCert || hasKey {
			// Add root CA if provided
			if hasCA {
				roots := x509.NewCertPool()
				if !roots.AppendCertsFromPEM(c.params.TLS.CABytes) {
					return nil, func() {}, fmt.Errorf("failed to parse CA bytes")
				}
				tlsConf.RootCAs = roots
			}
			// Add client certificate if provided
			if hasCert || hasKey {
				if hasCert != hasKey {
					return nil, func() {}, fmt.Errorf("both ssl cert and ssl key must be provided together")
				}
				cert, err := tls.X509KeyPair(c.params.TLS.CertBytes, c.params.TLS.KeyBytes)
				if err != nil {
					return nil, func() {}, fmt.Errorf("failed to parse ssl cert/key: %w", err)
				}
				tlsConf.Certificates = []tls.Certificate{cert}
			}

			// Register unique TLS config
			tlsName := mysqlTLSConfigName + strconv.FormatInt(atomic.AddInt64(&tlsConfigCounter, 1), 10)
			if err := mysql.RegisterTLSConfig(tlsName, tlsConf); err != nil {
				return nil, func() {}, fmt.Errorf("unable to register tls config: %w", err)
			}
			dsn += "?tls=" + tlsName
			tlsCleanup = func() { mysql.DeregisterTLSConfig(tlsName) }
		} else {
			// Use shared default TLS config
			dsn += "?tls=" + mysqlTLSConfigName
		}
	}

	// Open database connection
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		tlsCleanup()
		return nil, func() {}, fmt.Errorf("unable to open mysql connection: %w", err)
	}

	// Return DB with cleanup function
	cleanup := func() {
		_ = db.Close()
		tlsCleanup()
	}
	return db, cleanup, nil
}