package catalogpublication

import (
	"context"
	"encoding/json"
	"errors"
	"reflect"
	"regexp"
	"strings"
	"testing"
	"time"

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

func TestCandidateSourceVerificationLeaseRetainsExactSourceAndHasNoWireForm(
	t *testing.T,
) {
	authority, mock, closeDB, record, handle := sourceVerificationFixture(t)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)

	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	appKey := requiredAppKey(t, record, "io.example.alpha")
	claim, err := lease.BeginAppBundleKeyVerification(appKey)
	if err != nil {
		t.Fatal(err)
	}
	facts, err := claim.Facts()
	if err != nil {
		t.Fatal(err)
	}
	if facts.SourceCatalogSnapshot != record.CatalogSnapshotBinding ||
		facts.OrgID != record.OrgID ||
		facts.WriterCommandID != record.WriterCommandID ||
		facts.WriterReservationGeneration != record.WriterReservationGeneration ||
		facts.SourceRequestDigest != record.RequestDigest ||
		facts.RequiredBundleKeySetDigest != record.RequiredBundleKeySetDigest ||
		facts.BundleKey != appKey ||
		facts.AppID != appKey.SlotOwnerNamespace {
		t.Fatalf("locked source facts = %#v", facts)
	}
	if _, err := json.Marshal(lease); !errors.Is(
		err,
		errSourceVerificationHasNoJSON,
	) {
		t.Fatalf("lease JSON marshal = %v", err)
	}
	if _, err := json.Marshal(claim); !errors.Is(
		err,
		errSourceVerificationHasNoJSON,
	) {
		t.Fatalf("claim JSON marshal = %v", err)
	}
	if !lease.Active() || claim.Active() {
		t.Fatal("pending App claim did not retain only the source transaction")
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	if lease.Active() || claim.Active() {
		t.Fatal("source or owner capability remained active after close")
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationLeaseClaimsExactRequiredTenantKey(
	t *testing.T,
) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet.BundleKeys = []layoutbundlekeyv1.BundleKey{
		testRequiredTenantKey("io.example.tenant", "2"),
	}
	authority, mock, closeDB, record, handle :=
		sourceVerificationFixtureForRequest(t, request)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	tenantKey := requiredTenantKey(t, record, "io.example.tenant")
	claim, err := lease.BeginTenantBundleKeyVerification(tenantKey)
	if err != nil {
		t.Fatal(err)
	}
	facts, err := claim.Facts()
	if err != nil {
		t.Fatal(err)
	}
	if facts.SourceCatalogSnapshot != record.CatalogSnapshotBinding ||
		facts.OrgID != record.OrgID ||
		facts.WriterCommandID != record.WriterCommandID ||
		facts.WriterReservationGeneration != record.WriterReservationGeneration ||
		facts.SourceRequestDigest != record.RequestDigest ||
		facts.RequiredBundleKeySetDigest != record.RequiredBundleKeySetDigest ||
		facts.BundleKey != tenantKey {
		t.Fatalf("locked tenant source facts = %#v", facts)
	}
	if !lease.Active() || claim.Active() {
		t.Fatal("pending tenant claim did not retain only the source transaction")
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	if lease.Active() || claim.Active() {
		t.Fatal("tenant source or owner claim remained active after close")
	}
	if _, err := claim.Facts(); err == nil {
		t.Fatal("closed tenant claim still exposed source facts")
	}
	if _, err := lease.BeginTenantBundleKeyVerification(tenantKey); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("tenant claim after close = %v", err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationLeaseClaimsExactRequiredSystemKey(
	t *testing.T,
) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet.BundleKeys = []layoutbundlekeyv1.BundleKey{
		testRequiredSystemKey("3"),
	}
	authority, mock, closeDB, record, handle :=
		sourceVerificationFixtureForRequest(t, request)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	systemKey := requiredSystemKey(t, record, "3")
	claim, err := lease.BeginSystemBundleKeyVerification(systemKey)
	if err != nil {
		t.Fatal(err)
	}
	facts, err := claim.Facts()
	if err != nil {
		t.Fatal(err)
	}
	if facts.SourceCatalogSnapshot != record.CatalogSnapshotBinding ||
		facts.OrgID != record.OrgID ||
		facts.WriterCommandID != record.WriterCommandID ||
		facts.WriterReservationGeneration != record.WriterReservationGeneration ||
		facts.SourceRequestDigest != record.RequestDigest ||
		facts.RequiredBundleKeySetDigest != record.RequiredBundleKeySetDigest ||
		facts.BundleKey != systemKey {
		t.Fatalf("locked system source facts = %#v", facts)
	}
	if _, err := json.Marshal(claim); !errors.Is(
		err,
		errSourceVerificationHasNoJSON,
	) {
		t.Fatalf("system claim JSON marshal = %v", err)
	}
	if !lease.Active() || claim.Active() {
		t.Fatal("pending system claim did not retain only the source transaction")
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	if lease.Active() || claim.Active() {
		t.Fatal("system source or owner claim remained active after close")
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationCapabilitiesExposeNoStateOrTransaction(t *testing.T) {
	for _, valueType := range []reflect.Type{
		reflect.TypeOf(CandidateSourceVerificationLease{}),
		reflect.TypeOf(CandidateSourceOwnerVerificationCapability{}),
		reflect.TypeOf(CandidateSourceAppVerification{}),
		reflect.TypeOf(CandidateSourceTenantVerification{}),
		reflect.TypeOf(CandidateSourceSystemVerification{}),
	} {
		for index := 0; index < valueType.NumField(); index++ {
			if valueType.Field(index).PkgPath == "" {
				t.Fatalf(
					"%s field %q is exported",
					valueType,
					valueType.Field(index).Name,
				)
			}
		}
	}
	for _, claimType := range []reflect.Type{
		reflect.TypeOf((*CandidateSourceAppVerification)(nil)),
		reflect.TypeOf((*CandidateSourceTenantVerification)(nil)),
		reflect.TypeOf((*CandidateSourceSystemVerification)(nil)),
	} {
		for index := 0; index < claimType.NumMethod(); index++ {
			method := claimType.Method(index)
			for output := 0; output < method.Type.NumOut(); output++ {
				if strings.Contains(
					method.Type.Out(output).String(),
					"catalogverificationtx.Token",
				) {
					t.Fatalf("%s exposes the retained transaction", method.Name)
				}
			}
		}
	}
}

func TestCandidateSourceOwnerVerificationRejectsIncompleteRequiredSet(
	t *testing.T,
) {
	authority, mock, closeDB, record, handle := sourceVerificationFixture(t)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}

	verification, err := lease.CompleteOwnerVerification()
	if !errors.Is(err, ErrSourceNotReady) {
		t.Fatalf("incomplete owner verification = %v", err)
	}
	if verification.Capability != nil {
		t.Fatal("incomplete owner verification returned authority")
	}
	if !lease.Active() {
		t.Fatal("incomplete closure invalidated the retained source lease")
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceOwnerVerificationTreatsPendingClaimAsNotReady(
	t *testing.T,
) {
	authority, mock, closeDB, record, handle := sourceVerificationFixture(t)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	claim, err := lease.BeginAppBundleKeyVerification(
		requiredAppKey(t, record, "io.example.alpha"),
	)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := lease.CompleteOwnerVerification(); !errors.Is(
		err,
		ErrSourceNotReady,
	) {
		t.Fatalf("pending owner verification = %v, want ErrSourceNotReady", err)
	}
	if !lease.Active() {
		t.Fatal("pending owner closure invalidated the retained source lease")
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	claim.Abort()
	assertMockExpectations(t, mock)
}

func TestCandidateSourceOwnerVerificationCompletesEmptyRequiredSet(
	t *testing.T,
) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet = layoutbundlekeyv1.RequiredBundleKeySet{
		Version:    layoutbundlekeyv1.RequiredBundleKeySetVersion,
		BundleKeys: []layoutbundlekeyv1.BundleKey{},
	}
	authority, mock, closeDB, record, handle :=
		sourceVerificationFixtureForRequest(t, request)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}

	verification, err := lease.CompleteOwnerVerification()
	if err != nil {
		t.Fatal(err)
	}
	if verification.Capability == nil {
		t.Fatal("complete empty owner closure omitted its capability")
	}
	if verification.ReceiptSet.Receipts == nil ||
		len(verification.ReceiptSet.Receipts) != 0 {
		t.Fatalf("empty receipt set = %#v", verification.ReceiptSet.Receipts)
	}
	if verification.ReceiptSetDigest == "" {
		t.Fatal("complete empty owner closure omitted its receipt-set digest")
	}
	if err := verification.Capability.Validate(); err != nil {
		t.Fatalf("live empty owner closure = %v", err)
	}
	set, err := verification.Capability.ReceiptSet()
	if err != nil {
		t.Fatal(err)
	}
	if set.Receipts == nil || len(set.Receipts) != 0 {
		t.Fatalf("capability empty receipt set = %#v", set.Receipts)
	}
	digest, err := verification.Capability.ReceiptSetDigest()
	if err != nil {
		t.Fatal(err)
	}
	if digest != verification.ReceiptSetDigest {
		t.Fatalf("capability receipt-set digest = %q, want %q", digest, verification.ReceiptSetDigest)
	}
	verification.ReceiptSet.Receipts = append(
		verification.ReceiptSet.Receipts,
		layoutbundleownerverificationv1.ReceiptEntry{},
	)
	repeated, err := lease.CompleteOwnerVerification()
	if err != nil {
		t.Fatal(err)
	}
	if repeated.Capability != verification.Capability {
		t.Fatal("exact owner closure replay minted a second capability")
	}
	if repeated.ReceiptSet.Receipts == nil || len(repeated.ReceiptSet.Receipts) != 0 {
		t.Fatalf("mutated output changed sealed receipt set: %#v", repeated.ReceiptSet.Receipts)
	}
	if _, err := json.Marshal(verification.Capability); !errors.Is(
		err,
		errSourceVerificationHasNoJSON,
	) {
		t.Fatalf("owner closure JSON marshal = %v", err)
	}
	if err := json.Unmarshal([]byte("{}"), verification.Capability); !errors.Is(
		err,
		errSourceVerificationHasNoJSON,
	) {
		t.Fatalf("owner closure JSON unmarshal = %v", err)
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceOwnerVerificationCapabilityInvalidAfterSourceClose(
	t *testing.T,
) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet = layoutbundlekeyv1.RequiredBundleKeySet{
		Version:    layoutbundlekeyv1.RequiredBundleKeySetVersion,
		BundleKeys: []layoutbundlekeyv1.BundleKey{},
	}
	authority, mock, closeDB, record, handle :=
		sourceVerificationFixtureForRequest(t, request)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	verification, err := lease.CompleteOwnerVerification()
	if err != nil {
		t.Fatal(err)
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	if err := verification.Capability.Validate(); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("closed owner closure validation = %v", err)
	}
	if _, err := verification.Capability.ReceiptSet(); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("closed owner closure receipt set = %v", err)
	}
	if _, err := verification.Capability.ReceiptSetDigest(); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("closed owner closure receipt-set digest = %v", err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationLeaseRejectsWrongOrUnrequiredSystemKey(
	t *testing.T,
) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet.BundleKeys = append(
		request.RequiredBundleKeySet.BundleKeys,
		testRequiredSystemKey("3"),
	)
	authority, mock, closeDB, record, handle :=
		sourceVerificationFixtureForRequest(t, request)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}

	tenantKey := requiredTenantKey(t, record, "io.example.tenant")
	if _, err := lease.BeginSystemBundleKeyVerification(tenantKey); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("tenant key claimed as system = %v", err)
	}
	unknown := testRequiredSystemKey("4")
	if _, err := lease.BeginSystemBundleKeyVerification(unknown); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("unrequired system key = %v", err)
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationLeaseRejectsWrongOrUnrequiredTenantKey(
	t *testing.T,
) {
	authority, mock, closeDB, record, handle := sourceVerificationFixture(t)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}

	appKey := requiredAppKey(t, record, "io.example.alpha")
	if _, err := lease.BeginTenantBundleKeyVerification(appKey); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("App key claimed as tenant = %v", err)
	}
	unknown := requiredTenantKey(t, record, "io.example.tenant")
	unknown.LocalComponentKey = layoutbundlekeyv1.LocalComponentKey(
		"key256:" + strings.Repeat("4", 64),
	)
	if _, err := lease.BeginTenantBundleKeyVerification(unknown); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("unrequired tenant key = %v", err)
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationLeaseAllowsOnlyOnePendingOwnerClaim(
	t *testing.T,
) {
	t.Run("App blocks tenant", func(t *testing.T) {
		authority, mock, closeDB, record, handle := sourceVerificationFixture(t)
		defer closeDB()
		expectCandidateSourceVerificationLock(mock, record)
		lease, err := authority.BeginCandidateSourceVerification(
			context.Background(),
			handle,
		)
		if err != nil {
			t.Fatal(err)
		}
		appClaim, err := lease.BeginAppBundleKeyVerification(
			requiredAppKey(t, record, "io.example.alpha"),
		)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := lease.BeginTenantBundleKeyVerification(
			requiredTenantKey(t, record, "io.example.tenant"),
		); !errors.Is(err, ErrSourceConflict) {
			t.Fatalf("tenant claim while App claim is pending = %v", err)
		}
		if !lease.Active() || appClaim.Active() {
			t.Fatal("pending App claim did not retain only the source transaction")
		}

		mock.ExpectRollback()
		expectCandidateSourceSessionRestore(mock)
		if err := lease.Close(); err != nil {
			t.Fatal(err)
		}
		assertMockExpectations(t, mock)
	})

	t.Run("tenant blocks system", func(t *testing.T) {
		request := validCandidateSourceRequest()
		request.RequiredBundleKeySet.BundleKeys = []layoutbundlekeyv1.BundleKey{
			testRequiredTenantKey("io.example.tenant", "2"),
			testRequiredSystemKey("3"),
		}
		authority, mock, closeDB, record, handle :=
			sourceVerificationFixtureForRequest(t, request)
		defer closeDB()
		expectCandidateSourceVerificationLock(mock, record)
		lease, err := authority.BeginCandidateSourceVerification(
			context.Background(),
			handle,
		)
		if err != nil {
			t.Fatal(err)
		}
		tenantClaim, err := lease.BeginTenantBundleKeyVerification(
			requiredTenantKey(t, record, "io.example.tenant"),
		)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := lease.BeginSystemBundleKeyVerification(
			requiredSystemKey(t, record, "3"),
		); !errors.Is(err, ErrSourceConflict) {
			t.Fatalf("system claim while tenant claim is pending = %v", err)
		}
		if !lease.Active() || tenantClaim.Active() {
			t.Fatal("pending tenant claim did not retain only the source transaction")
		}

		mock.ExpectRollback()
		expectCandidateSourceSessionRestore(mock)
		if err := lease.Close(); err != nil {
			t.Fatal(err)
		}
		assertMockExpectations(t, mock)
	})
}

func TestCandidateSourceVerificationLeaseRetainsCanonicalOrderAcrossOwnerKinds(
	t *testing.T,
) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet.BundleKeys = append(
		request.RequiredBundleKeySet.BundleKeys,
		testRequiredSystemKey("3"),
	)
	authority, mock, closeDB, record, handle :=
		sourceVerificationFixtureForRequest(t, request)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}

	appKey := requiredAppKey(t, record, "io.example.alpha")
	tenantKey := requiredTenantKey(t, record, "io.example.tenant")
	systemKey := requiredSystemKey(t, record, "3")
	appIndex, appRequired := lease.required[appKey]
	tenantIndex, tenantRequired := lease.required[tenantKey]
	systemIndex, systemRequired := lease.required[systemKey]
	if !appRequired || !tenantRequired || !systemRequired ||
		appIndex >= tenantIndex || tenantIndex >= systemIndex {
		t.Fatalf(
			"canonical owner indexes = App %d, tenant %d, system %d",
			appIndex,
			tenantIndex,
			systemIndex,
		)
	}

	appClaim, err := lease.BeginAppBundleKeyVerification(appKey)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := lease.BeginTenantBundleKeyVerification(tenantKey); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("tenant claim while App claim is pending = %v", err)
	}
	if _, err := lease.BeginSystemBundleKeyVerification(systemKey); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("system claim while App claim is pending = %v", err)
	}
	if !lease.Active() || appClaim.Active() {
		t.Fatal("pending App claim did not retain only the source transaction")
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceTenantVerificationClaimAbortReleasesAllLocks(t *testing.T) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet.BundleKeys = []layoutbundlekeyv1.BundleKey{
		testRequiredTenantKey("io.example.tenant", "2"),
	}
	authority, mock, closeDB, record, handle :=
		sourceVerificationFixtureForRequest(t, request)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	claim, err := lease.BeginTenantBundleKeyVerification(
		requiredTenantKey(t, record, "io.example.tenant"),
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	claim.Abort()
	if lease.Active() || claim.Active() {
		t.Fatal("aborted tenant claim retained locks")
	}
	if _, err := claim.Facts(); err == nil {
		t.Fatal("aborted tenant claim still exposed source facts")
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationLeaseEnforcesMembershipOrderAndPendingExclusion(
	t *testing.T,
) {
	request := validCandidateSourceRequest()
	request.RequiredBundleKeySet.BundleKeys = append(
		request.RequiredBundleKeySet.BundleKeys,
		layoutbundlekeyv1.BundleKey{
			SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKindApp,
			SlotOwnerNamespace: "io.example.zeta",
			LocalComponentKey: layoutbundlekeyv1.LocalComponentKey(
				"key256:" + strings.Repeat("3", 64),
			),
		},
	)
	authority, mock, closeDB, record, handle := sourceVerificationFixtureForRequest(
		t,
		request,
	)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	defer lease.Close()

	unknown := layoutbundlekeyv1.BundleKey{
		SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKindApp,
		SlotOwnerNamespace: "io.example.zeta",
		LocalComponentKey: layoutbundlekeyv1.LocalComponentKey(
			"key256:" + strings.Repeat("4", 64),
		),
	}
	if _, err := lease.BeginAppBundleKeyVerification(unknown); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("unknown required key = %v", err)
	}

	zeta := requiredAppKey(t, record, "io.example.zeta")
	alpha := requiredAppKey(t, record, "io.example.alpha")
	if lease.required[alpha] >= lease.required[zeta] {
		t.Fatalf(
			"canonical owner indexes = alpha %d, zeta %d",
			lease.required[alpha],
			lease.required[zeta],
		)
	}
	if _, err := lease.BeginAppBundleKeyVerification(zeta); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("out-of-order required key = %v", err)
	}
	claim, err := lease.BeginAppBundleKeyVerification(alpha)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := lease.BeginAppBundleKeyVerification(zeta); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("second claim while alpha is pending = %v", err)
	}
	if !lease.Active() || claim.Active() {
		t.Fatal("pending alpha claim did not retain only the source transaction")
	}

	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	if err := lease.Close(); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceVerificationClaimAbortReleasesAllLocks(t *testing.T) {
	authority, mock, closeDB, record, handle := sourceVerificationFixture(t)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	lease, err := authority.BeginCandidateSourceVerification(
		context.Background(),
		handle,
	)
	if err != nil {
		t.Fatal(err)
	}
	claim, err := lease.BeginAppBundleKeyVerification(
		requiredAppKey(t, record, "io.example.alpha"),
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)
	claim.Abort()
	if lease.Active() || claim.Active() {
		t.Fatal("aborted claim retained locks")
	}
	if _, err := claim.Facts(); err == nil {
		t.Fatal("aborted claim still exposed source facts")
	}
	assertMockExpectations(t, mock)
}

func TestRevalidateCandidateSourceUsesRetainedLockPath(t *testing.T) {
	authority, mock, closeDB, record, handle := sourceVerificationFixture(t)
	defer closeDB()
	expectCandidateSourceVerificationLock(mock, record)
	mock.ExpectCommit()
	expectCandidateSourceSessionRestore(mock)
	if err := authority.RevalidateCandidateSource(
		context.Background(),
		handle,
	); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestRetainedSourceTransactionOutlivesCanceledAcquisitionContext(
	t *testing.T,
) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	ctx, cancel := context.WithCancel(context.Background())
	expectCandidateSourceSessionPessimistic(mock)
	mock.ExpectBegin()
	conn, tx, mode, err := beginRetainedSourceTransaction(ctx, db)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1")).WillReturnRows(
		sqlmock.NewRows([]string{"1"}).AddRow(1),
	)
	mock.ExpectRollback()
	expectCandidateSourceSessionRestore(mock)

	cancel()
	// Give database/sql's BeginTx cancellation watcher time to run. A
	// transaction begun with the request context would now be auto-rolled back.
	time.Sleep(20 * time.Millisecond)
	var one int
	if err := tx.QueryRowContext(
		context.Background(),
		"SELECT 1",
	).Scan(&one); err != nil {
		t.Fatalf("retained transaction after acquisition cancellation: %v", err)
	}
	if one != 1 {
		t.Fatalf("retained transaction probe = %d, want 1", one)
	}
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mode.Restore(context.Background(), conn); err != nil {
		t.Fatal(err)
	}
	if err := conn.Close(); err != nil {
		t.Fatal(err)
	}
	assertMockExpectations(t, mock)
}

func TestRetainedSourceTransactionDiscardsAmbiguousBeginFailure(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	expectCandidateSourceSessionPessimistic(mock)
	beginErr := errors.New("begin response lost")
	mock.ExpectBegin().WillReturnError(beginErr)

	if _, _, _, err := beginRetainedSourceTransaction(
		context.Background(),
		db,
	); !errors.Is(err, beginErr) {
		t.Fatalf("begin retained source transaction = %v, want begin failure", err)
	}
	if open := db.Stats().OpenConnections; open != 0 {
		t.Fatalf("ambiguous begin retained %d pooled connections, want 0", open)
	}
	assertMockExpectations(t, mock)
}

func sourceVerificationFixture(
	t *testing.T,
) (
	*Authority,
	sqlmock.Sqlmock,
	func(),
	candidateSourceRecord,
	*CatalogCandidateSourceHandle,
) {
	t.Helper()
	return sourceVerificationFixtureForRequest(t, validCandidateSourceRequest())
}

func sourceVerificationFixtureForRequest(
	t *testing.T,
	request CandidateSourceRequest,
) (
	*Authority,
	sqlmock.Sqlmock,
	func(),
	candidateSourceRecord,
	*CatalogCandidateSourceHandle,
) {
	t.Helper()
	authority, mock, closeDB := mockPublicationAuthority(t)
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		closeDB()
		t.Fatal(err)
	}
	record := candidateSourceRecordFromValidated(
		validated,
		1,
		CandidateSourceCatalogSuccessor,
	)
	return authority, mock, closeDB, record, authority.preparationFromRecord(record).Handle
}

func expectCandidateSourceVerificationLock(
	mock sqlmock.Sqlmock,
	record candidateSourceRecord,
) {
	expectCandidateSourceSessionPessimistic(mock)
	mock.ExpectBegin()
	expectSourceWriter(
		mock,
		record.OrgID,
		record.RuntimeAudience,
		record.WriterCommandID,
		record.ExpectedPriorCatalogRevision,
		0,
		record.WriterReservationGeneration,
	)
	expectSourceActiveAbsent(mock, record.OrgID)
	expectCandidateSourceRecord(mock, record)
	expectCandidateSourceKindVacant(mock, record.OrgID, record.WriterCommandID)
}

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

func expectCandidateSourceSessionRestore(mock sqlmock.Sqlmock) {
	mock.ExpectExec(regexp.QuoteMeta(
		"SET SESSION tidb_txn_mode = ?",
	)).WithArgs("optimistic").WillReturnResult(sqlmock.NewResult(0, 0))
}

func requiredAppKey(
	t *testing.T,
	record candidateSourceRecord,
	appID string,
) layoutbundlekeyv1.BundleKey {
	t.Helper()
	for _, key := range record.RequiredBundleKeySet.BundleKeys {
		if key.SlotOwnerKind == layoutbundlekeyv1.SlotOwnerKindApp &&
			key.SlotOwnerNamespace == appID {
			return key
		}
	}
	t.Fatalf("required App key %q is absent", appID)
	return layoutbundlekeyv1.BundleKey{}
}

func requiredTenantKey(
	t *testing.T,
	record candidateSourceRecord,
	namespace string,
) layoutbundlekeyv1.BundleKey {
	t.Helper()
	for _, key := range record.RequiredBundleKeySet.BundleKeys {
		if key.SlotOwnerKind == layoutbundlekeyv1.SlotOwnerKindTenant &&
			key.SlotOwnerNamespace == namespace {
			return key
		}
	}
	t.Fatalf("required tenant key %q is absent", namespace)
	return layoutbundlekeyv1.BundleKey{}
}

func testRequiredSystemKey(character string) layoutbundlekeyv1.BundleKey {
	return layoutbundlekeyv1.BundleKey{
		SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKindSystem,
		SlotOwnerNamespace: layoutbundlekeyv1.SystemSlotOwnerNamespace,
		LocalComponentKey: layoutbundlekeyv1.LocalComponentKey(
			"key256:" + strings.Repeat(character, 64),
		),
	}
}

func testRequiredTenantKey(
	namespace string,
	character string,
) layoutbundlekeyv1.BundleKey {
	return layoutbundlekeyv1.BundleKey{
		SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKindTenant,
		SlotOwnerNamespace: namespace,
		LocalComponentKey: layoutbundlekeyv1.LocalComponentKey(
			"key256:" + strings.Repeat(character, 64),
		),
	}
}

func requiredSystemKey(
	t *testing.T,
	record candidateSourceRecord,
	character string,
) layoutbundlekeyv1.BundleKey {
	t.Helper()
	key := testRequiredSystemKey(character)
	for _, required := range record.RequiredBundleKeySet.BundleKeys {
		if required == key {
			return required
		}
	}
	t.Fatalf("required system key %q is absent", character)
	return layoutbundlekeyv1.BundleKey{}
}