package common

import (
	"crypto/rand"
	"database/sql"
	"fmt"
	"net"
	"strings"
	"time"

	mysql "github.com/go-sql-driver/mysql"
	"github.com/tidbcloud/dataflow-service-e2e/internal/config"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
)

const markerDatabase = "e2e_essential"

func UniqueRunSuffix() string { return fmt.Sprintf("%d", time.Now().UnixNano()) }

func OpenMySQL(user, password, host, port, database string) (*sql.DB, error) {
	cfg := mysql.NewConfig()
	cfg.User, cfg.Passwd, cfg.Net, cfg.Addr, cfg.DBName = user, password, "tcp", net.JoinHostPort(host, port), database
	cfg.TLSConfig = "skip-verify"
	return sql.Open("mysql", cfg.FormatDSN())
}

func EnsureDatabase(user, password, host, port, database string) error {
	db, err := OpenMySQL(user, password, host, port, "")
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return err
	}
	_, err = db.Exec("CREATE DATABASE IF NOT EXISTS " + quoteIdentifier(database))
	return err
}

func WriteMarker(cluster *TiDBServerlessEntity, marker string) (int64, error) {
	if cluster == nil {
		return 0, fmt.Errorf("Essential cluster is nil")
	}
	if err := EnsureDatabase(cluster.User, cluster.Password, cluster.Host, cluster.Port, markerDatabase); err != nil {
		return 0, fmt.Errorf("create marker database: %w", err)
	}
	db, err := OpenMySQL(cluster.User, cluster.Password, cluster.Host, cluster.Port, markerDatabase)
	if err != nil {
		return 0, fmt.Errorf("connect to Essential cluster: %w", err)
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return 0, err
	}
	if _, err := db.Exec("CREATE TABLE IF NOT EXISTS lifecycle_marker (name VARCHAR(255) PRIMARY KEY)"); err != nil {
		return 0, err
	}
	if _, err := db.Exec("INSERT INTO lifecycle_marker(name) VALUES (?) ON DUPLICATE KEY UPDATE name=VALUES(name)", marker); err != nil {
		return 0, err
	}
	var ts int64
	if err := db.QueryRow("SELECT UNIX_TIMESTAMP(UTC_TIMESTAMP())").Scan(&ts); err != nil {
		return 0, err
	}
	var count int
	if err := db.QueryRow("SELECT COUNT(*) FROM lifecycle_marker WHERE name = ?", marker).Scan(&count); err != nil {
		return 0, err
	}
	if count != 1 {
		return 0, fmt.Errorf("marker %q was not written", marker)
	}
	return ts, nil
}

func VerifyRestoreData(cluster *TiDBServerlessEntity, baseline, marker string) error {
	db, err := OpenMySQL(cluster.User, cluster.Password, cluster.Host, cluster.Port, markerDatabase)
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return err
	}
	var baselineCount, markerCount int
	if err := db.QueryRow("SELECT COUNT(*) FROM lifecycle_marker WHERE name = ?", baseline).Scan(&baselineCount); err != nil {
		return err
	}
	if err := db.QueryRow("SELECT COUNT(*) FROM lifecycle_marker WHERE name = ?", marker).Scan(&markerCount); err != nil {
		return err
	}
	if baselineCount != 1 || markerCount != 0 {
		return fmt.Errorf("unexpected restore contents: baseline=%d marker=%d", baselineCount, markerCount)
	}
	return nil
}

func WriteMySQLMarker(source *MySQLEntity, database, marker string) error {
	host, port := splitMySQLEndpoint(source.Endpoint)
	if err := EnsureDatabase(source.User, source.Password, host, port, database); err != nil {
		return err
	}
	db, err := OpenMySQL(source.User, source.Password, host, port, database)
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return err
	}
	if _, err := db.Exec("CREATE TABLE IF NOT EXISTS lifecycle_marker (name VARCHAR(255) PRIMARY KEY)"); err != nil {
		return err
	}
	_, err = db.Exec("INSERT INTO lifecycle_marker(name) VALUES (?) ON DUPLICATE KEY UPDATE name=VALUES(name)", marker)
	return err
}

// PrepareMySQLMarkerTarget creates only the destination schema/table. It does
// not insert the marker, so CDC validators cannot pass by observing a
// pre-existing row on the sink.
func PrepareMySQLMarkerTarget(target *MySQLEntity, database string) error {
	host, port := splitMySQLEndpoint(target.Endpoint)
	if err := EnsureDatabase(target.User, target.Password, host, port, database); err != nil {
		return err
	}
	db, err := OpenMySQL(target.User, target.Password, host, port, database)
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return err
	}
	_, err = db.Exec("CREATE TABLE IF NOT EXISTS lifecycle_marker (name VARCHAR(255) PRIMARY KEY)")
	return err
}

func VerifyMySQLFixtureMarker(target *MySQLEntity, database, marker string) error {
	host, port := splitMySQLEndpoint(target.Endpoint)
	db, err := OpenMySQL(target.User, target.Password, host, port, database)
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return err
	}
	var count int
	if err := db.QueryRow("SELECT COUNT(*) FROM lifecycle_marker WHERE name = ?", marker).Scan(&count); err != nil {
		return err
	}
	if count != 1 {
		return fmt.Errorf("CDC MySQL sink marker %q count=%d", marker, count)
	}
	return nil
}

func VerifyMySQLMarker(cluster *TiDBServerlessEntity, database, marker string) error {
	db, err := OpenMySQL(cluster.User, cluster.Password, cluster.Host, cluster.Port, database)
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return err
	}
	var count int
	if err := db.QueryRow("SELECT COUNT(*) FROM lifecycle_marker WHERE name = ?", marker).Scan(&count); err != nil {
		return err
	}
	if count != 1 {
		return fmt.Errorf("DM target marker %q count=%d", marker, count)
	}
	return nil
}

func VerifyImportedRows(cluster *TiDBServerlessEntity) error {
	return VerifyImportedRowsIn(cluster, "bikeshare", []config.ImportDataTableName{{Table: "trips"}})
}

// VerifyImportedRowsIn asserts the imported contents of the given database:
// every declared table must exist and match its expected row count
// (expectedRows 0 = at least one row, positive = exact count).
func VerifyImportedRowsIn(cluster *TiDBServerlessEntity, database string, tables []config.ImportDataTableName) error {
	db, err := OpenMySQL(cluster.User, cluster.Password, cluster.Host, cluster.Port, database)
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.Ping(); err != nil {
		return err
	}
	for _, table := range tables {
		var rows int64
		if err := db.QueryRow("SELECT COUNT(*) FROM " + quoteIdentifier(table.Table)).Scan(&rows); err != nil {
			return err
		}
		if err := validateImportedRowCount(database, table.Table, rows, table.ExpectedRows); err != nil {
			return err
		}
	}
	return nil
}

// validateImportedRowCount checks a row count against the fixture expectation:
// exact match when expected > 0, at least one row otherwise.
func validateImportedRowCount(database, table string, got, expected int64) error {
	if expected > 0 && got != expected {
		return fmt.Errorf("%s.%s contains %d rows, want %d", database, table, got, expected)
	}
	if got == 0 {
		return fmt.Errorf("%s.%s contains zero rows", database, table)
	}
	return nil
}

func GenerateRestorePassword() (string, error) {
	const alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz23456789"
	b := make([]byte, 24)
	if _, err := rand.Read(b); err != nil {
		return "", err
	}
	for i := range b {
		b[i] = alphabet[int(b[i])%len(alphabet)]
	}
	return string(b), nil
}

func quoteIdentifier(value string) string { return "`" + strings.ReplaceAll(value, "`", "``") + "`" }

func splitMySQLEndpoint(endpoint string) (string, string) {
	if host, port, err := net.SplitHostPort(endpoint); err == nil {
		return host, port
	}
	return endpoint, "4000"
}