package persistence

import (
	"bytes"
	"context"
	"database/sql"
	"strings"
	"testing"

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

func TestIntegration_ConnectorServingEpochReproducibilitySchemaRetainsExactOwnerProofs(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"
		credentialGenerationTable = "_System_ConnectorCredentialGeneration"
		serviceGenerationTable    = "_System_ConnectorServiceConfigurationGeneration"
		epochTable                = "_System_ConnectorServingEpoch"
		bindingTable              = "_System_ConnectorServingEpochBinding"
		reproducibilityTable      = "_System_ConnectorServingEpochReproducibility"
		bindingOwnerTable         = "_System_ConnectorServingEpochBindingOwner"
		credentialProofTable      = "_System_ConnectorServingEpochCredentialGeneration"
	)

	repository := NewSchemaRepository(db, nil)
	createTables := func(pass int, names ...string) {
		t.Helper()
		for _, name := range names {
			definition, ok := byName[name]
			if !ok {
				t.Fatalf("%s absent from canonical schema catalog", name)
			}
			if err := repository.CreatePhysicalTable(context.Background(), definition); err != nil {
				t.Fatalf("CreatePhysicalTable(%s), pass %d: %v", name, pass, err)
			}
		}
	}

	// Reproduce an installed pre-extension database first. The additive tables
	// must not reinterpret or rewrite an existing unprofiled prepared candidate.
	createTables(1,
		packGenerationTable,
		definitionSetTable,
		credentialGenerationTable,
		serviceGenerationTable,
		epochTable,
		bindingTable,
	)

	digest := func(value string) string { return "sha256:" + strings.Repeat(value, 64) }
	packPayloadDigest := "nexus-pack-sha256-v1:" + strings.Repeat("1", 64)

	if _, err := db.Exec(
		`INSERT INTO _System_PackGeneration (Id, PackInstallId, PackName, Category)
		 VALUES (?, ?, ?, ?)`,
		"pack-generation-1", "pack-installation-1", "connector-pack", "ConnectorPack",
	); err != nil {
		t.Fatalf("insert PackGeneration: %v", err)
	}
	if _, err := db.Exec(
		`INSERT INTO _System_PackGenerationConnectorBindingDefinitionSet
		 (PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest)
		 VALUES (?, ?, ?, ?)`,
		"pack-generation-1",
		"nexuscrm.connector-binding-owner-definition-set/v1",
		uint64(1),
		digest("a"),
	); err != nil {
		t.Fatalf("insert binding-definition set: %v", err)
	}

	credentialReceipt := bytes.Repeat([]byte("credential-owner-receipt\x00"), 4_096)
	if len(credentialReceipt) <= 64*1024 {
		t.Fatalf("credential owner receipt probe is only %d bytes", len(credentialReceipt))
	}
	if _, err := db.Exec(
		`INSERT INTO _System_ConnectorCredentialGeneration (
			Id, StableCredentialIdentity, ProviderScopeId, ConnectorAccountId,
			SecretBackendProtocol, SecretBackendVersionRef,
			TransportConfiguration, TransportConfigurationDigest,
			ReceiptProtocolVersion, OwnerReceipt, OwnerReceiptDigest
		 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		"credential-generation-1",
		"credential-1",
		"provider-scope-1",
		"connector-account-1",
		"nexuscrm.secret-backend.v1",
		"secret-version-1",
		[]byte("non-secret-transport-configuration"),
		digest("b"),
		"nexuscrm.connector-cred-gen.v1",
		credentialReceipt,
		digest("c"),
	); err != nil {
		t.Fatalf("insert CredentialGeneration: %v", err)
	}

	serviceReceipt := []byte("service-configuration-owner-receipt-v1")
	if _, err := db.Exec(
		`INSERT INTO _System_ConnectorServiceConfigurationGeneration (
			Id, GenerationProtocolVersion, OrgId, AppInstallationId, AuthorizedWorkloadId,
			CatalogRevision, SchemaGeneration, EngineGeneration,
			SourceSnapshotDigest, CatalogViewDigest,
			CatalogReceiptProtocolVersion, CatalogReceiptCapabilityVersion, CatalogSecurityEpoch,
			ConfigurationOwnerRevisionId, ConfigurationOwnerRevisionDigest,
			GenerationDigest, ReceiptProtocolVersion, OwnerReceipt, OwnerReceiptDigest,
			AccountCount, BindingCount, CredentialSelectionCount
		 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		"service-configuration-1",
		"nexuscrm.connector-service-configuration-generation/v1",
		"org-1",
		"app-installation-1",
		"workload-1",
		uint64(11), uint64(12), uint64(13),
		digest("d"), digest("e"),
		"1.0", "catalog.consumer.integration-binding-view.v1", uint64(14),
		"owner-revision-1", digest("f"),
		digest("1"),
		"nexuscrm.connector-service-configuration-owner-receipt/v1",
		serviceReceipt,
		digest("2"),
		uint64(1), uint64(1), uint64(1),
	); err != nil {
		t.Fatalf("insert service-configuration generation: %v", err)
	}

	insertEpoch := `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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	scope := strings.Repeat("3", 64)
	epochArgs := func(id, epochDigest, l2ID, l2Digest string, l2Receipt []byte, l2ReceiptDigest string) []any {
		return []any{
			id, epochDigest, scope, nil, "prepared",
			"org-1", "app-installation-1", "workload-1",
			uint64(11), uint64(12), uint64(13),
			digest("d"), digest("e"),
			"1.0", "catalog.consumer.integration-binding-view.v1", uint64(14),
			l2ID, l2Digest, l2Receipt, l2ReceiptDigest,
		}
	}
	legacyReceipt := []byte("legacy-unprofiled-l2-receipt")
	if _, err := db.Exec(
		insertEpoch,
		epochArgs("legacy-epoch", digest("4"), "legacy-l2-generation", digest("5"), legacyReceipt, digest("6"))...,
	); err != nil {
		t.Fatalf("insert legacy ServingEpoch: %v", err)
	}
	if _, err := db.Exec(
		insertEpoch,
		epochArgs("profiled-epoch", digest("7"), "service-configuration-1", digest("1"), serviceReceipt, digest("2"))...,
	); err != nil {
		t.Fatalf("insert profiled ServingEpoch parent: %v", err)
	}

	insertBinding := `INSERT INTO _System_ConnectorServingEpochBinding (
		Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
		Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
		BusinessKey, DefinitionVersion, CredentialRefs
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	bindingArgs := func(id, epochID, identityDigest string) []any {
		return []any{
			id, epochID, uint64(0), identityDigest,
			"CONNECTOR_INBOUND_ROUTE", "connector-pack", "pack-installation-1",
			packPayloadDigest, "route-1", uint64(1), []byte("credential-1"),
		}
	}
	if _, err := db.Exec(insertBinding, bindingArgs("legacy-binding", "legacy-epoch", strings.Repeat("8", 64))...); err != nil {
		t.Fatalf("insert legacy ServingEpoch binding: %v", err)
	}
	if _, err := db.Exec(insertBinding, bindingArgs("profiled-binding", "profiled-epoch", strings.Repeat("9", 64))...); err != nil {
		t.Fatalf("insert profiled ServingEpoch binding: %v", err)
	}

	for pass := 1; pass <= 2; pass++ {
		createTables(pass, reproducibilityTable, bindingOwnerTable, credentialProofTable)
	}

	var (
		storedLegacyID      string
		storedLegacyReceipt []byte
		legacyHeaderCount   int
	)
	if err := db.QueryRow(
		`SELECT L2ConfigurationGenerationRef, L2VerificationReceipt
		   FROM _System_ConnectorServingEpoch WHERE Id = ?`,
		"legacy-epoch",
	).Scan(&storedLegacyID, &storedLegacyReceipt); err != nil {
		t.Fatalf("read legacy ServingEpoch after additive DDL: %v", err)
	}
	if storedLegacyID != "legacy-l2-generation" || !bytes.Equal(storedLegacyReceipt, legacyReceipt) {
		t.Fatalf("legacy ServingEpoch changed across additive DDL: id=%q receipt=%x", storedLegacyID, storedLegacyReceipt)
	}
	if err := db.QueryRow(
		`SELECT COUNT(*) FROM _System_ConnectorServingEpochReproducibility WHERE ServingEpochId = ?`,
		"legacy-epoch",
	).Scan(&legacyHeaderCount); err != nil {
		t.Fatalf("count legacy reproducibility headers: %v", err)
	}
	if legacyHeaderCount != 0 {
		t.Fatalf("legacy reproducibility header count = %d, want 0", legacyHeaderCount)
	}

	for _, expected := range []packGenerationConnectorBindingColumnShape{
		{table: reproducibilityTable, column: "ServingEpochId", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: reproducibilityTable, column: "ConnectorDigestProfileId", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: bindingOwnerTable, column: "ServingEpochBindingId", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: bindingOwnerTable, column: "PackGenerationId", columnType: "varchar(255)", collation: "utf8mb4_unicode_ci", nullable: false},
		{table: bindingOwnerTable, column: "DefinitionOwnerReceipt", columnType: "mediumblob", nullable: false},
		{table: credentialProofTable, column: "ServingEpochId", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: credentialProofTable, column: "CredentialGenerationId", columnType: "varchar(255)", collation: "utf8mb4_bin", nullable: false},
		{table: credentialProofTable, column: "OwnerReceipt", columnType: "mediumblob", nullable: false},
	} {
		requirePackGenerationConnectorBindingColumnShape(t, db, expected)
	}

	for _, expected := range []packGenerationConnectorBindingForeignKey{
		{table: reproducibilityTable, column: "ServingEpochId", parentTable: epochTable, parentColumn: "Id"},
		{table: bindingOwnerTable, column: "ServingEpochBindingId", parentTable: bindingTable, parentColumn: "Id"},
		{table: bindingOwnerTable, column: "PackGenerationId", parentTable: definitionSetTable, parentColumn: "PackGenerationId"},
		{table: credentialProofTable, column: "ServingEpochId", parentTable: epochTable, parentColumn: "Id"},
		{table: credentialProofTable, column: "CredentialGenerationId", parentTable: credentialGenerationTable, parentColumn: "Id"},
	} {
		requirePackGenerationConnectorBindingForeignKey(t, db, expected)
	}

	for _, expected := range []connectorServingEpochReproducibilityIndex{
		{table: reproducibilityTable, name: "PRIMARY", columns: "ServingEpochId", unique: true},
		{table: reproducibilityTable, name: "idx_connector_epoch_repro_profile", columns: "ConnectorDigestProfileId,ServingEpochId"},
		{table: bindingOwnerTable, name: "PRIMARY", columns: "ServingEpochBindingId", unique: true},
		{table: bindingOwnerTable, name: "idx_connector_epoch_binding_owner_pack", columns: "PackGenerationId,ServingEpochBindingId"},
		{table: credentialProofTable, name: "PRIMARY", columns: "ServingEpochId,CredentialOrdinal", unique: true},
		{table: credentialProofTable, name: "ux_connector_epoch_credential_generation", columns: "ServingEpochId,CredentialGenerationId", unique: true},
		{table: credentialProofTable, name: "idx_connector_epoch_credential_generation_ref", columns: "CredentialGenerationId,ServingEpochId"},
	} {
		requireConnectorServingEpochReproducibilityIndex(t, db, expected)
	}

	insertHeader := `INSERT INTO _System_ConnectorServingEpochReproducibility (
		ServingEpochId, ConnectorDigestProfileId,
		ServiceConfigurationGenerationProtocolVersion,
		ServiceConfigurationReceiptProtocolVersion
	) VALUES (?, ?, ?, ?)`
	headerArgs := func(epochID string) []any {
		return []any{
			epochID,
			"nexuscrm.connector-digest-profile.v2",
			"nexuscrm.connector-service-configuration-generation/v1",
			"nexuscrm.connector-service-configuration-owner-receipt/v1",
		}
	}
	if _, err := db.Exec(insertHeader, headerArgs("profiled-epoch")...); err != nil {
		t.Fatalf("insert reproducibility header: %v", err)
	}
	_, err = db.Exec(insertHeader, headerArgs("missing-epoch")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "header for missing ServingEpoch")
	_, err = db.Exec(insertHeader, headerArgs("PROFILED-EPOCH")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "header for an unpersisted case variant")

	insertBindingOwner := `INSERT INTO _System_ConnectorServingEpochBindingOwner (
		ServingEpochBindingId, PackGenerationId,
		DefinitionEnvelopeVersion, DefinitionReceiptVersion,
		CanonicalDefinitionDigest, DefinitionOwnerReceipt, DefinitionOwnerReceiptDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?)`
	bindingOwnerArgs := func(bindingID, generationID string) []any {
		return []any{
			bindingID, generationID,
			"nexuscrm.connector-binding-owner-definition/v1",
			"nexuscrm.connector-binding-owner-receipt/v1",
			digest("a"), []byte("binding-definition-owner-receipt"), digest("b"),
		}
	}
	if _, err := db.Exec(
		insertBindingOwner,
		bindingOwnerArgs("profiled-binding", "pack-generation-1")...,
	); err != nil {
		t.Fatalf("insert binding-owner proof: %v", err)
	}
	if _, err := db.Exec(
		insertBinding,
		[]any{
			"profiled-binding-without-owner", "profiled-epoch", uint64(1), strings.Repeat("a", 64),
			"CONNECTOR_INBOUND_ROUTE", "connector-pack", "pack-installation-1",
			packPayloadDigest, "route-2", uint64(2), []byte("credential-1"),
		}...,
	); err != nil {
		t.Fatalf("insert second profiled ServingEpoch binding: %v", err)
	}
	_, err = db.Exec(insertBindingOwner, bindingOwnerArgs("missing-binding", "pack-generation-1")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "binding-owner proof for missing binding")
	_, err = db.Exec(insertBindingOwner, bindingOwnerArgs("profiled-binding-without-owner", "missing-pack-generation")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "binding-owner proof for missing definition set")
	_, err = db.Exec(insertBindingOwner, bindingOwnerArgs("profiled-binding", "pack-generation-1")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1062, "duplicate binding-owner proof")

	insertCredentialProof := `INSERT INTO _System_ConnectorServingEpochCredentialGeneration (
		ServingEpochId, CredentialOrdinal, CredentialGenerationId,
		ReceiptProtocolVersion, TransportConfigurationDigest,
		OwnerReceipt, OwnerReceiptDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?)`
	credentialProofArgs := func(epochID string, ordinal uint64, generationID string) []any {
		return []any{
			epochID, ordinal, generationID,
			"nexuscrm.connector-cred-gen.v1", digest("b"), credentialReceipt, digest("c"),
		}
	}
	if _, err := db.Exec(
		insertCredentialProof,
		credentialProofArgs("profiled-epoch", 0, "credential-generation-1")...,
	); err != nil {
		t.Fatalf("insert credential-generation proof: %v", err)
	}
	_, err = db.Exec(insertCredentialProof, credentialProofArgs("missing-epoch", 0, "credential-generation-1")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "credential proof for missing ServingEpoch")
	_, err = db.Exec(insertCredentialProof, credentialProofArgs("profiled-epoch", 1, "missing-credential-generation")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1452, "credential proof for missing generation")
	_, err = db.Exec(insertCredentialProof, credentialProofArgs("profiled-epoch", 0, "credential-generation-1")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1062, "duplicate credential ordinal")
	_, err = db.Exec(insertCredentialProof, credentialProofArgs("profiled-epoch", 1, "credential-generation-1")...)
	requirePackGenerationConnectorBindingMySQLError(t, err, 1062, "duplicate credential generation")

	var (
		storedCredentialReceipt []byte
		storedReceiptBytes      int
	)
	if err := db.QueryRow(
		`SELECT OwnerReceipt, OCTET_LENGTH(OwnerReceipt)
		   FROM _System_ConnectorServingEpochCredentialGeneration
		  WHERE ServingEpochId = ? AND CredentialOrdinal = ?`,
		"profiled-epoch", uint64(0),
	).Scan(&storedCredentialReceipt, &storedReceiptBytes); err != nil {
		t.Fatalf("read exact credential-generation proof: %v", err)
	}
	if storedReceiptBytes != len(credentialReceipt) || !bytes.Equal(storedCredentialReceipt, credentialReceipt) {
		t.Fatalf("stored credential receipt = %d bytes/equal:%t, want %d/equal:true", storedReceiptBytes, bytes.Equal(storedCredentialReceipt, credentialReceipt), len(credentialReceipt))
	}

	for _, deletion := range []struct {
		query    string
		argument any
		name     string
	}{
		{`DELETE FROM _System_PackGenerationConnectorBindingDefinitionSet WHERE PackGenerationId = ?`, "pack-generation-1", "referenced definition set"},
		{`DELETE FROM _System_ConnectorCredentialGeneration WHERE Id = ?`, "credential-generation-1", "referenced credential generation"},
		{`DELETE FROM _System_ConnectorServingEpochBinding WHERE Id = ?`, "profiled-binding", "referenced ServingEpoch binding"},
		{`DELETE FROM _System_ConnectorServingEpoch WHERE Id = ?`, "profiled-epoch", "referenced ServingEpoch"},
	} {
		_, err := db.Exec(deletion.query, deletion.argument)
		requirePackGenerationConnectorBindingMySQLError(t, err, 1451, "delete "+deletion.name)
	}
}

type connectorServingEpochReproducibilityIndex struct {
	table   string
	name    string
	columns string
	unique  bool
}

func requireConnectorServingEpochReproducibilityIndex(
	t *testing.T,
	db *sql.DB,
	expected connectorServingEpochReproducibilityIndex,
) {
	t.Helper()

	var (
		columns   string
		nonUnique int
	)
	if err := db.QueryRow(
		`SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX SEPARATOR ','), MIN(NON_UNIQUE)
		   FROM INFORMATION_SCHEMA.STATISTICS
		  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?`,
		expected.table,
		expected.name,
	).Scan(&columns, &nonUnique); err != nil {
		t.Fatalf("inspect %s.%s index: %v", expected.table, expected.name, err)
	}
	wantNonUnique := 1
	if expected.unique {
		wantNonUnique = 0
	}
	if columns != expected.columns || nonUnique != wantNonUnique {
		t.Fatalf(
			"%s.%s index = columns %q/non-unique %d, want %q/%d",
			expected.table,
			expected.name,
			columns,
			nonUnique,
			expected.columns,
			wantNonUnique,
		)
	}
}