package persistence

import (
	"database/sql"
	"reflect"
	"testing"

	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
)

func TestRuntimeOutboundIntentOwnerIndexWarmUpgradeOnTiDB(t *testing.T) {
	for _, equivalentIndex := range []bool{false, true} {
		name := "missing"
		if equivalentIndex {
			name = "equivalent_index_under_another_name"
		}
		t.Run(name, func(t *testing.T) {
			db, cleanup := openSchemaDDLTiDB(t)
			defer cleanup()
			defs := runtimeOutboundOwnerMigrationCatalog(t)
			repository := NewSchemaRepository(db, nil)
			createRuntimeOutboundMigrationTables(t, repository, defs.legacyIntent, defs.generation)
			if equivalentIndex {
				if _, err := db.Exec(`ALTER TABLE _System_RuntimeOutboundIntent ADD INDEX prior_owner_lookup (SourceObject, SourceRecordId, AdmissionState)`); err != nil {
					t.Fatal(err)
				}
			}
			insertRuntimeOutboundMigrationRows(t, db)
			beforeIntents := snapshotRuntimeOutboundMigrationRows(t, db, `SELECT * FROM _System_RuntimeOutboundIntent ORDER BY Id`)
			if err := VerifyRuntimeOutboundIntentSchema(t.Context(), db); err == nil {
				t.Fatal("legacy schema unexpectedly admits owner-fenced Runtime work")
			}
			for pass := 1; pass <= 2; pass++ {
				createRuntimeOutboundMigrationTables(t, repository, defs.intent, defs.generation)
				if err := VerifyRuntimeOutboundIntentSchema(t.Context(), db); err != nil {
					t.Fatalf("readiness after warm upgrade pass %d: %v", pass, err)
				}
				if after := snapshotRuntimeOutboundMigrationRows(t, db, `SELECT * FROM _System_RuntimeOutboundIntent ORDER BY Id`); !reflect.DeepEqual(beforeIntents, after) {
					t.Fatalf("warm upgrade pass %d changed retained intent data", pass)
				}
			}
		})
	}
}

func TestRuntimeOutboundIntentOwnerIndexWarmUpgradeRejectsIndexDriftOnTiDB(t *testing.T) {
	for _, tc := range []struct{ name, ddl string }{
		{"wrong_order", `ADD INDEX idx_runtime_outbound_intent_owner (SourceRecordId, SourceObject, AdmissionState)`},
		{"unique", `ADD UNIQUE INDEX idx_runtime_outbound_intent_owner (SourceObject, SourceRecordId, AdmissionState)`},
		{"prefix", `ADD INDEX idx_runtime_outbound_intent_owner (SourceObject(16), SourceRecordId, AdmissionState)`},
		{"missing_column", `ADD INDEX idx_runtime_outbound_intent_owner (SourceObject, SourceRecordId)`},
		{"extra_column", `ADD INDEX idx_runtime_outbound_intent_owner (SourceObject, SourceRecordId, AdmissionState, Operation)`},
	} {
		t.Run(tc.name, func(t *testing.T) {
			db, cleanup := openSchemaDDLTiDB(t)
			defer cleanup()
			defs := runtimeOutboundOwnerMigrationCatalog(t)
			repository := NewSchemaRepository(db, nil)
			createRuntimeOutboundMigrationTables(t, repository, defs.legacyIntent, defs.generation)
			insertRuntimeOutboundMigrationRows(t, db)
			if _, err := db.Exec("ALTER TABLE _System_RuntimeOutboundIntent " + tc.ddl); err != nil {
				t.Fatal(err)
			}
			beforeDDL := snapshotRuntimeOutboundMigrationRows(t, db, `SHOW CREATE TABLE _System_RuntimeOutboundIntent`)
			beforeRows := snapshotRuntimeOutboundMigrationRows(t, db, `SELECT * FROM _System_RuntimeOutboundIntent ORDER BY Id`)
			if err := repository.CreatePhysicalTable(t.Context(), defs.intent); err == nil {
				t.Fatal("warm upgrade accepted an incompatible owner index")
			}
			if err := VerifyRuntimeOutboundIntentSchema(t.Context(), db); err == nil {
				t.Fatal("readiness accepted an incompatible owner index")
			}
			if after := snapshotRuntimeOutboundMigrationRows(t, db, `SHOW CREATE TABLE _System_RuntimeOutboundIntent`); !reflect.DeepEqual(beforeDDL, after) {
				t.Fatal("rejected warm upgrade changed the incompatible index")
			}
			if after := snapshotRuntimeOutboundMigrationRows(t, db, `SELECT * FROM _System_RuntimeOutboundIntent ORDER BY Id`); !reflect.DeepEqual(beforeRows, after) {
				t.Fatal("rejected warm upgrade changed retained intents")
			}
		})
	}
}

func TestRuntimeOutboundIntentOwnerIndexWarmUpgradeReportsDDLErrorOnTiDB(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	defs := runtimeOutboundOwnerMigrationCatalog(t)
	repository := NewSchemaRepository(db, nil)
	createRuntimeOutboundMigrationTables(t, repository, defs.legacyIntent)
	if _, err := db.Exec(`ALTER TABLE _System_RuntimeOutboundIntent DROP COLUMN SourceRecordId`); err != nil {
		t.Fatal(err)
	}
	if err := repository.CreatePhysicalTable(t.Context(), defs.intent); err == nil {
		t.Fatal("warm upgrade ignored failure to add its required owner index")
	}
}

type runtimeOutboundOwnerMigrationDefinitions struct {
	intent, legacyIntent, generation schema.TableDefinition
}

func runtimeOutboundOwnerMigrationCatalog(t *testing.T) runtimeOutboundOwnerMigrationDefinitions {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	var result runtimeOutboundOwnerMigrationDefinitions
	for _, definition := range definitions {
		switch definition.TableName {
		case "_System_RuntimeOutboundIntent":
			result.intent, result.legacyIntent = definition, definition
			result.legacyIntent.Indices = nil
			for _, index := range definition.Indices {
				if index.Name != "idx_runtime_outbound_intent_owner" {
					result.legacyIntent.Indices = append(result.legacyIntent.Indices, index)
				}
			}
		case "_System_RuntimeOutboundIntentGeneration":
			result.generation = definition
		}
	}
	if result.intent.TableName == "" || result.generation.TableName == "" ||
		len(result.intent.Indices) != len(result.legacyIntent.Indices)+1 {
		t.Fatal("canonical catalog is missing the outbound intent owner index")
	}
	return result
}

func createRuntimeOutboundMigrationTables(t *testing.T, repository *SchemaRepository, definitions ...schema.TableDefinition) {
	t.Helper()
	for _, definition := range definitions {
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create or upgrade %s: %v", definition.TableName, err)
		}
	}
}

func insertRuntimeOutboundMigrationRows(t *testing.T, db *sql.DB) {
	t.Helper()
	if _, err := db.Exec(`INSERT INTO _System_RuntimeOutboundIntent
(Id, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
IntentId, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation, SourceObject,
SourceRecordId, ConnectorOwnerKey, FlowDeveloperName, FlowActionName, ActionSlotId,
DestinationSlotId, Bindings, CanonicalPayload, AdmissionState, IntegrationCommandId)
VALUES ('retained-intent', 'org-1', 'installation-1', 'integration-1', REPEAT('a', 64), 41,
'intent-1', 'key-1', 1, CONCAT('sha256:', REPEAT('b', 64)), 'UPSERT', 'Record',
'record-1', 'owner-1', 'flow-1', 'action-1', 'slot-1', 'destination-1', JSON_ARRAY(),
?, 'ACCEPTED', 'integration-command-1')`, []byte(`{"retained":"payload"}`)); err != nil {
		t.Fatal(err)
	}
}

func snapshotRuntimeOutboundMigrationRows(t *testing.T, db *sql.DB, query string) [][][]byte {
	t.Helper()
	rows, err := db.Query(query)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	var result [][][]byte
	for rows.Next() {
		values := make([][]byte, len(columns))
		targets := make([]any, len(columns))
		for index := range values {
			targets[index] = &values[index]
		}
		if err := rows.Scan(targets...); err != nil {
			t.Fatal(err)
		}
		result = append(result, values)
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if len(result) == 0 {
		t.Fatal("migration snapshot unexpectedly contains no retained rows")
	}
	return result
}