package serving

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

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
)

func TestLoadPreparedUsesExactBinaryIDAndReconstructsNormalizedReceipts(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	candidate := servingEpochTestCandidate(t, "Epoch-Exact-Binary")
	expectPreparedEpochLoad(mock, candidate, false, nil)

	loaded, found, err := loadPrepared(t.Context(), db, candidate.ID(), false)
	if err != nil || !found || !loaded.equal(candidate) {
		t.Fatalf("loadPrepared() = equal:%t found:%t err:%v", loaded.equal(candidate), found, err)
	}
	if loaded.catalog.receipts.CatalogRevision != loaded.catalog.lineage.CatalogRevision ||
		loaded.catalog.receipts.SchemaGeneration != loaded.catalog.lineage.SchemaGeneration ||
		loaded.catalog.receipts.EngineGeneration != loaded.catalog.lineage.EngineGeneration {
		t.Fatal("normalized lineage tuple was not reconstructed into Catalog receipts")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestLoadPreparedRejectsPartialRuntimeTuple(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	candidate := servingEpochTestCandidate(t, "epoch-corrupt-runtime")
	expectPreparedEpochLoad(mock, candidate, false, func(index int, values []driver.Value) {
		if index == 1 {
			values[13] = nil // RuntimeDefinitionRef
		}
	})

	if _, _, err := loadPrepared(t.Context(), db, candidate.ID(), false); !errors.Is(err, ErrStoredCandidateInvalid) {
		t.Fatalf("loadPrepared() error = %v, want ErrStoredCandidateInvalid", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestPrepareConvergesOnlyTypedExactDuplicate(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	repository := newServingEpochTestRepository(t, db)
	candidate := servingEpochTestCandidate(t, "epoch-duplicate")
	expectServingSchema(t, mock, exactServingSchemaFixture(), 5)
	expectPessimisticBegin(mock)
	mock.ExpectExec(`^INSERT INTO _System_ConnectorServingEpoch `).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "duplicate primary key"})
	expectPreparedEpochLoad(mock, candidate, true, nil)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	stored, duplicate, err := repository.Prepare(t.Context(), candidate)
	if err != nil || !duplicate || !stored.equal(candidate) {
		t.Fatalf("Prepare() = equal:%t duplicate:%t err:%v", stored.equal(candidate), duplicate, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
	if isMySQLDuplicateKey(errors.New("Error 1062 duplicate entry")) {
		t.Fatal("untyped duplicate-looking text was classified as MySQL 1062")
	}
}

func TestPrepareRollsBackParentAndFirstBindingOnSecondBindingCollision(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	repository := newServingEpochTestRepository(t, db)
	candidate := servingEpochTestCandidate(t, "epoch-atomic-rollback")
	expectServingSchema(t, mock, exactServingSchemaFixture(), 5)
	expectPessimisticBegin(mock)
	mock.ExpectExec(`^INSERT INTO _System_ConnectorServingEpoch `).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`^INSERT INTO _System_ConnectorServingEpochBinding `).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`^INSERT INTO _System_ConnectorServingEpochBinding `).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "hostile binding primary-key collision"})
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	if _, _, err := repository.Prepare(t.Context(), candidate); !errors.Is(err, ErrIdentityConflict) {
		t.Fatalf("Prepare() error = %v, want ErrIdentityConflict", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestServingRepositoryRejectsMissingDependenciesAndInvalidInputBeforeSQL(t *testing.T) {
	if _, err := NewRepository(nil); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("NewRepository(nil) error = %v, want ErrDependencyMissing", err)
	}
	db, mock := newServingSchemaMock(t)
	repository := newServingEpochTestRepository(t, db)
	if _, _, err := repository.LoadPrepared(t.Context(), " epoch"); !errors.Is(err, ErrInvalidCandidate) {
		t.Fatalf("LoadPrepared(invalid ID) error = %v, want ErrInvalidCandidate", err)
	}
	invalid := servingEpochTestCandidate(t, "epoch-invalid-input")
	invalid.bindings[0].ordinal++
	if _, _, err := repository.Prepare(t.Context(), invalid); !errors.Is(err, ErrInvalidCandidate) {
		t.Fatalf("Prepare(invalid candidate) error = %v, want ErrInvalidCandidate", err)
	}
	if errors.Is(repository.VerifySchema(nil), ErrSchemaUnavailable) {
		t.Fatal("nil public context was misreported as catalog schema drift")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unexpected SQL before input validation: %v", err)
	}
}

func expectPreparedEpochLoad(
	mock sqlmock.Sqlmock,
	epoch PreparedEpoch,
	forUpdate bool,
	mutateBinding func(index int, values []driver.Value),
) {
	parentQuery := preparedEpochSelect
	bindingQuery := preparedBindingSelect
	if forUpdate {
		parentQuery += " FOR UPDATE"
		bindingQuery += " FOR UPDATE"
	}
	parentRows := sqlmock.NewRows([]string{
		"Id", "EpochDigest", "ServingScopeDigest", "ActiveScopeDigest", "State",
		"OrgId", "AppInstallationId", "AuthorizedWorkloadId",
		"CatalogRevision", "SchemaGeneration", "EngineGeneration",
		"SourceSnapshotDigest", "BindingViewDigest",
		"ReceiptProtocolVersion", "ReceiptCapabilityVersion", "SecurityEpoch",
		"L2ConfigurationGenerationRef", "L2ConfigurationDigest",
		"L2VerificationReceipt", "L2VerificationReceiptDigest",
	}).AddRow(
		epoch.id,
		epoch.epochDigest,
		epoch.servingScopeDigest,
		nil,
		epoch.state,
		epoch.catalog.orgID,
		epoch.catalog.appInstallationID,
		epoch.catalog.authorizedWorkloadID,
		int64(epoch.catalog.lineage.CatalogRevision),
		int64(epoch.catalog.lineage.SchemaGeneration),
		int64(epoch.catalog.lineage.EngineGeneration),
		epoch.catalog.lineage.SourceSnapshotDigest,
		epoch.catalog.lineage.ViewDigest,
		epoch.catalog.receipts.ProtocolVersion,
		epoch.catalog.receipts.CapabilityVersion,
		int64(epoch.catalog.receipts.SecurityEpoch),
		epoch.l2.configurationGenerationRef,
		epoch.l2.configurationDigest,
		epoch.l2.verificationReceipt,
		epoch.l2.verificationReceiptDigest,
	)
	mock.ExpectQuery(regexp.QuoteMeta(parentQuery)).
		WithArgs(epoch.id, epoch.id).
		WillReturnRows(parentRows)

	bindingRows := sqlmock.NewRows([]string{
		"Id", "ServingEpochId", "BindingOrdinal", "BindingIdentityDigest",
		"Kind", "OwnerPack", "PackInstallationId", "PackPayloadDigest",
		"BusinessKey", "DefinitionVersion", "CredentialRefs",
		"RuntimeCommandKind", "RuntimePayloadFormatVersion",
		"RuntimeDefinitionRef", "RuntimeDefinitionVersion",
	})
	for index := range epoch.bindings {
		binding := epoch.bindings[index]
		values := []driver.Value{
			binding.id,
			binding.servingEpochID,
			int64(binding.ordinal),
			binding.identityDigest,
			string(binding.kind),
			binding.ownerPack,
			binding.packInstallationID,
			binding.packPayloadDigest,
			binding.businessKey,
			int64(binding.definitionVersion),
			binding.credentialRefs,
			nullableString(binding.runtimeCommandKind),
			nullableUint32(binding.runtimePayloadFormatVersion),
			nullableString(binding.runtimeDefinitionRef),
			nullableString(binding.runtimeDefinitionVersion),
		}
		if mutateBinding != nil {
			mutateBinding(index, values)
		}
		bindingRows.AddRow(values...)
	}
	mock.ExpectQuery(regexp.QuoteMeta(bindingQuery)).
		WithArgs(epoch.id, epoch.id).
		WillReturnRows(bindingRows)
	expectEmptyProfiledProofGraph(mock, epoch.id, forUpdate)
}

func expectEmptyProfiledProofGraph(mock sqlmock.Sqlmock, epochID string, forUpdate bool) {
	headerQuery := profiledServingEpochHeaderSelect
	ownerQuery := profiledServingEpochBindingOwnerSelect
	credentialQuery := profiledServingEpochCredentialSelect
	if forUpdate {
		headerQuery += " FOR UPDATE"
		ownerQuery += " FOR UPDATE"
		credentialQuery += " FOR UPDATE"
	}
	mock.ExpectQuery(regexp.QuoteMeta(headerQuery)).
		WithArgs(epochID, epochID).
		WillReturnRows(sqlmock.NewRows([]string{
			"ServingEpochId", "ConnectorDigestProfileId",
			"ServiceConfigurationGenerationProtocolVersion",
			"ServiceConfigurationReceiptProtocolVersion",
		}))
	mock.ExpectQuery(regexp.QuoteMeta(ownerQuery)).
		WithArgs(epochID, epochID).
		WillReturnRows(sqlmock.NewRows([]string{
			"ServingEpochBindingId", "PackGenerationId",
			"DefinitionEnvelopeVersion", "DefinitionReceiptVersion",
			"CanonicalDefinitionDigest", "DefinitionOwnerReceipt",
			"DefinitionOwnerReceiptDigest",
		}))
	mock.ExpectQuery(regexp.QuoteMeta(credentialQuery)).
		WithArgs(epochID, epochID).
		WillReturnRows(sqlmock.NewRows([]string{
			"ServingEpochId", "CredentialOrdinal", "CredentialGenerationId",
			"ReceiptProtocolVersion", "TransportConfigurationDigest",
			"OwnerReceipt", "OwnerReceiptDigest",
		}))
}

func expectPessimisticBegin(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(regexp.QuoteMeta("SELECT @@SESSION.tidb_constraint_check_in_place_pessimistic")).
		WillReturnRows(sqlmock.NewRows([]string{"tidb_constraint_check_in_place_pessimistic"}).AddRow("OFF"))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = 'pessimistic'")).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_constraint_check_in_place_pessimistic = ON")).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectPessimisticCleanup(mock sqlmock.Sqlmock) {
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_constraint_check_in_place_pessimistic = OFF")).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = DEFAULT")).
		WillReturnResult(sqlmock.NewResult(0, 0))
}