package catalogtenant

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

	"github.com/DATA-DOG/go-sqlmock"
	tenantauthorityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/tenantlayoutbundleauthorityv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogadmissionfence"
)

const testCommand3 = tenantauthorityv1.CommandID(
	"01890f9e-7b2a-7000-8000-000000000003",
)

var (
	publicationAdmissionFenceLockQuery = regexp.QuoteMeta(`SELECT FenceKey, FenceGeneration
FROM _System_CatalogPublicationAdmissionFence
WHERE FenceKey = ? FOR UPDATE`)
	fenceLockQuery = regexp.QuoteMeta(`SELECT FenceKey, CommandSequence
FROM _System_CatalogTenantAuthorityFence WHERE FenceKey = ? FOR UPDATE`)
	commandLockQuery = regexp.QuoteMeta(`SELECT CommandId, CommandKind, RequestDigest, Result,
ObservedPriorPresent, ObservedPriorDigest, ResultingValuePresent, ResultingValueDigest,
ObservedNamespacePresent, ObservedNamespaceDigest,
ObservedBindingPresent, ObservedBindingDigest,
CommandSequence, OutcomeDigest
FROM _System_CatalogTenantAuthorityCommand WHERE CommandId = ? FOR UPDATE`)
	namespaceLockQuery = regexp.QuoteMeta(`SELECT TenantNamespaceId, ValueVersion, State,
ValueDigest, LastCommandId
FROM _System_CatalogTenantNamespaceReservation
WHERE TenantNamespaceId = ? FOR UPDATE`)
	bindingLockQuery = regexp.QuoteMeta(`SELECT OrgId, TenantNamespaceId, ValueVersion,
State, ValueDigest, LastCommandId
FROM _System_CatalogTenantOrgNamespaceBinding WHERE OrgId = ? FOR UPDATE`)
	otherActiveBindingLockQuery = regexp.QuoteMeta(`SELECT OrgId, TenantNamespaceId, ValueVersion,
State, ValueDigest, LastCommandId
FROM _System_CatalogTenantOrgNamespaceBinding
WHERE TenantNamespaceId = ? AND State = 'ACTIVE' AND OrgId <> ?
ORDER BY OrgId LIMIT 1 FOR UPDATE`)
	slotLockQuery = regexp.QuoteMeta(`SELECT TenantNamespaceId, LocalComponentKey,
BundleKeyVersion, SlotOwnerKind, ValueVersion, State, ValueDigest, LastCommandId
FROM _System_CatalogTenantLayoutBundleSlot
WHERE TenantNamespaceId = ? AND LocalComponentKey = ? FOR UPDATE`)
	activeSlotLockQuery = regexp.QuoteMeta(`SELECT TenantNamespaceId, LocalComponentKey,
BundleKeyVersion, SlotOwnerKind, ValueVersion, State, ValueDigest, LastCommandId
FROM _System_CatalogTenantLayoutBundleSlot
WHERE TenantNamespaceId = ? AND State = 'ACTIVE'
ORDER BY LocalComponentKey LIMIT 1 FOR UPDATE`)
	bindingTombstoneUpdateQuery = regexp.QuoteMeta(`UPDATE _System_CatalogTenantOrgNamespaceBinding
SET ValueVersion = ?, State = ?, ValueDigest = ?, LastCommandId = ?,
LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND TenantNamespaceId = ? AND State = 'ACTIVE'
AND ValueDigest = ? AND LastCommandId = ?`)
)

func TestNamespaceReservePersistsCurrentHistoryAndOutcomeAtomically(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := namespaceReserve(testNamespace)
	requestDigest, err := tenantauthorityv1.NamespaceCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	desiredDigest := mustNamespaceDigest(t, request.DesiredNamespaceValue)
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:     testCommand1,
		Kind:          commandKindNamespace,
		RequestDigest: requestDigest,
		Result:        ResultApplied,
		ResultingValue: ValueReference{
			Present: true,
			Digest:  desiredDigest,
		},
		Sequence: 1,
	})

	expectNewCommand(mock, testCommand1, 0)
	mock.ExpectQuery(namespaceLockQuery).WithArgs(testNamespace).
		WillReturnRows(sqlmock.NewRows(namespaceCurrentColumns()))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantNamespaceReservation").
		WithArgs(
			testNamespace,
			tenantauthorityv1.NamespaceReservationValueVersion,
			"ACTIVE",
			desiredDigest,
			string(testCommand1),
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantNamespaceHistory").
		WithArgs(
			string(testCommand1),
			"1",
			testNamespace,
			tenantauthorityv1.NamespaceReservationValueVersion,
			"ACTIVE",
			desiredDigest,
			nil,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	expectFinishCommand(mock, 0, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyNamespaceCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != namespaceOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, namespaceOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestOrgBindingCommandLocksNamespaceAndPersistsLineage(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := bindingBind(t, testOrg, testNamespace)
	requestDigest, err := tenantauthorityv1.OrgBindingCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	desiredDigest := mustBindingDigest(t, request.DesiredBindingValue)
	observedNamespace := ValueReference{
		Present: true,
		Digest:  request.ExpectedNamespaceReservationDigest,
	}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:         testCommand1,
		Kind:              commandKindBinding,
		RequestDigest:     requestDigest,
		Result:            ResultApplied,
		ResultingValue:    ValueReference{Present: true, Digest: desiredDigest},
		ObservedNamespace: observedNamespace,
		Sequence:          2,
	})

	expectNewCommand(mock, testCommand1, 1)
	expectActiveNamespace(mock, request.ExpectedNamespaceReservationDigest)
	mock.ExpectQuery(bindingLockQuery).WithArgs(testOrg).
		WillReturnRows(sqlmock.NewRows(bindingCurrentColumns()))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantOrgNamespaceBinding").
		WithArgs(
			testOrg,
			testNamespace,
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			desiredDigest,
			string(testCommand1),
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantOrgNamespaceBindingHistory").
		WithArgs(
			string(testCommand1),
			"2",
			testOrg,
			testNamespace,
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			desiredDigest,
			nil,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	expectFinishCommand(mock, 1, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyOrgBindingCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != bindingOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, bindingOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestLastOrgBindingCannotTombstoneWhileNamespaceHasActiveSlot(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := bindingTombstone(t, testOrg, testNamespace)
	requestDigest, err := tenantauthorityv1.OrgBindingCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	observedNamespace := ValueReference{
		Present: true,
		Digest:  request.ExpectedNamespaceReservationDigest,
	}
	observedBinding := bindingExpectedReference(request.ExpectedPriorBinding)
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:         testCommand1,
		Kind:              commandKindBinding,
		RequestDigest:     requestDigest,
		Result:            ResultActiveSlot,
		ObservedPrior:     observedBinding,
		ResultingValue:    observedBinding,
		ObservedNamespace: observedNamespace,
		Sequence:          3,
	})
	activeSlot := slotValue(testNamespace, tenantauthorityv1.SlotStateActive)
	activeSlotDigest := mustSlotDigest(t, activeSlot)

	expectNewCommand(mock, testCommand1, 2)
	expectActiveNamespace(mock, request.ExpectedNamespaceReservationDigest)
	mock.ExpectQuery(bindingLockQuery).WithArgs(testOrg).WillReturnRows(
		sqlmock.NewRows(bindingCurrentColumns()).AddRow(
			[]byte(testOrg),
			[]byte(testNamespace),
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			request.ExpectedPriorBinding.BindingValueDigest,
			string(testCommand2),
		),
	)
	mock.ExpectQuery(otherActiveBindingLockQuery).
		WithArgs(testNamespace, testOrg).
		WillReturnRows(sqlmock.NewRows(bindingCurrentColumns()))
	mock.ExpectQuery(activeSlotLockQuery).WithArgs(testNamespace).WillReturnRows(
		sqlmock.NewRows(slotCurrentColumns()).AddRow(
			[]byte(testNamespace),
			[]byte(testLocalKey),
			activeSlot.BundleKey.BundleKeyVersion,
			"tenant",
			tenantauthorityv1.LayoutBundleSlotValueVersion,
			"ACTIVE",
			activeSlotDigest,
			string(testCommand3),
		),
	)
	expectFinishCommand(mock, 2, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyOrgBindingCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != bindingOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, bindingOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestOrgBindingTombstoneRejectsCurrentBindingFromAnotherNamespace(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	const currentNamespace = "com.example.current"
	request := bindingTombstone(t, testOrg, testNamespace)
	requestDigest, err := tenantauthorityv1.OrgBindingCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	currentBinding := bindingValue(
		testOrg,
		currentNamespace,
		tenantauthorityv1.OrgBindingStateActive,
	)
	currentDigest := mustBindingDigest(t, currentBinding)
	observedNamespace := ValueReference{
		Present: true,
		Digest:  request.ExpectedNamespaceReservationDigest,
	}
	observedBinding := ValueReference{Present: true, Digest: currentDigest}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:         testCommand1,
		Kind:              commandKindBinding,
		RequestDigest:     requestDigest,
		Result:            ResultExpectedPriorMismatch,
		ObservedPrior:     observedBinding,
		ResultingValue:    observedBinding,
		ObservedNamespace: observedNamespace,
		Sequence:          3,
	})

	expectNewCommand(mock, testCommand1, 2)
	expectActiveNamespace(mock, request.ExpectedNamespaceReservationDigest)
	mock.ExpectQuery(bindingLockQuery).WithArgs(testOrg).WillReturnRows(
		sqlmock.NewRows(bindingCurrentColumns()).AddRow(
			[]byte(testOrg),
			[]byte(currentNamespace),
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			currentDigest,
			string(testCommand2),
		),
	)
	expectFinishCommand(mock, 2, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyOrgBindingCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != bindingOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, bindingOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestNonLastOrgBindingMayTombstoneWhileNamespaceHasActiveSlot(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	const otherOrg = "org-2"
	request := bindingTombstone(t, testOrg, testNamespace)
	requestDigest, err := tenantauthorityv1.OrgBindingCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	desiredDigest := mustBindingDigest(t, request.DesiredBindingValue)
	observedNamespace := ValueReference{
		Present: true,
		Digest:  request.ExpectedNamespaceReservationDigest,
	}
	observedBinding := bindingExpectedReference(request.ExpectedPriorBinding)
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:         testCommand1,
		Kind:              commandKindBinding,
		RequestDigest:     requestDigest,
		Result:            ResultApplied,
		ObservedPrior:     observedBinding,
		ResultingValue:    ValueReference{Present: true, Digest: desiredDigest},
		ObservedNamespace: observedNamespace,
		Sequence:          4,
	})
	otherBinding := bindingValue(
		otherOrg,
		testNamespace,
		tenantauthorityv1.OrgBindingStateActive,
	)
	otherBindingDigest := mustBindingDigest(t, otherBinding)
	activeSlot := slotValue(testNamespace, tenantauthorityv1.SlotStateActive)
	activeSlotDigest := mustSlotDigest(t, activeSlot)

	expectNewCommand(mock, testCommand1, 3)
	expectActiveNamespace(mock, request.ExpectedNamespaceReservationDigest)
	mock.ExpectQuery(bindingLockQuery).WithArgs(testOrg).WillReturnRows(
		sqlmock.NewRows(bindingCurrentColumns()).AddRow(
			[]byte(testOrg),
			[]byte(testNamespace),
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			request.ExpectedPriorBinding.BindingValueDigest,
			string(testCommand2),
		),
	)
	mock.ExpectQuery(otherActiveBindingLockQuery).
		WithArgs(testNamespace, testOrg).
		WillReturnRows(sqlmock.NewRows(bindingCurrentColumns()).AddRow(
			[]byte(otherOrg),
			[]byte(testNamespace),
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			otherBindingDigest,
			string(testCommand3),
		))
	mock.ExpectQuery(activeSlotLockQuery).WithArgs(testNamespace).WillReturnRows(
		sqlmock.NewRows(slotCurrentColumns()).AddRow(
			[]byte(testNamespace),
			[]byte(testLocalKey),
			activeSlot.BundleKey.BundleKeyVersion,
			"tenant",
			tenantauthorityv1.LayoutBundleSlotValueVersion,
			"ACTIVE",
			activeSlotDigest,
			string(testCommand3),
		),
	)
	mock.ExpectExec(bindingTombstoneUpdateQuery).
		WithArgs(
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"TOMBSTONE",
			desiredDigest,
			string(testCommand1),
			testOrg,
			testNamespace,
			request.ExpectedPriorBinding.BindingValueDigest,
			string(testCommand2),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantOrgNamespaceBindingHistory").
		WithArgs(
			string(testCommand1),
			"4",
			testOrg,
			testNamespace,
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"TOMBSTONE",
			desiredDigest,
			request.ExpectedPriorBinding.BindingValueDigest,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	expectFinishCommand(mock, 3, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyOrgBindingCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != bindingOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, bindingOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestSlotCommandLocksBothDependenciesAndPersistsLineage(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := slotCreate(t, testOrg, testNamespace)
	requestDigest, err := tenantauthorityv1.SlotCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	desiredDigest := mustSlotDigest(t, request.DesiredSlotValue)
	observedNamespace := ValueReference{
		Present: true,
		Digest:  request.ExpectedNamespaceReservationDigest,
	}
	observedBinding := ValueReference{
		Present: true,
		Digest:  request.ExpectedAuthorizingOrgBindingDigest,
	}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:         testCommand1,
		Kind:              commandKindSlot,
		RequestDigest:     requestDigest,
		Result:            ResultApplied,
		ResultingValue:    ValueReference{Present: true, Digest: desiredDigest},
		ObservedNamespace: observedNamespace,
		ObservedBinding:   observedBinding,
		Sequence:          3,
	})

	expectNewCommand(mock, testCommand1, 2)
	expectActiveNamespace(mock, request.ExpectedNamespaceReservationDigest)
	mock.ExpectQuery(bindingLockQuery).WithArgs(testOrg).WillReturnRows(
		sqlmock.NewRows(bindingCurrentColumns()).AddRow(
			[]byte(testOrg),
			[]byte(testNamespace),
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			request.ExpectedAuthorizingOrgBindingDigest,
			string(testCommand3),
		),
	)
	mock.ExpectQuery(slotLockQuery).
		WithArgs(testNamespace, string(testLocalKey)).
		WillReturnRows(sqlmock.NewRows(slotCurrentColumns()))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantLayoutBundleSlot").
		WithArgs(
			testNamespace,
			string(testLocalKey),
			request.DesiredSlotValue.BundleKey.BundleKeyVersion,
			"tenant",
			tenantauthorityv1.LayoutBundleSlotValueVersion,
			"ACTIVE",
			desiredDigest,
			string(testCommand1),
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantLayoutBundleSlotHistory").
		WithArgs(
			string(testCommand1),
			"3",
			testNamespace,
			string(testLocalKey),
			request.DesiredSlotValue.BundleKey.BundleKeyVersion,
			"tenant",
			tenantauthorityv1.LayoutBundleSlotValueVersion,
			"ACTIVE",
			desiredDigest,
			nil,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	expectFinishCommand(mock, 2, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplySlotCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != slotOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, slotOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestOrgBindingCommandRejectsTombstonedNamespaceDependency(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := bindingBind(t, testOrg, testNamespace)
	requestDigest, err := tenantauthorityv1.OrgBindingCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	tombstoneNamespace := namespaceValue(
		testNamespace,
		tenantauthorityv1.NamespaceStateTombstone,
	)
	tombstoneDigest := mustNamespaceDigest(t, tombstoneNamespace)
	observedNamespace := ValueReference{Present: true, Digest: tombstoneDigest}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:         testCommand1,
		Kind:              commandKindBinding,
		RequestDigest:     requestDigest,
		Result:            ResultNamespaceDependencyMismatch,
		ObservedNamespace: observedNamespace,
		Sequence:          2,
	})

	expectNewCommand(mock, testCommand1, 1)
	mock.ExpectQuery(namespaceLockQuery).WithArgs(testNamespace).WillReturnRows(
		sqlmock.NewRows(namespaceCurrentColumns()).AddRow(
			[]byte(testNamespace),
			tenantauthorityv1.NamespaceReservationValueVersion,
			"TOMBSTONE",
			tombstoneDigest,
			string(testCommand2),
		),
	)
	mock.ExpectQuery(bindingLockQuery).WithArgs(testOrg).
		WillReturnRows(sqlmock.NewRows(bindingCurrentColumns()))
	expectFinishCommand(mock, 1, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyOrgBindingCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != bindingOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, bindingOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestSlotCommandRejectsBindingForAnotherTenantNamespace(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := slotCreate(t, testOrg, testNamespace)
	requestDigest, err := tenantauthorityv1.SlotCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	const foreignNamespace = "com.example.foreign"
	foreignBinding := bindingValue(
		testOrg,
		foreignNamespace,
		tenantauthorityv1.OrgBindingStateActive,
	)
	foreignBindingDigest := mustBindingDigest(t, foreignBinding)
	observedNamespace := ValueReference{
		Present: true,
		Digest:  request.ExpectedNamespaceReservationDigest,
	}
	observedBinding := ValueReference{
		Present: true,
		Digest:  foreignBindingDigest,
	}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:         testCommand1,
		Kind:              commandKindSlot,
		RequestDigest:     requestDigest,
		Result:            ResultBindingDependencyMismatch,
		ObservedNamespace: observedNamespace,
		ObservedBinding:   observedBinding,
		Sequence:          3,
	})

	expectNewCommand(mock, testCommand1, 2)
	expectActiveNamespace(mock, request.ExpectedNamespaceReservationDigest)
	mock.ExpectQuery(bindingLockQuery).WithArgs(testOrg).WillReturnRows(
		sqlmock.NewRows(bindingCurrentColumns()).AddRow(
			[]byte(testOrg),
			[]byte(foreignNamespace),
			tenantauthorityv1.OrgNamespaceBindingValueVersion,
			"ACTIVE",
			foreignBindingDigest,
			string(testCommand3),
		),
	)
	mock.ExpectQuery(slotLockQuery).
		WithArgs(testNamespace, string(testLocalKey)).
		WillReturnRows(sqlmock.NewRows(slotCurrentColumns()))
	expectFinishCommand(mock, 2, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplySlotCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != slotOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, slotOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestExactReplayReturnsStoredOutcomeWithoutReadingCurrentState(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := namespaceReserve(testNamespace)
	requestDigest, err := tenantauthorityv1.NamespaceCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:     testCommand1,
		Kind:          commandKindNamespace,
		RequestDigest: requestDigest,
		Result:        ResultApplied,
		ResultingValue: ValueReference{
			Present: true,
			Digest:  mustNamespaceDigest(t, request.DesiredNamespaceValue),
		},
		Sequence: 4,
	})

	mock.ExpectBegin()
	expectPublicationAdmissionFence(mock)
	mock.ExpectQuery(fenceLockQuery).WithArgs(authorityFenceKey).WillReturnRows(
		sqlmock.NewRows([]string{"FenceKey", "CommandSequence"}).
			AddRow(authorityFenceKey, []byte("9")),
	)
	expectCommandRow(mock, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyNamespaceCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != namespaceOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, namespaceOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestTombstonedNamespaceCannotBeRevivedAndFailureIsTerminal(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	request := namespaceReserve(testNamespace)
	requestDigest, err := tenantauthorityv1.NamespaceCommandRequestDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	tombstoneValue := namespaceValue(
		testNamespace,
		tenantauthorityv1.NamespaceStateTombstone,
	)
	tombstoneDigest := mustNamespaceDigest(t, tombstoneValue)
	observed := ValueReference{Present: true, Digest: tombstoneDigest}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:      testCommand1,
		Kind:           commandKindNamespace,
		RequestDigest:  requestDigest,
		Result:         ResultExpectedPriorMismatch,
		ObservedPrior:  observed,
		ResultingValue: observed,
		Sequence:       1,
	})

	expectNewCommand(mock, testCommand1, 0)
	mock.ExpectQuery(namespaceLockQuery).WithArgs(testNamespace).WillReturnRows(
		sqlmock.NewRows(namespaceCurrentColumns()).AddRow(
			[]byte(testNamespace),
			tenantauthorityv1.NamespaceReservationValueVersion,
			"TOMBSTONE",
			tombstoneDigest,
			string(testCommand2),
		),
	)
	expectFinishCommand(mock, 0, command)
	mock.ExpectCommit()

	outcome, err := authority.ApplyNamespaceCommand(
		context.Background(),
		testCommand1,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if outcome != namespaceOutcome(command) {
		t.Fatalf("outcome = %#v, want %#v", outcome, namespaceOutcome(command))
	}
	assertExpectations(t, mock)
}

func TestCommandIDDigestReuseConflictsBeforeCurrentStateRead(t *testing.T) {
	authority, mock, closeDB := mockTenantAuthority(t)
	defer closeDB()
	original := namespaceReserve(testNamespace)
	originalDigest, err := tenantauthorityv1.NamespaceCommandRequestDigest(original)
	if err != nil {
		t.Fatal(err)
	}
	command := mustCompleteCommand(t, terminalCommand{
		CommandID:     testCommand1,
		Kind:          commandKindNamespace,
		RequestDigest: originalDigest,
		Result:        ResultApplied,
		ResultingValue: ValueReference{
			Present: true,
			Digest:  mustNamespaceDigest(t, original.DesiredNamespaceValue),
		},
		Sequence: 4,
	})
	changed := namespaceReserve("com.example.other")

	mock.ExpectBegin()
	expectPublicationAdmissionFence(mock)
	mock.ExpectQuery(fenceLockQuery).WithArgs(authorityFenceKey).WillReturnRows(
		sqlmock.NewRows([]string{"FenceKey", "CommandSequence"}).
			AddRow(authorityFenceKey, []byte("9")),
	)
	expectCommandRow(mock, command)
	mock.ExpectRollback()

	if _, err := authority.ApplyNamespaceCommand(
		context.Background(),
		testCommand1,
		changed,
	); !errors.Is(err, ErrConflict) {
		t.Fatalf("changed request replay = %v", err)
	}
	assertExpectations(t, mock)
}

func TestNewRejectsNilDatabase(t *testing.T) {
	if _, err := New(nil); err == nil {
		t.Fatal("New(nil) succeeded")
	}
}

func mockTenantAuthority(
	t *testing.T,
) (*Authority, sqlmock.Sqlmock, func()) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	authority, err := New(db)
	if err != nil {
		db.Close()
		t.Fatal(err)
	}
	return authority, mock, func() {
		_ = db.Close()
	}
}

func expectNewCommand(
	mock sqlmock.Sqlmock,
	commandID tenantauthorityv1.CommandID,
	priorSequence uint64,
) {
	mock.ExpectBegin()
	expectPublicationAdmissionFence(mock)
	mock.ExpectQuery(fenceLockQuery).WithArgs(authorityFenceKey).WillReturnRows(
		sqlmock.NewRows([]string{"FenceKey", "CommandSequence"}).
			AddRow(authorityFenceKey, []byte(uintText(priorSequence))),
	)
	mock.ExpectQuery(commandLockQuery).WithArgs(string(commandID)).
		WillReturnError(sql.ErrNoRows)
}

func expectPublicationAdmissionFence(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(publicationAdmissionFenceLockQuery).
		WithArgs(catalogadmissionfence.Key).
		WillReturnRows(sqlmock.NewRows([]string{
			"FenceKey",
			"FenceGeneration",
		}).AddRow(catalogadmissionfence.Key, []byte("0")))
}

func expectActiveNamespace(mock sqlmock.Sqlmock, digest string) {
	mock.ExpectQuery(namespaceLockQuery).WithArgs(testNamespace).WillReturnRows(
		sqlmock.NewRows(namespaceCurrentColumns()).AddRow(
			[]byte(testNamespace),
			tenantauthorityv1.NamespaceReservationValueVersion,
			"ACTIVE",
			digest,
			string(testCommand2),
		),
	)
}

func expectFinishCommand(
	mock sqlmock.Sqlmock,
	priorSequence uint64,
	command terminalCommand,
) {
	mock.ExpectExec("UPDATE _System_CatalogTenantAuthorityFence").
		WithArgs(
			uintText(command.Sequence),
			authorityFenceKey,
			uintText(priorSequence),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec("INSERT INTO _System_CatalogTenantAuthorityCommand").
		WithArgs(
			string(command.CommandID),
			string(command.Kind),
			command.RequestDigest,
			string(command.Result),
			command.ObservedPrior.Present,
			nullableReference(command.ObservedPrior),
			command.ResultingValue.Present,
			nullableReference(command.ResultingValue),
			command.ObservedNamespace.Present,
			nullableReference(command.ObservedNamespace),
			command.ObservedBinding.Present,
			nullableReference(command.ObservedBinding),
			uintText(command.Sequence),
			command.OutcomeDigest,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
}

func expectCommandRow(mock sqlmock.Sqlmock, command terminalCommand) {
	mock.ExpectQuery(commandLockQuery).WithArgs(string(command.CommandID)).WillReturnRows(
		sqlmock.NewRows(commandColumns()).AddRow(
			string(command.CommandID),
			string(command.Kind),
			command.RequestDigest,
			string(command.Result),
			command.ObservedPrior.Present,
			nullableReference(command.ObservedPrior),
			command.ResultingValue.Present,
			nullableReference(command.ResultingValue),
			command.ObservedNamespace.Present,
			nullableReference(command.ObservedNamespace),
			command.ObservedBinding.Present,
			nullableReference(command.ObservedBinding),
			[]byte(uintText(command.Sequence)),
			command.OutcomeDigest,
		),
	)
}

func mustCompleteCommand(t *testing.T, command terminalCommand) terminalCommand {
	t.Helper()
	completed, err := completeTerminalCommand(command)
	if err != nil {
		t.Fatal(err)
	}
	return completed
}

func commandColumns() []string {
	return []string{
		"CommandId", "CommandKind", "RequestDigest", "Result",
		"ObservedPriorPresent", "ObservedPriorDigest",
		"ResultingValuePresent", "ResultingValueDigest",
		"ObservedNamespacePresent", "ObservedNamespaceDigest",
		"ObservedBindingPresent", "ObservedBindingDigest",
		"CommandSequence", "OutcomeDigest",
	}
}

func namespaceCurrentColumns() []string {
	return []string{
		"TenantNamespaceId", "ValueVersion", "State", "ValueDigest", "LastCommandId",
	}
}

func bindingCurrentColumns() []string {
	return []string{
		"OrgId", "TenantNamespaceId", "ValueVersion", "State", "ValueDigest", "LastCommandId",
	}
}

func slotCurrentColumns() []string {
	return []string{
		"TenantNamespaceId", "LocalComponentKey", "BundleKeyVersion",
		"SlotOwnerKind", "ValueVersion", "State", "ValueDigest", "LastCommandId",
	}
}

func assertExpectations(t *testing.T, mock sqlmock.Sqlmock) {
	t.Helper()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}