package central

import (
	"context"
	"crypto/tls"
	"database/sql"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"crypto/x509"
	"strconv"
	"sync/atomic"

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

	"github.com/tidbcloud/clusterproxy/resolve"
	"github.com/tidbcloud/clusterproxy/route"
	"github.com/tidbcloud/clusterproxy/route/router"
)

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

var tlsConfigCounter int64

var ExcludeDatabases = []string{
	"__TiDB_BR_Temporary_mysql",
	"tidbcloud_dm_meta",
	"tidb_cdc",
	"lightning_task_info",
	"INFORMATION_SCHEMA",
	"mysql",
	"PERFORMANCE_SCHEMA",
	"METRICS_SCHEMA",
	"sys",
}

func isExcludedDatabase(name string) bool {
	lname := strings.ToLower(name)
	for _, ex := range ExcludeDatabases {
		if strings.ToLower(ex) == lname {
			return true
		}
	}
	return false
}

// escapeIdentifier safely quotes a MySQL identifier (database/table/schema name)
// by wrapping it in backticks and doubling any backticks inside the identifier.
// This prevents SQL injection when identifiers must be interpolated into
// statements (placeholders cannot be used for identifiers in MySQL).
func escapeIdentifier(name string) string {
	return "`" + strings.ReplaceAll(name, "`", "``") + "`"
}

// fetchTablesFromDB queries the given *sql.DB for tables in the provided
// database name. It escapes the identifier safely and ensures the rows are
// closed before returning. Returns a slice of table names or an error.
func fetchTablesFromDB(ctx context.Context, db *sql.DB, dbName string) ([]string, error) {
	query := fmt.Sprintf("SHOW TABLES FROM %s", escapeIdentifier(dbName))
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		log.Error("show tables query failed, skipping database", zap.String("db", dbName), zap.String("query", query), zap.Error(err))
		return nil, err
	}
	// Close immediately when the function returns
	defer rows.Close()

	var tables []string
	for rows.Next() {
		var tbl string
		if err := rows.Scan(&tbl); err != nil {
			log.Error("failed to scan table row", zap.String("db", dbName), zap.Error(err))
			return nil, fmt.Errorf("scan table row failed: %v", err)
		}
		tables = append(tables, tbl)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("rows iteration error: %v", err)
	}
	return tables, nil
}

func checkMySQLAccess(ctx context.Context, model *connectionModel) error {
	if model.User == "" || model.Password == "" || model.Url == "" {
		return fmt.Errorf("user, password and url are required field")
	}
	db, cleanup, err := openMySQLDB(model)
	if err != nil {
		return err
	}
	defer cleanup()

	if err := db.PingContext(ctx); err != nil {
		return fmt.Errorf("unable to connect sql url %s error %v", model.Url, err)
	}
	return nil
}

// openMySQLDB opens a *sql.DB using connection info in model.
// It registers a TLS config if needed and returns a cleanup func
// which must be called to close the DB and cleanup per-connection TLS.
func openMySQLDB(model *connectionModel) (*sql.DB, func(), error) {
	// Validate required fields
	if model.User == "" || model.Password == "" || model.Url == "" {
		return nil, func() {}, fmt.Errorf("user, password and url are required fields")
	}

	dsn := fmt.Sprintf("%s:%s@(%s)/", model.User, model.Password, model.Url)

	tlsCleanup := func() {}
	// Configure TLS if enabled
	if model.EnableTls {
		// Create custom TLS config with minimum TLS 1.2
		tlsConf := &tls.Config{MinVersion: tls.VersionTLS12}
		hasCA := len(model.MysqlParam.SSLCABytes) > 0
		hasCert := len(model.MysqlParam.SSLCertBytes) > 0
		hasKey := len(model.MysqlParam.SSLKeyBytes) > 0
		if hasCA || hasCert || hasKey {
			// Add root CA if provided
			if hasCA {
				roots := x509.NewCertPool()
				if !roots.AppendCertsFromPEM(model.MysqlParam.SSLCABytes) {
					return nil, func() {}, fmt.Errorf("failed to parse ssl 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(model.MysqlParam.SSLCertBytes, model.MysqlParam.SSLKeyBytes)
				if err != nil {
					return nil, func() {}, fmt.Errorf("failed to parse ssl cert/key: %v", 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: %v", 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 sql connection url %s error %v", model.Url, err)
	}

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

// getAllTables returns all non-system tables in the target MySQL/TiDB instance
// as fully-qualified names (schema.table). It supports the same TLS-bytes
// behavior as checkMySQLAccess by registering a temporary, unique TLS config
// when `model.EnableTls` is true and `model.MysqlParam` contains TLS material.
type tableInfo struct {
	Schema string `json:"schema"`
	Table  string `json:"table"`
}

func getAllTables(ctx context.Context, model *connectionModel) ([]tableInfo, error) {
	db, cleanup, err := openMySQLDB(model)
	if err != nil {
		return nil, err
	}
	defer cleanup()

	// 1) SHOW DATABASES
	rows, err := db.QueryContext(ctx, "SHOW DATABASES")
	if err != nil {
		return nil, fmt.Errorf("show databases failed: %v", err)
	}
	defer rows.Close()

	var databases []string
	for rows.Next() {
		var dbName string
		if err := rows.Scan(&dbName); err != nil {
			return nil, fmt.Errorf("scan database row failed: %v", err)
		}
		// skip system schemas and configured excludes
		if isExcludedDatabase(dbName) {
			continue
		}
		databases = append(databases, dbName)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("rows iteration error: %v", err)
	}
	log.Info("found databases", zap.Int("count", len(databases)), zap.Any("databases", databases))

	// 2) For each database run SHOW TABLES. We extract the per-db fetching into
	// fetchTablesFromDB so we can close rows immediately without using defer
	// inside the loop.
	var results []tableInfo
	for _, dbName := range databases {
		tbls, err := fetchTablesFromDB(ctx, db, dbName)
		if err != nil {
			log.Warn("failed to fetch tables from db", zap.String("db_name", dbName), zap.Error(err))
			continue
		}
		for _, t := range tbls {
			results = append(results, tableInfo{Schema: dbName, Table: t})
		}
	}

	return results, nil
}

func init() {
	router.RegisterRoute(getSourceTablesRoute)
	tlsConf := &tls.Config{MinVersion: tls.VersionTLS12}
	err := mysql.RegisterTLSConfig(mysqlTLSConfigName, tlsConf)
	if err != nil {
		log.Error("failed to register default mysql TLS config", zap.Error(err))
	}
}

var getSourceTablesRoute = &route.Route{
	Name:    "get-source-tables",
	Path:    "get_source_tables",
	Version: route.V1,
	Schema: &route.Schema{
		PathRegex:       "^$",
		RequiredHeaders: []string{},
		JSONBodyModel:   &connectionModel{},
	},
	GetHandler: getSourceTablesHandler,
}

func getSourceTablesHandler(c *route.Context, cl resolve.Cluster) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		defer r.Body.Close()
		if err != nil {
			http.Error(w, "failed to read body contents", http.StatusInternalServerError)
			return
		}

		model := &connectionModel{}
		if err := json.Unmarshal(body, model); err != nil {
			http.Error(w, fmt.Sprintf("failed to parse the body contents: %v", err), http.StatusBadRequest)
			return
		}

		// Use a short timeout for DB enumeration
		ctx, cancel := context.WithTimeout(r.Context(), 45*time.Second)
		defer cancel()

		tables, err := getAllTables(ctx, model)
		if err != nil {
			log.Warn("get source tables failed", zap.Error(err), zap.Any("connectionModel", model.ToLoggingString()))
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}

		resp, err := json.Marshal(tables)
		if err != nil {
			http.Error(w, fmt.Sprintf("failed to marshal response: %v", err), http.StatusInternalServerError)
			return
		}

		log.Info("get source tables response", zap.Int("table_count", len(tables)))
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write(resp)
	})
}