package persistence

import (
	"bytes"
	"context"
	"strings"
	"testing"
	"time"

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

func TestIntegration_ConnectorV2ExecutionSchemaEnforcesExactGenerationAndFencedAttempts(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()

	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	byName := make(map[string]schema.TableDefinition, len(definitions))
	for _, definition := range definitions {
		byName[definition.TableName] = definition
	}

	const (
		packGenerationTable = "_System_PackGeneration"
		definitionSetTable  = "_System_PackGenerationConnectorBindingDefinitionSet"
		epochTable          = "_System_ConnectorServingEpoch"
		bindingTable        = "_System_ConnectorServingEpochBinding"
		bindingOwnerTable   = "_System_ConnectorServingEpochBindingOwner"
		gateTable           = "_System_ConnectorAccountServingGate"
		executionTable      = "_System_ConnectorV2Execution"
		attemptTable        = "_System_ConnectorV2ExecutionAttempt"
	)
	tableNames := []string{
		packGenerationTable,
		definitionSetTable,
		epochTable,
		bindingTable,
		bindingOwnerTable,
		gateTable,
		executionTable,
		attemptTable,
	}
	repository := NewSchemaRepository(db, nil)
	for pass := 1; pass <= 2; pass++ {
		for _, tableName := range tableNames {
			definition, ok := byName[tableName]
			if !ok {
				t.Fatalf("%s absent from canonical schema catalog", tableName)
			}
			if err := repository.CreatePhysicalTable(context.Background(), definition); err != nil {
				t.Fatalf("CreatePhysicalTable(%s), pass %d: %v", tableName, pass, err)
			}
		}
	}

	for _, expected := range []packGenerationConnectorBindingColumnShape{
		{table: executionTable, column: "Id", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: executionTable, column: "PackGenerationId", columnType: "varchar(255)", collation: "utf8mb4_unicode_ci", nullable: false},
		{table: executionTable, column: "CanonicalInputs", columnType: "mediumblob", nullable: false},
		{table: executionTable, column: "ResultDigest", columnType: "char(71)", collation: "utf8mb4_bin", nullable: true},
		{table: executionTable, column: "RuntimeCommitReceipt", columnType: "char(71)", collation: "utf8mb4_bin", nullable: true},
		{table: attemptTable, column: "Id", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: attemptTable, column: "ExecutionId", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: attemptTable, column: "ClaimHolderId", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
	} {
		requirePackGenerationConnectorBindingColumnShape(t, db, expected)
	}

	for _, expected := range []packGenerationConnectorBindingForeignKey{
		{table: executionTable, column: "AccountServingGateId", parentTable: gateTable, parentColumn: "Id"},
		{table: executionTable, column: "ServingEpochId", parentTable: epochTable, parentColumn: "Id"},
		{table: executionTable, column: "ServingEpochBindingId", parentTable: bindingOwnerTable, parentColumn: "ServingEpochBindingId"},
		{table: executionTable, column: "PackGenerationId", parentTable: definitionSetTable, parentColumn: "PackGenerationId"},
		{table: attemptTable, column: "ExecutionId", parentTable: executionTable, parentColumn: "Id"},
	} {
		requirePackGenerationConnectorBindingForeignKey(t, db, expected)
	}

	for _, expected := range []struct {
		table   string
		name    string
		unique  bool
		columns []string
	}{
		{executionTable, "PRIMARY", true, []string{"Id"}},
		{executionTable, "ux_connector_v2_execution_ref", true, []string{"ExecutionRefDigest"}},
		{executionTable, "ux_connector_v2_execution_runtime_command", true, []string{"RuntimeCommandId"}},
		{executionTable, "idx_connector_v2_execution_claim", false, []string{"ConnectorAccountId", "State", "ClaimLeaseUntil", "Id"}},
		{executionTable, "idx_connector_v2_execution_epoch", false, []string{"ServingEpochId", "State", "Id"}},
		{executionTable, "idx_connector_v2_execution_gate", false, []string{"AccountServingGateId", "State", "Id"}},
		{attemptTable, "PRIMARY", true, []string{"Id"}},
		{attemptTable, "ux_connector_v2_execution_attempt_ordinal", true, []string{"ExecutionId", "AttemptOrdinal"}},
		{attemptTable, "ux_connector_v2_execution_attempt_fence", true, []string{"ExecutionId", "ClaimFenceToken"}},
		{attemptTable, "idx_connector_v2_execution_attempt_holder", false, []string{"ClaimHolderId", "LeaseUntil", "Id"}},
	} {
		requireConnectorV2IngressReceiptIndex(t, db, expected.table, expected.name, expected.unique, expected.columns)
	}

	digest := func(value string) string { return "sha256:" + strings.Repeat(value, 64) }
	const (
		packGenerationID = "pack-generation-1"
		packInstallID    = "pack-installation-1"
		epochID          = "v2-execution-epoch"
		bindingID        = "v2-execution-binding"
		gateID           = "v2-execution-gate"
		accountID        = "connector-account-1"
		scope            = "1111111111111111111111111111111111111111111111111111111111111111"
	)
	packPayloadDigest := "nexus-pack-sha256-v1:" + strings.Repeat("1", 64)
	if _, err := db.Exec(
		`INSERT INTO _System_PackGeneration (Id, PackInstallId, PackName, Category)
		 VALUES (?, ?, ?, ?)`,
		packGenerationID, packInstallID, "connector-pack", "ConnectorPack",
	); err != nil {
		t.Fatalf("insert PackGeneration: %v", err)
	}
	if _, err := db.Exec(
		`INSERT INTO _System_PackGenerationConnectorBindingDefinitionSet
		 (PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest)
		 VALUES (?, ?, ?, ?)`,
		packGenerationID,
		"nexuscrm.connector-binding-owner-definition-set/v1",
		uint64(1),
		digest("1"),
	); err != nil {
		t.Fatalf("insert binding-definition set: %v", err)
	}
	if _, err := db.Exec(
		`INSERT INTO _System_ConnectorServingEpoch (
			Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State,
			OrgId, AppInstallationId, AuthorizedWorkloadId,
			CatalogRevision, SchemaGeneration, EngineGeneration,
			SourceSnapshotDigest, BindingViewDigest,
			ReceiptProtocolVersion, ReceiptCapabilityVersion, SecurityEpoch,
			L2ConfigurationGenerationRef, L2ConfigurationDigest,
			L2VerificationReceipt, L2VerificationReceiptDigest
		 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		epochID, digest("2"), scope, scope, "active",
		"org-1", "app-installation-1", "workload-1",
		uint64(11), uint64(12), uint64(13),
		digest("3"), digest("4"),
		"1.0", "catalog.consumer.integration-binding-view.v1", uint64(14),
		"service-configuration-1", digest("5"),
		[]byte("service-configuration-owner-receipt"), digest("6"),
	); err != nil {
		t.Fatalf("insert active ServingEpoch: %v", err)
	}
	if _, err := db.Exec(
		`INSERT INTO _System_ConnectorServingEpochBinding (
			Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
			Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
			BusinessKey, DefinitionVersion, CredentialRefs,
			RuntimeCommandKind, RuntimePayloadFormatVersion,
			RuntimeDefinitionRef, RuntimeDefinitionVersion
		 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		bindingID, epochID, uint64(0), strings.Repeat("7", 64),
		"CONNECTOR_INBOUND_ROUTE", "connector-pack", packInstallID, packPayloadDigest,
		"route-1", uint64(1), []byte("credential-1"),
		"runtime.flow.execute", uint64(3), "flow-definition-1", "1",
	); err != nil {
		t.Fatalf("insert ServingEpoch binding: %v", err)
	}
	if _, err := db.Exec(
		`INSERT INTO _System_ConnectorServingEpochBindingOwner (
			ServingEpochBindingId, PackGenerationId,
			DefinitionEnvelopeVersion, DefinitionReceiptVersion,
			CanonicalDefinitionDigest, DefinitionOwnerReceipt, DefinitionOwnerReceiptDigest
		 ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
		bindingID, packGenerationID,
		"nexuscrm.connector-binding-owner-definition/v1",
		"nexuscrm.connector-binding-owner-receipt/v1",
		digest("8"), []byte("binding-definition-owner-receipt"), digest("9"),
	); err != nil {
		t.Fatalf("insert ServingEpoch binding owner: %v", err)
	}
	if _, err := db.Exec(
		`INSERT INTO _System_ConnectorAccountServingGate (
			Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
			ServingEpochId, ServingEpochDigest
		 ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
		gateID, scope, accountID, "v2_serving", int64(7), epochID, digest("2"),
	); err != nil {
		t.Fatalf("insert serving AccountServingGate: %v", err)
	}

	canonicalInputs := bytes.Repeat([]byte(`{"record":{"Name":"Ada"}}`), 4_096)
	if len(canonicalInputs) <= 64*1024 {
		t.Fatalf("canonical input probe is only %d bytes", len(canonicalInputs))
	}
	insertExecution := `INSERT INTO _System_ConnectorV2Execution (
		Id, ExecutionProtocolVersion, ServingScopeDigest, ConnectorAccountId,
		AccountServingGateId, AccountServingFenceToken,
		ServingEpochId, ServingEpochDigest, ServingEpochBindingId, PackGenerationId,
		DefinitionRef, DefinitionVersion, CanonicalInputs, ExternalEffectsMode,
		InputDigest, ExecutionRefDigest, RuntimeCommandId
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	executionArgs := func(id, executionRefDigest, runtimeCommandID string, inputs []byte) []any {
		return []any{
			id,
			"nexuscrm.connector-v2-execution.v1",
			scope,
			accountID,
			gateID,
			int64(7),
			epochID,
			digest("2"),
			bindingID,
			packGenerationID,
			"flow-definition-1",
			"1",
			inputs,
			"SUPPRESS_EXTERNAL_EFFECTS",
			digest("a"),
			executionRefDigest,
			runtimeCommandID,
		}
	}
	if _, err := db.Exec(
		insertExecution,
		executionArgs("v2-execution-1", digest("b"), "runtime-command-1", canonicalInputs)...,
	); err != nil {
		t.Fatalf("insert Connector V2 Execution: %v", err)
	}
	if _, err := db.Exec(
		insertExecution,
		executionArgs("V2-Execution-1", digest("c"), "Runtime-Command-1", []byte(`{}`))...,
	); err != nil {
		t.Fatalf("insert byte-distinct execution and Runtime command identities: %v", err)
	}

	var (
		storedInputs      []byte
		storedInputBytes  int
		storedState       string
		storedClaimFence  int64
		storedResult      *string
		storedCommitProof *string
	)
	if err := db.QueryRow(
		`SELECT CanonicalInputs, OCTET_LENGTH(CanonicalInputs), State, ClaimFenceToken,
		        ResultDigest, RuntimeCommitReceipt
		   FROM _System_ConnectorV2Execution WHERE Id = ?`,
		"v2-execution-1",
	).Scan(
		&storedInputs,
		&storedInputBytes,
		&storedState,
		&storedClaimFence,
		&storedResult,
		&storedCommitProof,
	); err != nil {
		t.Fatalf("read Connector V2 Execution: %v", err)
	}
	if !bytes.Equal(storedInputs, canonicalInputs) || storedInputBytes != len(canonicalInputs) {
		t.Fatalf("stored CanonicalInputs = %d/%d bytes, want exact %d bytes", len(storedInputs), storedInputBytes, len(canonicalInputs))
	}
	if storedState != "ready" || storedClaimFence != 0 || storedResult != nil || storedCommitProof != nil {
		t.Fatalf(
			"new execution state = %q fence=%d result=%v receipt=%v, want ready/0/nil/nil",
			storedState,
			storedClaimFence,
			storedResult,
			storedCommitProof,
		)
	}

	duplicateRef := executionArgs("v2-execution-duplicate-ref", digest("b"), "runtime-command-2", []byte(`{}`))
	_, err = db.Exec(insertExecution, duplicateRef...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1062, "duplicate ExecutionRefDigest")
	duplicateCommand := executionArgs("v2-execution-duplicate-command", digest("d"), "runtime-command-1", []byte(`{}`))
	_, err = db.Exec(insertExecution, duplicateCommand...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1062, "duplicate RuntimeCommandId")

	for _, missing := range []struct {
		name  string
		index int
		value string
	}{
		{name: "AccountServingGate", index: 4, value: "missing-gate"},
		{name: "ServingEpoch", index: 6, value: "missing-epoch"},
		{name: "ServingEpochBindingOwner", index: 8, value: "missing-binding-owner"},
		{name: "PackGeneration definition set", index: 9, value: "missing-pack-generation"},
	} {
		args := executionArgs("missing-"+strings.ReplaceAll(strings.ToLower(missing.name), " ", "-"), digest("e"), "missing-"+strings.ReplaceAll(strings.ToLower(missing.name), " ", "-"), []byte(`{}`))
		args[missing.index] = missing.value
		_, err = db.Exec(insertExecution, args...)
		requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "execution for missing "+missing.name)
	}

	insertAttempt := `INSERT INTO _System_ConnectorV2ExecutionAttempt (
		Id, ExecutionId, AttemptOrdinal, ClaimFenceToken, ClaimHolderId, LeaseUntil
	) VALUES (?, ?, ?, ?, ?, ?)`
	leaseUntil := time.Now().UTC().Add(time.Minute).Truncate(time.Microsecond)
	if _, err := db.Exec(insertAttempt, "execution-attempt-1", "v2-execution-1", int64(1), int64(1), "worker-1", leaseUntil); err != nil {
		t.Fatalf("insert execution attempt: %v", err)
	}
	_, err = db.Exec(insertAttempt, "execution-attempt-duplicate-ordinal", "v2-execution-1", int64(1), int64(2), "worker-2", leaseUntil)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1062, "duplicate execution attempt ordinal")
	_, err = db.Exec(insertAttempt, "execution-attempt-duplicate-fence", "v2-execution-1", int64(2), int64(1), "worker-2", leaseUntil)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1062, "duplicate execution claim fence")
	_, err = db.Exec(insertAttempt, "execution-attempt-missing-execution", "missing-execution", int64(1), int64(1), "worker-1", leaseUntil)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "attempt for missing execution")

	for _, deletion := range []struct {
		query    string
		argument string
		name     string
	}{
		{`DELETE FROM _System_ConnectorV2Execution WHERE Id = ?`, "v2-execution-1", "execution with retained attempt"},
		{`DELETE FROM _System_ConnectorAccountServingGate WHERE Id = ?`, gateID, "referenced AccountServingGate"},
		{`DELETE FROM _System_ConnectorServingEpochBindingOwner WHERE ServingEpochBindingId = ?`, bindingID, "referenced ServingEpochBindingOwner"},
		{`DELETE FROM _System_ConnectorServingEpoch WHERE Id = ?`, epochID, "referenced ServingEpoch"},
	} {
		_, err = db.Exec(deletion.query, deletion.argument)
		requirePackGenerationConnectorBindingMySQLError(t, err, 1451, "delete "+deletion.name)
	}
}