package catalogpublication

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"reflect"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	catalogpublicationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv1"
	layoutbundlekeyv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogadmissionfence"
	"github.com/pingcap/tidb/pkg/parser"
	_ "github.com/pingcap/tidb/pkg/parser/test_driver"
)

func TestCandidateSourceSchemaParsesAndStaysNonServing(t *testing.T) {
	p := parser.New()
	joined := strings.Join(SchemaStatementsForTest(), "\n")
	for _, required := range []string{
		"_System_CatalogPublicationAdmissionFence",
		"_System_CatalogCandidateSource",
		"_System_CatalogCandidateSourceRequiredKey",
		"_System_CatalogCandidateSourceAppSelection",
		"_System_CatalogCandidateSourceFailureOutcome",
		"_System_CatalogWriterGenerationTerminalHistory",
		"EXACT_CURRENT_CATALOG_REVALIDATION",
		"PUBLICATION_COMPLETED",
		"TERMINATED",
		"FailureOutcomeDigest",
		"WriterReservationGeneration",
	} {
		if !strings.Contains(joined, required) {
			t.Fatalf("candidate source schema is missing %q", required)
		}
	}
	for _, statement := range SchemaStatementsForTest() {
		if _, _, err := p.Parse(statement, "", ""); err != nil {
			t.Fatalf("schema does not parse with TiDB: %v\n%s", err, statement)
		}
		if strings.Contains(statement, "_System_CatalogCandidateSource") {
			for _, forbidden := range []string{
				"LayoutValidationViewDigest",
				"LayoutBundleOwnerVerificationReceiptSetDigest",
				"AuthorityEvidenceDigest",
				"LayoutActivationSetDigest",
				"ServingSelector",
			} {
				if strings.Contains(statement, forbidden) {
					t.Fatalf(
						"candidate source schema leaked post-evidence field %q",
						forbidden,
					)
				}
			}
		}
	}
	for _, forbidden := range []string{
		"_System_CatalogReceiptEnforcementGate",
		"AuthorityEvidenceDigest",
		"ServingSelector",
		"CurrentPointer",
	} {
		if strings.Contains(joined, forbidden) {
			t.Fatalf("candidate source schema leaked serving/evidence field %q", forbidden)
		}
	}
}

func TestPublicationSchemaRequiresExplicitLegacyBindingVersion(t *testing.T) {
	joined := strings.Join(SchemaStatementsForTest(), "\n")
	for _, required := range []string{
		"BindingVersion VARCHAR(64) NOT NULL",
		"LayoutBundleOwnerVerificationReceiptSetDigest VARCHAR(71) NULL",
		"ADD COLUMN IF NOT EXISTS BindingVersion VARCHAR(64) NULL AFTER PublicationDigest",
		"ADD COLUMN IF NOT EXISTS LayoutBundleOwnerVerificationReceiptSetDigest VARCHAR(71) NULL",
		"ALTER COLUMN BindingVersion DROP DEFAULT",
		"SET BindingVersion = '" + legacyPublicationBindingVersion + "'",
		"MODIFY COLUMN BindingVersion VARCHAR(64) NOT NULL",
	} {
		if !strings.Contains(joined, required) {
			t.Fatalf("publication schema is missing %q", required)
		}
	}
	if count := strings.Count(joined, "BindingVersion VARCHAR(64)"); count != 3 {
		t.Fatalf("BindingVersion declarations = %d, want fresh, expand, and explicit contraction declarations", count)
	}
	if count := strings.Count(joined, "LayoutBundleOwnerVerificationReceiptSetDigest VARCHAR(71)"); count != 2 {
		t.Fatalf("receipt-set declarations = %d, want fresh and expand declarations", count)
	}
	if strings.Contains(joined, "BindingVersion VARCHAR(64) NULL DEFAULT") ||
		strings.Contains(joined, "BindingVersion VARCHAR(64) NOT NULL DEFAULT") {
		t.Fatal("publication schema must not permit an implicit BindingVersion default")
	}
	create := strings.Index(joined, "CREATE TABLE IF NOT EXISTS _System_CatalogPublication (")
	expand := strings.Index(joined, "ALTER TABLE _System_CatalogPublication\nADD COLUMN IF NOT EXISTS BindingVersion")
	dropDefault := strings.Index(joined, "ALTER COLUMN BindingVersion DROP DEFAULT")
	classify := strings.Index(joined, "UPDATE _System_CatalogPublication\nSET BindingVersion = '")
	requireExplicit := strings.Index(joined, "MODIFY COLUMN BindingVersion VARCHAR(64) NOT NULL")
	if create < 0 || expand <= create || dropDefault <= expand || classify <= dropDefault || requireExplicit <= classify {
		t.Fatalf(
			"publication schema order is create=%d expand=%d drop-default=%d classify=%d require-explicit=%d",
			create,
			expand,
			dropDefault,
			classify,
			requireExplicit,
		)
	}
}

func TestCandidateSourceRelationMatrix(t *testing.T) {
	initial := validCandidateSourceRequest()
	relation, err := deriveCandidateSourceRelation(
		initial,
		0,
		activeCandidateSourceBaseline{},
	)
	if err != nil || relation != CandidateSourceCatalogSuccessor {
		t.Fatalf("initial relation = %q, %v", relation, err)
	}

	prior := Reference{Present: true, Digest: testDigest("8")}
	active := validBeginRequest().Candidate
	active.CatalogRevision = 7
	successor := cloneCandidateSourceRequestForTest(initial)
	successor.CatalogSnapshotBinding.CatalogRevision = 8
	successor.ExpectedPriorPublication = prior
	successor.ExpectedPriorCatalogRevision = 7
	baseline := activeCandidateSourceBaseline{
		active:  prior,
		binding: &active,
	}
	relation, err = deriveCandidateSourceRelation(successor, 7, baseline)
	if err != nil || relation != CandidateSourceCatalogSuccessor {
		t.Fatalf("successor relation = %q, %v", relation, err)
	}

	exact := cloneCandidateSourceRequestForTest(successor)
	exact.CatalogSnapshotBinding = snapshotBindingFromPublication(active)
	if _, err := deriveCandidateSourceRelation(exact, 7, baseline); err == nil {
		t.Fatal("legacy-v1 exact-current source was accepted")
	}
	baseline.receiptBackedV2 = true
	relation, err = deriveCandidateSourceRelation(exact, 7, baseline)
	if err != nil || relation != CandidateSourceExactCurrent {
		t.Fatalf("receipt-backed exact-current relation = %q, %v", relation, err)
	}

	for name, mutate := range map[string]func(*CandidateSourceRequest, *activeCandidateSourceBaseline){
		"third revision shape": func(
			request *CandidateSourceRequest,
			_ *activeCandidateSourceBaseline,
		) {
			request.CatalogSnapshotBinding.CatalogRevision = 9
		},
		"snapshot drift": func(
			request *CandidateSourceRequest,
			_ *activeCandidateSourceBaseline,
		) {
			request.CatalogSnapshotBinding.CatalogSnapshotDigest = testDigest("9")
		},
		"stale reference": func(
			_ *CandidateSourceRequest,
			value *activeCandidateSourceBaseline,
		) {
			value.active.Digest = testDigest("9")
		},
	} {
		t.Run(name, func(t *testing.T) {
			request := cloneCandidateSourceRequestForTest(exact)
			value := baseline
			mutate(&request, &value)
			if _, err := deriveCandidateSourceRelation(request, 7, value); err == nil {
				t.Fatal("invalid source relation was accepted")
			}
		})
	}
}

func TestCandidateSourceRequestBindsBothCompleteSets(t *testing.T) {
	request := validCandidateSourceRequest()
	forward, err := validateCandidateSourceRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	reversed := cloneCandidateSourceRequestForTest(request)
	reverseBundleKeys(reversed.RequiredBundleKeySet.BundleKeys)
	reverseAppSelections(reversed.SourceAppSelectionSet.Entries)
	retry, err := validateCandidateSourceRequest(reversed)
	if err != nil {
		t.Fatal(err)
	}
	if forward.requestDigest != retry.requestDigest ||
		!reflect.DeepEqual(forward.canonicalSet, retry.canonicalSet) ||
		!reflect.DeepEqual(forward.canonicalAppSet, retry.canonicalAppSet) {
		t.Fatal("input permutation changed canonical source identity")
	}

	for name, mutate := range map[string]func(*CandidateSourceRequest){
		"required membership": func(value *CandidateSourceRequest) {
			value.RequiredBundleKeySet.BundleKeys[0].LocalComponentKey =
				layoutbundlekeyv1.LocalComponentKey(
					"key256:" + strings.Repeat("9", 64),
				)
		},
		"App membership": func(value *CandidateSourceRequest) {
			value.SourceAppSelectionSet.Entries[0].AppReleaseDigest = testDigest("9")
		},
		"snapshot": func(value *CandidateSourceRequest) {
			value.CatalogSnapshotBinding.CatalogSnapshotDigest = testDigest("7")
		},
		"prior": func(value *CandidateSourceRequest) {
			value.ExpectedPriorCatalogRevision++
		},
	} {
		t.Run(name, func(t *testing.T) {
			changed := cloneCandidateSourceRequestForTest(request)
			mutate(&changed)
			got, err := validateCandidateSourceRequest(changed)
			if err != nil {
				t.Fatal(err)
			}
			if got.requestDigest == forward.requestDigest {
				t.Fatal("changed source retained request digest")
			}
		})
	}
}

func TestCandidateSourceRequestConformanceVector(t *testing.T) {
	request := CandidateSourceRequest{
		WriterCommandID: "018f0000-0000-7000-9000-000000000001",
		RuntimeAudience: "layout-runtime",
		CatalogSnapshotBinding: catalogpublicationv1.CatalogSnapshotBinding{
			OrgID:                 "org-conformance",
			CatalogRevision:       1,
			CatalogSnapshotDigest: testDigest("1"),
			SchemaGeneration:      2,
			EngineGeneration:      3,
		},
		RequiredBundleKeySet: layoutbundlekeyv1.RequiredBundleKeySet{
			Version:    layoutbundlekeyv1.RequiredBundleKeySetVersion,
			BundleKeys: []layoutbundlekeyv1.BundleKey{},
		},
		SourceAppSelectionSet: SourceAppSelectionSet{
			SourceAppSelectionSetVersion: SourceAppSelectionSetVersion,
			Entries:                      []SourceAppSelection{},
		},
	}
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	if len(validated.setBytes) != 74 ||
		len(validated.appSelectionBytes) != 46 ||
		validated.requestDigest !=
			"sha256:48a2212af1f7ec7a07ce2adb92652367d341ed1afc5fae816549c4346a574e64" {
		t.Fatalf(
			"conformance vector = required:%d App:%d digest:%s",
			len(validated.setBytes),
			len(validated.appSelectionBytes),
			validated.requestDigest,
		)
	}
}

func TestCandidateSourceRejectsCandidateSelectionsBeforeBeginTx(t *testing.T) {
	authority, mock, closeDB := mockPublicationAuthority(t)
	defer closeDB()
	request := validCandidateSourceRequest()
	request.SourceAppSelectionSet.Entries[0].SelectionKind =
		SourceAppSelectionKindCandidate
	request.SourceAppSelectionSet.Entries[0].AppLifecycleCandidateCommandID =
		request.WriterCommandID
	if _, err := authority.PrepareCandidateSource(
		context.Background(),
		request,
	); !errors.Is(err, ErrSourceNotReady) {
		t.Fatalf("CANDIDATE selection = %v", err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceFailureOutcomeIsClosedAndBounded(t *testing.T) {
	outcome := CatalogCandidateSourceFailureOutcomeV1{
		OrgID:                       strings.Repeat("o", 255),
		WriterCommandID:             "018f0000-0000-7000-9000-000000000001",
		WriterReservationGeneration: ^uint64(0),
		SourceRequestDigest:         testDigest("1"),
		ExpectedPriorPublication: Reference{
			Present: true,
			Digest:  testDigest("2"),
		},
		ExpectedPriorCatalogRevision: ^uint64(0),
		FailureClass:                 CandidateSourceSourceInvalidity,
		FailureCode:                  "A" + strings.Repeat("_", 63),
		FailureDetailDigest:          testDigest("3"),
		RetryDisposition:             CandidateSourceRetryWithNewOuterCommand,
		SourceTerminalState:          CandidateSourceStale,
	}
	size := candidateSourceFailureCanonicalSize(outcome)
	if size > maxCandidateSourceFailureBytes {
		t.Fatalf("maximum valid outcome uses %d bytes", size)
	}
	first, err := catalogCandidateSourceFailureOutcomeDigest(outcome)
	if err != nil {
		t.Fatal(err)
	}
	second, err := catalogCandidateSourceFailureOutcomeDigest(outcome)
	if err != nil || second != first {
		t.Fatalf("failure digest is not deterministic: %q, %q, %v", first, second, err)
	}

	for name, mutate := range map[string]func(*CatalogCandidateSourceFailureOutcomeV1){
		"class": func(value *CatalogCandidateSourceFailureOutcomeV1) {
			value.FailureClass = "TRANSIENT"
		},
		"code": func(value *CatalogCandidateSourceFailureOutcomeV1) {
			value.FailureCode = "lowercase"
		},
		"detail digest": func(value *CatalogCandidateSourceFailureOutcomeV1) {
			value.FailureDetailDigest = "bad"
		},
		"retry": func(value *CatalogCandidateSourceFailureOutcomeV1) {
			value.RetryDisposition = "RETRY_SAME_COMMAND"
		},
		"state": func(value *CatalogCandidateSourceFailureOutcomeV1) {
			value.SourceTerminalState = CandidateSourcePrepared
		},
	} {
		t.Run(name, func(t *testing.T) {
			changed := outcome
			mutate(&changed)
			if _, err := catalogCandidateSourceFailureOutcomeDigest(changed); err == nil {
				t.Fatal("invalid failure outcome was accepted")
			}
		})
	}
}

func TestCandidateSourceHandleHasNoWireFormAndIsAuthorityScoped(t *testing.T) {
	handleType := reflect.TypeOf(CatalogCandidateSourceHandle{})
	for index := 0; index < handleType.NumField(); index++ {
		if handleType.Field(index).PkgPath == "" {
			t.Fatalf("handle field %q is exported", handleType.Field(index).Name)
		}
	}
	handle := CatalogCandidateSourceHandle{}
	if _, err := json.Marshal(handle); !errors.Is(err, errSourceHandleHasNoJSON) {
		t.Fatalf("value JSON marshal = %v", err)
	}
	if err := json.Unmarshal([]byte(`{}`), &handle); !errors.Is(
		err,
		errSourceHandleHasNoJSON,
	) {
		t.Fatalf("JSON unmarshal = %v", err)
	}

	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	first, _ := New(db)
	second, _ := New(db)
	valid := &CatalogCandidateSourceHandle{
		authority:             first.identity,
		orgID:                 "org-1",
		writerCommandID:       "018f0000-0000-7000-9000-000000000001",
		requestDigest:         testDigest("1"),
		reservationGeneration: 1,
	}
	if err := first.validateHandleIdentity(valid); err != nil {
		t.Fatal(err)
	}
	if err := second.validateHandleIdentity(valid); !errors.Is(
		err,
		ErrSourceConflict,
	) {
		t.Fatalf("foreign Authority handle = %v", err)
	}
}

func TestPrepareCandidateSourcePersistsWithoutCreatingPublicationState(t *testing.T) {
	authority, mock, closeDB := mockPublicationAuthority(t)
	defer closeDB()
	request := validCandidateSourceRequest()
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	record := candidateSourceRecordFromValidated(
		validated,
		1,
		CandidateSourceCatalogSuccessor,
	)

	mock.ExpectBegin()
	expectSourcePreparationFence(mock)
	expectSourceWriter(
		mock,
		record.OrgID,
		record.RuntimeAudience,
		"",
		0,
		0,
		0,
	)
	expectCandidateSourceStateAbsent(mock, record.OrgID, record.WriterCommandID)
	expectSourceActiveAbsent(mock, record.OrgID)
	expectCandidateSourceAbsent(mock, record.OrgID, record.WriterCommandID)
	expectCandidateSourceKindVacant(mock, record.OrgID, record.WriterCommandID)
	mock.ExpectExec("UPDATE _System_CatalogOrgState").
		WithArgs(
			record.WriterCommandID,
			"1",
			record.OrgID,
			"0",
			"0",
			"0",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	expectCandidateSourceInsert(mock, record)
	mock.ExpectCommit()

	prepared, err := authority.PrepareCandidateSource(
		context.Background(),
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if prepared.State != CandidateSourcePrepared ||
		prepared.Relation != CandidateSourceCatalogSuccessor ||
		prepared.PreparationOutcomeDigest != record.PreparationOutcomeDigest ||
		prepared.Handle == nil {
		t.Fatalf("preparation = %#v", prepared)
	}
	assertMockExpectations(t, mock)
}

func TestPrepareCandidateSourceRecoversOnlyExactPreparedRequest(t *testing.T) {
	request := validCandidateSourceRequest()
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	record := candidateSourceRecordFromValidated(
		validated,
		1,
		CandidateSourceCatalogSuccessor,
	)

	t.Run("exact", func(t *testing.T) {
		authority, mock, closeDB := mockPublicationAuthority(t)
		defer closeDB()
		mock.ExpectBegin()
		expectSourcePreparationFence(mock)
		expectSourceWriter(
			mock,
			record.OrgID,
			record.RuntimeAudience,
			record.WriterCommandID,
			0,
			0,
			1,
		)
		expectCandidateSourceState(mock, record)
		expectSourceActiveAbsent(mock, record.OrgID)
		expectCandidateSourceRecord(mock, record)
		expectCandidateSourceKindVacant(mock, record.OrgID, record.WriterCommandID)
		mock.ExpectCommit()
		got, err := authority.PrepareCandidateSource(
			context.Background(),
			request,
		)
		if err != nil {
			t.Fatal(err)
		}
		if got.Handle == nil ||
			got.PreparationOutcomeDigest != record.PreparationOutcomeDigest {
			t.Fatalf("exact recovery = %#v", got)
		}
		assertMockExpectations(t, mock)
	})

	t.Run("changed", func(t *testing.T) {
		authority, mock, closeDB := mockPublicationAuthority(t)
		defer closeDB()
		changed := cloneCandidateSourceRequestForTest(request)
		changed.SourceAppSelectionSet.Entries[0].AppReleaseDigest = testDigest("9")
		mock.ExpectBegin()
		expectSourcePreparationFence(mock)
		expectSourceWriter(
			mock,
			record.OrgID,
			record.RuntimeAudience,
			record.WriterCommandID,
			0,
			0,
			1,
		)
		expectCandidateSourceState(mock, record)
		expectSourceActiveAbsent(mock, record.OrgID)
		expectCandidateSourceRecord(mock, record)
		mock.ExpectRollback()
		if _, err := authority.PrepareCandidateSource(
			context.Background(),
			changed,
		); !errors.Is(err, ErrSourceConflict) {
			t.Fatalf("changed recovery = %v", err)
		}
		assertMockExpectations(t, mock)
	})
}

func TestTerminateCandidateSourceCommitsTypedFailureAndReplays(t *testing.T) {
	authority, mock, closeDB := mockPublicationAuthority(t)
	defer closeDB()
	request := validCandidateSourceRequest()
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	record := candidateSourceRecordFromValidated(
		validated,
		1,
		CandidateSourceCatalogSuccessor,
	)
	handle := authority.preparationFromRecord(record).Handle
	failure := validCandidateSourceFailure()
	outcome, digest, err := candidateSourceFailureOutcome(
		record,
		failure,
		CandidateSourceTerminated,
	)
	if err != nil {
		t.Fatal(err)
	}

	mock.ExpectBegin()
	expectSourceWriter(
		mock,
		record.OrgID,
		record.RuntimeAudience,
		record.WriterCommandID,
		0,
		0,
		1,
	)
	expectCandidateSourceState(mock, record)
	expectSourceActiveAbsent(mock, record.OrgID)
	expectCandidateSourceRecord(mock, record)
	expectCandidateSourceKindVacant(mock, record.OrgID, record.WriterCommandID)
	expectCandidateSourceFailureInsert(mock, outcome, digest)
	mock.ExpectExec("UPDATE _System_CatalogOrgState").
		WithArgs(record.OrgID, record.WriterCommandID, "1").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec("INSERT INTO _System_CatalogWriterGenerationTerminalHistory").
		WithArgs(
			record.OrgID,
			record.WriterCommandID,
			"1",
			writerTerminalReleased,
			digest,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec("UPDATE _System_CatalogCandidateSource").
		WithArgs(
			string(CandidateSourceTerminated),
			digest,
			record.OrgID,
			record.WriterCommandID,
			record.RequestDigest,
			"1",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	got, gotDigest, err := authority.TerminateCandidateSource(
		context.Background(),
		handle,
		failure,
	)
	if err != nil || got != outcome || gotDigest != digest {
		t.Fatalf("terminate = %#v, %q, %v", got, gotDigest, err)
	}

	record.State = CandidateSourceTerminated
	record.FailureOutcome = &outcome
	record.FailureOutcomeDigest = digest
	record.WriterTerminalAction = writerTerminalReleased
	record.WriterTerminalOutcome = digest
	mock.ExpectBegin()
	expectSourceWriter(
		mock,
		record.OrgID,
		record.RuntimeAudience,
		"",
		0,
		0,
		1,
	)
	expectCandidateSourceState(mock, record)
	expectCandidateSourceRecord(mock, record)
	expectCandidateSourceKindVacant(mock, record.OrgID, record.WriterCommandID)
	mock.ExpectCommit()
	retriedOutcome, retriedDigest, err := authority.TerminateCandidateSource(
		context.Background(),
		handle,
		failure,
	)
	if err != nil || retriedOutcome != outcome || retriedDigest != digest {
		t.Fatalf(
			"terminal transition replay = %#v, %q, %v",
			retriedOutcome,
			retriedDigest,
			err,
		)
	}

	mock.ExpectBegin()
	expectSourcePreparationFence(mock)
	expectSourceWriter(
		mock,
		record.OrgID,
		record.RuntimeAudience,
		"",
		0,
		0,
		1,
	)
	expectCandidateSourceState(mock, record)
	expectCandidateSourceRecord(mock, record)
	expectCandidateSourceKindVacant(mock, record.OrgID, record.WriterCommandID)
	mock.ExpectCommit()
	replayed, err := authority.PrepareCandidateSource(
		context.Background(),
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if replayed.State != CandidateSourceTerminated ||
		replayed.Handle != nil ||
		replayed.FailureOutcome == nil ||
		*replayed.FailureOutcome != outcome ||
		replayed.FailureOutcomeDigest != digest {
		t.Fatalf("terminal replay = %#v", replayed)
	}
	assertMockExpectations(t, mock)
}

func TestInvalidateStaleCandidateSourceRefusesLiveSource(t *testing.T) {
	authority, mock, closeDB := mockPublicationAuthority(t)
	defer closeDB()
	request := validCandidateSourceRequest()
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	record := candidateSourceRecordFromValidated(
		validated,
		1,
		CandidateSourceCatalogSuccessor,
	)
	handle := authority.preparationFromRecord(record).Handle

	mock.ExpectBegin()
	expectSourceWriter(
		mock,
		record.OrgID,
		record.RuntimeAudience,
		record.WriterCommandID,
		0,
		0,
		1,
	)
	expectCandidateSourceState(mock, record)
	expectSourceActiveAbsent(mock, record.OrgID)
	expectCandidateSourceRecord(mock, record)
	expectCandidateSourceKindVacant(mock, record.OrgID, record.WriterCommandID)
	mock.ExpectRollback()
	if _, _, err := authority.InvalidateStaleCandidateSource(
		context.Background(),
		handle,
		validCandidateSourceFailure(),
	); !errors.Is(err, ErrSourceConflict) {
		t.Fatalf("live stale invalidation = %v", err)
	}
	assertMockExpectations(t, mock)
}

func TestPublicationCompletedSourceRejectsLegacyV1TerminalPair(t *testing.T) {
	authority, mock, closeDB := mockPublicationAuthority(t)
	defer closeDB()
	request := validCandidateSourceRequest()
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	record := candidateSourceRecordFromValidated(
		validated,
		1,
		CandidateSourceCatalogSuccessor,
	)
	publicationRequest := validBeginRequest()
	candidateDigest, publicationRequestDigest, err := validateBeginRequest(
		publicationRequest,
	)
	if err != nil {
		t.Fatal(err)
	}
	command := commandFromRequest(
		publicationRequest,
		candidateDigest,
		publicationRequestDigest,
	)
	command.State = CommandCompleted
	outcome := Outcome{
		OrgID:                      record.OrgID,
		CommandID:                  record.WriterCommandID,
		CandidatePublicationDigest: candidateDigest,
		Result:                     ResultSelected,
		ResultingActive: Reference{
			Present: true,
			Digest:  candidateDigest,
		},
	}
	outcome.OutcomeDigest, err = canonicalOutcomeDigest(outcome)
	if err != nil {
		t.Fatal(err)
	}
	record.State = CandidateSourcePublicationCompleted
	record.AdoptedPublicationRequest = publicationRequestDigest
	record.AdoptedCandidatePublication = candidateDigest
	record.CatalogPublicationOutcome = outcome.OutcomeDigest
	record.WriterTerminalAction = writerTerminalCommitted
	record.WriterTerminalOutcome = outcome.OutcomeDigest

	mock.ExpectBegin()
	expectSourcePreparationFence(mock)
	expectSourceWriter(
		mock,
		record.OrgID,
		record.RuntimeAudience,
		"",
		1,
		1,
		1,
	)
	expectCandidateSourceState(mock, record)
	expectCandidateSourceRecord(mock, record)
	mock.ExpectQuery("SELECT OrgId.*_System_CatalogAppOperation").
		WithArgs(record.WriterCommandID).
		WillReturnError(sql.ErrNoRows)
	expectCommandRow(mock, command)
	expectPublicationRow(mock, record.OrgID, candidateDigest, publicationRequest)
	mock.ExpectRollback()

	if _, err := authority.PrepareCandidateSource(
		context.Background(),
		request,
	); !errors.Is(err, ErrSourceCorrupt) {
		t.Fatalf("legacy-v1 publication terminal replay = %v", err)
	}
	assertMockExpectations(t, mock)
}

func TestCandidateSourceRejectsPublicationCommandKind(t *testing.T) {
	authority, mock, closeDB := mockPublicationAuthority(t)
	defer closeDB()
	request := validCandidateSourceRequest()
	publication := validBeginRequest()
	candidateDigest, requestDigest, err := validateBeginRequest(publication)
	if err != nil {
		t.Fatal(err)
	}
	command := commandFromRequest(publication, candidateDigest, requestDigest)

	mock.ExpectBegin()
	expectSourcePreparationFence(mock)
	expectSourceWriter(
		mock,
		request.CatalogSnapshotBinding.OrgID,
		request.RuntimeAudience,
		"",
		0,
		0,
		0,
	)
	expectCandidateSourceStateAbsent(
		mock,
		request.CatalogSnapshotBinding.OrgID,
		request.WriterCommandID,
	)
	expectSourceActiveAbsent(mock, request.CatalogSnapshotBinding.OrgID)
	expectCandidateSourceAbsent(
		mock,
		request.CatalogSnapshotBinding.OrgID,
		request.WriterCommandID,
	)
	expectCommandRow(mock, command)
	mock.ExpectRollback()
	if _, err := authority.PrepareCandidateSource(
		context.Background(),
		request,
	); !errors.Is(err, ErrSourceConflict) {
		t.Fatalf("publication command collision = %v", err)
	}
	assertMockExpectations(t, mock)
}

func validCandidateSourceRequest() CandidateSourceRequest {
	publication := validBeginRequest()
	return CandidateSourceRequest{
		WriterCommandID: publication.CommandID,
		RuntimeAudience: publication.RuntimeAudience,
		CatalogSnapshotBinding: catalogpublicationv1.CatalogSnapshotBinding{
			OrgID:                 publication.Candidate.OrgID,
			CatalogRevision:       publication.Candidate.CatalogRevision,
			CatalogSnapshotDigest: publication.Candidate.CatalogSnapshotDigest,
			SchemaGeneration:      publication.Candidate.SchemaGeneration,
			EngineGeneration:      publication.Candidate.EngineGeneration,
		},
		RequiredBundleKeySet: layoutbundlekeyv1.RequiredBundleKeySet{
			Version: layoutbundlekeyv1.RequiredBundleKeySetVersion,
			BundleKeys: []layoutbundlekeyv1.BundleKey{
				{
					SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKindTenant,
					SlotOwnerNamespace: "io.example.tenant",
					LocalComponentKey: layoutbundlekeyv1.LocalComponentKey(
						"key256:" + strings.Repeat("2", 64),
					),
				},
				{
					SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKindApp,
					SlotOwnerNamespace: "io.example.alpha",
					LocalComponentKey: layoutbundlekeyv1.LocalComponentKey(
						"key256:" + strings.Repeat("1", 64),
					),
				},
			},
		},
		SourceAppSelectionSet: SourceAppSelectionSet{
			SourceAppSelectionSetVersion: SourceAppSelectionSetVersion,
			Entries: []SourceAppSelection{
				{
					AppID:             "io.example.zeta",
					AppInstallationID: "018f0000-0000-7000-9000-000000000012",
					AppReleaseDigest:  testDigest("b"),
					SelectionKind:     SourceAppSelectionKindActive,
				},
				{
					AppID:             "io.example.alpha",
					AppInstallationID: "018f0000-0000-7000-9000-000000000011",
					AppReleaseDigest:  testDigest("a"),
					SelectionKind:     SourceAppSelectionKindActive,
				},
			},
		},
	}
}

func validCandidateSourceFailure() CandidateSourceFailure {
	return CandidateSourceFailure{
		FailureClass:        CandidateSourceSemanticInvalidity,
		FailureCode:         "OWNER_EVIDENCE_INVALID",
		FailureDetailDigest: testDigest("d"),
		RetryDisposition:    CandidateSourceRetryWithNewOuterCommand,
	}
}

func candidateSourceRecordFromValidated(
	value validatedCandidateSourceRequest,
	generation uint64,
	relation CandidateSourceRelation,
) candidateSourceRecord {
	record := candidateSourceRecord{
		OrgID:                         value.request.CatalogSnapshotBinding.OrgID,
		WriterCommandID:               value.request.WriterCommandID,
		RequestDigest:                 value.requestDigest,
		RuntimeAudience:               value.request.RuntimeAudience,
		CatalogSnapshotBinding:        value.request.CatalogSnapshotBinding,
		RequiredBundleKeySet:          cloneRequiredSet(value.canonicalSet),
		RequiredBundleKeySetDigest:    value.setDigest,
		RequiredBundleKeySetCanonical: append([]byte(nil), value.setBytes...),
		ExpectedPriorPublication:      value.request.ExpectedPriorPublication,
		ExpectedPriorCatalogRevision:  value.request.ExpectedPriorCatalogRevision,
		SourceAppSelectionSet:         cloneSourceAppSelectionSet(value.canonicalAppSet),
		SourceAppSelectionCanonical:   append([]byte(nil), value.appSelectionBytes...),
		WriterReservationGeneration:   generation,
		Relation:                      relation,
		State:                         CandidateSourcePrepared,
	}
	record.PreparationOutcomeDigest = candidateSourcePreparationOutcomeDigest(record)
	return record
}

func cloneCandidateSourceRequestForTest(
	value CandidateSourceRequest,
) CandidateSourceRequest {
	value.RequiredBundleKeySet = cloneRequiredSet(value.RequiredBundleKeySet)
	value.SourceAppSelectionSet = cloneSourceAppSelectionSet(
		value.SourceAppSelectionSet,
	)
	return value
}

func reverseBundleKeys(values []layoutbundlekeyv1.BundleKey) {
	for left, right := 0, len(values)-1; left < right; left, right = left+1, right-1 {
		values[left], values[right] = values[right], values[left]
	}
}

func reverseAppSelections(values []SourceAppSelection) {
	for left, right := 0, len(values)-1; left < right; left, right = left+1, right-1 {
		values[left], values[right] = values[right], values[left]
	}
}

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

func expectSourceWriter(
	mock sqlmock.Sqlmock,
	orgID string,
	runtimeAudience string,
	activeCommand string,
	catalogRevision uint64,
	securityEpoch uint64,
	generation uint64,
) {
	mock.ExpectQuery("SELECT OrgId, RuntimeAudience, CatalogRevision").
		WithArgs(orgID).
		WillReturnRows(sqlmock.NewRows([]string{
			"OrgId",
			"RuntimeAudience",
			"CatalogRevision",
			"LastSecurityEpoch",
			"ActiveOperationId",
			"WriterGeneration",
		}).AddRow(
			orgID,
			runtimeAudience,
			[]byte(decimalText(catalogRevision)),
			[]byte(decimalText(securityEpoch)),
			sourceNullableString(activeCommand),
			[]byte(decimalText(generation)),
		))
}

func expectSourceActiveAbsent(mock sqlmock.Sqlmock, orgID string) {
	mock.ExpectQuery("SELECT OrgId, ActivePublicationPresent").
		WithArgs(orgID).
		WillReturnError(sql.ErrNoRows)
}

func expectCandidateSourceState(
	mock sqlmock.Sqlmock,
	record candidateSourceRecord,
) {
	mock.ExpectQuery("SELECT State").
		WithArgs(record.OrgID, record.WriterCommandID).
		WillReturnRows(sqlmock.NewRows([]string{"State"}).
			AddRow(string(record.State)))
}

func expectCandidateSourceStateAbsent(
	mock sqlmock.Sqlmock,
	orgID string,
	commandID string,
) {
	mock.ExpectQuery("SELECT State").
		WithArgs(orgID, commandID).
		WillReturnError(sql.ErrNoRows)
}

func expectCandidateSourceAbsent(
	mock sqlmock.Sqlmock,
	orgID string,
	commandID string,
) {
	mock.ExpectQuery("SELECT OrgId, WriterCommandId, RequestDigest").
		WithArgs(orgID, commandID).
		WillReturnError(sql.ErrNoRows)
}

func expectCandidateSourceKindVacant(
	mock sqlmock.Sqlmock,
	orgID string,
	commandID string,
) {
	mock.ExpectQuery("SELECT OrgId, CommandId, RequestDigest").
		WithArgs(orgID, commandID).
		WillReturnError(sql.ErrNoRows)
	mock.ExpectQuery("SELECT OrgId.*_System_CatalogAppOperation").
		WithArgs(commandID).
		WillReturnError(sql.ErrNoRows)
}

func expectCandidateSourceRecord(
	mock sqlmock.Sqlmock,
	record candidateSourceRecord,
) {
	priorDigest := any(nil)
	if record.ExpectedPriorPublication.Present {
		priorDigest = record.ExpectedPriorPublication.Digest
	}
	failureDigest := any(nil)
	adoptedRequest := any(nil)
	adoptedCandidate := any(nil)
	publicationOutcome := any(nil)
	if record.State == CandidateSourceTerminated ||
		record.State == CandidateSourceStale {
		failureDigest = record.FailureOutcomeDigest
	}
	if record.State == CandidateSourcePublicationCompleted {
		adoptedRequest = record.AdoptedPublicationRequest
		adoptedCandidate = record.AdoptedCandidatePublication
		publicationOutcome = record.CatalogPublicationOutcome
	}
	mock.ExpectQuery("SELECT OrgId, WriterCommandId, RequestDigest").
		WithArgs(record.OrgID, record.WriterCommandID).
		WillReturnRows(sqlmock.NewRows([]string{
			"OrgId",
			"WriterCommandId",
			"RequestDigest",
			"PreparationOutcomeDigest",
			"RuntimeAudience",
			"CatalogRevision",
			"CatalogSnapshotDigest",
			"SchemaGeneration",
			"EngineGeneration",
			"RequiredBundleKeySetVersion",
			"RequiredBundleKeySetDigest",
			"RequiredBundleKeySetCanonicalBytes",
			"ExpectedPriorPresent",
			"ExpectedPriorDigest",
			"ExpectedPriorCatalogRevision",
			"SourceAppSelectionSetVersion",
			"SourceAppSelectionSetCanonicalBytes",
			"WriterReservationGeneration",
			"SourceRelation",
			"State",
			"FailureOutcomeDigest",
			"AdoptedPublicationRequestDigest",
			"AdoptedCandidatePublicationDigest",
			"CatalogPublicationOutcomeDigest",
		}).AddRow(
			record.OrgID,
			record.WriterCommandID,
			record.RequestDigest,
			record.PreparationOutcomeDigest,
			record.RuntimeAudience,
			[]byte(decimalText(record.CatalogSnapshotBinding.CatalogRevision)),
			record.CatalogSnapshotBinding.CatalogSnapshotDigest,
			[]byte(decimalText(record.CatalogSnapshotBinding.SchemaGeneration)),
			[]byte(decimalText(record.CatalogSnapshotBinding.EngineGeneration)),
			record.RequiredBundleKeySet.Version,
			record.RequiredBundleKeySetDigest,
			record.RequiredBundleKeySetCanonical,
			record.ExpectedPriorPublication.Present,
			priorDigest,
			[]byte(decimalText(record.ExpectedPriorCatalogRevision)),
			record.SourceAppSelectionSet.SourceAppSelectionSetVersion,
			record.SourceAppSelectionCanonical,
			[]byte(decimalText(record.WriterReservationGeneration)),
			string(record.Relation),
			string(record.State),
			failureDigest,
			adoptedRequest,
			adoptedCandidate,
			publicationOutcome,
		))

	keyRows := sqlmock.NewRows([]string{
		"KeyOrdinal",
		"SlotOwnerKind",
		"SlotOwnerNamespace",
		"LocalComponentKey",
	})
	for index, key := range record.RequiredBundleKeySet.BundleKeys {
		keyRows.AddRow(
			[]byte(decimalText(uint64(index))),
			string(key.SlotOwnerKind),
			key.SlotOwnerNamespace,
			string(key.LocalComponentKey),
		)
	}
	mock.ExpectQuery("SELECT KeyOrdinal, SlotOwnerKind, SlotOwnerNamespace").
		WithArgs(record.OrgID, record.WriterCommandID).
		WillReturnRows(keyRows)

	selectionRows := sqlmock.NewRows([]string{
		"SelectionOrdinal",
		"AppId",
		"AppInstallationId",
		"AppReleaseDigest",
		"SelectionKind",
		"AppLifecycleCandidateCommandId",
	})
	for index, selection := range record.SourceAppSelectionSet.Entries {
		selectionRows.AddRow(
			[]byte(decimalText(uint64(index))),
			selection.AppID,
			selection.AppInstallationID,
			selection.AppReleaseDigest,
			string(selection.SelectionKind),
			sourceNullableString(selection.AppLifecycleCandidateCommandID),
		)
	}
	mock.ExpectQuery("SELECT SelectionOrdinal, AppId, AppInstallationId").
		WithArgs(record.OrgID, record.WriterCommandID).
		WillReturnRows(selectionRows)

	if record.FailureOutcome == nil {
		mock.ExpectQuery("SELECT OrgId, WriterCommandId, WriterReservationGeneration").
			WithArgs(
				record.OrgID,
				record.WriterCommandID,
				decimalText(record.WriterReservationGeneration),
			).
			WillReturnError(sql.ErrNoRows)
	} else {
		outcome := *record.FailureOutcome
		mock.ExpectQuery("SELECT OrgId, WriterCommandId, WriterReservationGeneration").
			WithArgs(
				record.OrgID,
				record.WriterCommandID,
				decimalText(record.WriterReservationGeneration),
			).
			WillReturnRows(sqlmock.NewRows([]string{
				"OrgId",
				"WriterCommandId",
				"WriterReservationGeneration",
				"SourceRequestDigest",
				"ExpectedPriorPresent",
				"ExpectedPriorDigest",
				"ExpectedPriorCatalogRevision",
				"FailureClass",
				"FailureCode",
				"FailureDetailDigest",
				"RetryDisposition",
				"SourceTerminalState",
				"FailureOutcomeDigest",
			}).AddRow(
				outcome.OrgID,
				outcome.WriterCommandID,
				[]byte(decimalText(outcome.WriterReservationGeneration)),
				outcome.SourceRequestDigest,
				outcome.ExpectedPriorPublication.Present,
				nullableReference(outcome.ExpectedPriorPublication),
				[]byte(decimalText(outcome.ExpectedPriorCatalogRevision)),
				string(outcome.FailureClass),
				outcome.FailureCode,
				outcome.FailureDetailDigest,
				string(outcome.RetryDisposition),
				string(outcome.SourceTerminalState),
				record.FailureOutcomeDigest,
			))
	}

	if record.State == CandidateSourcePrepared {
		mock.ExpectQuery("SELECT OrgId, WriterCommandId, WriterReservationGeneration").
			WithArgs(
				record.OrgID,
				record.WriterCommandID,
				decimalText(record.WriterReservationGeneration),
			).
			WillReturnError(sql.ErrNoRows)
	} else {
		mock.ExpectQuery("SELECT OrgId, WriterCommandId, WriterReservationGeneration").
			WithArgs(
				record.OrgID,
				record.WriterCommandID,
				decimalText(record.WriterReservationGeneration),
			).
			WillReturnRows(sqlmock.NewRows([]string{
				"OrgId",
				"WriterCommandId",
				"WriterReservationGeneration",
				"TerminalAction",
				"TerminalOutcomeDigest",
			}).AddRow(
				record.OrgID,
				record.WriterCommandID,
				[]byte(decimalText(record.WriterReservationGeneration)),
				record.WriterTerminalAction,
				record.WriterTerminalOutcome,
			))
	}
}

func expectCandidateSourceInsert(
	mock sqlmock.Sqlmock,
	record candidateSourceRecord,
) {
	mock.ExpectExec("INSERT INTO _System_CatalogCandidateSource").
		WithArgs(
			record.OrgID,
			record.WriterCommandID,
			record.RequestDigest,
			record.PreparationOutcomeDigest,
			record.RuntimeAudience,
			decimalText(record.CatalogSnapshotBinding.CatalogRevision),
			record.CatalogSnapshotBinding.CatalogSnapshotDigest,
			decimalText(record.CatalogSnapshotBinding.SchemaGeneration),
			decimalText(record.CatalogSnapshotBinding.EngineGeneration),
			record.RequiredBundleKeySet.Version,
			record.RequiredBundleKeySetDigest,
			record.RequiredBundleKeySetCanonical,
			record.ExpectedPriorPublication.Present,
			nullableReference(record.ExpectedPriorPublication),
			decimalText(record.ExpectedPriorCatalogRevision),
			record.SourceAppSelectionSet.SourceAppSelectionSetVersion,
			record.SourceAppSelectionCanonical,
			decimalText(record.WriterReservationGeneration),
			string(record.Relation),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	for index, key := range record.RequiredBundleKeySet.BundleKeys {
		mock.ExpectExec("INSERT INTO _System_CatalogCandidateSourceRequiredKey").
			WithArgs(
				record.OrgID,
				record.WriterCommandID,
				decimalText(uint64(index)),
				string(key.SlotOwnerKind),
				key.SlotOwnerNamespace,
				string(key.LocalComponentKey),
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
	for index, selection := range record.SourceAppSelectionSet.Entries {
		mock.ExpectExec("INSERT INTO _System_CatalogCandidateSourceAppSelection").
			WithArgs(
				record.OrgID,
				record.WriterCommandID,
				decimalText(uint64(index)),
				selection.AppID,
				selection.AppInstallationID,
				selection.AppReleaseDigest,
				string(selection.SelectionKind),
				sourceNullableString(selection.AppLifecycleCandidateCommandID),
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
}

func expectCandidateSourceFailureInsert(
	mock sqlmock.Sqlmock,
	outcome CatalogCandidateSourceFailureOutcomeV1,
	digest string,
) {
	mock.ExpectExec("INSERT INTO _System_CatalogCandidateSourceFailureOutcome").
		WithArgs(
			outcome.OrgID,
			outcome.WriterCommandID,
			decimalText(outcome.WriterReservationGeneration),
			outcome.SourceRequestDigest,
			outcome.ExpectedPriorPublication.Present,
			nullableReference(outcome.ExpectedPriorPublication),
			decimalText(outcome.ExpectedPriorCatalogRevision),
			string(outcome.FailureClass),
			outcome.FailureCode,
			outcome.FailureDetailDigest,
			string(outcome.RetryDisposition),
			string(outcome.SourceTerminalState),
			digest,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
}