package persistence

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

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

func TestIntegrationOutboundDependencyWarmUpgradeOnTiDB(t *testing.T) {
	for _, partial := range []bool{false, true} {
		name := "legacy"
		if partial {
			name = "resume_after_first_column"
		}
		t.Run(name, func(t *testing.T) {
			db, cleanup := openSchemaDDLTiDB(t)
			defer cleanup()
			canonical, legacy := integrationOutboundDependencyMigrationDefinitions(t)
			repository := NewSchemaRepository(db, nil)
			if err := repository.createPhysicalTable(t.Context(), legacy, true); err != nil {
				t.Fatalf("create legacy outbound command table: %v", err)
			}
			if partial {
				if _, err := db.Exec(`ALTER TABLE _System_IntegrationOutboundCommand ADD COLUMN DependencyKind VARCHAR(64) NULL`); err != nil {
					t.Fatal(err)
				}
			}
			insertIntegrationOutboundMigrationRow(t, db)
			before := integrationOutboundMigrationRow(t, db)

			for pass := 1; pass <= 2; pass++ {
				if err := repository.CreatePhysicalTable(t.Context(), canonical); err != nil {
					t.Fatalf("warm upgrade pass %d: %v", pass, err)
				}
				if after := integrationOutboundMigrationRow(t, db); !reflect.DeepEqual(before, after) {
					t.Fatalf("warm upgrade pass %d changed retained command: before=%#v after=%#v", pass, before, after)
				}
				assertIntegrationOutboundDependencyPhysicalSchema(t, db)
			}
		})
	}
}

func TestIntegrationOutboundDependencyWarmUpgradeRejectsColumnDriftBeforeMutationOnTiDB(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	canonical, legacy := integrationOutboundDependencyMigrationDefinitions(t)
	repository := NewSchemaRepository(db, nil)
	if err := repository.createPhysicalTable(t.Context(), legacy, true); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ALTER TABLE _System_IntegrationOutboundCommand ADD COLUMN DependencyKey VARCHAR(255) COLLATE utf8mb4_general_ci NULL`); err != nil {
		t.Fatal(err)
	}
	if err := repository.CreatePhysicalTable(t.Context(), canonical); err == nil || !strings.Contains(err.Error(), "DependencyKey") {
		t.Fatalf("warm upgrade accepted incompatible DependencyKey: %v", err)
	}
	for _, column := range []string{"DependencyKind", "DependencyReadinessKeys", "DependencyBlockedAt"} {
		if _, found, err := inspectIntegrationOutboundDependencyColumn(t.Context(), db, canonical.TableName, column); err != nil {
			t.Fatal(err)
		} else if found {
			t.Fatalf("failed preflight added %s", column)
		}
	}
	columns, _, _, err := inspectManagedPhysicalIndex(t.Context(), db, canonical.TableName, integrationOutboundDependencyIndex)
	if err != nil {
		t.Fatal(err)
	}
	if len(columns) != 0 {
		t.Fatalf("failed preflight added dependency index: %v", columns)
	}
}

func TestIntegrationOutboundDependencyWarmUpgradeRejectsIndexDriftOnTiDB(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	canonical, legacy := integrationOutboundDependencyMigrationDefinitions(t)
	repository := NewSchemaRepository(db, nil)
	if err := repository.createPhysicalTable(t.Context(), legacy, true); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ALTER TABLE _System_IntegrationOutboundCommand
ADD COLUMN DependencyKind VARCHAR(64) NULL,
ADD COLUMN DependencyKey VARCHAR(255) COLLATE utf8mb4_bin NULL,
	ADD COLUMN DependencyReadinessKeys JSON NULL,
	ADD COLUMN DependencyBlockedAt DATETIME(6) NULL`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ALTER TABLE _System_IntegrationOutboundCommand
ADD INDEX idx_integration_outbound_dependency (DependencyKey, ScopeDigest, DependencyKind, State, Id)`); err != nil {
		t.Fatal(err)
	}
	var tableName, before string
	if err := db.QueryRow(`SHOW CREATE TABLE _System_IntegrationOutboundCommand`).Scan(&tableName, &before); err != nil {
		t.Fatal(err)
	}
	if err := repository.CreatePhysicalTable(t.Context(), canonical); err == nil || !strings.Contains(err.Error(), "dependency index") {
		t.Fatalf("warm upgrade accepted incompatible dependency index: %v", err)
	}
	var after string
	if err := db.QueryRow(`SHOW CREATE TABLE _System_IntegrationOutboundCommand`).Scan(&tableName, &after); err != nil {
		t.Fatal(err)
	}
	if before != after {
		t.Fatal("failed warm upgrade changed incompatible physical schema")
	}
}

func integrationOutboundDependencyMigrationDefinitions(t *testing.T) (schema.TableDefinition, schema.TableDefinition) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	for _, definition := range definitions {
		if definition.TableName != integrationOutboundCommandTable {
			continue
		}
		legacy := definition
		legacy.Columns = nil
		for _, column := range definition.Columns {
			if !containsExactString(integrationOutboundDependencyColumnNames, column.Name) {
				legacy.Columns = append(legacy.Columns, column)
			}
		}
		legacy.Indices = nil
		for _, index := range definition.Indices {
			if index.Name != integrationOutboundDependencyIndex && index.Name != integrationOutboundReconcileIndex {
				legacy.Indices = append(legacy.Indices, index)
			}
		}
		return definition, legacy
	}
	t.Fatal("canonical catalog is missing Integration outbound command table")
	return schema.TableDefinition{}, schema.TableDefinition{}
}

func containsExactString(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}

func insertIntegrationOutboundMigrationRow(t *testing.T, db *sql.DB) {
	t.Helper()
	_, err := db.Exec(`INSERT INTO _System_IntegrationOutboundCommand
(Id, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
 IntentId, LinkKey, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation,
 SourceObject, SourceRecordId, ConnectorOwnerKey, FlowDeveloperName, FlowActionName,
 ActionSlotId, DestinationSlotId, Bindings, CanonicalPayload)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, JSON_ARRAY(), ?)`,
		"retained-command", "org-1", "installation-1", "workload-1", strings.Repeat("a", 64), 27,
		"intent-1", "link-1", "idempotency-1", 1, "sha256:"+strings.Repeat("b", 64), "UPSERT",
		"Record", "record-1", "owner-1", "flow-1", "action-1", "slot-1", "destination-1",
		[]byte(`{"retained":true}`),
	)
	if err != nil {
		t.Fatal(err)
	}
}

func integrationOutboundMigrationRow(t *testing.T, db *sql.DB) []byte {
	t.Helper()
	var payload []byte
	if err := db.QueryRow(`SELECT CanonicalPayload FROM _System_IntegrationOutboundCommand WHERE Id = 'retained-command'`).Scan(&payload); err != nil {
		t.Fatal(err)
	}
	return payload
}

func assertIntegrationOutboundDependencyPhysicalSchema(t *testing.T, db *sql.DB) {
	t.Helper()
	canonical, _ := integrationOutboundDependencyMigrationDefinitions(t)
	columns, indexes, err := integrationOutboundDependencyDeclaration(canonical)
	if err != nil {
		t.Fatal(err)
	}
	for _, column := range columns {
		physical, found, err := inspectIntegrationOutboundDependencyColumn(t.Context(), db, canonical.TableName, column.Name)
		if err != nil || !found {
			t.Fatalf("inspect %s: found=%t err=%v", column.Name, found, err)
		}
		if err := validateIntegrationOutboundDependencyColumn(column, physical); err != nil {
			t.Fatal(err)
		}
	}
	for _, index := range indexes {
		if err := VerifyExactTableIndex(t.Context(), db, canonical.TableName, index.Name, false, index.Columns); err != nil {
			t.Fatal(err)
		}
	}
	var nonNull int
	if err := db.QueryRow(`SELECT (DependencyKind IS NOT NULL) + (DependencyKey IS NOT NULL) +
       (DependencyReadinessKeys IS NOT NULL) + (DependencyBlockedAt IS NOT NULL)
FROM _System_IntegrationOutboundCommand WHERE Id = 'retained-command'`).Scan(&nonNull); err != nil {
		t.Fatal(err)
	}
	if nonNull != 0 {
		t.Fatalf("new dependency columns on retained row are non-NULL: %d", nonNull)
	}
}