package catalogpublication

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogwriter"
)

// PrepareCandidateSource durably records and reserves one exact Catalog source
// prefix. Its short transaction acquires the global identity-admission fence,
// org writer, active pointer, and source row in that order. The
// global fence makes the cross-table source/App command identity check atomic;
// it is never retained by owner verification. Prepare never creates publication
// state or advances a serving selector.
func (a *Authority) PrepareCandidateSource(
	ctx context.Context,
	request CandidateSourceRequest,
) (CandidateSourcePreparation, error) {
	// This validation rejects every CANDIDATE App selection before BeginTx.
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		return CandidateSourcePreparation{}, err
	}
	orgID := validated.request.CatalogSnapshotBinding.OrgID
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return CandidateSourcePreparation{}, err
	}
	defer rollbackTx(tx)

	writer, err := lockCandidateSourcePreparationPrefix(ctx, tx, orgID)
	if err != nil {
		return CandidateSourcePreparation{}, err
	}
	peekedState, peekErr := peekCandidateSourceState(
		ctx, tx, orgID, request.WriterCommandID,
	)
	if peekErr == nil && peekedState != CandidateSourcePrepared {
		// Immutable terminal recovery deliberately does not read the possibly
		// newer active pointer. The writer lock makes this source-state peek
		// stable; the terminal row is then re-read under its own lock.
		record, err := readCandidateSourceDB(
			ctx, tx, orgID, request.WriterCommandID, true,
		)
		if err != nil {
			return CandidateSourcePreparation{}, err
		}
		return a.completeCandidateSourceRetryLocked(
			ctx,
			tx,
			validated,
			writer,
			activeState{},
			activeCandidateSourceBaseline{},
			record,
		)
	}
	if peekErr != nil && !errors.Is(peekErr, sql.ErrNoRows) {
		return CandidateSourcePreparation{}, peekErr
	}
	state, baseline, err := lockCandidateSourceActiveBaseline(ctx, tx, orgID)
	if err != nil {
		return CandidateSourcePreparation{}, err
	}
	record, err := readCandidateSourceDB(
		ctx, tx, orgID, request.WriterCommandID, true,
	)
	if err == nil {
		return a.completeCandidateSourceRetryLocked(
			ctx, tx, validated, writer, state, baseline, record,
		)
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return CandidateSourcePreparation{}, err
	}
	if err := rejectSourceCommandKindCollisionLocked(
		ctx, tx, orgID, request.WriterCommandID, false,
	); err != nil {
		return CandidateSourcePreparation{}, err
	}
	if writer.RuntimeAudience != validated.request.RuntimeAudience ||
		writer.ActiveCommandID != "" {
		return CandidateSourcePreparation{}, ErrSourceConflict
	}
	if err := validateWriterCoherence(writer, baseline.binding); err != nil {
		return CandidateSourcePreparation{}, mapSourcePublicationError(err)
	}
	relation, err := deriveCandidateSourceRelation(
		validated.request,
		writer.CatalogRevision,
		baseline,
	)
	if err != nil {
		return CandidateSourcePreparation{}, fmt.Errorf("%w: %v", ErrSourceConflict, err)
	}
	generation, err := catalogwriter.ReserveGeneration(
		ctx, tx, writer, request.WriterCommandID,
	)
	if err != nil {
		return CandidateSourcePreparation{}, mapSourceWriterError(err)
	}

	record = candidateSourceRecord{
		OrgID:                         orgID,
		WriterCommandID:               request.WriterCommandID,
		RequestDigest:                 validated.requestDigest,
		RuntimeAudience:               validated.request.RuntimeAudience,
		CatalogSnapshotBinding:        validated.request.CatalogSnapshotBinding,
		RequiredBundleKeySet:          cloneRequiredSet(validated.canonicalSet),
		RequiredBundleKeySetDigest:    validated.setDigest,
		RequiredBundleKeySetCanonical: append([]byte(nil), validated.setBytes...),
		ExpectedPriorPublication:      validated.request.ExpectedPriorPublication,
		ExpectedPriorCatalogRevision:  validated.request.ExpectedPriorCatalogRevision,
		SourceAppSelectionSet:         cloneSourceAppSelectionSet(validated.canonicalAppSet),
		SourceAppSelectionCanonical:   append([]byte(nil), validated.appSelectionBytes...),
		WriterReservationGeneration:   generation,
		Relation:                      relation,
		State:                         CandidateSourcePrepared,
	}
	record.PreparationOutcomeDigest = candidateSourcePreparationOutcomeDigest(record)
	if err := insertCandidateSource(ctx, tx, record); err != nil {
		return CandidateSourcePreparation{}, err
	}
	if err := tx.Commit(); err != nil {
		return CandidateSourcePreparation{}, err
	}
	return a.preparationFromRecord(record), nil
}

func (a *Authority) completeCandidateSourceRetryLocked(
	ctx context.Context,
	tx *sql.Tx,
	validated validatedCandidateSourceRequest,
	writer catalogwriter.State,
	state activeState,
	baseline activeCandidateSourceBaseline,
	record candidateSourceRecord,
) (CandidateSourcePreparation, error) {
	if record.RequestDigest != validated.requestDigest {
		return CandidateSourcePreparation{}, ErrSourceConflict
	}
	switch record.State {
	case CandidateSourcePrepared:
		if err := rejectSourceCommandKindCollisionLocked(
			ctx, tx, record.OrgID, record.WriterCommandID, true,
		); err != nil {
			return CandidateSourcePreparation{}, err
		}
		if err := validateCandidateSourceLifecycle(
			record, writer, state, baseline,
		); err != nil {
			return CandidateSourcePreparation{}, err
		}
	case CandidateSourcePublicationCompleted:
		if err := a.validateCompletedCandidateSourceLocked(
			ctx, tx, writer, record,
		); err != nil {
			return CandidateSourcePreparation{}, err
		}
	case CandidateSourceTerminated, CandidateSourceStale:
		if err := validateFailedCandidateSourceLocked(
			ctx, tx, writer, record,
		); err != nil {
			return CandidateSourcePreparation{}, err
		}
	default:
		return CandidateSourcePreparation{}, fmt.Errorf(
			"%w: unknown source state %q",
			ErrSourceCorrupt,
			record.State,
		)
	}
	if err := tx.Commit(); err != nil {
		return CandidateSourcePreparation{}, err
	}
	return a.preparationFromRecord(record), nil
}

// RevalidateCandidateSource proves that the opaque handle still names the
// locked PREPARED record and exact live writer reservation.
func (a *Authority) RevalidateCandidateSource(
	ctx context.Context,
	handle *CatalogCandidateSourceHandle,
) error {
	lease, err := a.BeginCandidateSourceVerification(ctx, handle)
	if err != nil {
		return err
	}
	defer lease.Close()
	return lease.commitRevalidation()
}

// TerminateCandidateSource atomically records a typed immutable failure,
// releases the exact live generation, appends generation history, and moves
// PREPARED to TERMINATED.
func (a *Authority) TerminateCandidateSource(
	ctx context.Context,
	handle *CatalogCandidateSourceHandle,
	failure CandidateSourceFailure,
) (CatalogCandidateSourceFailureOutcomeV1, string, error) {
	if err := a.validateHandleIdentity(handle); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := validateCandidateSourceFailureInput(failure); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	return a.failCandidateSource(
		ctx,
		handle,
		CandidateSourceRequest{},
		false,
		failure,
		CandidateSourceTerminated,
	)
}

// InvalidateStaleCandidateSource atomically records a typed immutable failure
// only after the exact live source fence is proven stale.
func (a *Authority) InvalidateStaleCandidateSource(
	ctx context.Context,
	handle *CatalogCandidateSourceHandle,
	failure CandidateSourceFailure,
) (CatalogCandidateSourceFailureOutcomeV1, string, error) {
	if err := a.validateHandleIdentity(handle); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := validateCandidateSourceFailureInput(failure); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	return a.failCandidateSource(
		ctx,
		handle,
		CandidateSourceRequest{},
		false,
		failure,
		CandidateSourceStale,
	)
}

// InvalidateStaleCandidateSourceRequest performs the same proven-stale
// transition from the complete exact request after process restart. It never
// trusts or reconstructs a serialized handle.
func (a *Authority) InvalidateStaleCandidateSourceRequest(
	ctx context.Context,
	request CandidateSourceRequest,
	failure CandidateSourceFailure,
) (CatalogCandidateSourceFailureOutcomeV1, string, error) {
	if err := validateCandidateSourceFailureInput(failure); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	return a.failCandidateSource(
		ctx,
		nil,
		request,
		true,
		failure,
		CandidateSourceStale,
	)
}

func (a *Authority) failCandidateSource(
	ctx context.Context,
	handle *CatalogCandidateSourceHandle,
	request CandidateSourceRequest,
	useRequest bool,
	failure CandidateSourceFailure,
	terminalState CandidateSourceState,
) (CatalogCandidateSourceFailureOutcomeV1, string, error) {
	var validated validatedCandidateSourceRequest
	var err error
	orgID, commandID := "", ""
	if useRequest {
		validated, err = validateCandidateSourceRequest(request)
		if err != nil {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", err
		}
		orgID = validated.request.CatalogSnapshotBinding.OrgID
		commandID = validated.request.WriterCommandID
	} else {
		orgID = handle.orgID
		commandID = handle.writerCommandID
	}

	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	defer rollbackTx(tx)
	writer, err := lockCandidateSourcePrefix(ctx, tx, orgID)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	peekedState, err := peekCandidateSourceState(ctx, tx, orgID, commandID)
	if errors.Is(err, sql.ErrNoRows) {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", ErrSourceNotFound
	}
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if peekedState != CandidateSourcePrepared {
		record, err := readCandidateSourceDB(ctx, tx, orgID, commandID, true)
		if err != nil {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", err
		}
		if err := matchCandidateSourceFailureCaller(
			record, handle, validated, useRequest,
		); err != nil {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", err
		}
		if record.State != terminalState {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", ErrSourceNotReady
		}
		wanted, wantedDigest, err := candidateSourceFailureOutcome(
			record, failure, terminalState,
		)
		if err != nil {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", err
		}
		if record.FailureOutcome == nil ||
			record.FailureOutcomeDigest != wantedDigest ||
			*record.FailureOutcome != wanted {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", ErrSourceConflict
		}
		if err := validateFailedCandidateSourceLocked(
			ctx, tx, writer, record,
		); err != nil {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", err
		}
		if err := tx.Commit(); err != nil {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", err
		}
		return wanted, wantedDigest, nil
	}
	state, baseline, err := lockCandidateSourceActiveBaseline(ctx, tx, orgID)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	record, err := readCandidateSourceDB(ctx, tx, orgID, commandID, true)
	if errors.Is(err, sql.ErrNoRows) {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", ErrSourceNotFound
	}
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := rejectSourceCommandKindCollisionLocked(
		ctx, tx, orgID, commandID, true,
	); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := matchCandidateSourceFailureCaller(
		record, handle, validated, useRequest,
	); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if record.State != CandidateSourcePrepared {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", ErrSourceNotReady
	}
	if err := validateCandidateSourceWriter(record, writer); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if terminalState == CandidateSourceStale {
		lifecycleErr := validateCandidateSourceLifecycle(
			record, writer, state, baseline,
		)
		if lifecycleErr == nil {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", fmt.Errorf(
				"%w: source remains valid",
				ErrSourceConflict,
			)
		}
		if !errors.Is(lifecycleErr, ErrSourceConflict) {
			return CatalogCandidateSourceFailureOutcomeV1{}, "", lifecycleErr
		}
	}

	outcome, digest, err := candidateSourceFailureOutcome(
		record, failure, terminalState,
	)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := insertCandidateSourceFailureOutcome(ctx, tx, outcome, digest); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := catalogwriter.ReleaseGeneration(
		ctx,
		tx,
		record.OrgID,
		record.WriterCommandID,
		record.WriterReservationGeneration,
	); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", mapSourceWriterError(err)
	}
	if err := insertWriterGenerationTerminalHistory(
		ctx, tx, record, writerTerminalReleased, digest,
	); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := completeCandidateSourceFailure(
		ctx, tx, record, terminalState, digest,
	); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	if err := tx.Commit(); err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, "", err
	}
	return outcome, digest, nil
}

func matchCandidateSourceFailureCaller(
	record candidateSourceRecord,
	handle *CatalogCandidateSourceHandle,
	validated validatedCandidateSourceRequest,
	useRequest bool,
) error {
	if useRequest {
		if record.RequestDigest != validated.requestDigest {
			return ErrSourceConflict
		}
		return nil
	}
	return validateHandleRecord(handle, record)
}

func lockCandidateSourcePrefix(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
) (catalogwriter.State, error) {
	writer, err := catalogwriter.Lock(ctx, tx, orgID)
	if errors.Is(err, sql.ErrNoRows) {
		return catalogwriter.State{}, fmt.Errorf(
			"%w: org writer is not provisioned",
			ErrSourceNotReady,
		)
	}
	if err != nil {
		return catalogwriter.State{}, err
	}
	return writer, nil
}

func lockCandidateSourcePreparationPrefix(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
) (catalogwriter.State, error) {
	if err := lockCandidateSourcePreparationAdmission(ctx, tx); err != nil {
		return catalogwriter.State{}, err
	}
	return lockCandidateSourcePrefix(ctx, tx, orgID)
}

func lockCandidateSourceActiveBaseline(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
) (activeState, activeCandidateSourceBaseline, error) {
	state, err := readActiveStateDB(ctx, tx, orgID, true)
	if errors.Is(err, sql.ErrNoRows) {
		return activeState{}, activeCandidateSourceBaseline{}, nil
	}
	if err != nil {
		return activeState{}, activeCandidateSourceBaseline{}, err
	}
	binding, err := lockedActiveBinding(ctx, tx, orgID, state)
	if err != nil {
		return activeState{}, activeCandidateSourceBaseline{}, err
	}
	// Current durable publications are catalogpublicationv1. The later v2
	// adoption/cutover slice will decode its versioned receipt-backed binding
	// here; until then exact-current admission fails closed.
	return state, activeCandidateSourceBaseline{
		active:          state.Active,
		binding:         binding,
		receiptBackedV2: false,
	}, nil
}

func validateCandidateSourceLifecycle(
	record candidateSourceRecord,
	writer catalogwriter.State,
	state activeState,
	baseline activeCandidateSourceBaseline,
) error {
	if record.State != CandidateSourcePrepared {
		return fmt.Errorf("%w: source is %s", ErrSourceNotReady, record.State)
	}
	if err := validateCandidateSourceWriter(record, writer); err != nil {
		return err
	}
	if state.Active != baseline.active {
		return fmt.Errorf("%w: active source baseline differs", ErrSourceCorrupt)
	}
	if err := validateWriterCoherence(writer, baseline.binding); err != nil {
		return mapSourcePublicationError(err)
	}
	request := CandidateSourceRequest{
		WriterCommandID:              record.WriterCommandID,
		RuntimeAudience:              record.RuntimeAudience,
		CatalogSnapshotBinding:       record.CatalogSnapshotBinding,
		RequiredBundleKeySet:         cloneRequiredSet(record.RequiredBundleKeySet),
		ExpectedPriorPublication:     record.ExpectedPriorPublication,
		ExpectedPriorCatalogRevision: record.ExpectedPriorCatalogRevision,
		SourceAppSelectionSet:        cloneSourceAppSelectionSet(record.SourceAppSelectionSet),
	}
	relation, err := deriveCandidateSourceRelation(
		request,
		writer.CatalogRevision,
		baseline,
	)
	if err != nil {
		return fmt.Errorf("%w: %v", ErrSourceConflict, err)
	}
	if relation != record.Relation {
		return fmt.Errorf(
			"%w: durable source relation differs from its exact baseline",
			ErrSourceCorrupt,
		)
	}
	return nil
}

func validateCandidateSourceWriter(
	record candidateSourceRecord,
	writer catalogwriter.State,
) error {
	if writer.RuntimeAudience != record.RuntimeAudience {
		return fmt.Errorf(
			"%w: writer RuntimeAudience differs from durable source",
			ErrSourceCorrupt,
		)
	}
	if writer.ActiveCommandID != record.WriterCommandID ||
		writer.WriterGeneration != record.WriterReservationGeneration {
		return ErrSourceConflict
	}
	return nil
}

func validateFailedCandidateSourceLocked(
	ctx context.Context,
	tx *sql.Tx,
	writer catalogwriter.State,
	record candidateSourceRecord,
) error {
	if record.State != CandidateSourceTerminated &&
		record.State != CandidateSourceStale {
		return ErrSourceNotReady
	}
	if err := rejectSourceCommandKindCollisionLocked(
		ctx, tx, record.OrgID, record.WriterCommandID, true,
	); err != nil {
		return err
	}
	if writer.RuntimeAudience != record.RuntimeAudience {
		return fmt.Errorf(
			"%w: writer RuntimeAudience differs from terminal source",
			ErrSourceCorrupt,
		)
	}
	if writer.ActiveCommandID == record.WriterCommandID {
		return fmt.Errorf(
			"%w: terminal source command still owns a writer reservation",
			ErrSourceCorrupt,
		)
	}
	if writer.WriterGeneration < record.WriterReservationGeneration {
		return fmt.Errorf(
			"%w: terminal source generation is ahead of the org writer",
			ErrSourceCorrupt,
		)
	}
	if record.FailureOutcome == nil ||
		record.WriterTerminalAction != writerTerminalReleased ||
		record.WriterTerminalOutcome != record.FailureOutcomeDigest {
		return fmt.Errorf(
			"%w: terminal failure linkage is incomplete",
			ErrSourceCorrupt,
		)
	}
	return nil
}

func (a *Authority) validateCompletedCandidateSourceLocked(
	ctx context.Context,
	tx *sql.Tx,
	writer catalogwriter.State,
	record candidateSourceRecord,
) error {
	if record.State != CandidateSourcePublicationCompleted {
		return ErrSourceNotReady
	}
	if err := rejectSourceAppOperationLocked(
		ctx, tx, record.WriterCommandID, true,
	); err != nil {
		return err
	}
	command, err := readCommandDB(
		ctx, tx, record.OrgID, record.WriterCommandID, true,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf(
			"%w: publication-completed source has no publication command",
			ErrSourceCorrupt,
		)
	}
	if err != nil {
		return err
	}
	if command.State != CommandCompleted ||
		command.RuntimeAudience != record.RuntimeAudience ||
		command.ExpectedPrior != record.ExpectedPriorPublication ||
		command.RequestDigest != record.AdoptedPublicationRequest ||
		command.CandidatePublicationDigest != record.AdoptedCandidatePublication {
		return fmt.Errorf(
			"%w: completed source and publication command differ",
			ErrSourceCorrupt,
		)
	}
	publication, receiptBackedV2, err := readSourceAdoptedPublication(
		ctx,
		tx,
		record.OrgID,
		record.AdoptedCandidatePublication,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf(
			"%w: adopted candidate publication is missing",
			ErrSourceCorrupt,
		)
	}
	if err != nil {
		return err
	}
	if !receiptBackedV2 {
		return fmt.Errorf(
			"%w: publication-completed source cannot adopt a legacy v1 publication",
			ErrSourceCorrupt,
		)
	}
	if snapshotBindingFromPublication(publication.Binding) !=
		record.CatalogSnapshotBinding {
		return fmt.Errorf(
			"%w: adopted publication and source Catalog snapshot differ",
			ErrSourceCorrupt,
		)
	}
	outcome, err := readOutcomeDB(
		ctx, tx, record.OrgID, record.WriterCommandID, true,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf(
			"%w: publication-completed source has no immutable outcome",
			ErrSourceCorrupt,
		)
	}
	if err != nil {
		return err
	}
	if outcome.CandidatePublicationDigest != record.AdoptedCandidatePublication ||
		outcome.OutcomeDigest != record.CatalogPublicationOutcome ||
		(outcome.Result != ResultSelected &&
			outcome.Result != ResultExpectedPriorMismatch) {
		return fmt.Errorf(
			"%w: completed source and publication outcome differ",
			ErrSourceCorrupt,
		)
	}
	historyOutcome, err := readSourcePublicationHistoryOutcome(
		ctx, tx, record.OrgID, record.WriterCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf(
			"%w: publication-completed source has no immutable history",
			ErrSourceCorrupt,
		)
	}
	if err != nil {
		return err
	}
	if historyOutcome != outcome {
		return fmt.Errorf(
			"%w: completed publication outcome and history differ",
			ErrSourceCorrupt,
		)
	}
	expectedAction := writerTerminalReleased
	if outcome.Result == ResultSelected {
		expectedAction = writerTerminalCommitted
	}
	if record.WriterTerminalAction != expectedAction ||
		record.WriterTerminalOutcome != outcome.OutcomeDigest {
		return fmt.Errorf(
			"%w: completed source writer terminal history differs",
			ErrSourceCorrupt,
		)
	}
	if outcome.Result == ResultSelected &&
		writer.CatalogRevision != publication.Binding.CatalogRevision {
		return fmt.Errorf(
			"%w: selected source writer lineage differs from publication",
			ErrSourceCorrupt,
		)
	}
	if writer.RuntimeAudience != record.RuntimeAudience ||
		writer.ActiveCommandID == record.WriterCommandID {
		return fmt.Errorf(
			"%w: completed source retains or conflicts with writer identity",
			ErrSourceCorrupt,
		)
	}
	if writer.WriterGeneration < record.WriterReservationGeneration {
		return fmt.Errorf(
			"%w: completed source generation is ahead of the org writer",
			ErrSourceCorrupt,
		)
	}
	return nil
}

func readSourceAdoptedPublication(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	digest string,
) (Publication, bool, error) {
	publication, err := readPublicationDB(ctx, tx, orgID, digest, true)
	if err != nil {
		return Publication{}, false, err
	}
	// The current publication store contains only immutable v1 bindings.
	// Versioned v2 decoding and its receipt-set presence check arrive with the
	// atomic finalizer/cutover gate. Until then a forged
	// PUBLICATION_COMPLETED row cannot turn a legacy v1 command into an
	// authorized source-backed terminal pair.
	return publication, false, nil
}

func readSourcePublicationHistoryOutcome(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
) (Outcome, error) {
	var outcome Outcome
	var rawResult string
	var observedPresent, resultingPresent bool
	var observedDigest, resultingDigest sql.NullString
	err := tx.QueryRowContext(ctx, `SELECT OrgId, CommandId,
CandidatePublicationDigest, Result, ObservedPriorPresent, ObservedPriorDigest,
ResultingActivePresent, ResultingActiveDigest, OutcomeDigest
FROM _System_CatalogPublicationHistory
WHERE OrgId = ? AND CommandId = ? FOR UPDATE`, orgID, commandID).Scan(
		&outcome.OrgID,
		&outcome.CommandID,
		&outcome.CandidatePublicationDigest,
		&rawResult,
		&observedPresent,
		&observedDigest,
		&resultingPresent,
		&resultingDigest,
		&outcome.OutcomeDigest,
	)
	if err != nil {
		return Outcome{}, err
	}
	if outcome.OrgID != orgID || outcome.CommandID != commandID {
		return Outcome{}, fmt.Errorf(
			"%w: publication history identity differs",
			ErrSourceCorrupt,
		)
	}
	outcome.Result = Result(rawResult)
	outcome.ObservedPrior, err = durableReference(
		"source history ObservedPrior", observedPresent, observedDigest,
	)
	if err != nil {
		return Outcome{}, sourceCorrupt(err)
	}
	outcome.ResultingActive, err = durableReference(
		"source history ResultingActive", resultingPresent, resultingDigest,
	)
	if err != nil {
		return Outcome{}, sourceCorrupt(err)
	}
	if err := validateDurableOutcome(outcome); err != nil {
		return Outcome{}, sourceCorrupt(err)
	}
	return outcome, nil
}

func (a *Authority) preparationFromRecord(
	record candidateSourceRecord,
) CandidateSourcePreparation {
	result := CandidateSourcePreparation{
		State:                           record.State,
		Relation:                        record.Relation,
		PreparationOutcomeDigest:        record.PreparationOutcomeDigest,
		FailureOutcomeDigest:            record.FailureOutcomeDigest,
		CatalogPublicationOutcomeDigest: record.CatalogPublicationOutcome,
	}
	if record.FailureOutcome != nil {
		value := *record.FailureOutcome
		result.FailureOutcome = &value
	}
	if record.State == CandidateSourcePrepared {
		result.Handle = &CatalogCandidateSourceHandle{
			authority:             a.identity,
			orgID:                 record.OrgID,
			writerCommandID:       record.WriterCommandID,
			requestDigest:         record.RequestDigest,
			reservationGeneration: record.WriterReservationGeneration,
		}
	}
	return result
}

func (a *Authority) validateHandleIdentity(
	handle *CatalogCandidateSourceHandle,
) error {
	if handle == nil ||
		handle.authority == nil ||
		handle.authority != a.identity ||
		handle.orgID == "" ||
		!validCommandID(handle.writerCommandID) ||
		handle.requestDigest == "" ||
		handle.reservationGeneration == 0 {
		return ErrSourceConflict
	}
	return nil
}

func validateHandleRecord(
	handle *CatalogCandidateSourceHandle,
	record candidateSourceRecord,
) error {
	if handle.orgID != record.OrgID ||
		handle.writerCommandID != record.WriterCommandID ||
		handle.requestDigest != record.RequestDigest ||
		handle.reservationGeneration != record.WriterReservationGeneration {
		return ErrSourceConflict
	}
	return nil
}

func mapSourceWriterError(err error) error {
	switch {
	case errors.Is(err, catalogwriter.ErrConflict),
		errors.Is(err, catalogwriter.ErrLineageExhausted):
		return ErrSourceConflict
	default:
		return err
	}
}

func mapSourcePublicationError(err error) error {
	switch {
	case errors.Is(err, ErrNotReady):
		return fmt.Errorf("%w: %v", ErrSourceNotReady, err)
	case errors.Is(err, ErrCorrupt):
		return fmt.Errorf("%w: %v", ErrSourceCorrupt, err)
	default:
		return err
	}
}

func rejectSourceCommandKindCollisionLocked(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	asCorruption bool,
) error {
	if _, err := readCommandDB(ctx, tx, orgID, commandID, true); err == nil {
		return sourceCommandKindCollision(asCorruption, "publication")
	} else if !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	return rejectSourceAppOperationLocked(ctx, tx, commandID, asCorruption)
}

func rejectSourceAppOperationLocked(
	ctx context.Context,
	tx *sql.Tx,
	commandID string,
	asCorruption bool,
) error {
	var operationOrgID string
	err := tx.QueryRowContext(ctx, `SELECT OrgId
FROM _System_CatalogAppOperation WHERE OperationId = ? FOR UPDATE`, commandID).Scan(
		&operationOrgID,
	)
	if err == nil {
		return sourceCommandKindCollision(asCorruption, "App operation")
	}
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	var mysqlError *mysql.MySQLError
	if errors.As(err, &mysqlError) && mysqlError.Number == 1146 {
		// catalogpublication can be provisioned without catalogapp. In that
		// topology no App-operation journal can exist.
		return nil
	}
	return err
}

func sourceCommandKindCollision(asCorruption bool, kind string) error {
	if asCorruption {
		return fmt.Errorf(
			"%w: candidate-source command collides with a %s journal",
			ErrSourceCorrupt,
			kind,
		)
	}
	return ErrSourceConflict
}