package database

import (
	"database/sql"
	"embed"
	"encoding/json"
	"fmt"
	"io/fs"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/store/automation"
	"github.com/tidbcloud/AgentTeam/internal/store/content"
	"gorm.io/gorm"
)

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

type automationMigration struct {
	Version string
	Name    string
	Path    string
	SQL     string
}

var automationRequiredColumns = map[string][]string{
	"content_authorities": {
		"id", "workspace_id", "domain", "binding_key", "state", "generation",
		"checkpoint", "updated_by", "created_at", "updated_at",
	},
	"content_version_operations": {
		"id", "workspace_id", "content_id", "operation_key", "request_digest",
		"operation_namespace", "operation_version", "expected_head_version_id",
		"bounds_policy_namespace", "bounds_policy_version", "max_files", "max_file_bytes",
		"max_total_bytes", "max_manifest_bytes",
		"expected_generation", "actor", "model", "checkpoint", "version_id",
		"root_digest", "size_bytes", "file_count", "summary_bytes", "created_at",
		"content_revision", "retained_version_count", "stored_payload_bytes",
	},
	"schema_migrations": {"version", "name", "applied_at"},
	"automation_content_publications": {
		"workspace_id", "content_id", "content_version_id", "repository_owner",
		"repository_name", "ref", "changed_paths_json", "actor_id", "created_at",
	},
	"automation_content_definitions": {
		"workspace_id", "content_id", "content_version_id", "definition_path", "digest", "size_bytes", "created_at",
	},
	"automation_runs": {
		"id", "workspace_id", "idempotency_key", "source_kind", "repository_owner",
		"content_id", "content_version_id", "repository_name", "repository_full_name", "definition_path", "git_ref",
		"commit_sha", "definition_id", "name", "run_name", "status", "input_json",
		"output_json", "error_json", "trigger_kind", "trigger_ref", "target_kind",
		"target_ref", "owner_principal_id", "created_by_principal_id", "current_job_id",
		"summary", "metadata", "started_at", "completed_at", "created_at", "updated_at",
	},
	"automation_run_slots": {
		"workspace_id", "slot_index", "run_id", "created_at", "updated_at",
	},
	"automation_repo_status_outbox": {
		"id", "workspace_id", "run_id", "run_created_at", "repository_owner", "repository_name",
		"commit_sha", "context", "state", "description", "sequence", "status",
		"attempts", "max_attempts", "next_attempt_at", "lease_owner",
		"lease_expires_at", "last_error", "delivered_at", "created_at", "updated_at",
	},
	"automation_job_runs": {
		"id", "workspace_id", "run_id", "job_id", "job_type", "status", "attempt",
		"max_attempts", "runs_on_json", "input_json", "output_json", "error_json",
		"depends_on_json", "lease_owner", "lease_until", "scheduled_at", "started_at",
		"completed_at", "created_at", "updated_at",
	},
	"automation_events": {
		"id", "workspace_id", "run_id", "job_run_id", "event_type", "actor_principal_id",
		"resource_type", "resource_id", "summary", "metadata_json", "created_at",
	},
	"automation_jobs": {
		"id", "workspace_id", "run_id", "job_run_id", "queue_name", "status", "priority",
		"scheduled_at", "not_before", "lease_owner", "lease_until", "attempt", "max_attempts",
		"idempotency_key", "runner_group", "runner_group_key", "labels_json", "label_key",
		"label_key_hash", "label_count", "shard_bucket", "created_at", "updated_at",
	},
	"automation_waits": {
		"id", "workspace_id", "run_id", "job_run_id", "wait_kind", "status", "active_key",
		"signal_key", "resource_type", "resource_id", "resume_event_type",
		"input_json", "result_json", "created_by", "resolved_by",
		"created_at", "updated_at", "resolved_at", "canceled_at",
	},
	"automation_runners": {
		"workspace_id", "runner_id", "name", "runner_group", "labels_json", "status",
		"last_seen_at", "revoked_at", "registered_at", "created_at", "updated_at",
	},
	"automation_runner_tokens": {
		"token_hash", "workspace_id", "runner_id", "token_kind", "token_scope",
		"session_id", "session_token_hash", "expires_at", "revoked_at", "used_at",
		"created_at", "updated_at",
	},
	"automation_runner_sessions": {
		"workspace_id", "runner_id", "session_id", "lease_owner", "runner_group",
		"labels_json", "status", "heartbeat_at", "expires_at", "revoked_at", "created_at",
		"updated_at",
	},
	"automation_actions_runner_credentials": {
		"client_id", "workspace_id", "runner_id", "official_runner_id",
		"public_key", "created_at", "updated_at", "revoked_at",
	},
	"automation_log_chunks": {
		"id", "workspace_id", "run_id", "job_run_id", "stream", "sequence", "object_key",
		"size_bytes", "sha256", "content_type", "created_at",
	},
	"automation_artifacts": {
		"id", "workspace_id", "run_id", "job_run_id", "name", "object_key", "object_url",
		"size_bytes", "sha256", "content_type", "expires_at", "created_at", "updated_at",
	},
	"automation_secrets": {
		"workspace_id", "repository_owner", "repository_name", "name", "value_ciphertext",
		"created_by", "updated_by", "created_at", "updated_at",
	},
	"automation_pinned_repos": {
		"workspace_id", "principal_id", "owner", "repo", "created_at", "updated_at",
	},
}

var automationRequiredIndexes = map[string][]string{
	"content_authorities": {
		"uk_content_authority_binding", "idx_content_authority_state",
	},
	"content_version_operations": {
		"uk_content_version_operation",
	},
	"automation_runs": {
		"uk_automation_runs_workspace_idem",
		"idx_automation_runs_workspace_updated",
		"idx_automation_runs_workspace_status_updated",
		"idx_automation_runs_status_updated",
	},
	"automation_run_slots": {
		"uk_automation_run_slots_run",
	},
	"automation_repo_status_outbox": {
		"uk_automation_repo_status_outbox_commit_context",
		"idx_automation_repo_status_outbox_status_next",
	},
	"automation_job_runs": {
		"idx_automation_job_runs_run_created",
		"idx_automation_job_runs_run_status",
		"idx_automation_job_runs_status_updated",
	},
	"automation_events": {
		"idx_automation_events_run_created",
	},
	"automation_jobs": {
		"uk_automation_jobs_idem",
		"idx_automation_jobs_run_status",
		"idx_automation_jobs_claim_lookup",
		"idx_automation_jobs_workspace_claim_lookup",
		"idx_automation_jobs_ready_workspace_lookup",
		"idx_automation_jobs_lease_recover",
	},
	"automation_waits": {
		"uk_automation_waits_active",
		"idx_automation_waits_signal_status",
		"idx_automation_waits_run_status_created",
	},
	"automation_runner_tokens": {
		"idx_automation_runner_tokens_runner_expiry",
		"idx_automation_runner_tokens_session_lookup",
	},
	"automation_runner_sessions": {
		"idx_automation_runner_sessions_runner_expiry",
	},
	"automation_actions_runner_credentials": {
		"idx_automation_actions_runner_credentials_runner",
		"idx_automation_actions_runner_credentials_official_runner_id",
		"idx_automation_actions_runner_credentials_revoked_at",
	},
	"automation_log_chunks": {
		"uk_automation_log_chunks_sequence",
	},
	"automation_artifacts": {
		"idx_automation_artifacts_run_job_name",
	},
	"automation_pinned_repos": {
		"idx_automation_pinned_repos_updated_at",
	},
}

var automationRequiredIndexColumns = map[string]map[string][]string{
	"content_authorities": {
		"uk_content_authority_binding": {"workspace_id", "domain", "binding_key"},
	},
	"content_version_operations": {
		"uk_content_version_operation": {"workspace_id", "content_id", "operation_key"},
	},
	"automation_runner_tokens": {
		"idx_automation_runner_tokens_session_lookup": {
			"workspace_id",
			"runner_id",
			"token_kind",
			"session_id",
			"revoked_at",
			"expires_at",
			"updated_at",
		},
	},
	"automation_jobs": {
		"idx_automation_jobs_workspace_claim_lookup": {
			"workspace_id",
			"queue_name",
			"shard_bucket",
			"status",
			"runner_group_key",
			"label_key_hash",
			"scheduled_at",
			"priority",
			"id",
		},
		"idx_automation_jobs_ready_workspace_lookup": {
			"status",
			"queue_name",
			"scheduled_at",
			"workspace_id",
		},
	},
	"automation_repo_status_outbox": {
		"uk_automation_repo_status_outbox_commit_context": {
			"workspace_id",
			"repository_owner",
			"repository_name",
			"commit_sha",
			"context",
		},
		"idx_automation_repo_status_outbox_status_next": {
			"status",
			"next_attempt_at",
			"id",
		},
	},
	"automation_waits": {
		"idx_automation_waits_signal_status": {
			"signal_key",
			"status",
			"updated_at",
			"id",
		},
	},
}

const (
	automationJobClaimBackfillBatchSize   = 500
	automationRunSlotBackfillBatchSize    = 500
	automationWaitSignalBackfillBatchSize = 500
)

func applyAutomationMigrations(db *gorm.DB) error {
	if db == nil {
		return fmt.Errorf("meshx database is nil")
	}
	if err := ensureSchemaMigrationsTable(db); err != nil {
		return err
	}
	applied, err := appliedAutomationMigrations(db)
	if err != nil {
		return err
	}
	if err := reconcileWorkflowAutomationSchema(db, applied); err != nil {
		return err
	}
	migrations, err := readAutomationMigrations()
	if err != nil {
		return err
	}
	for _, migration := range migrations {
		if applied[migration.Version] {
			continue
		}
		if err := executeAutomationMigration(db, migration); err != nil {
			return err
		}
		if err := recordAutomationMigration(db, migration); err != nil {
			return err
		}
	}
	return nil
}

func ensureSchemaMigrationsTable(db *gorm.DB) error {
	err := db.Exec(`
CREATE TABLE IF NOT EXISTS schema_migrations (
  version VARCHAR(64) NOT NULL,
  name VARCHAR(255) NOT NULL,
  applied_at TIMESTAMP NOT NULL,
  PRIMARY KEY (version)
)`).Error
	if err != nil {
		return fmt.Errorf("create meshx schema migrations table: %w", err)
	}
	return nil
}

func appliedAutomationMigrations(db *gorm.DB) (map[string]bool, error) {
	rows, err := db.Raw("SELECT version FROM schema_migrations").Rows()
	if err != nil {
		return nil, fmt.Errorf("read meshx schema migrations: %w", err)
	}
	defer rows.Close()

	applied := map[string]bool{}
	for rows.Next() {
		var version string
		if err := rows.Scan(&version); err != nil {
			return nil, fmt.Errorf("scan meshx schema migration version: %w", err)
		}
		applied[strings.TrimSpace(version)] = true
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate meshx schema migration versions: %w", err)
	}
	return applied, nil
}

func readAutomationMigrations() ([]automationMigration, error) {
	entries, err := fs.ReadDir(automationMigrationFiles, "migrations")
	if err != nil {
		return nil, fmt.Errorf("read meshx automation migrations: %w", err)
	}
	migrations := make([]automationMigration, 0, len(entries))
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") {
			continue
		}
		version, name, ok := parseMigrationFilename(entry.Name())
		if !ok {
			return nil, fmt.Errorf("invalid meshx automation migration filename %q", entry.Name())
		}
		path := filepath.Join("migrations", entry.Name())
		content, err := fs.ReadFile(automationMigrationFiles, path)
		if err != nil {
			return nil, fmt.Errorf("read meshx automation migration %s: %w", path, err)
		}
		migrations = append(migrations, automationMigration{Version: version, Name: name, Path: path, SQL: string(content)})
	}
	sort.Slice(migrations, func(i int, j int) bool {
		return migrations[i].Version < migrations[j].Version
	})
	for i := 1; i < len(migrations); i++ {
		if migrations[i-1].Version == migrations[i].Version {
			return nil, fmt.Errorf("duplicate meshx automation migration version %s", migrations[i].Version)
		}
	}
	return migrations, nil
}

func parseMigrationFilename(name string) (string, string, bool) {
	base := strings.TrimSuffix(name, ".sql")
	parts := strings.SplitN(base, "_", 2)
	if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" || strings.TrimSpace(parts[1]) == "" {
		return "", "", false
	}
	for _, r := range parts[0] {
		if r < '0' || r > '9' {
			return "", "", false
		}
	}
	return parts[0], parts[1], true
}

func executeAutomationMigration(db *gorm.DB, migration automationMigration) error {
	sqlText := renderAutomationMigrationSQL(db.Dialector.Name(), migration.SQL)
	tableModels := automationTableModels()
	for _, statement := range splitSQLStatements(sqlText) {
		if parseAutomationJobClaimBackfillDirective(statement) {
			if err := backfillAutomationJobClaimFields(db); err != nil {
				return fmt.Errorf("backfill automation job claim fields in %s: %w", migration.Path, err)
			}
			continue
		}
		if parseAutomationRunSlotsBackfillDirective(statement) {
			if err := backfillAutomationRunSlots(db); err != nil {
				return fmt.Errorf("backfill automation active run slots in %s: %w", migration.Path, err)
			}
			continue
		}
		if parseAutomationWaitSignalKeyBackfillDirective(statement) {
			if err := backfillAutomationWaitSignalKeys(db); err != nil {
				return fmt.Errorf("backfill automation wait signal keys in %s: %w", migration.Path, err)
			}
			continue
		}
		if parsePreviouslyNativeContentAuthorityDirective(statement) {
			if err := normalizePreviouslyNativeContentAuthorities(db); err != nil {
				return fmt.Errorf("normalize previously native content authorities in %s: %w", migration.Path, err)
			}
			continue
		}
		if parseFinalizeContentAuthoritySchemaDirective(statement) {
			if err := finalizeContentAuthoritySchema(db); err != nil {
				return fmt.Errorf("finalize content authority schema in %s: %w", migration.Path, err)
			}
			continue
		}
		if parseAgentOwnedMemoriesBackfillDirective(statement) {
			if err := backfillAgentOwnedMemories(db); err != nil {
				return fmt.Errorf("backfill agent-owned memories in %s: %w", migration.Path, err)
			}
			continue
		}
		dropIndex := parseDropIndexDirective(statement)
		if dropIndex.table != "" {
			model := tableModels[dropIndex.table]
			if model == nil {
				model = dropIndex.table
			}
			if db.Migrator().HasIndex(model, dropIndex.index) {
				if err := db.Migrator().DropIndex(model, dropIndex.index); err != nil {
					return fmt.Errorf("drop meshx automation index %s.%s in %s: %w", dropIndex.table, dropIndex.index, migration.Path, err)
				}
			}
			continue
		}
		dropColumn := parseDropColumnDirective(statement)
		if dropColumn.table != "" {
			if strings.EqualFold(db.Dialector.Name(), "sqlite") {
				if dropColumn.table == "memories" && dropColumn.column == "workspace_id" && db.Migrator().HasColumn(dropColumn.table, dropColumn.column) {
					sqlStatement := stripSQLComments(statement)
					if strings.TrimSpace(sqlStatement) != "" {
						if err := db.Exec(sqlStatement).Error; err != nil {
							return fmt.Errorf("drop meshx automation column %s.%s in %s: %w", dropColumn.table, dropColumn.column, migration.Path, err)
						}
					}
				}
				continue
			}
			model := tableModels[dropColumn.table]
			if model == nil {
				model = dropColumn.table
			}
			if db.Migrator().HasColumn(model, dropColumn.column) {
				if err := db.Migrator().DropColumn(model, dropColumn.column); err != nil {
					return fmt.Errorf("drop meshx automation column %s.%s in %s: %w", dropColumn.table, dropColumn.column, migration.Path, err)
				}
			}
			continue
		}
		directive := parseAddColumnDirective(statement)
		if directive.table != "" {
			model := tableModels[directive.table]
			if model == nil {
				model = directive.table
			}
			if db.Migrator().HasColumn(model, directive.column) {
				continue
			}
		}
		alterColumn := parseAlterColumnDirective(statement)
		if alterColumn.table != "" {
			if strings.EqualFold(db.Dialector.Name(), "sqlite") {
				continue
			}
			model := tableModels[alterColumn.table]
			if model == nil {
				model = alterColumn.table
			}
			if err := db.Migrator().AlterColumn(model, alterColumn.column); err != nil {
				return fmt.Errorf("alter meshx automation column %s.%s in %s: %w", alterColumn.table, alterColumn.column, migration.Path, err)
			}
			continue
		}
		sqlStatement := stripSQLComments(statement)
		if strings.TrimSpace(sqlStatement) == "" {
			continue
		}
		if err := db.Exec(sqlStatement).Error; err != nil {
			return fmt.Errorf("apply meshx automation migration %s: %w", migration.Path, err)
		}
	}
	return nil
}

func renderAutomationMigrationSQL(dialect string, sqlText string) string {
	autoIncrementID := "id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY"
	switch strings.ToLower(strings.TrimSpace(dialect)) {
	case "postgres":
		autoIncrementID = "id BIGSERIAL PRIMARY KEY"
	case "sqlite":
		autoIncrementID = "id INTEGER PRIMARY KEY AUTOINCREMENT"
	}
	return strings.ReplaceAll(sqlText, "{{AUTOINCREMENT_ID_COLUMN}}", autoIncrementID)
}

type addColumnDirective struct {
	table  string
	column string
}

type alterColumnDirective struct {
	table  string
	column string
}

type dropIndexDirective struct {
	table string
	index string
}

type dropColumnDirective struct {
	table  string
	column string
}

func parseAddColumnDirective(statement string) addColumnDirective {
	for _, line := range strings.Split(statement, "\n") {
		line = strings.TrimSpace(line)
		if !strings.HasPrefix(line, "-- +meshx AddColumn ") {
			continue
		}
		fields := strings.Fields(strings.TrimPrefix(line, "-- +meshx AddColumn "))
		if len(fields) == 2 {
			return addColumnDirective{table: fields[0], column: fields[1]}
		}
	}
	return addColumnDirective{}
}

func parseAlterColumnDirective(statement string) alterColumnDirective {
	for _, line := range strings.Split(statement, "\n") {
		line = strings.TrimSpace(line)
		if !strings.HasPrefix(line, "-- +meshx AlterColumn ") {
			continue
		}
		fields := strings.Fields(strings.TrimPrefix(line, "-- +meshx AlterColumn "))
		if len(fields) == 2 {
			return alterColumnDirective{table: fields[0], column: strings.TrimSuffix(fields[1], ";")}
		}
	}
	return alterColumnDirective{}
}

func parseDropIndexDirective(statement string) dropIndexDirective {
	for _, line := range strings.Split(statement, "\n") {
		line = strings.TrimSpace(line)
		if !strings.HasPrefix(line, "-- +meshx DropIndex ") {
			continue
		}
		fields := strings.Fields(strings.TrimPrefix(line, "-- +meshx DropIndex "))
		if len(fields) == 2 {
			return dropIndexDirective{table: fields[0], index: strings.TrimSuffix(fields[1], ";")}
		}
	}
	return dropIndexDirective{}
}

func parseDropColumnDirective(statement string) dropColumnDirective {
	for _, line := range strings.Split(statement, "\n") {
		line = strings.TrimSpace(line)
		if !strings.HasPrefix(line, "-- +meshx DropColumn ") {
			continue
		}
		fields := strings.Fields(strings.TrimPrefix(line, "-- +meshx DropColumn "))
		if len(fields) == 2 {
			return dropColumnDirective{table: fields[0], column: strings.TrimSuffix(fields[1], ";")}
		}
	}
	return dropColumnDirective{}
}

func parseAutomationJobClaimBackfillDirective(statement string) bool {
	for _, line := range strings.Split(statement, "\n") {
		if strings.TrimSpace(line) == "-- +meshx BackfillAutomationJobClaimFields" {
			return true
		}
	}
	return false
}

func parseAutomationRunSlotsBackfillDirective(statement string) bool {
	for _, line := range strings.Split(statement, "\n") {
		if strings.TrimSpace(line) == "-- +meshx BackfillAutomationRunSlots" {
			return true
		}
	}
	return false
}

func parseAutomationWaitSignalKeyBackfillDirective(statement string) bool {
	for _, line := range strings.Split(statement, "\n") {
		if strings.TrimSpace(line) == "-- +meshx BackfillAutomationWaitSignalKeys" {
			return true
		}
	}
	return false
}

func parsePreviouslyNativeContentAuthorityDirective(statement string) bool {
	for _, line := range strings.Split(statement, "\n") {
		if strings.TrimSpace(line) == "-- +meshx NormalizePreviouslyNativeContentAuthorities" {
			return true
		}
	}
	return false
}

func normalizePreviouslyNativeContentAuthorities(db *gorm.DB) error {
	if !db.Migrator().HasColumn("content_authorities", "native_activated") {
		return nil
	}
	return db.Exec(`UPDATE content_authorities
SET state = 'native_active'
WHERE binding_key <> '*'
  AND state = 'write_fenced'
  AND native_activated = TRUE`).Error
}

func parseFinalizeContentAuthoritySchemaDirective(statement string) bool {
	for _, line := range strings.Split(statement, "\n") {
		if strings.TrimSpace(line) == "-- +meshx FinalizeContentAuthoritySchema" {
			return true
		}
	}
	return false
}

// finalizeContentAuthoritySchema removes the two migration-only marker
// columns. SQLite needs an explicit model-owned replacement table because its
// generic column migrator does not reliably remove unquoted legacy columns.
// The bounded INSERT ... SELECT names checkpoint so activation proof cannot be
// lost during the rebuild. Other dialects can drop the columns in place.
func finalizeContentAuthoritySchema(db *gorm.DB) error {
	legacyColumns := []string{"native_activated", "recovery_point_id"}
	if !hasLegacyContentAuthorityColumns(db) {
		return db.AutoMigrate(&content.Authority{})
	}
	if strings.EqualFold(db.Dialector.Name(), "sqlite") {
		return rebuildSQLiteContentAuthoritySchema(db)
	}

	for _, column := range legacyColumns {
		if !db.Migrator().HasColumn("content_authorities", column) {
			continue
		}
		if err := db.Migrator().DropColumn(&content.Authority{}, column); err != nil {
			return err
		}
	}
	return db.AutoMigrate(&content.Authority{})
}

func hasLegacyContentAuthorityColumns(db *gorm.DB) bool {
	return db.Migrator().HasColumn("content_authorities", "native_activated") ||
		db.Migrator().HasColumn("content_authorities", "recovery_point_id")
}

func rebuildSQLiteContentAuthoritySchema(db *gorm.DB) error {
	const legacyTable = "content_authorities__legacy_0025"
	return db.Transaction(func(tx *gorm.DB) error {
		// Renaming preserves standalone index names, so release the two names
		// before creating the current model-owned table inside this transaction.
		for _, index := range []string{"idx_content_authority_state", "uk_content_authority_binding"} {
			if err := tx.Exec("DROP INDEX IF EXISTS " + index).Error; err != nil {
				return err
			}
		}
		if err := tx.Exec("ALTER TABLE content_authorities RENAME TO " + legacyTable).Error; err != nil {
			return err
		}
		if err := tx.Migrator().CreateTable(&content.Authority{}); err != nil {
			return err
		}
		if err := tx.Exec(`INSERT INTO content_authorities
  (id, workspace_id, domain, binding_key, state, generation, checkpoint, updated_by, created_at, updated_at)
SELECT id, workspace_id, domain, binding_key, state, generation, checkpoint, updated_by, created_at, updated_at
FROM ` + legacyTable).Error; err != nil {
			return err
		}
		return tx.Exec("DROP TABLE " + legacyTable).Error
	})
}

func parseAgentOwnedMemoriesBackfillDirective(statement string) bool {
	for _, line := range strings.Split(statement, "\n") {
		if strings.TrimSpace(line) == "-- +meshx BackfillAgentOwnedMemories" {
			return true
		}
	}
	return false
}

func backfillAgentOwnedMemories(db *gorm.DB) error {
	if !db.Migrator().HasTable("memory_agent_bindings") {
		return nil
	}
	statements := []string{
		`DELETE FROM memory_agent_bindings
WHERE NOT EXISTS (
  SELECT 1 FROM memories WHERE memories.id = memory_agent_bindings.memory_id
)`,
		`DELETE FROM memory_grants
WHERE NOT EXISTS (
  SELECT 1 FROM memory_agent_bindings WHERE memory_agent_bindings.memory_id = memory_grants.memory_id
)`,
		`DELETE FROM memory_pins
WHERE NOT EXISTS (
  SELECT 1 FROM memory_agent_bindings WHERE memory_agent_bindings.memory_id = memory_pins.memory_id
)`,
		`DELETE FROM resource_repositories
WHERE resource_type = 'memory'
  AND NOT EXISTS (
    SELECT 1 FROM memory_agent_bindings WHERE memory_agent_bindings.memory_id = resource_repositories.resource_id
  )`,
		`DELETE FROM resource_activity_events
WHERE asset_type = 'memory'
  AND NOT EXISTS (
    SELECT 1 FROM memory_agent_bindings WHERE memory_agent_bindings.memory_id = resource_activity_events.asset_id
  )`,
		`DELETE FROM memories
WHERE NOT EXISTS (
  SELECT 1 FROM memory_agent_bindings WHERE memory_agent_bindings.memory_id = memories.id
)`,
		`UPDATE memories
SET
  agent_id = (
    SELECT memory_agent_bindings.agent_id
    FROM memory_agent_bindings
    WHERE memory_agent_bindings.memory_id = memories.id
    ORDER BY memory_agent_bindings.updated_at DESC, memory_agent_bindings.id DESC
    LIMIT 1
  ),
  branch = COALESCE(NULLIF((
    SELECT memory_agent_bindings.branch
    FROM memory_agent_bindings
    WHERE memory_agent_bindings.memory_id = memories.id
    ORDER BY memory_agent_bindings.updated_at DESC, memory_agent_bindings.id DESC
    LIMIT 1
  ), ''), 'main'),
  import_existing_local_content = COALESCE((
    SELECT memory_agent_bindings.import_existing_local_content
    FROM memory_agent_bindings
    WHERE memory_agent_bindings.memory_id = memories.id
    ORDER BY memory_agent_bindings.updated_at DESC, memory_agent_bindings.id DESC
    LIMIT 1
  ), FALSE),
  replace_local_content_on_switch = COALESCE((
    SELECT memory_agent_bindings.replace_local_content_on_switch
    FROM memory_agent_bindings
    WHERE memory_agent_bindings.memory_id = memories.id
    ORDER BY memory_agent_bindings.updated_at DESC, memory_agent_bindings.id DESC
    LIMIT 1
  ), FALSE),
  updated_by = COALESCE(NULLIF((
    SELECT memory_agent_bindings.updated_by
    FROM memory_agent_bindings
    WHERE memory_agent_bindings.memory_id = memories.id
    ORDER BY memory_agent_bindings.updated_at DESC, memory_agent_bindings.id DESC
    LIMIT 1
  ), ''), updated_by),
  updated_at = COALESCE((
    SELECT memory_agent_bindings.updated_at
    FROM memory_agent_bindings
    WHERE memory_agent_bindings.memory_id = memories.id
    ORDER BY memory_agent_bindings.updated_at DESC, memory_agent_bindings.id DESC
    LIMIT 1
  ), updated_at)
WHERE EXISTS (
  SELECT 1 FROM memory_agent_bindings WHERE memory_agent_bindings.memory_id = memories.id
)`,
	}
	for _, statement := range statements {
		if err := db.Exec(statement).Error; err != nil {
			return err
		}
	}
	if err := db.Migrator().DropTable("memory_agent_bindings"); err != nil {
		return fmt.Errorf("drop memory_agent_bindings: %w", err)
	}
	return nil
}

func backfillAutomationJobClaimFields(db *gorm.DB) error {
	if !db.Migrator().HasTable(&automation.AutomationJob{}) {
		return nil
	}
	for _, column := range []string{"runner_group_key", "label_key", "label_key_hash", "label_count"} {
		if !db.Migrator().HasColumn(&automation.AutomationJob{}, column) {
			return nil
		}
	}

	var lastID int64
	for {
		var jobs []automation.AutomationJob
		err := db.Select("id", "runner_group", "labels_json").
			Where(`id > ? AND ((TRIM(runner_group) <> '' AND runner_group_key = '') OR (TRIM(label_key) <> '' AND label_key_hash = '') OR (TRIM(labels_json) NOT IN ('', '[]', 'null') AND (label_key = '' OR label_key_hash = '' OR label_count = 0)))`, lastID).
			Order("id ASC").
			Limit(automationJobClaimBackfillBatchSize).
			Find(&jobs).Error
		if err != nil {
			return fmt.Errorf("load automation jobs missing claim fields: %w", err)
		}
		if len(jobs) == 0 {
			return nil
		}
		lastID = jobs[len(jobs)-1].ID
		if err := db.Transaction(func(tx *gorm.DB) error {
			for _, job := range jobs {
				automation.NormalizeAutomationJobClaimFields(&job)
				err := tx.Model(&automation.AutomationJob{}).
					Where("id = ?", job.ID).
					Updates(map[string]any{
						"runner_group_key": job.RunnerGroupKey,
						"label_key":        job.LabelKey,
						"label_key_hash":   job.LabelKeyHash,
						"label_count":      job.LabelCount,
					}).Error
				if err != nil {
					return fmt.Errorf("update automation job %d claim fields: %w", job.ID, err)
				}
			}
			return nil
		}); err != nil {
			return err
		}
	}
}

func backfillAutomationWaitSignalKeys(db *gorm.DB) error {
	if !db.Migrator().HasTable(&automation.AutomationWait{}) || !db.Migrator().HasColumn(&automation.AutomationWait{}, "signal_key") {
		return nil
	}
	var lastID string
	for {
		var waits []automation.AutomationWait
		q := db.Select("id", "workspace_id", "wait_kind", "resource_type", "resource_id", "resume_event_type", "input_json").
			Where("signal_key = ''").
			Where("workspace_id <> '' AND wait_kind <> '' AND resource_type <> '' AND resume_event_type <> ''").
			Order("id ASC").
			Limit(automationWaitSignalBackfillBatchSize)
		if lastID != "" {
			q = q.Where("id > ?", lastID)
		}
		if err := q.Find(&waits).Error; err != nil {
			return fmt.Errorf("load automation waits missing signal keys: %w", err)
		}
		if len(waits) == 0 {
			return nil
		}
		lastID = waits[len(waits)-1].ID
		if err := db.Transaction(func(tx *gorm.DB) error {
			for _, wait := range waits {
				resourceID := automationWaitSignalBackfillResourceID(wait)
				signalKey := automation.AutomationWaitSignalKey(wait.WorkspaceID, automation.AutomationWaitSignal{
					WaitKind:     wait.WaitKind,
					EventType:    wait.ResumeEventType,
					ResourceType: wait.ResourceType,
					ResourceID:   resourceID,
				})
				if signalKey == "" {
					continue
				}
				updates := map[string]any{"signal_key": signalKey}
				if resourceID != strings.TrimSpace(wait.ResourceID) {
					updates["resource_id"] = resourceID
				}
				if err := tx.Model(&automation.AutomationWait{}).Where("id = ?", wait.ID).Updates(updates).Error; err != nil {
					return fmt.Errorf("update automation wait %s signal key: %w", wait.ID, err)
				}
			}
			return nil
		}); err != nil {
			return err
		}
	}
}

func automationWaitSignalBackfillResourceID(wait automation.AutomationWait) string {
	resourceID := strings.TrimSpace(wait.ResourceID)
	if wait.WaitKind != automation.AutomationWaitKindAgentTask {
		return resourceID
	}
	if parsed := automationWaitAgentTaskResourceIDFromInput(wait.InputJSON); parsed != "" {
		return parsed
	}
	return resourceID
}

func automationWaitAgentTaskResourceIDFromInput(raw string) string {
	var job struct {
		Steps []struct {
			Uses string         `json:"uses"`
			With map[string]any `json:"with"`
		} `json:"steps"`
	}
	if err := json.Unmarshal([]byte(raw), &job); err != nil {
		return ""
	}
	for _, step := range job.Steps {
		if strings.TrimSpace(step.Uses) != "agentteam/agent-task" {
			continue
		}
		return firstAutomationWaitInputString(step.With, "task_id", "taskId", "task-id", "resource_id", "resourceId", "resource-id")
	}
	return ""
}

func firstAutomationWaitInputString(values map[string]any, keys ...string) string {
	for _, key := range keys {
		if value, ok := values[key].(string); ok {
			if trimmed := strings.TrimSpace(value); trimmed != "" {
				return trimmed
			}
		}
	}
	return ""
}

func backfillAutomationRunSlots(db *gorm.DB) error {
	if !db.Migrator().HasTable(&automation.AutomationRun{}) || !db.Migrator().HasTable(&automation.AutomationRunSlot{}) {
		return nil
	}
	nextSlot := map[string]int{}
	var lastID string
	for {
		var runs []automation.AutomationRun
		q := db.Select("id", "workspace_id", "created_at").
			Where("status NOT IN ?", []string{automation.AutomationRunStatusSucceeded, automation.AutomationRunStatusFailed, automation.AutomationRunStatusCanceled}).
			Order("id ASC").
			Limit(automationRunSlotBackfillBatchSize)
		if lastID != "" {
			q = q.Where("id > ?", lastID)
		}
		if err := q.Find(&runs).Error; err != nil {
			return fmt.Errorf("load active automation runs for slot backfill: %w", err)
		}
		if len(runs) == 0 {
			return nil
		}
		lastID = runs[len(runs)-1].ID
		if err := backfillAutomationRunSlotBatch(db, runs, nextSlot); err != nil {
			return err
		}
	}
}

func backfillAutomationRunSlotBatch(db *gorm.DB, runs []automation.AutomationRun, nextSlot map[string]int) error {
	runIDs := make([]string, 0, len(runs))
	workspaceIDs := make([]string, 0, len(runs))
	seenWorkspaces := map[string]struct{}{}
	for _, run := range runs {
		workspaceID := strings.TrimSpace(run.WorkspaceID)
		runID := strings.TrimSpace(run.ID)
		if workspaceID == "" || runID == "" {
			continue
		}
		runIDs = append(runIDs, runID)
		if _, ok := seenWorkspaces[workspaceID]; !ok {
			seenWorkspaces[workspaceID] = struct{}{}
			workspaceIDs = append(workspaceIDs, workspaceID)
		}
	}
	if len(runIDs) == 0 {
		return nil
	}

	return db.Transaction(func(tx *gorm.DB) error {
		type existingSlot struct {
			RunID string
		}
		var existingSlots []existingSlot
		if err := tx.Model(&automation.AutomationRunSlot{}).Select("run_id").Where("run_id IN ?", runIDs).Find(&existingSlots).Error; err != nil {
			return fmt.Errorf("load existing automation run slots: %w", err)
		}
		existingRunIDs := make(map[string]struct{}, len(existingSlots))
		for _, slot := range existingSlots {
			existingRunIDs[strings.TrimSpace(slot.RunID)] = struct{}{}
		}

		uninitializedWorkspaces := make([]string, 0, len(workspaceIDs))
		for _, workspaceID := range workspaceIDs {
			if _, ok := nextSlot[workspaceID]; ok {
				continue
			}
			nextSlot[workspaceID] = 0
			uninitializedWorkspaces = append(uninitializedWorkspaces, workspaceID)
		}
		if len(uninitializedWorkspaces) > 0 {
			type slotMax struct {
				WorkspaceID  string
				MaxSlotIndex sql.NullInt64
			}
			var maxes []slotMax
			if err := tx.Model(&automation.AutomationRunSlot{}).
				Select("workspace_id, MAX(slot_index) AS max_slot_index").
				Where("workspace_id IN ?", uninitializedWorkspaces).
				Group("workspace_id").
				Scan(&maxes).Error; err != nil {
				return fmt.Errorf("load automation run slot maxima: %w", err)
			}
			for _, max := range maxes {
				if max.MaxSlotIndex.Valid {
					nextSlot[strings.TrimSpace(max.WorkspaceID)] = int(max.MaxSlotIndex.Int64) + 1
				}
			}
		}

		now := time.Now().UTC()
		slots := make([]automation.AutomationRunSlot, 0, len(runs)-len(existingRunIDs))
		for _, run := range runs {
			workspaceID := strings.TrimSpace(run.WorkspaceID)
			runID := strings.TrimSpace(run.ID)
			if workspaceID == "" || runID == "" {
				continue
			}
			if _, ok := existingRunIDs[runID]; ok {
				continue
			}
			createdAt := run.CreatedAt
			if createdAt.IsZero() {
				createdAt = now
			}
			slotIndex := nextSlot[workspaceID]
			nextSlot[workspaceID] = slotIndex + 1
			slots = append(slots, automation.AutomationRunSlot{WorkspaceID: workspaceID, SlotIndex: slotIndex, RunID: runID, CreatedAt: createdAt, UpdatedAt: createdAt})
		}
		if len(slots) == 0 {
			return nil
		}
		if err := tx.Create(&slots).Error; err != nil {
			return fmt.Errorf("backfill automation run slots: %w", err)
		}
		return nil
	})
}

func splitSQLStatements(sqlText string) []string {
	var statements []string
	var current strings.Builder
	var quote rune
	for _, r := range sqlText {
		switch r {
		case '\'', '"':
			if quote == 0 {
				quote = r
			} else if quote == r {
				quote = 0
			}
			current.WriteRune(r)
		case ';':
			if quote != 0 {
				current.WriteRune(r)
				continue
			}
			if statement := strings.TrimSpace(current.String()); statement != "" {
				statements = append(statements, statement)
			}
			current.Reset()
		default:
			current.WriteRune(r)
		}
	}
	if statement := strings.TrimSpace(current.String()); statement != "" {
		statements = append(statements, statement)
	}
	return statements
}

func stripSQLComments(statement string) string {
	lines := strings.Split(statement, "\n")
	out := lines[:0]
	for _, line := range lines {
		if strings.HasPrefix(strings.TrimSpace(line), "--") {
			continue
		}
		out = append(out, line)
	}
	return strings.TrimSpace(strings.Join(out, "\n"))
}

func recordAutomationMigration(db *gorm.DB, migration automationMigration) error {
	err := db.Exec(
		"INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
		migration.Version,
		migration.Name,
		time.Now().UTC(),
	).Error
	if err != nil {
		return fmt.Errorf("record meshx automation migration %s: %w", migration.Path, err)
	}
	return nil
}

// VerifySchema verifies the MeshX runtime schema without reading user data. It
// checks only bounded schema metadata: tables, columns, and indexes.
func VerifySchema(db *gorm.DB) error {
	if db == nil {
		return fmt.Errorf("meshx database is nil")
	}
	migrator := db.Migrator()
	tableModels := automationTableModels()
	for _, table := range resourceContentSchemaTables {
		tableModels[table.table] = table.model
	}

	var missingTables []string
	if !migrator.HasTable("schema_migrations") {
		missingTables = append(missingTables, "schema_migrations")
	}
	for _, table := range automationSchemaTables {
		if !migrator.HasTable(table.model) {
			missingTables = append(missingTables, table.table)
		}
	}
	for _, table := range resourceContentSchemaTables {
		if !migrator.HasTable(table.model) {
			missingTables = append(missingTables, table.table)
		}
	}

	var missingColumns []string
	for _, requirements := range []map[string][]string{automationRequiredColumns, resourceContentRequiredColumns} {
		for table, columns := range requirements {
			model := tableModels[table]
			if model == nil {
				model = table
			}
			for _, column := range columns {
				if !migrator.HasColumn(model, column) {
					missingColumns = append(missingColumns, table+"."+column)
				}
			}
		}
	}

	var missingIndexes []string
	for _, requirements := range []map[string][]string{automationRequiredIndexes, resourceContentRequiredIndexes} {
		for table, indexes := range requirements {
			model := tableModels[table]
			if model == nil {
				model = table
			}
			for _, index := range indexes {
				if !migrator.HasIndex(model, index) {
					missingIndexes = append(missingIndexes, table+"."+index)
				}
			}
		}
	}

	var mismatchedIndexes []string
	for _, requirements := range []map[string]map[string][]string{automationRequiredIndexColumns, resourceContentRequiredIndexColumns} {
		for table, indexes := range requirements {
			model := tableModels[table]
			if model == nil {
				model = table
			}
			for index, wantColumns := range indexes {
				if !migrator.HasIndex(model, index) {
					continue
				}
				gotColumns, err := automationIndexColumns(db, table, index)
				if err != nil {
					return err
				}
				if !sameStringSlice(gotColumns, wantColumns) {
					mismatchedIndexes = append(mismatchedIndexes, fmt.Sprintf("%s.%s columns %s want %s", table, index, strings.Join(gotColumns, ","), strings.Join(wantColumns, ",")))
				}
			}
		}
	}

	parts := make([]string, 0, 4)
	if len(missingTables) > 0 {
		parts = append(parts, "missing tables: "+strings.Join(missingTables, ", "))
	}
	if len(missingColumns) > 0 {
		parts = append(parts, "missing columns: "+strings.Join(missingColumns, ", "))
	}
	if len(missingIndexes) > 0 {
		parts = append(parts, "missing indexes: "+strings.Join(missingIndexes, ", "))
	}
	if len(mismatchedIndexes) > 0 {
		parts = append(parts, "mismatched indexes: "+strings.Join(mismatchedIndexes, ", "))
	}
	if len(parts) > 0 {
		return fmt.Errorf("meshx schema mismatch (%s)", strings.Join(parts, "; "))
	}
	return nil
}