package swmigration

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"time"

	"gorm.io/gorm"
	scripts "kra/deploy/migrations"
	"kra/pkg/database/migration"
	"kra/pkg/module"
)

// Recognized identifies an old SW database or a previous automatic upgrade.
// Database names are not used to infer ownership of unrelated schemas.
func Recognized(db *gorm.DB) (bool, error) {
	if db.Dialector.Name() != "mysql" {
		return false, nil
	}
	tables, err := tableSet(db)
	if err != nil {
		return false, err
	}
	if tables["sw_kra_state"] || tables["sys_user"] && tables["tab_sw_report_day"] && tables["tab_sw_report_server_day"] {
		return true, nil
	}
	if !tables[TableName] {
		return false, nil
	}
	// The state table also exists in new template installations. Only a
	// recorded legacy workflow identifies a previously upgraded SW database.
	var count int64
	if err := db.Table(TableName).Where("name=?", workflowName).Count(&count).Error; err != nil {
		return false, err
	}
	return count > 0, nil
}

func tableSet(db *gorm.DB) (map[string]bool, error) {
	var names []string
	if err := db.Raw("SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() AND table_type='BASE TABLE'").Scan(&names).Error; err != nil {
		return nil, err
	}
	set := map[string]bool{}
	for _, name := range names {
		set[name] = true
	}
	return set, nil
}

func Run(ctx context.Context, db *gorm.DB, catalog module.Catalog, backupRoot string, logger *slog.Logger) (runErr error) {
	if logger == nil {
		logger = slog.Default()
	}
	db = db.Session(&gorm.Session{NewDB: true}).WithContext(ctx)
	pool, err := db.DB()
	if err != nil {
		return err
	}
	conn, err := pool.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()
	var schema, serverUUID string
	if err = conn.QueryRowContext(ctx, "SELECT DATABASE(),@@server_uuid").Scan(&schema, &serverUUID); err != nil {
		return err
	}
	if schema == "" {
		return fmt.Errorf("automatic upgrade requires a selected database")
	}
	var lockName string
	if err = conn.QueryRowContext(ctx, "SELECT CONCAT('sw_to_kra:',MD5(DATABASE()))").Scan(&lockName); err != nil {
		return err
	}
	for {
		var locked sql.NullInt64
		if err = conn.QueryRowContext(ctx, "SELECT GET_LOCK(?,10)", lockName).Scan(&locked); err != nil {
			return err
		}
		if locked.Valid && locked.Int64 == 1 {
			break
		}
		if !locked.Valid {
			return fmt.Errorf("database migration lock is unavailable")
		}
		logger.Info("waiting for another database upgrade", "database", schema)
	}
	defer func() {
		clean, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
		defer cancel()
		_, _ = conn.ExecContext(clean, "DO RELEASE_LOCK(?)", lockName)
	}()
	if err = StateMigration().Migrate(db); err != nil {
		return err
	}
	state := statePO{Name: workflowName}
	err = db.First(&state, "name=?", workflowName).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		state = statePO{Name: workflowName, Version: workflowVersion, Phase: "backup", DatabaseIdentity: serverUUID + ":" + schema, StartedAt: time.Now(), UpdatedAt: time.Now()}
		initial, readErr := tableSet(db)
		if readErr != nil {
			return readErr
		}
		state.SourceWasLegacy = initial["sys_user"]
		if initial["sw_kra_state"] {
			var phase string
			if err := db.Table("sw_kra_state").Select("phase").Where("id=1").Scan(&phase).Error; err != nil {
				return err
			}
			state.SourceWasLegacy = state.SourceWasLegacy || phase != "completed"
		}
		if err = db.Create(&state).Error; err != nil {
			return err
		}
	} else if err != nil {
		return err
	}
	if state.Version != workflowVersion || state.DatabaseIdentity != serverUUID+":"+schema {
		return fmt.Errorf("automatic upgrade identity/version does not match this database")
	}
	started := time.Now()
	logger.Info("automatic business database upgrade started", "database", schema, "phase", state.Phase)
	defer func() {
		status := "completed"
		if runErr != nil {
			status = "failed"
			clean, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
			defer cancel()
			_ = db.Session(&gorm.Session{NewDB: true}).WithContext(clean).Model(&statePO{}).Where("name=?", workflowName).Updates(map[string]any{"last_error": runErr.Error(), "updated_at": time.Now()}).Error
		}
		logger.Info("automatic business database upgrade finished", "database", schema, "status", status, "elapsed_ms", time.Since(started).Milliseconds())
	}()
	save := func() error { state.UpdatedAt = time.Now(); return db.Save(&state).Error }
	stage := func(name string, work func() error) error {
		state.Phase = name
		state.LastError = ""
		if err := save(); err != nil {
			return err
		}
		begin := time.Now()
		logger.Info("database upgrade stage started", "stage", name)
		done := make(chan struct{})
		stopped := make(chan struct{})
		go func() {
			defer close(stopped)
			ticker := time.NewTicker(10 * time.Second)
			defer ticker.Stop()
			for {
				select {
				case <-done:
					return
				case <-ticker.C:
					logger.Info("database upgrade is still running", "stage", name, "elapsed_ms", time.Since(begin).Milliseconds())
				}
			}
		}()
		err := work()
		close(done)
		<-stopped
		if err != nil {
			return fmt.Errorf("database upgrade %s: %w", name, err)
		}
		logger.Info("database upgrade stage completed", "stage", name, "elapsed_ms", time.Since(begin).Milliseconds())
		return nil
	}
	resumingCleanup := state.CleanupBackupPath != "" && state.VerifiedAt != nil
	wasComplete := state.CompletedAt != nil
	if !resumingCleanup && !wasComplete {
		if state.BackupPath == "" {
			if err := stage("backup", func() error {
				var e error
				state.BackupPath, state.BackupSHA256, e = backupDatabase(ctx, db, backupRoot, state.DatabaseIdentity, "source", nil, logger)
				return e
			}); err != nil {
				return err
			}
			if err := save(); err != nil {
				return err
			}
		} else if err := verifyBackup(state.BackupPath, state.BackupSHA256); err != nil {
			return err
		}
		if err := stage("framework", func() error { return migrateFramework(ctx, db, conn, logger) }); err != nil {
			return err
		}
	}
	if err := stage("migrations", func() error {
		if err := catalog.CheckPrerequisites(db); err != nil {
			return err
		}
		return migration.Run(db, catalog.MigrationSteps())
	}); err != nil {
		return err
	}
	if err := stage("verify", func() error {
		if err := migration.CheckApplied(db, catalog.MigrationSteps()); err != nil {
			return err
		}
		return catalog.CheckSchema(db)
	}); err != nil {
		return err
	}
	remaining, err := remainingTargets(db)
	if err != nil {
		return err
	}
	if len(remaining) > 0 {
		if wasComplete {
			return fmt.Errorf("retired tables reappeared after a completed upgrade; automatic deletion refused")
		}
		if !resumingCleanup && !wasComplete {
			if err := verifyLegacyChecks(db); err != nil {
				return err
			}
		}
		if state.SourceWasLegacy && !state.TasksRestored {
			if err := stage("tasks", func() error {
				return db.Transaction(func(tx *gorm.DB) error {
					if err := tx.Exec("UPDATE sys_timed_tasks t JOIN sw_kra_id_map m ON m.entity='task' AND m.new_id=t.id JOIN sj_job j ON CAST(j.id AS CHAR)=m.old_id SET t.enabled=(j.job_status=1) WHERE j.deleted=0").Error; err != nil {
						return err
					}
					return tx.Model(&statePO{}).Where("name=?", workflowName).Update("tasks_restored", true).Error
				})
			}); err != nil {
				return err
			}
			state.TasksRestored = true
		}
		if state.CleanupBackupPath == "" {
			if err := stage("cleanup-backup", func() error {
				var e error
				state.CleanupBackupPath, state.CleanupSHA256, e = backupDatabase(ctx, db, backupRoot, state.DatabaseIdentity, "cleanup", remaining, logger)
				return e
			}); err != nil {
				return err
			}
			if err := save(); err != nil {
				return err
			}
		} else if err := verifyBackup(state.CleanupBackupPath, state.CleanupSHA256); err != nil {
			return err
		}
		// The journal survives deletion of all old migration helpers. A restart
		// can finish a partially completed DROP list using the verified archive.
		now := time.Now()
		state.VerifiedAt = &now
		state.VerifiedMigrations = migrationHash(catalog.MigrationSteps())
		if err := stage("cleanup", func() error {
			// CALL can commit before a killed process reaches DROP PROCEDURE.
			// Remove only our own shipped routine before checking dependencies.
			if _, err := conn.ExecContext(ctx, "DROP PROCEDURE IF EXISTS sw_kra_migrate_20261103"); err != nil {
				return err
			}
			return executeScript(ctx, conn, scripts.CleanupSQL(), true, logger)
		}); err != nil {
			return err
		}
	}
	if err := migration.CheckApplied(db, catalog.MigrationSteps()); err != nil {
		return err
	}
	if err := catalog.CheckSchema(db); err != nil {
		return err
	}
	remaining, err = remainingTargets(db)
	if err != nil {
		return err
	}
	if len(remaining) != 0 {
		return fmt.Errorf("legacy cleanup left %d tables", len(remaining))
	}
	now := time.Now()
	state.Phase = "complete"
	if state.CompletedAt == nil {
		state.CompletedAt = &now
	}
	state.VerifiedAt = &now
	state.LastError = ""
	state.VerifiedMigrations = migrationHash(catalog.MigrationSteps())
	return save()
}

func migrationHash(steps []migration.Step) string {
	ids := make([]string, len(steps))
	for i, step := range steps {
		ids[i] = step.ID
	}
	sort.Strings(ids)
	sum := sha256.Sum256([]byte(strings.Join(ids, "\n")))
	return hex.EncodeToString(sum[:])
}

func remainingTargets(db *gorm.DB) (map[string]bool, error) {
	tables, err := tableSet(db)
	if err != nil {
		return nil, err
	}
	remaining := map[string]bool{}
	for _, name := range scripts.CleanupTables() {
		if tables[name] {
			remaining[name] = true
		}
	}
	return remaining, nil
}

func executeScript(ctx context.Context, conn *sql.Conn, source string, cleanup bool, logger *slog.Logger) error {
	statements, err := scripts.Statements(source)
	if err != nil {
		return err
	}
	for _, statement := range statements {
		upper := strings.ToUpper(strings.TrimSpace(statement))
		if strings.HasPrefix(upper, "USE ") {
			continue
		}
		if cleanup && strings.HasPrefix(upper, "SET @SW_CLEANUP_EXECUTE") {
			statement = "SET @sw_cleanup_execute=1"
		}
		rows, err := conn.QueryContext(ctx, statement)
		if err != nil {
			return err
		}
		for {
			columns, err := rows.Columns()
			if err != nil {
				rows.Close()
				return err
			}
			for rows.Next() {
				values := make([]any, len(columns))
				dest := make([]any, len(columns))
				for i := range values {
					dest[i] = &values[i]
				}
				if err := rows.Scan(dest...); err != nil {
					rows.Close()
					return err
				}
				// Result sets contain progress/metadata only, never copied rows.
				if len(columns) > 0 && (columns[0] == "dropped_table" || columns[0] == "result") {
					attrs := make([]any, 0, len(columns)*2)
					for i, name := range columns {
						value := values[i]
						if b, ok := value.([]byte); ok {
							value = string(b)
						}
						attrs = append(attrs, name, value)
					}
					logger.Info("database migration SQL result", attrs...)
				}
			}
			if err := rows.Err(); err != nil {
				rows.Close()
				return err
			}
			if !rows.NextResultSet() {
				break
			}
		}
		if err := rows.Err(); err != nil {
			rows.Close()
			return err
		}
		rows.Close()
	}
	return nil
}

func verifyLegacyChecks(db *gorm.DB) error {
	tables, err := tableSet(db)
	if err != nil {
		return err
	}
	if !tables["sw_kra_state"] || !tables["sw_kra_checks"] {
		return fmt.Errorf("legacy migration evidence is missing; cleanup refused")
	}
	var completed int64
	if err := db.Table("sw_kra_state").Where("id=1 AND version=? AND phase='completed' AND completed_at IS NOT NULL", "20261103_framework_migration_v5").Count(&completed).Error; err != nil {
		return err
	}
	if completed != 1 {
		return fmt.Errorf("legacy framework migration is incomplete")
	}
	if err := verifyCheckRows(db); err != nil {
		return err
	}
	if tables["legacy_sw_20261103_sys_user"] && tables["sw_kra_id_map"] {
		var archived, mapped int64
		if err := db.Table("legacy_sw_20261103_sys_user").Where("del_flag='0'").Count(&archived).Error; err != nil {
			return err
		}
		if err := db.Table("sw_kra_id_map").Where("entity='user'").Count(&mapped).Error; err != nil {
			return err
		}
		if archived != mapped {
			return fmt.Errorf("legacy accounts changed during conversion; cleanup refused")
		}
	}
	return nil
}

func verifyCheckRows(db *gorm.DB) error {
	var invalid, backups, checks int64
	if err := db.Table("sw_kra_checks").Where("passed<>1 OR expected_value IS NULL OR actual_value IS NULL OR expected_value<>actual_value").Count(&invalid).Error; err != nil {
		return err
	}
	if err := db.Table("sw_kra_checks").Where("stage='backup'").Distinct("check_name").Count(&backups).Error; err != nil {
		return err
	}
	if err := db.Table("sw_kra_checks").Where("stage='migration'").Distinct("check_name").Count(&checks).Error; err != nil {
		return err
	}
	if invalid != 0 || backups != 52 || checks != 24 {
		return fmt.Errorf("legacy migration validation is incomplete; cleanup refused")
	}
	return nil
}