package catalogpublication

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

	layoutbundlekeyv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
)

const (
	writerTerminalCommitted = "COMMITTED"
	writerTerminalReleased  = "RELEASED"
)

func peekCandidateSourceState(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
) (CandidateSourceState, error) {
	var rawState string
	err := tx.QueryRowContext(ctx, `SELECT State
FROM _System_CatalogCandidateSource
WHERE OrgId = ? AND WriterCommandId = ?`, orgID, commandID).Scan(&rawState)
	if err != nil {
		return "", err
	}
	state := CandidateSourceState(rawState)
	switch state {
	case CandidateSourcePrepared,
		CandidateSourcePublicationCompleted,
		CandidateSourceTerminated,
		CandidateSourceStale:
		return state, nil
	default:
		return "", fmt.Errorf(
			"%w: unknown source state %q",
			ErrSourceCorrupt,
			rawState,
		)
	}
}

func readCandidateSourceDB(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	forUpdate bool,
) (candidateSourceRecord, error) {
	statement := `SELECT OrgId, WriterCommandId, RequestDigest, PreparationOutcomeDigest,
RuntimeAudience, CatalogRevision, CatalogSnapshotDigest, SchemaGeneration, EngineGeneration,
RequiredBundleKeySetVersion, RequiredBundleKeySetDigest, RequiredBundleKeySetCanonicalBytes,
ExpectedPriorPresent, ExpectedPriorDigest, ExpectedPriorCatalogRevision,
SourceAppSelectionSetVersion, SourceAppSelectionSetCanonicalBytes,
WriterReservationGeneration, SourceRelation, State, FailureOutcomeDigest,
AdoptedPublicationRequestDigest, AdoptedCandidatePublicationDigest,
CatalogPublicationOutcomeDigest
FROM _System_CatalogCandidateSource WHERE OrgId = ? AND WriterCommandId = ?` + lockSuffix(forUpdate)
	var record candidateSourceRecord
	var rawCatalogRevision, rawSchemaGeneration, rawEngineGeneration []byte
	var rawExpectedPriorCatalogRevision, rawWriterGeneration []byte
	var priorPresent bool
	var priorDigest sql.NullString
	var rawRelation, rawState string
	var failureDigest, adoptedRequest, adoptedCandidate, publicationOutcome sql.NullString
	err := tx.QueryRowContext(ctx, statement, orgID, commandID).Scan(
		&record.OrgID,
		&record.WriterCommandID,
		&record.RequestDigest,
		&record.PreparationOutcomeDigest,
		&record.RuntimeAudience,
		&rawCatalogRevision,
		&record.CatalogSnapshotBinding.CatalogSnapshotDigest,
		&rawSchemaGeneration,
		&rawEngineGeneration,
		&record.RequiredBundleKeySet.Version,
		&record.RequiredBundleKeySetDigest,
		&record.RequiredBundleKeySetCanonical,
		&priorPresent,
		&priorDigest,
		&rawExpectedPriorCatalogRevision,
		&record.SourceAppSelectionSet.SourceAppSelectionSetVersion,
		&record.SourceAppSelectionCanonical,
		&rawWriterGeneration,
		&rawRelation,
		&rawState,
		&failureDigest,
		&adoptedRequest,
		&adoptedCandidate,
		&publicationOutcome,
	)
	if err != nil {
		return candidateSourceRecord{}, err
	}
	if record.OrgID != orgID || record.WriterCommandID != commandID {
		return candidateSourceRecord{}, fmt.Errorf(
			"%w: source identity does not exactly match its lookup",
			ErrSourceCorrupt,
		)
	}
	record.CatalogSnapshotBinding.OrgID = record.OrgID
	record.CatalogSnapshotBinding.CatalogRevision, err = parseSourceUint(
		"source CatalogRevision", rawCatalogRevision,
	)
	if err != nil {
		return candidateSourceRecord{}, err
	}
	record.CatalogSnapshotBinding.SchemaGeneration, err = parseSourceUint(
		"source SchemaGeneration", rawSchemaGeneration,
	)
	if err != nil {
		return candidateSourceRecord{}, err
	}
	record.CatalogSnapshotBinding.EngineGeneration, err = parseSourceUint(
		"source EngineGeneration", rawEngineGeneration,
	)
	if err != nil {
		return candidateSourceRecord{}, err
	}
	record.ExpectedPriorCatalogRevision, err = parseSourceUint(
		"source ExpectedPriorCatalogRevision", rawExpectedPriorCatalogRevision,
	)
	if err != nil {
		return candidateSourceRecord{}, err
	}
	record.WriterReservationGeneration, err = parseSourceUint(
		"source WriterReservationGeneration", rawWriterGeneration,
	)
	if err != nil || record.WriterReservationGeneration == 0 {
		return candidateSourceRecord{}, fmt.Errorf(
			"%w: invalid writer reservation generation",
			ErrSourceCorrupt,
		)
	}
	record.ExpectedPriorPublication, err = durableReference(
		"source ExpectedPriorPublication", priorPresent, priorDigest,
	)
	if err != nil {
		return candidateSourceRecord{}, sourceCorrupt(err)
	}
	record.Relation = CandidateSourceRelation(rawRelation)
	record.State = CandidateSourceState(rawState)
	record.FailureOutcomeDigest = failureDigest.String
	record.AdoptedPublicationRequest = adoptedRequest.String
	record.AdoptedCandidatePublication = adoptedCandidate.String
	record.CatalogPublicationOutcome = publicationOutcome.String
	if err := validateCandidateSourceStateShape(
		record,
		failureDigest.Valid,
		adoptedRequest.Valid,
		adoptedCandidate.Valid,
		publicationOutcome.Valid,
	); err != nil {
		return candidateSourceRecord{}, err
	}

	record.RequiredBundleKeySet.BundleKeys, err = readCandidateSourceKeys(
		ctx, tx, orgID, commandID, forUpdate,
	)
	if err != nil {
		return candidateSourceRecord{}, err
	}
	record.SourceAppSelectionSet.Entries, err = readCandidateSourceAppSelections(
		ctx, tx, orgID, commandID, forUpdate,
	)
	if err != nil {
		return candidateSourceRecord{}, err
	}

	failure, failureErr := readCandidateSourceFailureOutcomeDB(
		ctx, tx, orgID, commandID, record.WriterReservationGeneration, forUpdate,
	)
	switch record.State {
	case CandidateSourceTerminated, CandidateSourceStale:
		if errors.Is(failureErr, sql.ErrNoRows) {
			return candidateSourceRecord{}, fmt.Errorf(
				"%w: terminal source has no immutable failure outcome",
				ErrSourceCorrupt,
			)
		}
		if failureErr != nil {
			return candidateSourceRecord{}, failureErr
		}
		record.FailureOutcome = &failure
	default:
		if failureErr == nil {
			return candidateSourceRecord{}, fmt.Errorf(
				"%w: non-failure source has an immutable failure outcome",
				ErrSourceCorrupt,
			)
		}
		if !errors.Is(failureErr, sql.ErrNoRows) {
			return candidateSourceRecord{}, failureErr
		}
	}

	action, terminalDigest, historyErr := readWriterGenerationTerminalHistory(
		ctx, tx, orgID, commandID, record.WriterReservationGeneration, forUpdate,
	)
	if record.State == CandidateSourcePrepared {
		if historyErr == nil {
			return candidateSourceRecord{}, fmt.Errorf(
				"%w: prepared source has terminal writer-generation history",
				ErrSourceCorrupt,
			)
		}
		if !errors.Is(historyErr, sql.ErrNoRows) {
			return candidateSourceRecord{}, historyErr
		}
	} else {
		if errors.Is(historyErr, sql.ErrNoRows) {
			return candidateSourceRecord{}, fmt.Errorf(
				"%w: terminal source has no writer-generation history",
				ErrSourceCorrupt,
			)
		}
		if historyErr != nil {
			return candidateSourceRecord{}, historyErr
		}
		record.WriterTerminalAction = action
		record.WriterTerminalOutcome = terminalDigest
	}

	if err := validateDurableCandidateSource(record); err != nil {
		return candidateSourceRecord{}, err
	}
	return record, nil
}

func readCandidateSourceKeys(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	forUpdate bool,
) ([]layoutbundlekeyv1.BundleKey, error) {
	statement := `SELECT KeyOrdinal, SlotOwnerKind, SlotOwnerNamespace, LocalComponentKey
FROM _System_CatalogCandidateSourceRequiredKey
WHERE OrgId = ? AND WriterCommandId = ? ORDER BY KeyOrdinal ASC` + lockSuffix(forUpdate)
	rows, err := tx.QueryContext(ctx, statement, orgID, commandID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	keys := make([]layoutbundlekeyv1.BundleKey, 0)
	for rows.Next() {
		if len(keys) == layoutbundlekeyv1.MaxRequiredBundleKeys {
			return nil, fmt.Errorf(
				"%w: required key membership exceeds %d entries",
				ErrSourceCorrupt,
				layoutbundlekeyv1.MaxRequiredBundleKeys,
			)
		}
		var rawOrdinal []byte
		var kind, namespace, localKey string
		if err := rows.Scan(&rawOrdinal, &kind, &namespace, &localKey); err != nil {
			return nil, err
		}
		ordinal, err := parseSourceUint("source required-key ordinal", rawOrdinal)
		if err != nil || ordinal != uint64(len(keys)) {
			return nil, fmt.Errorf(
				"%w: required key ordinals are not exact and contiguous",
				ErrSourceCorrupt,
			)
		}
		keys = append(keys, layoutbundlekeyv1.BundleKey{
			SlotOwnerKind:      layoutbundlekeyv1.SlotOwnerKind(kind),
			SlotOwnerNamespace: namespace,
			LocalComponentKey:  layoutbundlekeyv1.LocalComponentKey(localKey),
		})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return keys, nil
}

func readCandidateSourceAppSelections(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	forUpdate bool,
) ([]SourceAppSelection, error) {
	statement := `SELECT SelectionOrdinal, AppId, AppInstallationId, AppReleaseDigest,
SelectionKind, AppLifecycleCandidateCommandId
FROM _System_CatalogCandidateSourceAppSelection
WHERE OrgId = ? AND WriterCommandId = ? ORDER BY SelectionOrdinal ASC` + lockSuffix(forUpdate)
	rows, err := tx.QueryContext(ctx, statement, orgID, commandID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	selections := make([]SourceAppSelection, 0)
	for rows.Next() {
		if len(selections) == MaxSourceAppSelectionEntries {
			return nil, fmt.Errorf(
				"%w: source App selection membership exceeds %d entries",
				ErrSourceCorrupt,
				MaxSourceAppSelectionEntries,
			)
		}
		var rawOrdinal []byte
		var selection SourceAppSelection
		var rawKind string
		var candidateCommand sql.NullString
		if err := rows.Scan(
			&rawOrdinal,
			&selection.AppID,
			&selection.AppInstallationID,
			&selection.AppReleaseDigest,
			&rawKind,
			&candidateCommand,
		); err != nil {
			return nil, err
		}
		ordinal, err := parseSourceUint("source App selection ordinal", rawOrdinal)
		if err != nil || ordinal != uint64(len(selections)) {
			return nil, fmt.Errorf(
				"%w: source App selection ordinals are not exact and contiguous",
				ErrSourceCorrupt,
			)
		}
		selection.SelectionKind = SourceAppSelectionKind(rawKind)
		switch selection.SelectionKind {
		case SourceAppSelectionKindActive:
			if candidateCommand.Valid {
				return nil, fmt.Errorf(
					"%w: ACTIVE source App selection has a candidate command",
					ErrSourceCorrupt,
				)
			}
		case SourceAppSelectionKindCandidate:
			if !candidateCommand.Valid || candidateCommand.String != commandID {
				return nil, fmt.Errorf(
					"%w: CANDIDATE source App selection lacks its exact writer command",
					ErrSourceCorrupt,
				)
			}
			selection.AppLifecycleCandidateCommandID = candidateCommand.String
		default:
			return nil, fmt.Errorf(
				"%w: unknown source App selection kind %q",
				ErrSourceCorrupt,
				rawKind,
			)
		}
		selections = append(selections, selection)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return selections, nil
}

func readCandidateSourceFailureOutcomeDB(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	generation uint64,
	forUpdate bool,
) (CatalogCandidateSourceFailureOutcomeV1, error) {
	statement := `SELECT OrgId, WriterCommandId, WriterReservationGeneration,
SourceRequestDigest, ExpectedPriorPresent, ExpectedPriorDigest,
ExpectedPriorCatalogRevision, FailureClass, FailureCode, FailureDetailDigest,
RetryDisposition, SourceTerminalState, FailureOutcomeDigest
FROM _System_CatalogCandidateSourceFailureOutcome
WHERE OrgId = ? AND WriterCommandId = ? AND WriterReservationGeneration = ?` +
		lockSuffix(forUpdate)
	var outcome CatalogCandidateSourceFailureOutcomeV1
	var rawGeneration, rawPriorRevision []byte
	var priorPresent bool
	var priorDigest sql.NullString
	var rawClass, rawRetry, rawState, storedDigest string
	err := tx.QueryRowContext(
		ctx,
		statement,
		orgID,
		commandID,
		decimalText(generation),
	).Scan(
		&outcome.OrgID,
		&outcome.WriterCommandID,
		&rawGeneration,
		&outcome.SourceRequestDigest,
		&priorPresent,
		&priorDigest,
		&rawPriorRevision,
		&rawClass,
		&outcome.FailureCode,
		&outcome.FailureDetailDigest,
		&rawRetry,
		&rawState,
		&storedDigest,
	)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, err
	}
	if outcome.OrgID != orgID || outcome.WriterCommandID != commandID {
		return CatalogCandidateSourceFailureOutcomeV1{}, fmt.Errorf(
			"%w: failure outcome identity differs",
			ErrSourceCorrupt,
		)
	}
	outcome.WriterReservationGeneration, err = parseSourceUint(
		"failure outcome WriterReservationGeneration",
		rawGeneration,
	)
	if err != nil || outcome.WriterReservationGeneration != generation {
		return CatalogCandidateSourceFailureOutcomeV1{}, fmt.Errorf(
			"%w: failure outcome writer generation differs",
			ErrSourceCorrupt,
		)
	}
	outcome.ExpectedPriorPublication, err = durableReference(
		"failure outcome ExpectedPriorPublication", priorPresent, priorDigest,
	)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, sourceCorrupt(err)
	}
	outcome.ExpectedPriorCatalogRevision, err = parseSourceUint(
		"failure outcome ExpectedPriorCatalogRevision", rawPriorRevision,
	)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, err
	}
	outcome.FailureClass = CandidateSourceFailureClass(rawClass)
	outcome.RetryDisposition = CandidateSourceRetryDisposition(rawRetry)
	outcome.SourceTerminalState = CandidateSourceState(rawState)
	computed, err := catalogCandidateSourceFailureOutcomeDigest(outcome)
	if err != nil {
		return CatalogCandidateSourceFailureOutcomeV1{}, fmt.Errorf(
			"%w: invalid failure outcome: %v",
			ErrSourceCorrupt,
			err,
		)
	}
	if computed != storedDigest {
		return CatalogCandidateSourceFailureOutcomeV1{}, fmt.Errorf(
			"%w: failure outcome digest differs",
			ErrSourceCorrupt,
		)
	}
	return outcome, nil
}

func readWriterGenerationTerminalHistory(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	generation uint64,
	forUpdate bool,
) (string, string, error) {
	statement := `SELECT OrgId, WriterCommandId, WriterReservationGeneration,
TerminalAction, TerminalOutcomeDigest
FROM _System_CatalogWriterGenerationTerminalHistory
WHERE OrgId = ? AND WriterCommandId = ? AND WriterReservationGeneration = ?` +
		lockSuffix(forUpdate)
	var storedOrg, storedCommand, action, outcomeDigest string
	var rawGeneration []byte
	err := tx.QueryRowContext(
		ctx,
		statement,
		orgID,
		commandID,
		decimalText(generation),
	).Scan(
		&storedOrg,
		&storedCommand,
		&rawGeneration,
		&action,
		&outcomeDigest,
	)
	if err != nil {
		return "", "", err
	}
	storedGeneration, err := parseSourceUint(
		"writer terminal WriterReservationGeneration", rawGeneration,
	)
	if err != nil {
		return "", "", err
	}
	if storedOrg != orgID || storedCommand != commandID ||
		storedGeneration != generation {
		return "", "", fmt.Errorf(
			"%w: writer terminal history identity differs",
			ErrSourceCorrupt,
		)
	}
	if action != writerTerminalCommitted && action != writerTerminalReleased {
		return "", "", fmt.Errorf(
			"%w: writer terminal history action is unknown",
			ErrSourceCorrupt,
		)
	}
	if err := validateSHA256Digest(outcomeDigest); err != nil {
		return "", "", fmt.Errorf(
			"%w: writer terminal outcome digest is invalid",
			ErrSourceCorrupt,
		)
	}
	return action, outcomeDigest, nil
}

func validateDurableCandidateSource(record candidateSourceRecord) error {
	if record.WriterReservationGeneration == 0 {
		return fmt.Errorf(
			"%w: invalid writer reservation generation",
			ErrSourceCorrupt,
		)
	}
	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),
	}
	validated, err := validateCandidateSourceRequest(request)
	if err != nil {
		return fmt.Errorf("%w: invalid source prefix: %v", ErrSourceCorrupt, err)
	}
	if validated.requestDigest != record.RequestDigest ||
		validated.setDigest != record.RequiredBundleKeySetDigest ||
		!bytes.Equal(validated.setBytes, record.RequiredBundleKeySetCanonical) ||
		!bytes.Equal(validated.appSelectionBytes, record.SourceAppSelectionCanonical) {
		return fmt.Errorf(
			"%w: source digest or canonical membership does not match durable facts",
			ErrSourceCorrupt,
		)
	}
	switch record.Relation {
	case CandidateSourceCatalogSuccessor, CandidateSourceExactCurrent:
	default:
		return fmt.Errorf(
			"%w: unknown source relation %q",
			ErrSourceCorrupt,
			record.Relation,
		)
	}
	if candidateSourcePreparationOutcomeDigest(record) != record.PreparationOutcomeDigest {
		return fmt.Errorf(
			"%w: preparation outcome digest does not match durable source",
			ErrSourceCorrupt,
		)
	}
	switch record.State {
	case CandidateSourcePrepared:
		if record.WriterTerminalAction != "" || record.WriterTerminalOutcome != "" {
			return fmt.Errorf(
				"%w: prepared source has terminal writer history",
				ErrSourceCorrupt,
			)
		}
	case CandidateSourcePublicationCompleted:
		if err := validateSHA256Digest(record.AdoptedPublicationRequest); err != nil {
			return fmt.Errorf("%w: adopted publication request digest is invalid", ErrSourceCorrupt)
		}
		if err := validateSHA256Digest(record.AdoptedCandidatePublication); err != nil {
			return fmt.Errorf("%w: adopted candidate publication digest is invalid", ErrSourceCorrupt)
		}
		if err := validateSHA256Digest(record.CatalogPublicationOutcome); err != nil {
			return fmt.Errorf("%w: Catalog publication outcome digest is invalid", ErrSourceCorrupt)
		}
		if record.WriterTerminalOutcome != record.CatalogPublicationOutcome {
			return fmt.Errorf(
				"%w: publication source and writer terminal outcomes differ",
				ErrSourceCorrupt,
			)
		}
	case CandidateSourceTerminated, CandidateSourceStale:
		if record.FailureOutcome == nil {
			return fmt.Errorf("%w: terminal source has no failure outcome", ErrSourceCorrupt)
		}
		outcome := *record.FailureOutcome
		computed, err := catalogCandidateSourceFailureOutcomeDigest(outcome)
		if err != nil {
			return fmt.Errorf("%w: invalid failure outcome: %v", ErrSourceCorrupt, err)
		}
		if computed != record.FailureOutcomeDigest ||
			outcome.OrgID != record.OrgID ||
			outcome.WriterCommandID != record.WriterCommandID ||
			outcome.WriterReservationGeneration != record.WriterReservationGeneration ||
			outcome.SourceRequestDigest != record.RequestDigest ||
			outcome.ExpectedPriorPublication != record.ExpectedPriorPublication ||
			outcome.ExpectedPriorCatalogRevision != record.ExpectedPriorCatalogRevision ||
			outcome.SourceTerminalState != record.State ||
			record.WriterTerminalAction != writerTerminalReleased ||
			record.WriterTerminalOutcome != record.FailureOutcomeDigest {
			return fmt.Errorf(
				"%w: terminal failure linkage differs from durable source",
				ErrSourceCorrupt,
			)
		}
	default:
		return fmt.Errorf("%w: unknown source state %q", ErrSourceCorrupt, record.State)
	}
	return nil
}

func validateCandidateSourceStateShape(
	record candidateSourceRecord,
	failurePresent bool,
	adoptedRequestPresent bool,
	adoptedCandidatePresent bool,
	publicationOutcomePresent bool,
) error {
	switch record.State {
	case CandidateSourcePrepared:
		if failurePresent || adoptedRequestPresent || adoptedCandidatePresent ||
			publicationOutcomePresent {
			return fmt.Errorf("%w: prepared source has terminal linkage", ErrSourceCorrupt)
		}
	case CandidateSourcePublicationCompleted:
		if failurePresent || !adoptedRequestPresent || !adoptedCandidatePresent ||
			!publicationOutcomePresent {
			return fmt.Errorf(
				"%w: publication-completed source linkage is incomplete",
				ErrSourceCorrupt,
			)
		}
	case CandidateSourceTerminated, CandidateSourceStale:
		if !failurePresent || adoptedRequestPresent || adoptedCandidatePresent ||
			publicationOutcomePresent {
			return fmt.Errorf("%w: failure terminal linkage is invalid", ErrSourceCorrupt)
		}
	default:
		return fmt.Errorf("%w: unknown source state %q", ErrSourceCorrupt, record.State)
	}
	return nil
}

func insertCandidateSource(
	ctx context.Context,
	tx *sql.Tx,
	record candidateSourceRecord,
) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogCandidateSource
(OrgId, WriterCommandId, RequestDigest, PreparationOutcomeDigest, RuntimeAudience,
CatalogRevision, CatalogSnapshotDigest, SchemaGeneration, EngineGeneration,
RequiredBundleKeySetVersion, RequiredBundleKeySetDigest, RequiredBundleKeySetCanonicalBytes,
ExpectedPriorPresent, ExpectedPriorDigest, ExpectedPriorCatalogRevision,
SourceAppSelectionSetVersion, SourceAppSelectionSetCanonicalBytes,
WriterReservationGeneration, SourceRelation, State, FailureOutcomeDigest,
AdoptedPublicationRequestDigest, AdoptedCandidatePublicationDigest,
CatalogPublicationOutcomeDigest, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'PREPARED',
NULL, NULL, NULL, NULL, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
		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),
	)
	if err != nil {
		return err
	}
	if err := requireOneSource(result); err != nil {
		return err
	}

	for index, key := range record.RequiredBundleKeySet.BundleKeys {
		result, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogCandidateSourceRequiredKey
(OrgId, WriterCommandId, KeyOrdinal, SlotOwnerKind, SlotOwnerNamespace, LocalComponentKey, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
			record.OrgID,
			record.WriterCommandID,
			decimalText(uint64(index)),
			string(key.SlotOwnerKind),
			key.SlotOwnerNamespace,
			string(key.LocalComponentKey),
		)
		if err != nil {
			return err
		}
		if err := requireOneSource(result); err != nil {
			return err
		}
	}

	for index, selection := range record.SourceAppSelectionSet.Entries {
		result, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogCandidateSourceAppSelection
(OrgId, WriterCommandId, SelectionOrdinal, AppId, AppInstallationId, AppReleaseDigest,
SelectionKind, AppLifecycleCandidateCommandId, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
			record.OrgID,
			record.WriterCommandID,
			decimalText(uint64(index)),
			selection.AppID,
			selection.AppInstallationID,
			selection.AppReleaseDigest,
			string(selection.SelectionKind),
			sourceNullableString(selection.AppLifecycleCandidateCommandID),
		)
		if err != nil {
			return err
		}
		if err := requireOneSource(result); err != nil {
			return err
		}
	}
	return nil
}

func insertCandidateSourceFailureOutcome(
	ctx context.Context,
	tx *sql.Tx,
	outcome CatalogCandidateSourceFailureOutcomeV1,
	digest string,
) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogCandidateSourceFailureOutcome
(OrgId, WriterCommandId, WriterReservationGeneration, SourceRequestDigest,
ExpectedPriorPresent, ExpectedPriorDigest, ExpectedPriorCatalogRevision,
FailureClass, FailureCode, FailureDetailDigest, RetryDisposition,
SourceTerminalState, FailureOutcomeDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		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,
	)
	if err != nil {
		return err
	}
	return requireOneSource(result)
}

func insertWriterGenerationTerminalHistory(
	ctx context.Context,
	tx *sql.Tx,
	record candidateSourceRecord,
	action string,
	outcomeDigest string,
) error {
	if action != writerTerminalCommitted && action != writerTerminalReleased {
		return ErrSourceConflict
	}
	if err := validateSHA256Digest(outcomeDigest); err != nil {
		return err
	}
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogWriterGenerationTerminalHistory
(OrgId, WriterCommandId, WriterReservationGeneration, TerminalAction,
TerminalOutcomeDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		record.OrgID,
		record.WriterCommandID,
		decimalText(record.WriterReservationGeneration),
		action,
		outcomeDigest,
	)
	if err != nil {
		return err
	}
	return requireOneSource(result)
}

func completeCandidateSourceFailure(
	ctx context.Context,
	tx *sql.Tx,
	record candidateSourceRecord,
	state CandidateSourceState,
	outcomeDigest string,
) error {
	if record.State != CandidateSourcePrepared ||
		(state != CandidateSourceTerminated && state != CandidateSourceStale) {
		return ErrSourceConflict
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogCandidateSource
SET State = ?, FailureOutcomeDigest = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND WriterCommandId = ? AND State = 'PREPARED'
AND RequestDigest = ? AND WriterReservationGeneration = ?`,
		string(state),
		outcomeDigest,
		record.OrgID,
		record.WriterCommandID,
		record.RequestDigest,
		decimalText(record.WriterReservationGeneration),
	)
	if err != nil {
		return err
	}
	return requireOneSource(result)
}

func parseSourceUint(label string, raw []byte) (uint64, error) {
	value, err := parseDurableUint64(label, raw)
	return value, sourceCorrupt(err)
}

func sourceNullableString(value string) any {
	if value == "" {
		return nil
	}
	return value
}

func requireOneSource(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return ErrSourceConflict
	}
	return nil
}

func sourceCorrupt(err error) error {
	if err == nil || errors.Is(err, ErrSourceCorrupt) {
		return err
	}
	return fmt.Errorf("%w: %v", ErrSourceCorrupt, err)
}