package database

import (
	"fmt"
	"strings"
	"testing"

	"github.com/stretchr/testify/require"
	"gorm.io/gorm"
)

func TestMigrateReconcilesWorkflowAutomationSchemas(t *testing.T) {
	tests := []struct {
		name    string
		prepare func(*testing.T, *gorm.DB)
		verify  func(*testing.T, *gorm.DB)
	}{
		{
			name: "fresh empty",
			verify: func(t *testing.T, db *gorm.DB) {
				require.True(t, db.Migrator().HasTable("automation_runs"))
				require.False(t, db.Migrator().HasTable("workflow_runs"))
			},
		},
		{
			name: "legacy workflow 0001 through 0008",
			prepare: func(t *testing.T, db *gorm.DB) {
				prepareLegacyWorkflowSchema(t, db, "0008")
				require.NoError(t, db.Exec(`INSERT INTO workflow_runs
					(id, workspace_id, name, input_json, output_json, error_json, summary, metadata, created_at, updated_at)
					VALUES ('run-legacy', 'ws-legacy', 'Legacy', '{}', '{}', '{}', '', '', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`).Error)
			},
			verify: func(t *testing.T, db *gorm.DB) {
				var count int64
				require.NoError(t, db.Table("automation_runs").Where("id = ? AND workspace_id = ?", "run-legacy", "ws-legacy").Count(&count).Error)
				require.EqualValues(t, 1, count)
				require.False(t, db.Migrator().HasTable("workflow_runs"))
				require.True(t, db.Migrator().HasColumn("automation_runs", "definition_path"))
				require.True(t, db.Migrator().HasIndex("automation_runs", "idx_automation_runs_workspace_updated"))
			},
		},
		{
			name: "c partial workflow 0001 through automation 0010",
			prepare: func(t *testing.T, db *gorm.DB) {
				prepareLegacyWorkflowSchema(t, db, "0010")
			},
			verify: func(t *testing.T, db *gorm.DB) {
				require.True(t, db.Migrator().HasTable("automation_runner_tokens"))
				require.True(t, db.Migrator().HasColumn("automation_runner_tokens", "token_scope"))
				require.True(t, db.Migrator().HasTable("automation_actions_runner_credentials"))
			},
		},
		{
			name: "interrupted reconciliation",
			prepare: func(t *testing.T, db *gorm.DB) {
				prepareLegacyWorkflowSchema(t, db, "0008")
				require.NoError(t, db.Migrator().RenameTable("workflow_runs", "automation_runs"))
				require.NoError(t, db.Migrator().RenameColumn("automation_runs", "workflow_path", "definition_path"))
				require.NoError(t, db.Migrator().RenameColumn("automation_runs", "workflow_id", "definition_id"))
			},
			verify: func(t *testing.T, db *gorm.DB) {
				require.False(t, db.Migrator().HasTable("workflow_jobs"))
				require.True(t, db.Migrator().HasTable("automation_jobs"))
				require.True(t, db.Migrator().HasColumn("automation_runs", "definition_path"))
			},
		},
		{
			name: "automation only repeat",
			prepare: func(t *testing.T, db *gorm.DB) {
				require.NoError(t, Migrate(db))
			},
			verify: func(t *testing.T, db *gorm.DB) {
				var markerTables int64
				require.NoError(t, db.Raw("SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'schema_reconciliations'").Scan(&markerTables).Error)
				require.Zero(t, markerTables)
				require.True(t, db.Migrator().HasTable("automation_runs"))
				require.False(t, db.Migrator().HasTable("workflow_runs"))
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dsn := "file:" + strings.ReplaceAll(tt.name, " ", "_") + "?mode=memory&cache=shared"
			db, err := open(Config{DSN: dsn}, false)
			require.NoError(t, err)
			t.Cleanup(func() { require.NoError(t, Close(db)) })
			if tt.prepare != nil {
				tt.prepare(t, db)
			}

			require.NoError(t, Migrate(db))
			require.NoError(t, VerifySchema(db))
			tt.verify(t, db)
			require.NoError(t, Migrate(db))
		})
	}
}

func TestMigrateRejectsWorkflowAutomationTableConflict(t *testing.T) {
	db, err := open(Config{DSN: "file:workflow_automation_conflict?mode=memory&cache=shared"}, false)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, Close(db)) })
	require.NoError(t, Migrate(db))
	require.NoError(t, db.Exec("CREATE TABLE workflow_runs (id VARCHAR(64) PRIMARY KEY)").Error)

	err = Migrate(db)
	require.ErrorContains(t, err, "workflow_runs and automation_runs")
}

func TestMigrateRejectsAppliedWorkflowMigrationWithMissingTable(t *testing.T) {
	db, err := open(Config{DSN: "file:workflow_automation_missing?mode=memory&cache=shared"}, false)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, Close(db)) })
	require.NoError(t, ensureSchemaMigrationsTable(db))
	require.NoError(t, db.Exec("INSERT INTO schema_migrations (version, name, applied_at) VALUES ('0001', 'workflow_runtime', CURRENT_TIMESTAMP)").Error)

	err = Migrate(db)
	require.ErrorContains(t, err, "neither workflow_runs nor automation_runs")
}

func prepareLegacyWorkflowSchema(t *testing.T, db *gorm.DB, lastVersion string) {
	t.Helper()
	require.NoError(t, ensureSchemaMigrationsTable(db))
	migrations, err := readAutomationMigrations()
	require.NoError(t, err)
	for _, migration := range migrations {
		if migration.Version > lastVersion {
			break
		}
		if migration.Version <= "0008" && migration.Version != "0005" && migration.Version != "0006" {
			executeLegacyWorkflowMigration(t, db, migration)
		}
		if migration.Version == "0010" {
			require.NoError(t, executeAutomationMigration(db, migration))
		}
		require.NoError(t, db.Exec(
			"INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
			migration.Version,
			strings.ReplaceAll(migration.Name, "automation", "workflow"),
		).Error)
	}
}

func executeLegacyWorkflowMigration(t *testing.T, db *gorm.DB, migration automationMigration) {
	t.Helper()
	sqlText := strings.NewReplacer(
		"automation_", "workflow_",
		"definition_path", "workflow_path",
		"definition_id", "workflow_id",
	).Replace(renderAutomationMigrationSQL(db.Dialector.Name(), migration.SQL))
	for _, statement := range splitSQLStatements(sqlText) {
		if strings.Contains(statement, "+meshx Backfill") {
			continue
		}
		directive := parseAddColumnDirective(statement)
		if directive.table != "" && db.Migrator().HasColumn(directive.table, directive.column) {
			continue
		}
		sqlStatement := stripSQLComments(statement)
		if strings.TrimSpace(sqlStatement) == "" {
			continue
		}
		require.NoError(t, db.Exec(sqlStatement).Error, fmt.Sprintf("apply legacy fixture migration %s", migration.Path))
	}
}