package serviceconfiguration

import (
	"database/sql/driver"
	"errors"
	"regexp"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
)

func TestPublishRejectsStaleCatalogViewDigestBeforeCredentialSQL(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New(): %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	repository, err := NewRepository(db)
	if err != nil {
		t.Fatalf("NewRepository(): %v", err)
	}
	view := serviceConfigurationTestView()
	view.Bindings[0].DefinitionVersion++
	if view.Lineage.ViewDigest == catalogv1.ComputeIntegrationBindingViewDigest(view) {
		t.Fatal("test setup did not stale the Catalog view digest")
	}

	if _, _, err := repository.Publish(
		t.Context(),
		serviceConfigurationTestInput("service-configuration-stale-catalog-publish"),
		view,
	); !errors.Is(err, ErrInvalidGeneration) {
		t.Fatalf("Publish(stale Catalog digest) error = %v, want ErrInvalidGeneration", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("stale Catalog digest reached credential SQL: %v", err)
	}
}

func TestLoadServiceConfigurationExactReconstructsCompleteGraph(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New(): %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	want, err := deriveGeneration(
		serviceConfigurationTestInput("service-configuration-load-exact"),
		serviceConfigurationTestView(),
		serviceConfigurationTestCredentials(t),
	)
	if err != nil {
		t.Fatalf("deriveGeneration(): %v", err)
	}
	expectServiceConfigurationGenerationLoad(mock, want, false)

	loaded, found, err := loadServiceConfigurationExact(t.Context(), db, want.id, false)
	if err != nil || !found || !loaded.equal(want) {
		t.Fatalf(
			"loadServiceConfigurationExact() = equal:%t found:%t err:%v",
			loaded.equal(want),
			found,
			err,
		)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestInsertServiceConfigurationBindingsPreservesNonNullEmptyConfiguration(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New(): %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	generation, err := deriveGeneration(
		serviceConfigurationTestInput("service-configuration-empty-blob"),
		serviceConfigurationTestView(),
		serviceConfigurationTestCredentials(t),
	)
	if err != nil {
		t.Fatalf("deriveGeneration(): %v", err)
	}
	bindings := []BindingConfiguration{
		generation.accounts[0].bindings[0],
		generation.accounts[0].bindings[1],
	}
	if bindings[0].canonicalConfiguration == nil || len(bindings[0].canonicalConfiguration) != 0 {
		t.Fatalf(
			"derived empty CanonicalConfiguration = %#v, want non-nil empty bytes",
			bindings[0].canonicalConfiguration,
		)
	}

	mock.ExpectBegin()
	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatalf("BeginTx(): %v", err)
	}
	query := `INSERT INTO _System_ConnectorServiceConfigurationBinding
       (Id, AccountConfigurationId, CatalogBindingOrdinal,
        ConfigurationRevisionId, ConfigurationRevisionDigest,
        ConfigurationFormatVersion, CanonicalConfiguration)
VALUES ` + serviceConfigurationValueTuples(2, 7)
	mock.ExpectExec(regexp.QuoteMeta(query)).WithArgs(
		bindings[0].rowID,
		bindings[0].accountRowID,
		bindings[0].catalogBindingOrdinal,
		bindings[0].configurationRevision.RevisionID,
		bindings[0].configurationRevision.ValueDigest,
		bindings[0].configurationFormatVersion,
		serviceConfigurationNonNullEmptyBytes{},
		bindings[1].rowID,
		bindings[1].accountRowID,
		bindings[1].catalogBindingOrdinal,
		bindings[1].configurationRevision.RevisionID,
		bindings[1].configurationRevision.ValueDigest,
		bindings[1].configurationFormatVersion,
		bindings[1].canonicalConfiguration,
	).WillReturnResult(sqlmock.NewResult(0, 2))
	if err := insertServiceConfigurationBindings(t.Context(), tx, generation); err != nil {
		t.Fatalf("insertServiceConfigurationBindings(): %v", err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatalf("Rollback(): %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestRepositoryClassifiesStoredConversionAndTypedChildConflicts(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New(): %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	mock.ExpectQuery(regexp.QuoteMeta(serviceConfigurationAccountSelect)).
		WithArgs("generation-invalid-row", "generation-invalid-row").
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "AccountOrdinal", "ConnectorAccountId", "ProviderScopeId",
			"AccountRevisionId", "AccountRevisionDigest",
			"ProviderRevisionId", "ProviderRevisionDigest",
		}).AddRow(
			"row-id",
			"not-an-integer",
			"account-a",
			"provider-a",
			"account-revision",
			serviceConfigurationDigest("1"),
			"provider-revision",
			serviceConfigurationDigest("2"),
		))
	if _, _, err := loadServiceConfigurationAccounts(
		t.Context(),
		db,
		"generation-invalid-row",
		false,
	); !errors.Is(err, ErrStoredGenerationInvalid) {
		t.Fatalf(
			"loadServiceConfigurationAccounts(conversion failure) error = %v, want ErrStoredGenerationInvalid",
			err,
		)
	}
	typedDuplicate := &mysql.MySQLError{Number: 1062, Message: "duplicate child key"}
	if err := serviceConfigurationChildInsertError("binding", typedDuplicate); !errors.Is(err, ErrIdentityConflict) {
		t.Fatalf("typed child duplicate error = %v, want ErrIdentityConflict", err)
	}
	if err := serviceConfigurationChildInsertError(
		"binding",
		errors.New("Error 1062 duplicate child key"),
	); errors.Is(err, ErrIdentityConflict) {
		t.Fatalf("untyped duplicate-looking error = %v, want ordinary insert failure", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func expectServiceConfigurationGenerationLoad(
	mock sqlmock.Sqlmock,
	generation Generation,
	forUpdate bool,
) {
	parentQuery := serviceConfigurationGenerationSelect
	accountQuery := serviceConfigurationAccountSelect
	bindingQuery := serviceConfigurationBindingSelect
	selectionQuery := serviceConfigurationCredentialSelectionSelect
	if forUpdate {
		parentQuery += " FOR UPDATE"
		accountQuery += " FOR UPDATE"
		bindingQuery += " FOR UPDATE"
		selectionQuery += " FOR UPDATE"
	}
	mock.ExpectQuery(regexp.QuoteMeta(parentQuery)).
		WithArgs(generation.id, generation.id).
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "GenerationProtocolVersion",
			"OrgId", "AppInstallationId", "AuthorizedWorkloadId",
			"CatalogRevision", "SchemaGeneration", "EngineGeneration",
			"SourceSnapshotDigest", "CatalogViewDigest",
			"CatalogReceiptProtocolVersion", "CatalogReceiptCapabilityVersion",
			"CatalogSecurityEpoch",
			"ConfigurationOwnerRevisionId", "ConfigurationOwnerRevisionDigest",
			"GenerationDigest", "ReceiptProtocolVersion",
			"OwnerReceipt", "OwnerReceiptDigest",
			"AccountCount", "BindingCount", "CredentialSelectionCount",
		}).AddRow(
			generation.id,
			string(generation.generationProtocolVersion),
			generation.catalog.orgID,
			generation.catalog.appInstallationID,
			generation.catalog.authorizedWorkloadID,
			generation.catalog.lineage.CatalogRevision,
			generation.catalog.lineage.SchemaGeneration,
			generation.catalog.lineage.EngineGeneration,
			generation.catalog.lineage.SourceSnapshotDigest,
			generation.catalog.lineage.ViewDigest,
			generation.catalog.receipts.ProtocolVersion,
			generation.catalog.receipts.CapabilityVersion,
			generation.catalog.receipts.SecurityEpoch,
			generation.ownerRevision.RevisionID,
			generation.ownerRevision.ValueDigest,
			generation.generationDigest,
			string(generation.receiptProtocolVersion),
			generation.ownerReceipt,
			generation.ownerReceiptDigest,
			generation.accountCount,
			generation.bindingCount,
			generation.credentialSelectionCount,
		))

	accountRows := sqlmock.NewRows([]string{
		"Id", "AccountOrdinal", "ConnectorAccountId", "ProviderScopeId",
		"AccountRevisionId", "AccountRevisionDigest",
		"ProviderRevisionId", "ProviderRevisionDigest",
	})
	for _, account := range generation.accounts {
		accountRows.AddRow(
			account.rowID,
			account.ordinal,
			account.connectorAccountID,
			account.providerScopeID,
			account.accountRevision.RevisionID,
			account.accountRevision.ValueDigest,
			account.providerRevision.RevisionID,
			account.providerRevision.ValueDigest,
		)
	}
	mock.ExpectQuery(regexp.QuoteMeta(accountQuery)).
		WithArgs(generation.id, generation.id).
		WillReturnRows(accountRows)

	bindingRows := sqlmock.NewRows([]string{
		"Id", "AccountConfigurationId", "CatalogBindingOrdinal",
		"ConfigurationRevisionId", "ConfigurationRevisionDigest",
		"ConfigurationFormatVersion", "CanonicalConfiguration",
	})
	selectionRows := sqlmock.NewRows([]string{
		"Id", "BindingConfigurationId", "CredentialOrdinal",
		"CredentialRef", "CredentialGenerationId",
	})
	for _, account := range generation.accounts {
		for _, binding := range account.bindings {
			bindingRows.AddRow(
				binding.rowID,
				binding.accountRowID,
				binding.catalogBindingOrdinal,
				binding.configurationRevision.RevisionID,
				binding.configurationRevision.ValueDigest,
				binding.configurationFormatVersion,
				binding.canonicalConfiguration,
			)
			for _, selection := range binding.credentialSelections {
				selectionRows.AddRow(
					selection.rowID,
					selection.bindingRowID,
					selection.ordinal,
					selection.credentialRef,
					selection.credentialGenerationID,
				)
			}
		}
	}
	mock.ExpectQuery(regexp.QuoteMeta(bindingQuery)).
		WithArgs(generation.id, generation.id).
		WillReturnRows(bindingRows)
	mock.ExpectQuery(regexp.QuoteMeta(selectionQuery)).
		WithArgs(generation.id, generation.id).
		WillReturnRows(selectionRows)
}

type serviceConfigurationNonNullEmptyBytes struct{}

func (serviceConfigurationNonNullEmptyBytes) Match(value driver.Value) bool {
	bytes, ok := value.([]byte)
	return ok && bytes != nil && len(bytes) == 0
}