package mysql

import (
	"crypto/md5"
	"embed"
	"fmt"
	"io/fs"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/jmoiron/sqlx"
	"github.com/tidbcloud/pov-service/internal/pkg/log"
)

//go:embed migrations/*.sql
var migrationFiles embed.FS

// Constants for migration system
const (
	// ChecksumLength is the expected length of MD5 checksums
	ChecksumLength = 32

	// MigrationTableName is the name of the table that tracks applied migrations
	MigrationTableName = "schema_migrations"

	// MigrationFileExtension is the expected extension for migration files
	MigrationFileExtension = ".sql"

	// MigrationFilenamePattern is the regex pattern for migration filenames
	MigrationFilenamePattern = `^(\d+)_(.+)\.sql$`

	// MigrationLockName is the name of the MySQL advisory lock for migrations
	MigrationLockName = "tidbcloud_migration_lock"

	// DefaultLockTimeoutSeconds is the default timeout for acquiring migration lock
	DefaultLockTimeoutSeconds = 300 // 5 minutes
)

// Migration represents a single database migration
type Migration struct {
	Version   int
	Name      string
	Filename  string
	Content   string
	Checksum  string
	AppliedAt *time.Time
}

// MigrationConfig holds configuration options for migration execution
type MigrationConfig struct {
	// IgnoreChecksumMismatch allows continuing when checksums don't match
	// This should only be used in development or emergency situations
	IgnoreChecksumMismatch bool

	// StrictValidation enables additional validation checks
	StrictValidation bool

	// DryRun shows what migrations would be applied without executing them
	DryRun bool

	// LockTimeoutSeconds is the timeout for acquiring the migration advisory lock
	// Set to 0 to disable locking (not recommended for multi-replica deployments)
	LockTimeoutSeconds int
}

// ChecksumMismatchError represents a checksum validation error
type ChecksumMismatchError struct {
	Version          int
	Name             string
	ExpectedChecksum string
	ActualChecksum   string
}

func (e ChecksumMismatchError) Error() string {
	return fmt.Sprintf("migration %d (%s) checksum mismatch: expected %s, got %s - this indicates the migration file was modified after being applied",
		e.Version, e.Name, e.ExpectedChecksum, e.ActualChecksum)
}

// MigrationManager handles database migrations
type MigrationManager struct {
	db     *sqlx.DB
	logger log.Logger
	config MigrationConfig
}

// NewMigrationManager creates a new migration manager with default configuration
func NewMigrationManager(db *sqlx.DB, logger log.Logger) *MigrationManager {
	return &MigrationManager{
		db:     db,
		logger: logger,
		config: MigrationConfig{
			IgnoreChecksumMismatch: false,
			StrictValidation:       true,
			DryRun:                 false,
			LockTimeoutSeconds:     DefaultLockTimeoutSeconds,
		},
	}
}

// NewMigrationManagerWithConfig creates a new migration manager with custom configuration
func NewMigrationManagerWithConfig(db *sqlx.DB, logger log.Logger, config MigrationConfig) *MigrationManager {
	return &MigrationManager{
		db:     db,
		logger: logger,
		config: config,
	}
}

// RunMigrations executes all pending migrations
// RunMigrations applies pending database migrations.
// It creates the migration tracking table if it doesn't exist,
// loads migrations from embedded files, identifies pending migrations,
// and applies them in version order.
// Uses MySQL advisory locks to coordinate execution across multiple replicas.
func (m *MigrationManager) RunMigrations() error {
	// Acquire advisory lock to prevent concurrent migrations from multiple replicas
	if err := m.acquireMigrationLock(); err != nil {
		return fmt.Errorf("failed to acquire migration lock: %w", err)
	}
	defer func() {
		if err := m.releaseMigrationLock(); err != nil {
			m.logger.Warnf("Failed to release migration lock: %v", err)
		}
	}()

	// Create migration tracking table if it doesn't exist
	if err := m.createMigrationTable(); err != nil {
		return fmt.Errorf("failed to create migration table: %w", err)
	}

	// Load migrations from embedded files
	migrations, err := m.loadMigrations()
	if err != nil {
		return fmt.Errorf("failed to load migrations: %w", err)
	}

	// Get applied migrations from database
	appliedMigrations, err := m.getAppliedMigrations()
	if err != nil {
		return fmt.Errorf("failed to get applied migrations: %w", err)
	}

	// Validate checksums of applied migrations
	if err := m.validateAppliedMigrations(migrations, appliedMigrations); err != nil {
		return fmt.Errorf("checksum validation failed: %w", err)
	}

	// Find pending migrations
	pendingMigrations := m.findPendingMigrations(migrations, appliedMigrations)

	if len(pendingMigrations) == 0 {
		m.logger.Infof("No pending migrations to apply")
		return nil
	}

	m.logger.Infof("Found %d pending migrations to apply", len(pendingMigrations))

	// If dry run, just report what would be done
	if m.config.DryRun {
		m.logger.Infof("DRY RUN: Would apply the following migrations:")
		for _, migration := range pendingMigrations {
			m.logger.Infof("  - Migration %d: %s", migration.Version, migration.Name)
		}
		return nil
	}

	// Apply pending migrations
	for _, migration := range pendingMigrations {
		if err := m.applyMigration(migration); err != nil {
			return fmt.Errorf("failed to apply migration %d (%s): %w", migration.Version, migration.Name, err)
		}
		m.logger.Infof("Applied migration %d: %s", migration.Version, migration.Name)
	}

	m.logger.Infof("All migrations applied successfully")
	return nil
}

// validateAppliedMigrations validates that applied migrations haven't been modified
func (m *MigrationManager) validateAppliedMigrations(allMigrations []Migration, appliedMigrations map[int]Migration) error {
	var validationErrors []error

	// Create a map for quick lookup of current migrations
	currentMigrations := make(map[int]Migration)
	for _, migration := range allMigrations {
		currentMigrations[migration.Version] = migration
	}

	// Check each applied migration
	for version, appliedMigration := range appliedMigrations {
		currentMigration, exists := currentMigrations[version]
		if !exists {
			// Applied migration no longer exists in code
			err := fmt.Errorf("migration %d (%s) was applied but no longer exists in migration files",
				version, appliedMigration.Name)
			validationErrors = append(validationErrors, err)
			continue
		}

		// Validate checksum
		if appliedMigration.Checksum != currentMigration.Checksum {
			checksumErr := ChecksumMismatchError{
				Version:          version,
				Name:             appliedMigration.Name,
				ExpectedChecksum: appliedMigration.Checksum,
				ActualChecksum:   currentMigration.Checksum,
			}
			validationErrors = append(validationErrors, checksumErr)
		}

		// Additional strict validation checks
		if m.config.StrictValidation {
			if appliedMigration.Name != currentMigration.Name {
				err := fmt.Errorf("migration %d name mismatch: applied as '%s', current file shows '%s'",
					version, appliedMigration.Name, currentMigration.Name)
				validationErrors = append(validationErrors, err)
			}

			if appliedMigration.Filename != currentMigration.Filename {
				err := fmt.Errorf("migration %d filename mismatch: applied as '%s', current file is '%s'",
					version, appliedMigration.Filename, currentMigration.Filename)
				validationErrors = append(validationErrors, err)
			}
		}
	}

	// Handle validation errors
	if len(validationErrors) > 0 {
		if m.config.IgnoreChecksumMismatch {
			// Log warnings but don't fail
			m.logger.Warnf("Found %d migration validation issues (ignored due to configuration):", len(validationErrors))
			for _, err := range validationErrors {
				m.logger.Warnf("  - %v", err)
			}
			return nil
		} else {
			// Fail with detailed error information
			var errorMsg strings.Builder
			errorMsg.WriteString(fmt.Sprintf("Found %d migration validation errors:\n", len(validationErrors)))
			for i, err := range validationErrors {
				errorMsg.WriteString(fmt.Sprintf("  %d. %v\n", i+1, err))
			}
			errorMsg.WriteString("\nThis indicates migration files were modified after being applied to the database.\n")
			errorMsg.WriteString("To ignore these errors (NOT RECOMMENDED), set IgnoreChecksumMismatch=true in MigrationConfig.")

			return fmt.Errorf("%s", errorMsg.String())
		}
	}

	return nil
}

// createMigrationTable creates the schema_migrations table for tracking applied migrations
func (m *MigrationManager) createMigrationTable() error {
	query := fmt.Sprintf(`
		CREATE TABLE IF NOT EXISTS %s (
			version INT PRIMARY KEY COMMENT 'Migration version number',
			name VARCHAR(255) NOT NULL COMMENT 'Migration name/description',
			filename VARCHAR(255) NOT NULL COMMENT 'Migration filename',
			checksum VARCHAR(%d) NOT NULL COMMENT 'MD5 checksum of migration content',
			applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 'When the migration was applied',
			
			INDEX idx_applied_at (applied_at)
		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci 
		COMMENT='Tracks applied database migrations'
	`, MigrationTableName, ChecksumLength)

	_, err := m.db.Exec(query)
	return err
}

// loadMigrations loads all migration files from the embedded filesystem
func (m *MigrationManager) loadMigrations() ([]Migration, error) {
	var migrations []Migration

	err := fs.WalkDir(migrationFiles, "migrations", func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}

		if d.IsDir() || !strings.HasSuffix(path, MigrationFileExtension) {
			return nil
		}

		// Parse migration filename
		filename := filepath.Base(path)
		version, name, err := m.parseMigrationFilename(filename)
		if err != nil {
			return fmt.Errorf("invalid migration filename %s: %w", filename, err)
		}

		// Read migration content
		content, err := migrationFiles.ReadFile(path)
		if err != nil {
			return fmt.Errorf("failed to read migration file %s: %w", path, err)
		}

		// Calculate checksum
		checksum := fmt.Sprintf("%x", md5.Sum(content))

		migrations = append(migrations, Migration{
			Version:  version,
			Name:     name,
			Filename: filename,
			Content:  string(content),
			Checksum: checksum,
		})

		return nil
	})

	if err != nil {
		return nil, err
	}

	// Sort migrations by version
	sort.Slice(migrations, func(i, j int) bool {
		return migrations[i].Version < migrations[j].Version
	})

	return migrations, nil
}

// parseMigrationFilename parses migration filename to extract version and name
// Expected format: {version}_{name}.sql (e.g., "001_create_workload_tests.sql")
func (m *MigrationManager) parseMigrationFilename(filename string) (int, string, error) {
	// Validate file extension
	if !strings.HasSuffix(filename, MigrationFileExtension) {
		return 0, "", fmt.Errorf("migration file must have %s extension", MigrationFileExtension)
	}

	// Match pattern: {version}_{name}.sql
	re := regexp.MustCompile(MigrationFilenamePattern)
	matches := re.FindStringSubmatch(filename)
	if len(matches) != 3 {
		return 0, "", fmt.Errorf("filename must match pattern: {version}_{name}%s", MigrationFileExtension)
	}

	version, err := strconv.Atoi(matches[1])
	if err != nil {
		return 0, "", fmt.Errorf("invalid version number: %s", matches[1])
	}

	// Validate version is positive
	if version <= 0 {
		return 0, "", fmt.Errorf("migration version must be positive, got: %d", version)
	}

	name := strings.ReplaceAll(matches[2], "_", " ")

	return version, name, nil
}

// getAppliedMigrations retrieves all applied migrations from the database
func (m *MigrationManager) getAppliedMigrations() (map[int]Migration, error) {
	appliedMigrations := make(map[int]Migration)

	query := fmt.Sprintf(`
		SELECT version, name, filename, checksum, applied_at 
		FROM %s 
		ORDER BY version
	`, MigrationTableName)

	rows, err := m.db.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	for rows.Next() {
		var migration Migration
		err := rows.Scan(
			&migration.Version,
			&migration.Name,
			&migration.Filename,
			&migration.Checksum,
			&migration.AppliedAt,
		)
		if err != nil {
			return nil, err
		}

		appliedMigrations[migration.Version] = migration
	}

	return appliedMigrations, rows.Err()
}

// findPendingMigrations compares available migrations with applied migrations to find pending ones
func (m *MigrationManager) findPendingMigrations(allMigrations []Migration, appliedMigrations map[int]Migration) []Migration {
	var pendingMigrations []Migration

	for _, migration := range allMigrations {
		if _, exists := appliedMigrations[migration.Version]; exists {
			// Migration already applied, skip it
			// Note: Checksum validation is now handled separately in validateAppliedMigrations
			continue
		}

		// Migration not applied yet
		pendingMigrations = append(pendingMigrations, migration)
	}

	return pendingMigrations
}

// applyMigration executes a single migration within a transaction
func (m *MigrationManager) applyMigration(migration Migration) error {
	// Start transaction
	tx, err := m.db.Beginx()
	if err != nil {
		return fmt.Errorf("failed to start transaction: %w", err)
	}
	defer tx.Rollback()

	// Execute migration SQL
	if err := runSQLStatementsInTx(tx, migration.Content); err != nil {
		return fmt.Errorf("failed to execute migration SQL: %w", err)
	}

	// Record migration as applied
	insertQuery := fmt.Sprintf(`
		INSERT INTO %s (version, name, filename, checksum, applied_at)
		VALUES (?, ?, ?, ?, ?)
	`, MigrationTableName)
	_, err = tx.Exec(insertQuery, migration.Version, migration.Name, migration.Filename, migration.Checksum, time.Now())
	if err != nil {
		return fmt.Errorf("failed to record migration: %w", err)
	}

	// Commit transaction
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("failed to commit migration transaction: %w", err)
	}

	return nil
}

// GetMigrationStatus returns the current migration status
func (m *MigrationManager) GetMigrationStatus() ([]Migration, error) {
	// Load all available migrations
	allMigrations, err := m.loadMigrations()
	if err != nil {
		return nil, fmt.Errorf("failed to load migrations: %w", err)
	}

	// Get applied migrations
	appliedMigrations, err := m.getAppliedMigrations()
	if err != nil {
		return nil, fmt.Errorf("failed to get applied migrations: %w", err)
	}

	// Merge information
	var status []Migration
	for _, migration := range allMigrations {
		if applied, exists := appliedMigrations[migration.Version]; exists {
			migration.AppliedAt = applied.AppliedAt
		}
		status = append(status, migration)
	}

	return status, nil
}

// GetConfig returns the current migration configuration
func (m *MigrationManager) GetConfig() MigrationConfig {
	return m.config
}

// acquireMigrationLock acquires a MySQL advisory lock for migration coordination
func (m *MigrationManager) acquireMigrationLock() error {
	if m.config.LockTimeoutSeconds <= 0 {
		// Locking disabled
		return nil
	}

	m.logger.Infof("Acquiring migration lock '%s' with timeout %d seconds",
		MigrationLockName, m.config.LockTimeoutSeconds)

	var lockResult int
	query := "SELECT GET_LOCK(?, ?)"
	err := m.db.Get(&lockResult, query, MigrationLockName, m.config.LockTimeoutSeconds)
	if err != nil {
		return fmt.Errorf("failed to execute GET_LOCK: %w", err)
	}

	switch lockResult {
	case 1:
		m.logger.Infof("Successfully acquired migration lock")
		return nil
	case 0:
		return fmt.Errorf("timeout waiting for migration lock after %d seconds - another instance may be running migrations",
			m.config.LockTimeoutSeconds)
	default:
		return fmt.Errorf("failed to acquire migration lock (result: %d) - check database connection", lockResult)
	}
}

// releaseMigrationLock releases the MySQL advisory lock
func (m *MigrationManager) releaseMigrationLock() error {
	if m.config.LockTimeoutSeconds <= 0 {
		// Locking disabled
		return nil
	}

	m.logger.Infof("Releasing migration lock '%s'", MigrationLockName)

	var releaseResult int
	query := "SELECT RELEASE_LOCK(?)"
	err := m.db.Get(&releaseResult, query, MigrationLockName)
	if err != nil {
		return fmt.Errorf("failed to execute RELEASE_LOCK: %w", err)
	}

	switch releaseResult {
	case 1:
		m.logger.Infof("Successfully released migration lock")
		return nil
	case 0:
		m.logger.Warnf("Lock was not held by this connection")
		return nil
	default:
		return fmt.Errorf("failed to release migration lock (result: %d)", releaseResult)
	}
}