package persistence

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

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
)

func TestIntegration_ConnectorServiceConfigurationSchemaPreservesExactGeneration(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 (
		credentialGenerationTable = "_System_ConnectorCredentialGeneration"
		generationTable           = "_System_ConnectorServiceConfigurationGeneration"
		accountTable              = "_System_ConnectorServiceConfigurationAccount"
		bindingTable              = "_System_ConnectorServiceConfigurationBinding"
		selectionTable            = "_System_ConnectorServiceConfigurationCredentialSelection"
	)
	tableNames := []string{
		credentialGenerationTable,
		generationTable,
		accountTable,
		bindingTable,
		selectionTable,
	}
	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)
			}
		}
	}

	requireConnectorServiceConfigurationBinaryColumns(t, db, map[string][]string{
		generationTable: {
			"Id", "GenerationProtocolVersion", "OrgId", "AppInstallationId", "AuthorizedWorkloadId",
			"SourceSnapshotDigest", "CatalogViewDigest", "CatalogReceiptProtocolVersion",
			"CatalogReceiptCapabilityVersion", "ConfigurationOwnerRevisionId",
			"ConfigurationOwnerRevisionDigest", "GenerationDigest", "ReceiptProtocolVersion",
			"OwnerReceiptDigest",
		},
		accountTable: {
			"Id", "ServiceConfigurationGenerationId", "ConnectorAccountId", "ProviderScopeId",
			"AccountRevisionId", "AccountRevisionDigest", "ProviderRevisionId", "ProviderRevisionDigest",
		},
		bindingTable: {
			"Id", "AccountConfigurationId", "ConfigurationRevisionId",
			"ConfigurationRevisionDigest", "ConfigurationFormatVersion",
		},
		selectionTable: {
			"Id", "BindingConfigurationId", "CredentialRef", "CredentialGenerationId",
		},
	})

	for _, foreignKey := range []connectorServiceConfigurationForeignKey{
		{accountTable, "ServiceConfigurationGenerationId", generationTable, "Id"},
		{bindingTable, "AccountConfigurationId", accountTable, "Id"},
		{selectionTable, "BindingConfigurationId", bindingTable, "Id"},
		{selectionTable, "CredentialGenerationId", credentialGenerationTable, "Id"},
	} {
		requireConnectorServiceConfigurationForeignKey(t, db, foreignKey)
	}

	digest := func(value string) string {
		return "sha256:" + strings.Repeat(value, 64)
	}
	insertCredentialGeneration := `INSERT INTO _System_ConnectorCredentialGeneration (
		Id, StableCredentialIdentity, ProviderScopeId, ConnectorAccountId,
		SecretBackendProtocol, SecretBackendVersionRef,
		TransportConfiguration, TransportConfigurationDigest,
		ReceiptProtocolVersion, OwnerReceipt, OwnerReceiptDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	insertExactCredentialGeneration := func(id, stableIdentity, receiptDigest string) {
		t.Helper()
		if _, err := db.Exec(
			insertCredentialGeneration,
			id,
			stableIdentity,
			"provider-scope-1",
			"connector-account-1",
			"nexuscrm.secret-backend.v1",
			"secret-version-"+id,
			[]byte("transport-configuration-v1\x00"+id),
			digest("a"),
			"nexuscrm.connector-cred-gen.v1",
			[]byte("credential-generation-receipt-v1\x00"+id),
			receiptDigest,
		); err != nil {
			t.Fatalf("insert CredentialGeneration %q: %v", id, err)
		}
	}
	insertExactCredentialGeneration("credential-generation-1", "credential-ref-1", digest("b"))
	insertExactCredentialGeneration("credential-generation-2", "Credential-ref-1", digest("c"))
	insertExactCredentialGeneration("credential-generation-3", "credential-ref-3", digest("n"))

	insertGeneration := `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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	ownerReceipt := []byte("connector-service-configuration-receipt-v1\x00generation-1")
	generationArgs := func(id, generationDigest, receiptDigest string) []any {
		return []any{
			id,
			"nexuscrm.connector-service-configuration-generation/v1",
			"org-1",
			"app-installation-1",
			"authorized-workload-1",
			uint64(11),
			uint64(12),
			uint64(13),
			digest("o"),
			digest("p"),
			"1.0",
			"catalog.consumer.integration-binding-view.v1",
			uint64(14),
			"owner-revision-1",
			digest("d"),
			generationDigest,
			"nexuscrm.connector-service-configuration-owner-receipt/v1",
			ownerReceipt,
			receiptDigest,
			uint64(2),
			uint64(1),
			uint64(2),
		}
	}
	if _, err := db.Exec(insertGeneration, generationArgs("configuration-generation-1", digest("e"), digest("f"))...); err != nil {
		t.Fatalf("insert service-configuration generation: %v", err)
	}
	if _, err := db.Exec(insertGeneration, generationArgs("Configuration-Generation-1", digest("g"), digest("h"))...); err != nil {
		t.Fatalf("insert byte-distinct service-configuration generation: %v", err)
	}
	if _, err := db.Exec(insertGeneration, generationArgs("configuration-generation-1", digest("i"), digest("j"))...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate exact configuration generation error = %v, want duplicate-key rejection", err)
	}

	var (
		storedOwnerReceipt     []byte
		storedOwnerReceiptSize int
		ownerReceiptIsNull     bool
	)
	if err := db.QueryRow(
		`SELECT OwnerReceipt, OCTET_LENGTH(OwnerReceipt), OwnerReceipt IS NULL
		   FROM _System_ConnectorServiceConfigurationGeneration
		  WHERE Id = ?`,
		"configuration-generation-1",
	).Scan(&storedOwnerReceipt, &storedOwnerReceiptSize, &ownerReceiptIsNull); err != nil {
		t.Fatalf("read exact service-configuration owner receipt: %v", err)
	}
	if ownerReceiptIsNull || storedOwnerReceiptSize != len(ownerReceipt) ||
		!equalConnectorServiceConfigurationBytes(storedOwnerReceipt, ownerReceipt) {
		t.Fatalf(
			"stored owner receipt = %x/%d null:%t, want %x/%d null:false",
			storedOwnerReceipt,
			storedOwnerReceiptSize,
			ownerReceiptIsNull,
			ownerReceipt,
			len(ownerReceipt),
		)
	}

	insertAccount := `INSERT INTO _System_ConnectorServiceConfigurationAccount (
		Id, ServiceConfigurationGenerationId, AccountOrdinal,
		ConnectorAccountId, ProviderScopeId,
		AccountRevisionId, AccountRevisionDigest,
		ProviderRevisionId, ProviderRevisionDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
	accountArgs := func(id, generationID string, ordinal uint64, accountID string) []any {
		return []any{
			id,
			generationID,
			ordinal,
			accountID,
			"provider-scope-1",
			"account-revision-1",
			digest("k"),
			"provider-revision-1",
			digest("l"),
		}
	}
	if _, err := db.Exec(insertAccount, accountArgs("configuration-account-1", "configuration-generation-1", 0, "connector-account-1")...); err != nil {
		t.Fatalf("insert first service-configuration account: %v", err)
	}
	if _, err := db.Exec(insertAccount, accountArgs("configuration-account-2", "configuration-generation-1", 1, "Connector-account-1")...); err != nil {
		t.Fatalf("insert byte-distinct service-configuration account: %v", err)
	}
	if _, err := db.Exec(insertAccount, accountArgs("account-ordinal-conflict", "configuration-generation-1", 0, "connector-account-2")...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate generation/account ordinal error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(insertAccount, accountArgs("account-identity-conflict", "configuration-generation-1", 2, "connector-account-1")...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate generation/account identity error = %v, want duplicate-key rejection", err)
	}
	_, err = db.Exec(insertAccount, accountArgs("account-missing-generation", "missing-generation", 0, "connector-account-3")...)
	requireConnectorServiceConfigurationMySQLError(t, err, 1452, "account for a missing configuration generation")
	_, err = db.Exec(insertAccount, accountArgs("account-case-mismatch", "CONFIGURATION-GENERATION-1", 0, "connector-account-4")...)
	requireConnectorServiceConfigurationMySQLError(t, err, 1452, "account for an unpersisted configuration-generation case variant")

	insertBinding := `INSERT INTO _System_ConnectorServiceConfigurationBinding (
		Id, AccountConfigurationId, CatalogBindingOrdinal,
		ConfigurationRevisionId, ConfigurationRevisionDigest,
		ConfigurationFormatVersion, CanonicalConfiguration
	) VALUES (?, ?, ?, ?, ?, ?, ?)`
	bindingArgs := func(
		id, accountID string,
		ordinal uint64,
		configuration []byte,
	) []any {
		return []any{
			id,
			accountID,
			ordinal,
			"configuration-revision-1",
			digest("m"),
			"nexuscrm.connector-route-configuration/v1",
			configuration,
		}
	}
	if _, err := db.Exec(
		insertBinding,
		bindingArgs("configuration-binding-1", "configuration-account-1", 0, []byte{})...,
	); err != nil {
		t.Fatalf("insert service-configuration binding with empty canonical bytes: %v", err)
	}
	var (
		storedConfiguration     []byte
		storedConfigurationSize int
		configurationIsNull     bool
	)
	if err := db.QueryRow(
		`SELECT CanonicalConfiguration, OCTET_LENGTH(CanonicalConfiguration), CanonicalConfiguration IS NULL
		   FROM _System_ConnectorServiceConfigurationBinding
		  WHERE Id = ?`,
		"configuration-binding-1",
	).Scan(&storedConfiguration, &storedConfigurationSize, &configurationIsNull); err != nil {
		t.Fatalf("read empty canonical service configuration: %v", err)
	}
	if configurationIsNull || storedConfigurationSize != 0 || len(storedConfiguration) != 0 {
		t.Fatalf(
			"stored empty canonical configuration = %x/%d null:%t, want empty/0/null:false",
			storedConfiguration,
			storedConfigurationSize,
			configurationIsNull,
		)
	}
	if _, err := db.Exec(
		insertBinding,
		bindingArgs("binding-ordinal-conflict", "configuration-account-1", 0, []byte("other"))...,
	); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate account/binding ordinal error = %v, want duplicate-key rejection", err)
	}
	_, err = db.Exec(
		insertBinding,
		bindingArgs("binding-missing-account", "missing-account", 0, []byte("other"))...,
	)
	requireConnectorServiceConfigurationMySQLError(t, err, 1452, "binding for a missing configuration account")
	_, err = db.Exec(
		insertBinding,
		bindingArgs("binding-case-mismatch", "CONFIGURATION-ACCOUNT-1", 0, []byte("other"))...,
	)
	requireConnectorServiceConfigurationMySQLError(t, err, 1452, "binding for an unpersisted configuration-account case variant")
	_, err = db.Exec(
		insertBinding,
		"binding-null-configuration",
		"configuration-account-2",
		uint64(0),
		"configuration-revision-1",
		digest("m"),
		"nexuscrm.connector-route-configuration/v1",
		nil,
	)
	requireConnectorServiceConfigurationMySQLError(t, err, 1048, "binding with SQL NULL canonical configuration")

	insertSelection := `INSERT INTO _System_ConnectorServiceConfigurationCredentialSelection (
		Id, BindingConfigurationId, CredentialOrdinal, CredentialRef, CredentialGenerationId
	) VALUES (?, ?, ?, ?, ?)`
	selectionArgs := func(
		id, bindingID string,
		ordinal uint64,
		credentialRef, credentialGenerationID string,
	) []any {
		return []any{id, bindingID, ordinal, credentialRef, credentialGenerationID}
	}
	if _, err := db.Exec(
		insertSelection,
		selectionArgs("credential-selection-1", "configuration-binding-1", 0, "credential-ref-1", "credential-generation-1")...,
	); err != nil {
		t.Fatalf("insert first service-configuration credential selection: %v", err)
	}
	if _, err := db.Exec(
		insertSelection,
		selectionArgs("credential-selection-2", "configuration-binding-1", 1, "Credential-ref-1", "credential-generation-2")...,
	); err != nil {
		t.Fatalf("insert byte-distinct service-configuration credential selection: %v", err)
	}
	if _, err := db.Exec(
		insertSelection,
		selectionArgs("credential-ordinal-conflict", "configuration-binding-1", 0, "credential-ref-2", "credential-generation-3")...,
	); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate binding/credential ordinal error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(
		insertSelection,
		selectionArgs("credential-ref-conflict", "configuration-binding-1", 2, "credential-ref-1", "credential-generation-3")...,
	); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate binding/credential reference error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(
		insertSelection,
		selectionArgs("credential-generation-conflict", "configuration-binding-1", 2, "credential-ref-2", "credential-generation-1")...,
	); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate binding/credential generation error = %v, want duplicate-key rejection", err)
	}
	_, err = db.Exec(
		insertSelection,
		selectionArgs("selection-missing-binding", "missing-binding", 0, "credential-ref-2", "credential-generation-3")...,
	)
	requireConnectorServiceConfigurationMySQLError(t, err, 1452, "credential selection for a missing configuration binding")
	_, err = db.Exec(
		insertSelection,
		selectionArgs("selection-missing-credential", "configuration-binding-1", 2, "credential-ref-2", "missing-credential-generation")...,
	)
	requireConnectorServiceConfigurationMySQLError(t, err, 1452, "credential selection for a missing credential generation")
	_, err = db.Exec(
		insertSelection,
		selectionArgs("selection-credential-case-mismatch", "configuration-binding-1", 2, "credential-ref-2", "CREDENTIAL-GENERATION-1")...,
	)
	requireConnectorServiceConfigurationMySQLError(t, err, 1452, "credential selection for an unpersisted credential-generation case variant")

	_, err = db.Exec(`DELETE FROM _System_ConnectorServiceConfigurationGeneration WHERE Id = ?`, "configuration-generation-1")
	requireConnectorServiceConfigurationMySQLError(t, err, 1451, "delete a configuration generation retained by an account")
	_, err = db.Exec(`DELETE FROM _System_ConnectorServiceConfigurationAccount WHERE Id = ?`, "configuration-account-1")
	requireConnectorServiceConfigurationMySQLError(t, err, 1451, "delete a configuration account retained by a binding")
	_, err = db.Exec(`DELETE FROM _System_ConnectorServiceConfigurationBinding WHERE Id = ?`, "configuration-binding-1")
	requireConnectorServiceConfigurationMySQLError(t, err, 1451, "delete a configuration binding retained by a credential selection")
	_, err = db.Exec(`DELETE FROM _System_ConnectorCredentialGeneration WHERE Id = ?`, "credential-generation-1")
	requireConnectorServiceConfigurationMySQLError(t, err, 1451, "delete a credential generation retained by a configuration selection")
}

func requireConnectorServiceConfigurationBinaryColumns(
	t *testing.T,
	db *sql.DB,
	columnsByTable map[string][]string,
) {
	t.Helper()
	for tableName, columnNames := range columnsByTable {
		for _, columnName := range columnNames {
			var collation sql.NullString
			if err := db.QueryRow(
				`SELECT COLLATION_NAME
				   FROM INFORMATION_SCHEMA.COLUMNS
				  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
				tableName,
				columnName,
			).Scan(&collation); err != nil {
				t.Fatalf("inspect %s.%s physical collation: %v", tableName, columnName, err)
			}
			if !collation.Valid || collation.String != "utf8mb4_bin" {
				t.Fatalf(
					"%s.%s physical collation = %q valid:%t, want utf8mb4_bin",
					tableName,
					columnName,
					collation.String,
					collation.Valid,
				)
			}
		}
	}
}

type connectorServiceConfigurationForeignKey struct {
	table, column, parentTable, parentColumn string
}

func requireConnectorServiceConfigurationForeignKey(
	t *testing.T,
	db *sql.DB,
	expected connectorServiceConfigurationForeignKey,
) {
	t.Helper()

	var (
		currentDatabase  string
		referencedSchema string
		referencedTable  string
		referencedColumn string
		deleteRule       string
		updateRule       string
		foreignKeyCount  int
	)
	if err := db.QueryRow(`SELECT DATABASE()`).Scan(&currentDatabase); err != nil {
		t.Fatalf("read current database: %v", err)
	}
	if err := db.QueryRow(
		`SELECT COUNT(*)
		   FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
		  WHERE CONSTRAINT_SCHEMA = DATABASE()
		    AND TABLE_SCHEMA = DATABASE()
		    AND TABLE_NAME = ?
		    AND COLUMN_NAME = ?
		    AND REFERENCED_TABLE_NAME IS NOT NULL`,
		expected.table,
		expected.column,
	).Scan(&foreignKeyCount); err != nil {
		t.Fatalf("count %s.%s foreign keys: %v", expected.table, expected.column, err)
	}
	if foreignKeyCount != 1 {
		t.Fatalf("%s.%s foreign-key count = %d, want 1", expected.table, expected.column, foreignKeyCount)
	}
	if err := db.QueryRow(
		`SELECT k.REFERENCED_TABLE_SCHEMA, k.REFERENCED_TABLE_NAME, k.REFERENCED_COLUMN_NAME,
		        r.DELETE_RULE, r.UPDATE_RULE
		   FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
		   JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS r
		     ON r.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
		    AND r.CONSTRAINT_NAME = k.CONSTRAINT_NAME
		    AND r.TABLE_NAME = k.TABLE_NAME
		  WHERE k.CONSTRAINT_SCHEMA = DATABASE()
		    AND k.TABLE_SCHEMA = DATABASE()
		    AND k.TABLE_NAME = ?
		    AND k.COLUMN_NAME = ?
		    AND k.REFERENCED_TABLE_NAME IS NOT NULL`,
		expected.table,
		expected.column,
	).Scan(
		&referencedSchema,
		&referencedTable,
		&referencedColumn,
		&deleteRule,
		&updateRule,
	); err != nil {
		t.Fatalf("inspect %s.%s foreign key: %v", expected.table, expected.column, err)
	}
	if referencedSchema != currentDatabase || referencedTable != expected.parentTable ||
		referencedColumn != expected.parentColumn {
		t.Fatalf(
			"%s.%s foreign key references %s.%s(%s), want %s.%s(%s)",
			expected.table,
			expected.column,
			referencedSchema,
			referencedTable,
			referencedColumn,
			currentDatabase,
			expected.parentTable,
			expected.parentColumn,
		)
	}
	if !connectorServiceConfigurationRestrictiveRule(deleteRule) ||
		!connectorServiceConfigurationRestrictiveRule(updateRule) {
		t.Fatalf(
			"%s.%s foreign-key rules = delete %q update %q, want restrictive",
			expected.table,
			expected.column,
			deleteRule,
			updateRule,
		)
	}
}

func connectorServiceConfigurationRestrictiveRule(rule string) bool {
	return strings.EqualFold(rule, "RESTRICT") || strings.EqualFold(rule, "NO ACTION")
}

func requireConnectorServiceConfigurationMySQLError(
	t *testing.T,
	err error,
	number uint16,
	operation string,
) {
	t.Helper()
	if err == nil {
		t.Fatalf("%s succeeded, want MySQL %d rejection", operation, number)
	}
	var mysqlErr *mysql.MySQLError
	if !errors.As(err, &mysqlErr) || mysqlErr.Number != number {
		t.Fatalf("%s error = %v, want MySQL %d", operation, err, number)
	}
}

func equalConnectorServiceConfigurationBytes(left, right []byte) bool {
	if len(left) != len(right) {
		return false
	}
	for index := range left {
		if left[index] != right[index] {
			return false
		}
	}
	return true
}