package catalogverificationtx

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

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

const (
	testTenantOwnerOrgID       = "org-tenant-1"
	testTenantOwnerNamespace   = "io.example.tenant"
	testTenantNamespaceCommand = "018f0000-0000-7000-9000-000000000101"
	testTenantBindingCommand   = "018f0000-0000-7000-9000-000000000102"
	testTenantSlotCommand      = "018f0000-0000-7000-9000-000000000103"
)

const tenantOwnerNamespaceQuery = `SELECT TenantNamespaceId, ValueVersion, State,
ValueDigest, LastCommandId
FROM _System_CatalogTenantNamespaceReservation
WHERE TenantNamespaceId = ? FOR UPDATE`

const tenantOwnerBindingQuery = `SELECT OrgId, TenantNamespaceId, ValueVersion,
State, ValueDigest, LastCommandId
FROM _System_CatalogTenantOrgNamespaceBinding WHERE OrgId = ? FOR UPDATE`

const tenantOwnerSlotQuery = `SELECT TenantNamespaceId, LocalComponentKey,
BundleKeyVersion, SlotOwnerKind, ValueVersion, State, ValueDigest, LastCommandId
FROM _System_CatalogTenantLayoutBundleSlot
WHERE TenantNamespaceId = ? AND LocalComponentKey = ? FOR UPDATE`

const tenantOwnerNamespaceHistoryQuery = `SELECT CommandId, CommandSequence,
TenantNamespaceId, ValueVersion, State, ValueDigest, PriorValueDigest
FROM _System_CatalogTenantNamespaceHistory
WHERE TenantNamespaceId = ? ORDER BY CommandSequence FOR UPDATE`

const tenantOwnerBindingHistoryQuery = `SELECT CommandId, CommandSequence,
OrgId, TenantNamespaceId, ValueVersion, State, ValueDigest, PriorValueDigest
FROM _System_CatalogTenantOrgNamespaceBindingHistory
WHERE OrgId = ? ORDER BY CommandSequence FOR UPDATE`

const tenantOwnerSlotHistoryQuery = `SELECT CommandId, CommandSequence,
TenantNamespaceId, LocalComponentKey, BundleKeyVersion, SlotOwnerKind,
ValueVersion, State, ValueDigest, PriorValueDigest
FROM _System_CatalogTenantLayoutBundleSlotHistory
WHERE TenantNamespaceId = ? AND LocalComponentKey = ? ORDER BY CommandSequence FOR UPDATE`

const tenantOwnerCommandQuery = `SELECT CommandId, CommandKind, RequestDigest, Result,
ObservedPriorPresent, ObservedPriorDigest, ResultingValuePresent, ResultingValueDigest,
ObservedNamespacePresent, ObservedNamespaceDigest,
ObservedBindingPresent, ObservedBindingDigest,
CommandSequence, OutcomeDigest
FROM _System_CatalogTenantAuthorityCommand WHERE CommandId = ? FOR UPDATE`

const tenantOwnerRetirementQuery = `SELECT CommandId
FROM _System_CatalogTenantAuthorityCommand
WHERE Result = 'APPLIED' AND ResultingValuePresent = TRUE
AND ResultingValueDigest IN (?, ?, ?)
ORDER BY CommandSequence FOR UPDATE`

type tenantOwnerFixture struct {
	request          TenantOwnerRequest
	namespace        tenantauthorityv1.TenantNamespaceReservationV1
	namespaceDigest  string
	binding          tenantauthorityv1.TenantOrgNamespaceBindingV1
	bindingDigest    string
	slot             tenantauthorityv1.TenantLayoutBundleSlotV1
	slotDigest       string
	namespaceCommand tenantCommand
	bindingCommand   tenantCommand
	slotCommand      tenantCommand
}

func TestVerifyTenantOwnerLocksExactActiveTripleAndCreationLineage(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	db, mock, token := newTenantOwnerVerificationToken(t)
	defer db.Close()

	expectTenantOwnerCurrentRows(mock, fixture)
	expectTenantOwnerHistories(mock, fixture)
	expectTenantOwnerCommand(mock, fixture.namespaceCommand)
	expectTenantOwnerCommand(mock, fixture.bindingCommand)
	expectTenantOwnerCommand(mock, fixture.slotCommand)
	expectNoTenantOwnerRetirement(mock, fixture)

	required := requiredBundleKeySetForTest(fixture.request.BundleKey)
	lease, err := NewLease(token, required)
	if err != nil {
		t.Fatal(err)
	}
	facts := tenantSourceFactsForTest(t, TenantSourceFacts{
		OrgID:     fixture.request.OrgID,
		BundleKey: fixture.request.BundleKey,
	}, required)
	claim, err := lease.BeginTenantClaim(
		0,
		facts,
		fixture.request,
	)
	if err != nil {
		t.Fatal(err)
	}
	fact, err := VerifyTenantOwnerClaim(
		context.Background(),
		db,
		claim,
	)
	if err != nil {
		t.Fatal(err)
	}
	if fact.SlotCreationCommandSequence != 3 ||
		fact.NamespaceReservation != fixture.namespace ||
		fact.NamespaceReservationDigest != fixture.namespaceDigest ||
		fact.OrgBinding != fixture.binding ||
		fact.OrgBindingDigest != fixture.bindingDigest ||
		fact.Slot != fixture.slot ||
		fact.SlotDigest != fixture.slotDigest {
		t.Fatalf("tenant owner fact = %#v", fact)
	}
	if _, err := VerifyTenantOwnerClaim(
		context.Background(),
		db,
		claim,
	); !errors.Is(err, ErrClosed) {
		t.Fatalf("tenant claim verification replay = %v, want ErrClosed", err)
	}
	if err := CompleteTenantClaim(claim, receiptForTenantFacts(facts)); err != nil {
		t.Fatal(err)
	}
	if !claim.Active() {
		t.Fatal("completed tenant claim is not active")
	}
	if err := CompleteTenantClaim(
		claim,
		receiptForTenantFacts(facts),
	); !errors.Is(err, ErrClosed) {
		t.Fatalf("second tenant claim completion = %v, want ErrClosed", err)
	}

	mock.ExpectRollback()
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	if claim.Active() {
		t.Fatal("tenant claim remained active after source lease close")
	}
	assertTenantOwnerExpectations(t, mock)
}

func TestVerifyTenantOwnerRejectsAbsentCurrentFacts(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	tests := []struct {
		name   string
		expect func(sqlmock.Sqlmock)
	}{
		{
			name: "namespace",
			expect: func(mock sqlmock.Sqlmock) {
				mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerNamespaceQuery)).
					WithArgs(testTenantOwnerNamespace).
					WillReturnRows(sqlmock.NewRows(tenantOwnerNamespaceColumns()))
			},
		},
		{
			name: "binding",
			expect: func(mock sqlmock.Sqlmock) {
				expectTenantOwnerNamespaceCurrent(
					mock,
					fixture.namespace,
					fixture.namespaceDigest,
					testTenantNamespaceCommand,
				)
				mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerBindingQuery)).
					WithArgs(testTenantOwnerOrgID).
					WillReturnRows(sqlmock.NewRows(tenantOwnerBindingColumns()))
			},
		},
		{
			name: "slot",
			expect: func(mock sqlmock.Sqlmock) {
				expectTenantOwnerNamespaceCurrent(
					mock,
					fixture.namespace,
					fixture.namespaceDigest,
					testTenantNamespaceCommand,
				)
				expectTenantOwnerBindingCurrent(
					mock,
					fixture.binding,
					fixture.bindingDigest,
					testTenantBindingCommand,
				)
				mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerSlotQuery)).
					WithArgs(
						testTenantOwnerNamespace,
						string(fixture.request.BundleKey.LocalComponentKey),
					).
					WillReturnRows(sqlmock.NewRows(tenantOwnerSlotColumns()))
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, token := newTenantOwnerVerificationToken(t)
			defer db.Close()
			test.expect(mock)

			if _, err := VerifyTenantOwner(
				context.Background(),
				token,
				db,
				fixture.request,
			); !errors.Is(err, ErrTenantOwnerNotFound) {
				t.Fatalf("absent %s error = %v, want ErrTenantOwnerNotFound", test.name, err)
			}

			closeTenantOwnerVerificationToken(t, mock, token)
		})
	}
}

func TestVerifyTenantOwnerRejectsTombstonedCurrentFacts(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	tests := []struct {
		name   string
		expect func(*testing.T, sqlmock.Sqlmock)
	}{
		{
			name: "namespace",
			expect: func(t *testing.T, mock sqlmock.Sqlmock) {
				value := fixture.namespace
				value.State = tenantauthorityv1.NamespaceStateTombstone
				expectTenantOwnerNamespaceCurrent(
					mock,
					value,
					mustTenantNamespaceDigest(t, value),
					"018f0000-0000-7000-9000-000000000104",
				)
			},
		},
		{
			name: "binding",
			expect: func(t *testing.T, mock sqlmock.Sqlmock) {
				expectTenantOwnerNamespaceCurrent(
					mock,
					fixture.namespace,
					fixture.namespaceDigest,
					testTenantNamespaceCommand,
				)
				value := fixture.binding
				value.State = tenantauthorityv1.OrgBindingStateTombstone
				expectTenantOwnerBindingCurrent(
					mock,
					value,
					mustTenantBindingDigest(t, value),
					"018f0000-0000-7000-9000-000000000105",
				)
			},
		},
		{
			name: "slot",
			expect: func(t *testing.T, mock sqlmock.Sqlmock) {
				expectTenantOwnerNamespaceCurrent(
					mock,
					fixture.namespace,
					fixture.namespaceDigest,
					testTenantNamespaceCommand,
				)
				expectTenantOwnerBindingCurrent(
					mock,
					fixture.binding,
					fixture.bindingDigest,
					testTenantBindingCommand,
				)
				value := fixture.slot
				value.State = tenantauthorityv1.SlotStateTombstone
				expectTenantOwnerSlotCurrent(
					mock,
					value,
					mustTenantSlotDigest(t, value),
					"018f0000-0000-7000-9000-000000000106",
				)
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, token := newTenantOwnerVerificationToken(t)
			defer db.Close()
			test.expect(t, mock)

			if _, err := VerifyTenantOwner(
				context.Background(),
				token,
				db,
				fixture.request,
			); !errors.Is(err, ErrTenantOwnerConflict) {
				t.Fatalf("tombstoned %s error = %v, want ErrTenantOwnerConflict", test.name, err)
			}

			closeTenantOwnerVerificationToken(t, mock, token)
		})
	}
}

func TestVerifyTenantOwnerRejectsForeignNamespaceBinding(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	db, mock, token := newTenantOwnerVerificationToken(t)
	defer db.Close()

	expectTenantOwnerNamespaceCurrent(
		mock,
		fixture.namespace,
		fixture.namespaceDigest,
		testTenantNamespaceCommand,
	)
	foreign := fixture.binding
	foreign.TenantNamespaceID = "io.example.foreign"
	expectTenantOwnerBindingCurrent(
		mock,
		foreign,
		mustTenantBindingDigest(t, foreign),
		testTenantBindingCommand,
	)

	if _, err := VerifyTenantOwner(
		context.Background(),
		token,
		db,
		fixture.request,
	); !errors.Is(err, ErrTenantOwnerConflict) {
		t.Fatalf("foreign namespace binding error = %v, want ErrTenantOwnerConflict", err)
	}

	closeTenantOwnerVerificationToken(t, mock, token)
}

func TestVerifyTenantOwnerRejectsCurrentDigestMismatch(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	db, mock, token := newTenantOwnerVerificationToken(t)
	defer db.Close()

	expectTenantOwnerNamespaceCurrent(
		mock,
		fixture.namespace,
		fixture.namespaceDigest,
		testTenantNamespaceCommand,
	)
	expectTenantOwnerBindingCurrent(
		mock,
		fixture.binding,
		fixture.bindingDigest,
		testTenantBindingCommand,
	)
	expectTenantOwnerSlotCurrent(
		mock,
		fixture.slot,
		tenantOwnerTestDigest("0"),
		testTenantSlotCommand,
	)

	if _, err := VerifyTenantOwner(
		context.Background(),
		token,
		db,
		fixture.request,
	); !errors.Is(err, ErrTenantOwnerCorrupt) {
		t.Fatalf("slot digest mismatch error = %v, want ErrTenantOwnerCorrupt", err)
	}

	closeTenantOwnerVerificationToken(t, mock, token)
}

func TestVerifyTenantOwnerRejectsDuplicateCreationHistory(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	db, mock, token := newTenantOwnerVerificationToken(t)
	defer db.Close()

	expectTenantOwnerCurrentRows(mock, fixture)
	rows := sqlmock.NewRows(tenantOwnerNamespaceHistoryColumns()).
		AddRow(
			testTenantNamespaceCommand,
			[]byte("1"),
			[]byte(fixture.namespace.TenantNamespaceID),
			fixture.namespace.ValueVersion,
			string(fixture.namespace.State),
			fixture.namespaceDigest,
			nil,
		).
		AddRow(
			"018f0000-0000-7000-9000-000000000104",
			[]byte("4"),
			[]byte(fixture.namespace.TenantNamespaceID),
			fixture.namespace.ValueVersion,
			string(tenantauthorityv1.NamespaceStateTombstone),
			tenantOwnerTestDigest("4"),
			fixture.namespaceDigest,
		)
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerNamespaceHistoryQuery)).
		WithArgs(fixture.namespace.TenantNamespaceID).
		WillReturnRows(rows)

	if _, err := VerifyTenantOwner(
		context.Background(),
		token,
		db,
		fixture.request,
	); !errors.Is(err, ErrTenantOwnerCorrupt) {
		t.Fatalf("duplicate namespace history error = %v, want ErrTenantOwnerCorrupt", err)
	}

	closeTenantOwnerVerificationToken(t, mock, token)
}

func TestVerifyTenantOwnerRequiresCausalCreationOrder(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	tests := []struct {
		name              string
		namespaceSequence uint64
		bindingSequence   uint64
		slotSequence      uint64
	}{
		{
			name:              "binding precedes namespace",
			namespaceSequence: 2,
			bindingSequence:   1,
			slotSequence:      3,
		},
		{
			name:              "slot precedes binding",
			namespaceSequence: 1,
			bindingSequence:   3,
			slotSequence:      2,
		},
		{
			name:              "dependency sequences are equal",
			namespaceSequence: 1,
			bindingSequence:   1,
			slotSequence:      2,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, token := newTenantOwnerVerificationToken(t)
			defer db.Close()

			expectTenantOwnerCurrentRows(mock, fixture)
			expectTenantOwnerHistoriesWithSequences(
				mock,
				fixture,
				test.namespaceSequence,
				test.bindingSequence,
				test.slotSequence,
			)

			if _, err := VerifyTenantOwner(
				context.Background(),
				token,
				db,
				fixture.request,
			); !errors.Is(err, ErrTenantOwnerCorrupt) {
				t.Fatalf(
					"causally impossible creation order error = %v, want ErrTenantOwnerCorrupt",
					err,
				)
			}

			closeTenantOwnerVerificationToken(t, mock, token)
		})
	}
}

func TestVerifyTenantOwnerRequiresAppliedCreationCommand(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	db, mock, token := newTenantOwnerVerificationToken(t)
	defer db.Close()

	expectTenantOwnerCurrentRows(mock, fixture)
	expectTenantOwnerHistories(mock, fixture)
	notApplied := fixture.namespaceCommand
	notApplied.result = "EXPECTED_PRIOR_MISMATCH"
	notApplied.resultingValue = tenantReference{}
	notApplied.outcomeDigest = mustTenantCommandOutcomeDigest(t, notApplied)
	expectTenantOwnerCommand(mock, notApplied)

	if _, err := VerifyTenantOwner(
		context.Background(),
		token,
		db,
		fixture.request,
	); !errors.Is(err, ErrTenantOwnerCorrupt) {
		t.Fatalf("non-APPLIED creation error = %v, want ErrTenantOwnerCorrupt", err)
	}

	closeTenantOwnerVerificationToken(t, mock, token)
}

func TestVerifyTenantOwnerRejectsSuccessfulRetirementOutcome(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	db, mock, token := newTenantOwnerVerificationToken(t)
	defer db.Close()

	expectTenantOwnerCurrentRows(mock, fixture)
	expectTenantOwnerHistories(mock, fixture)
	expectTenantOwnerCommand(mock, fixture.namespaceCommand)
	expectTenantOwnerCommand(mock, fixture.bindingCommand)
	expectTenantOwnerCommand(mock, fixture.slotCommand)
	expectTenantOwnerRetirement(mock, fixture)

	if _, err := VerifyTenantOwner(
		context.Background(),
		token,
		db,
		fixture.request,
	); !errors.Is(err, ErrTenantOwnerCorrupt) {
		t.Fatalf("retirement outcome error = %v, want ErrTenantOwnerCorrupt", err)
	}

	closeTenantOwnerVerificationToken(t, mock, token)
}

func TestVerifyTenantOwnerClaimRejectsUseAfterLeaseClose(t *testing.T) {
	fixture := newTenantOwnerFixture(t)
	db, mock, token := newTenantOwnerVerificationToken(t)
	defer db.Close()
	required := requiredBundleKeySetForTest(fixture.request.BundleKey)
	lease, err := NewLease(token, required)
	if err != nil {
		t.Fatal(err)
	}
	facts := tenantSourceFactsForTest(t, TenantSourceFacts{
		OrgID:     fixture.request.OrgID,
		BundleKey: fixture.request.BundleKey,
	}, required)
	claim, err := lease.BeginTenantClaim(
		0,
		facts,
		fixture.request,
	)
	if err != nil {
		t.Fatal(err)
	}

	mock.ExpectRollback()
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	if _, err := VerifyTenantOwnerClaim(
		context.Background(),
		db,
		claim,
	); !errors.Is(err, ErrClosed) {
		t.Fatalf("verification after source close = %v, want ErrClosed", err)
	}
	assertTenantOwnerExpectations(t, mock)
}

func newTenantOwnerFixture(t *testing.T) tenantOwnerFixture {
	t.Helper()
	key := layoutbundlekeyv1.BundleKey{
		SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKindTenant,
		SlotOwnerNamespace: testTenantOwnerNamespace,
		LocalComponentKey: layoutbundlekeyv1.LocalComponentKey(
			"key256:" + strings.Repeat("1", 64),
		),
	}
	fixture := tenantOwnerFixture{
		request: TenantOwnerRequest{
			OrgID:     testTenantOwnerOrgID,
			BundleKey: key,
		},
		namespace: tenantauthorityv1.TenantNamespaceReservationV1{
			ValueVersion:      tenantauthorityv1.NamespaceReservationValueVersion,
			TenantNamespaceID: testTenantOwnerNamespace,
			State:             tenantauthorityv1.NamespaceStateActive,
		},
		binding: tenantauthorityv1.TenantOrgNamespaceBindingV1{
			ValueVersion:      tenantauthorityv1.OrgNamespaceBindingValueVersion,
			OrgID:             testTenantOwnerOrgID,
			TenantNamespaceID: testTenantOwnerNamespace,
			State:             tenantauthorityv1.OrgBindingStateActive,
		},
		slot: tenantauthorityv1.TenantLayoutBundleSlotV1{
			ValueVersion: tenantauthorityv1.LayoutBundleSlotValueVersion,
			BundleKey:    tenantBundleKey(key),
			State:        tenantauthorityv1.SlotStateActive,
		},
	}
	fixture.namespaceDigest = mustTenantNamespaceDigest(t, fixture.namespace)
	fixture.bindingDigest = mustTenantBindingDigest(t, fixture.binding)
	fixture.slotDigest = mustTenantSlotDigest(t, fixture.slot)

	namespaceRequest := tenantauthorityv1.TenantNamespaceCommandRequest{
		Operation:               tenantauthorityv1.NamespaceOperationReserve,
		TargetTenantNamespaceID: fixture.namespace.TenantNamespaceID,
		DesiredNamespaceValue:   fixture.namespace,
		ExpectedPriorNamespace: tenantauthorityv1.TenantNamespaceExpectedPrior{
			Variant: tenantauthorityv1.NamespaceExpectedPriorAbsent,
		},
	}
	namespaceRequestDigest, err := tenantauthorityv1.NamespaceCommandRequestDigest(
		namespaceRequest,
	)
	if err != nil {
		t.Fatal(err)
	}
	fixture.namespaceCommand = tenantCommand{
		commandID:     testTenantNamespaceCommand,
		kind:          "NAMESPACE",
		requestDigest: namespaceRequestDigest,
		result:        "APPLIED",
		resultingValue: tenantReference{
			present: true,
			digest:  fixture.namespaceDigest,
		},
		sequence: 1,
	}
	fixture.namespaceCommand.outcomeDigest = mustTenantCommandOutcomeDigest(
		t,
		fixture.namespaceCommand,
	)

	bindingRequest := tenantauthorityv1.TenantOrgNamespaceBindingCommandRequest{
		Operation:                          tenantauthorityv1.BindingOperationBind,
		TargetOrgID:                        fixture.binding.OrgID,
		DesiredBindingValue:                fixture.binding,
		ExpectedNamespaceReservationDigest: fixture.namespaceDigest,
		ExpectedPriorBinding: tenantauthorityv1.TenantOrgBindingExpectedPrior{
			Variant: tenantauthorityv1.BindingExpectedPriorAbsent,
		},
	}
	bindingRequestDigest, err := tenantauthorityv1.OrgBindingCommandRequestDigest(
		bindingRequest,
	)
	if err != nil {
		t.Fatal(err)
	}
	fixture.bindingCommand = tenantCommand{
		commandID:     testTenantBindingCommand,
		kind:          "ORG_BINDING",
		requestDigest: bindingRequestDigest,
		result:        "APPLIED",
		resultingValue: tenantReference{
			present: true,
			digest:  fixture.bindingDigest,
		},
		observedNamespace: tenantReference{
			present: true,
			digest:  fixture.namespaceDigest,
		},
		sequence: 2,
	}
	fixture.bindingCommand.outcomeDigest = mustTenantCommandOutcomeDigest(
		t,
		fixture.bindingCommand,
	)

	slotRequest := tenantauthorityv1.TenantLayoutBundleSlotCommandRequest{
		Operation:                           tenantauthorityv1.SlotOperationCreate,
		AuthorizingOrgID:                    fixture.binding.OrgID,
		TargetBundleKey:                     fixture.slot.BundleKey,
		DesiredSlotValue:                    fixture.slot,
		ExpectedNamespaceReservationDigest:  fixture.namespaceDigest,
		ExpectedAuthorizingOrgBindingDigest: fixture.bindingDigest,
		ExpectedPriorSlot: tenantauthorityv1.TenantSlotExpectedPrior{
			Variant: tenantauthorityv1.SlotExpectedPriorAbsent,
		},
	}
	slotRequestDigest, err := tenantauthorityv1.SlotCommandRequestDigest(slotRequest)
	if err != nil {
		t.Fatal(err)
	}
	fixture.slotCommand = tenantCommand{
		commandID:     testTenantSlotCommand,
		kind:          "SLOT",
		requestDigest: slotRequestDigest,
		result:        "APPLIED",
		resultingValue: tenantReference{
			present: true,
			digest:  fixture.slotDigest,
		},
		observedNamespace: tenantReference{
			present: true,
			digest:  fixture.namespaceDigest,
		},
		observedBinding: tenantReference{
			present: true,
			digest:  fixture.bindingDigest,
		},
		sequence: 3,
	}
	fixture.slotCommand.outcomeDigest = mustTenantCommandOutcomeDigest(
		t,
		fixture.slotCommand,
	)
	return fixture
}

func newTenantOwnerVerificationToken(
	t *testing.T,
) (*sql.DB, sqlmock.Sqlmock, *Token) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		db.Close()
		t.Fatal(err)
	}
	token := &Token{db: db, tx: tx, active: true}
	return db, mock, token
}

func expectTenantOwnerCurrentRows(
	mock sqlmock.Sqlmock,
	fixture tenantOwnerFixture,
) {
	expectTenantOwnerNamespaceCurrent(
		mock,
		fixture.namespace,
		fixture.namespaceDigest,
		testTenantNamespaceCommand,
	)
	expectTenantOwnerBindingCurrent(
		mock,
		fixture.binding,
		fixture.bindingDigest,
		testTenantBindingCommand,
	)
	expectTenantOwnerSlotCurrent(
		mock,
		fixture.slot,
		fixture.slotDigest,
		testTenantSlotCommand,
	)
}

func expectTenantOwnerNamespaceCurrent(
	mock sqlmock.Sqlmock,
	value tenantauthorityv1.TenantNamespaceReservationV1,
	digest string,
	commandID string,
) {
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerNamespaceQuery)).
		WithArgs(value.TenantNamespaceID).
		WillReturnRows(sqlmock.NewRows(tenantOwnerNamespaceColumns()).AddRow(
			[]byte(value.TenantNamespaceID),
			value.ValueVersion,
			string(value.State),
			digest,
			commandID,
		))
}

func expectTenantOwnerBindingCurrent(
	mock sqlmock.Sqlmock,
	value tenantauthorityv1.TenantOrgNamespaceBindingV1,
	digest string,
	commandID string,
) {
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerBindingQuery)).
		WithArgs(value.OrgID).
		WillReturnRows(sqlmock.NewRows(tenantOwnerBindingColumns()).AddRow(
			[]byte(value.OrgID),
			[]byte(value.TenantNamespaceID),
			value.ValueVersion,
			string(value.State),
			digest,
			commandID,
		))
}

func expectTenantOwnerSlotCurrent(
	mock sqlmock.Sqlmock,
	value tenantauthorityv1.TenantLayoutBundleSlotV1,
	digest string,
	commandID string,
) {
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerSlotQuery)).
		WithArgs(
			value.BundleKey.SlotOwnerNamespace,
			string(value.BundleKey.LocalComponentKey),
		).
		WillReturnRows(sqlmock.NewRows(tenantOwnerSlotColumns()).AddRow(
			[]byte(value.BundleKey.SlotOwnerNamespace),
			[]byte(value.BundleKey.LocalComponentKey),
			value.BundleKey.BundleKeyVersion,
			string(value.BundleKey.SlotOwnerKind),
			value.ValueVersion,
			string(value.State),
			digest,
			commandID,
		))
}

func expectTenantOwnerHistories(
	mock sqlmock.Sqlmock,
	fixture tenantOwnerFixture,
) {
	expectTenantOwnerHistoriesWithSequences(mock, fixture, 1, 2, 3)
}

func expectTenantOwnerHistoriesWithSequences(
	mock sqlmock.Sqlmock,
	fixture tenantOwnerFixture,
	namespaceSequence uint64,
	bindingSequence uint64,
	slotSequence uint64,
) {
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerNamespaceHistoryQuery)).
		WithArgs(fixture.namespace.TenantNamespaceID).
		WillReturnRows(sqlmock.NewRows(tenantOwnerNamespaceHistoryColumns()).AddRow(
			testTenantNamespaceCommand,
			[]byte(tenantUintText(namespaceSequence)),
			[]byte(fixture.namespace.TenantNamespaceID),
			fixture.namespace.ValueVersion,
			string(fixture.namespace.State),
			fixture.namespaceDigest,
			nil,
		))
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerBindingHistoryQuery)).
		WithArgs(fixture.binding.OrgID).
		WillReturnRows(sqlmock.NewRows(tenantOwnerBindingHistoryColumns()).AddRow(
			testTenantBindingCommand,
			[]byte(tenantUintText(bindingSequence)),
			[]byte(fixture.binding.OrgID),
			[]byte(fixture.binding.TenantNamespaceID),
			fixture.binding.ValueVersion,
			string(fixture.binding.State),
			fixture.bindingDigest,
			nil,
		))
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerSlotHistoryQuery)).
		WithArgs(
			fixture.slot.BundleKey.SlotOwnerNamespace,
			string(fixture.slot.BundleKey.LocalComponentKey),
		).
		WillReturnRows(sqlmock.NewRows(tenantOwnerSlotHistoryColumns()).AddRow(
			testTenantSlotCommand,
			[]byte(tenantUintText(slotSequence)),
			[]byte(fixture.slot.BundleKey.SlotOwnerNamespace),
			[]byte(fixture.slot.BundleKey.LocalComponentKey),
			fixture.slot.BundleKey.BundleKeyVersion,
			string(fixture.slot.BundleKey.SlotOwnerKind),
			fixture.slot.ValueVersion,
			string(fixture.slot.State),
			fixture.slotDigest,
			nil,
		))
}

func expectTenantOwnerCommand(mock sqlmock.Sqlmock, command tenantCommand) {
	priorPresent, priorDigest := tenantReferenceSQL(command.observedPrior)
	resultingPresent, resultingDigest := tenantReferenceSQL(command.resultingValue)
	namespacePresent, namespaceDigest := tenantReferenceSQL(command.observedNamespace)
	bindingPresent, bindingDigest := tenantReferenceSQL(command.observedBinding)
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerCommandQuery)).
		WithArgs(string(command.commandID)).
		WillReturnRows(sqlmock.NewRows(tenantOwnerCommandColumns()).AddRow(
			string(command.commandID),
			command.kind,
			command.requestDigest,
			command.result,
			priorPresent,
			priorDigest,
			resultingPresent,
			resultingDigest,
			namespacePresent,
			namespaceDigest,
			bindingPresent,
			bindingDigest,
			[]byte(tenantUintText(command.sequence)),
			command.outcomeDigest,
		))
}

func expectNoTenantOwnerRetirement(
	mock sqlmock.Sqlmock,
	fixture tenantOwnerFixture,
) {
	namespaceDigest, bindingDigest, slotDigest := tenantOwnerTombstoneDigests(
		nil,
		fixture,
	)
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerRetirementQuery)).
		WithArgs(namespaceDigest, bindingDigest, slotDigest).
		WillReturnRows(sqlmock.NewRows([]string{"CommandId"}))
}

func expectTenantOwnerRetirement(
	mock sqlmock.Sqlmock,
	fixture tenantOwnerFixture,
) {
	namespaceDigest, bindingDigest, slotDigest := tenantOwnerTombstoneDigests(
		nil,
		fixture,
	)
	mock.ExpectQuery(regexp.QuoteMeta(tenantOwnerRetirementQuery)).
		WithArgs(namespaceDigest, bindingDigest, slotDigest).
		WillReturnRows(sqlmock.NewRows([]string{"CommandId"}).AddRow(
			"018f0000-0000-7000-9000-000000000104",
		))
}

func tenantOwnerTombstoneDigests(
	t *testing.T,
	fixture tenantOwnerFixture,
) (string, string, string) {
	namespace := fixture.namespace
	namespace.State = tenantauthorityv1.NamespaceStateTombstone
	binding := fixture.binding
	binding.State = tenantauthorityv1.OrgBindingStateTombstone
	slot := fixture.slot
	slot.State = tenantauthorityv1.SlotStateTombstone
	if t == nil {
		namespaceDigest, namespaceErr := tenantauthorityv1.NamespaceValueDigest(namespace)
		bindingDigest, bindingErr := tenantauthorityv1.OrgBindingValueDigest(binding)
		slotDigest, slotErr := tenantauthorityv1.SlotValueDigest(slot)
		if namespaceErr != nil || bindingErr != nil || slotErr != nil {
			panic("valid tenant owner fixture could not derive tombstone digests")
		}
		return namespaceDigest, bindingDigest, slotDigest
	}
	return mustTenantNamespaceDigest(t, namespace),
		mustTenantBindingDigest(t, binding),
		mustTenantSlotDigest(t, slot)
}

func tenantReferenceSQL(reference tenantReference) (bool, any) {
	if !reference.present {
		return false, nil
	}
	return true, reference.digest
}

func closeTenantOwnerVerificationToken(
	t *testing.T,
	mock sqlmock.Sqlmock,
	token *Token,
) {
	t.Helper()
	mock.ExpectRollback()
	if err := token.Close(); err != nil {
		t.Fatal(err)
	}
	assertTenantOwnerExpectations(t, mock)
}

func mustTenantNamespaceDigest(
	t *testing.T,
	value tenantauthorityv1.TenantNamespaceReservationV1,
) string {
	t.Helper()
	digest, err := tenantauthorityv1.NamespaceValueDigest(value)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func mustTenantBindingDigest(
	t *testing.T,
	value tenantauthorityv1.TenantOrgNamespaceBindingV1,
) string {
	t.Helper()
	digest, err := tenantauthorityv1.OrgBindingValueDigest(value)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func mustTenantSlotDigest(
	t *testing.T,
	value tenantauthorityv1.TenantLayoutBundleSlotV1,
) string {
	t.Helper()
	digest, err := tenantauthorityv1.SlotValueDigest(value)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func mustTenantCommandOutcomeDigest(t *testing.T, command tenantCommand) string {
	t.Helper()
	digest, err := tenantCommandOutcomeDigest(command)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

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

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

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

func tenantOwnerNamespaceHistoryColumns() []string {
	return []string{
		"CommandId",
		"CommandSequence",
		"TenantNamespaceId",
		"ValueVersion",
		"State",
		"ValueDigest",
		"PriorValueDigest",
	}
}

func tenantOwnerBindingHistoryColumns() []string {
	return []string{
		"CommandId",
		"CommandSequence",
		"OrgId",
		"TenantNamespaceId",
		"ValueVersion",
		"State",
		"ValueDigest",
		"PriorValueDigest",
	}
}

func tenantOwnerSlotHistoryColumns() []string {
	return []string{
		"CommandId",
		"CommandSequence",
		"TenantNamespaceId",
		"LocalComponentKey",
		"BundleKeyVersion",
		"SlotOwnerKind",
		"ValueVersion",
		"State",
		"ValueDigest",
		"PriorValueDigest",
	}
}

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

func tenantOwnerTestDigest(digit string) string {
	return "sha256:" + strings.Repeat(digit, 64)
}

func tenantUintText(value uint64) string {
	const digits = "0123456789"
	if value == 0 {
		return "0"
	}
	var encoded [20]byte
	index := len(encoded)
	for value > 0 {
		index--
		encoded[index] = digits[value%10]
		value /= 10
	}
	return string(encoded[index:])
}

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