chio_store_sqlite/

admission_operation_store.rs

use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{SystemTime, UNIX_EPOCH};

use chio_core::canonical::canonical_json_bytes;
use chio_core::economic_continuity::VerifiedEconomicStateBatchAdvance;
use chio_core::receipt::{body::ChioReceipt, lineage::ChildRequestReceipt};
use chio_core::{sha256_hex, StoreMutationFence};
#[cfg(test)]
use chio_credit::obligation::CreditExposureReservationRequest;
use chio_credit::obligation::{
    CreditExposureReservationRecordV1, ObligationAtomV1, ObligationDispositionRecordV1,
    ObligationSettlementLifecycleV1,
};
use chio_kernel::admission_operation::{
    AdmissionAttachment, AdmissionBeginResult, AdmissionCaptureError, AdmissionCommandResult,
    AdmissionDigest, AdmissionIdentifier, AdmissionOperationCommand, AdmissionOperationError,
    AdmissionOperationId, AdmissionOperationKind, AdmissionOperationState, AdmissionOperationStore,
    AdmissionOperationStoreError, AdmissionOperationV1, AdmissionProjectionCapabilities,
    AdmissionProjectionContext, AdmissionProjectionManifestV1, AdmissionProjectionRecordKind,
    AdmissionRecoveryLease, AdmissionReplayClassification, AdmissionReplayKey, AdmissionTerminal,
    AdmissionTerminalProjection, AdmissionTerminalReplay, CanonicalAdmissionProjectionRecord,
    CanonicalAdmissionTerminalProjection, PersistedAdmissionOperationV1,
    QualifiedAdmissionOperationStore, SideEffectClass, SignedAdmissionTerminalProjectionV1,
    UntrustedAdmissionRecoveryClaim, VerifiedAdmissionTerminalProjectionRecordV1,
    VerifiedAdmissionTerminalProjectionV1,
};
use chio_kernel::budget_store::{
    BudgetAuthorizeHoldDecision, BudgetAuthorizeHoldRequest, BudgetCaptureInvocationRequest,
    BudgetReconcileHoldRequest, BudgetStoreError,
};
use chio_kernel::payment::{PaymentJournalRecord, PaymentJournalTransition};
use chio_kernel::receipt_store::{
    AdmissionPaymentJournalAdvance, AdmissionPaymentJournalError, AdmissionPaymentSettlement,
    AdmissionPaymentSettlementBegin, AuthorizationReceiptConsumption, PendingSettlementObservation,
    ReceiptStore, ReceiptStoreError,
};
use rusqlite::{params, Connection, OptionalExtension, Row, Transaction, TransactionBehavior};
use serde::{Deserialize, Serialize};

use crate::serving_owner::{SqliteServingOwner, SqliteServingOwnerError};

mod commit_chain;
mod credit_exposure;
mod errors;
mod factor_assignment;
mod obligation;
mod participant;
mod projection;
mod schema;
mod store;
mod threshold_approval;

use commit_chain::append_operation_commit;
pub(crate) use commit_chain::{
    append_operation_commit_with_participant, load_admission_commit_head,
    verify_admission_commit_chain, verify_admission_commit_suffix, AdmissionCommitHead,
    GENESIS_CHAIN_DIGEST,
};
pub use credit_exposure::CreditExposureAccountSnapshot;
pub(crate) use credit_exposure::{
    apply_credit_exposure_terminal_tx, load_credit_exposure_reservation_tx,
    reserve_credit_exposure_tx,
};
use errors::*;
pub use factor_assignment::{
    DurableFactorAssignmentResultV1, FactorAssignmentAuthorityRegistryV1,
    FactorAssignmentAuthoritySetHeadV1, FactorAssignmentCommitV1,
    FactorAssignmentSigningAuthorityV1, FactorAssignmentVerificationAuthorityV1,
    SqliteFactorAssignmentStore, StoredFactorAssignmentResultV1,
};
use obligation::load_durable_obligation;
pub(crate) use participant::{
    advance_budget_authorization_tx, advance_budget_capture_tx, advance_tool_outcome_tx,
    append_participant_update_tx, finalize_channel_reservation_operation_tx,
    verify_budget_authorization_replay_tx, verify_participant_recovery_tx,
    BudgetAuthorizationAdvance,
};
use participant::{
    ensure_no_reserved_terminal_stage, qualify_generic_channel_command,
    validate_payment_reconcile_binding, verify_payment_terminal_source,
    verify_payment_write_context,
};
use projection::{
    ensure_projection_absent, full_projection_capabilities, insert_terminal_projection,
    insert_verified_terminal_projection, projected_terminal_state, terminal_from_operation,
    validate_canonical_projection_size, verify_exact_signed_terminal_replay,
    verify_exact_terminal_replay, verify_stored_terminal_projection,
};
use schema::{coordinator_lease_id_for_epoch, recovery_claim_digest, verify_latest_commit};
pub(crate) use schema::{
    initialize_admission_operation_schema, validate_trusted_time, verify_active_owner,
    verify_admission_operation_invariants, verify_trusted_time,
};

const ADMISSION_OPERATION_SCHEMA_KEY: &str = "admission_operation";
pub(crate) const ADMISSION_OPERATION_SUPPORTED_SCHEMA_VERSION: i32 = 9;
const ADMISSION_OPERATION_SCHEMA_ANCHORS: &[&str] = &[
    "admission_operations",
    "admission_operation_commits",
    "threshold_approval_proposals",
    "chio_serving_owner",
    "capability_grant_budgets",
];
const MAX_PERSISTED_OPERATION_BYTES: usize = 256 * 1024;
const MAX_TERMINAL_PROJECTION_BYTES: usize = 4 * 1024 * 1024;
const MAX_TERMINAL_MANIFEST_BYTES: usize = 256 * 1024;
const MAX_TERMINAL_RECORD_BYTES: usize = 1024 * 1024;
const MAX_TERMINAL_RECORDS: usize = 32;
const MAX_RECOVERY_BATCH: usize = 256;
const MAX_TRUSTED_UNIX_MS: u64 = (1_u64 << 53) - 1;
const MAX_TRUSTED_CLOCK_SKEW_MS: u64 = 5 * 60 * 1_000;
const MAX_RECOVERY_LEASE_DURATION_MS: u64 = 5 * 60 * 1_000;
const COMBINED_CAPTURE_OPERATION_MUTATION_KIND: &str = "compare_and_swap";

const ADMISSION_OPERATION_SCHEMA: &str = include_str!("admission_operation_store.sql");

#[derive(Clone)]
pub struct SqliteAdmissionOperationStore {
    connection: Arc<Mutex<Connection>>,
    serving_owner: Arc<SqliteServingOwner>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DurableObligationV1 {
    atom: ObligationAtomV1,
    disposition: ObligationDispositionRecordV1,
    settlement_lifecycle: ObligationSettlementLifecycleV1,
    head_sequence: u64,
    head_digest: String,
    snapshot_version: u64,
    resource_fence: u64,
}

impl DurableObligationV1 {
    #[must_use]
    pub const fn atom(&self) -> &ObligationAtomV1 {
        &self.atom
    }

    #[must_use]
    pub const fn disposition(&self) -> &ObligationDispositionRecordV1 {
        &self.disposition
    }

    #[must_use]
    pub const fn settlement_lifecycle(&self) -> &ObligationSettlementLifecycleV1 {
        &self.settlement_lifecycle
    }

    #[must_use]
    pub const fn head_sequence(&self) -> u64 {
        self.head_sequence
    }

    #[must_use]
    pub fn head_digest(&self) -> &str {
        &self.head_digest
    }

    #[must_use]
    pub const fn snapshot_version(&self) -> u64 {
        self.snapshot_version
    }

    #[must_use]
    pub const fn resource_fence(&self) -> u64 {
        self.resource_fence
    }
}

impl SqliteAdmissionOperationStore {
    pub(crate) fn open_alongside(
        connection: Arc<Mutex<Connection>>,
        serving_owner: Arc<SqliteServingOwner>,
    ) -> Self {
        Self {
            connection,
            serving_owner,
        }
    }

    fn connection(&self) -> Result<MutexGuard<'_, Connection>, AdmissionOperationStoreError> {
        self.connection.lock().map_err(|_| {
            AdmissionOperationStoreError::Invariant(
                "sqlite admission operation lock poisoned".to_string(),
            )
        })
    }

    fn begin_read<'a>(
        &self,
        connection: &'a mut Connection,
    ) -> Result<Transaction<'a>, AdmissionOperationStoreError> {
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Deferred)
            .map_err(sqlite_error)?;
        verify_active_owner(&transaction, &self.serving_owner, None)?;
        self.serving_owner
            .verify_authority_anchor(&transaction)
            .map_err(map_owner_error)?;
        Ok(transaction)
    }

    fn begin_write<'a>(
        &self,
        connection: &'a mut Connection,
        fence: Option<&StoreMutationFence>,
    ) -> Result<Transaction<'a>, AdmissionOperationStoreError> {
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(sqlite_error)?;
        verify_active_owner(&transaction, &self.serving_owner, fence)?;
        self.serving_owner
            .verify_authority_anchor(&transaction)
            .map_err(map_owner_error)?;
        Ok(transaction)
    }

    fn sync_after_write(
        &self,
        connection: &Connection,
    ) -> Result<(), AdmissionOperationStoreError> {
        self.serving_owner
            .sync_authority_anchor(connection)
            .map_err(map_owner_error)
    }

    fn commit_write(
        &self,
        transaction: Transaction<'_>,
    ) -> Result<(), AdmissionOperationStoreError> {
        transaction.commit().map_err(|error| {
            map_owner_error(self.serving_owner.outcome_unknown(format!(
                "sqlite admission operation commit outcome is unknown: {error}"
            )))
        })
    }

    pub fn load_obligation(
        &self,
        obligation_id: &str,
    ) -> Result<Option<DurableObligationV1>, AdmissionOperationStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        load_durable_obligation(&transaction, obligation_id)
    }

    #[cfg(test)]
    pub(crate) fn provision_credit_exposure_account(
        &self,
        request: &CreditExposureReservationRequest,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<CreditExposureAccountSnapshot, AdmissionOperationStoreError> {
        if active_fence != &self.serving_owner.fence {
            return Err(AdmissionOperationStoreError::Fenced);
        }
        request
            .validate()
            .map_err(|error| invariant(error.to_string()))?;
        request
            .authorities
            .ensure_current_at(trusted_now_unix_ms / 1_000)
            .map_err(|error| invariant(error.to_string()))?;
        request
            .credit_facility_bind
            .ensure_current_at(trusted_now_unix_ms)
            .map_err(|error| invariant(error.to_string()))?;
        let bind = request.credit_facility_bind.body();
        let reservation = CreditExposureReservationRecordV1::prepare_reserved(
            request,
            bind.expected_exposure_version()
                .checked_add(1)
                .ok_or_else(|| invariant("credit exposure account version overflowed"))?,
            bind.expected_exposure_fence()
                .checked_add(1)
                .ok_or_else(|| invariant("credit exposure resource fence overflowed"))?,
        )
        .map_err(|error| invariant(error.to_string()))?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, Some(active_fence))?;
        let snapshot = credit_exposure::initialize_credit_exposure_account_tx(
            &transaction,
            &reservation,
            0,
            0,
            active_fence,
            trusted_now_unix_ms,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(snapshot)
    }

    pub fn load_credit_exposure_account(
        &self,
        debtor_id: &str,
        scope_digest: &str,
        currency: &str,
    ) -> Result<Option<CreditExposureAccountSnapshot>, AdmissionOperationStoreError> {
        AdmissionIdentifier::try_new("credit_exposure_debtor_id", debtor_id.to_owned())?;
        AdmissionDigest::try_new("credit_exposure_scope_digest", scope_digest.to_owned())?;
        if currency.len() != 3 || !currency.bytes().all(|byte| byte.is_ascii_uppercase()) {
            return Err(invariant("credit exposure currency is invalid"));
        }
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        credit_exposure::load_credit_exposure_account_tx(
            &transaction,
            debtor_id,
            scope_digest,
            currency,
        )
    }

    pub fn load_credit_exposure_reservation(
        &self,
        operation_id: &str,
    ) -> Result<Option<CreditExposureReservationRecordV1>, AdmissionOperationStoreError> {
        AdmissionDigest::try_new("credit_exposure_operation_id", operation_id.to_owned())?;
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        load_credit_exposure_reservation_tx(&transaction, operation_id)
    }

    pub fn capture_invocation_and_commit_dispatch(
        &self,
        operation: &AdmissionOperationV1,
        recovery_lease: &AdmissionRecoveryLease,
        request: BudgetCaptureInvocationRequest,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<
        (
            chio_kernel::budget_store::BudgetInvocationCaptureDecision,
            AdmissionOperationV1,
        ),
        AdmissionCaptureError,
    > {
        if active_fence != &self.serving_owner.fence
            || recovery_lease.store_fence() != active_fence
            || operation.state() != AdmissionOperationState::CapturePending
            || operation.binding().capability_id().as_str() != request.capability_id
            || operation
                .budget_hold_id()
                .is_none_or(|hold_id| hold_id.as_str() != request.hold_id)
        {
            return Err(AdmissionCaptureError::Fenced);
        }
        let budget = crate::budget_store::SqliteBudgetStore::open_alongside(
            self.connection.clone(),
            self.serving_owner.clone(),
        );
        budget
            .capture_composite_invocation_and_commit_dispatch(
                request,
                crate::budget_store::AdmissionCaptureBinding {
                    operation,
                    recovery_lease,
                    trusted_now_unix_ms,
                },
            )
            .map_err(map_budget_capture_error)
    }

    #[allow(clippy::too_many_arguments)]
    pub fn authorize_budget_and_commit_admission(
        &self,
        operation: &AdmissionOperationV1,
        recovery_lease: &AdmissionRecoveryLease,
        request: BudgetAuthorizeHoldRequest,
        payment_journal: Option<PaymentJournalRecord>,
        credit_exposure: Option<chio_credit::obligation::CreditExposureReservationRequest>,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<(BudgetAuthorizeHoldDecision, AdmissionOperationV1), AdmissionCaptureError> {
        if active_fence != &self.serving_owner.fence || recovery_lease.store_fence() != active_fence
        {
            return Err(AdmissionCaptureError::Fenced);
        }
        let budget = crate::budget_store::SqliteBudgetStore::open_alongside(
            self.connection.clone(),
            self.serving_owner.clone(),
        );
        budget
            .authorize_composite_hold_and_commit_admission(
                request,
                crate::budget_store::AdmissionAuthorizationBinding {
                    operation,
                    recovery_lease,
                    payment_journal: payment_journal.as_ref(),
                    credit_exposure: credit_exposure.as_ref(),
                    trusted_now_unix_ms,
                },
            )
            .map_err(map_budget_capture_error)
    }

    pub fn load_payment_journal(
        &self,
        operation_id: &str,
        active_fence: &StoreMutationFence,
    ) -> Result<Option<PaymentJournalRecord>, AdmissionPaymentJournalError> {
        if active_fence != &self.serving_owner.fence {
            return Err(AdmissionPaymentJournalError::Fenced);
        }
        let mut connection = self.connection().map_err(map_payment_operation_error)?;
        let transaction = self
            .begin_read(&mut connection)
            .map_err(map_payment_operation_error)?;
        let journal = crate::budget_store::load_payment_journal(&transaction, operation_id)
            .map_err(map_payment_budget_error)?;
        transaction
            .commit()
            .map_err(|error| AdmissionPaymentJournalError::Invariant(error.to_string()))?;
        Ok(journal)
    }

    pub fn advance_payment_journal(
        &self,
        advance: AdmissionPaymentJournalAdvance<'_>,
    ) -> Result<PaymentJournalRecord, AdmissionPaymentJournalError> {
        let AdmissionPaymentJournalAdvance {
            operation,
            recovery_lease,
            expected,
            transition,
            release_evidence,
            active_fence,
            trusted_now_unix_ms,
        } = advance;
        if active_fence != &self.serving_owner.fence
            || recovery_lease.store_fence() != active_fence
            || expected.operation_id != operation.binding().operation_id().as_str()
        {
            return Err(AdmissionPaymentJournalError::Fenced);
        }
        let mut connection = self.connection().map_err(map_payment_operation_error)?;
        let transaction = self
            .begin_write(&mut connection, Some(active_fence))
            .map_err(map_payment_operation_error)?;
        verify_payment_write_context(
            &transaction,
            &self.serving_owner,
            operation,
            recovery_lease,
            active_fence,
            trusted_now_unix_ms,
        )?;
        let (updated, changed) = crate::budget_store::advance_payment_journal(
            &transaction,
            expected,
            transition,
            release_evidence,
            trusted_now_unix_ms,
        )
        .map_err(map_payment_budget_error)?;
        if changed {
            self.serving_owner
                .append_global_commit(
                    &transaction,
                    "payment_journal_transition",
                    "payment",
                    &updated.operation_id,
                    updated.journal_version,
                )
                .map_err(map_payment_owner_error)?;
        }
        self.commit_write(transaction)
            .map_err(map_payment_operation_error)?;
        if changed {
            self.sync_after_write(&connection)
                .map_err(map_payment_operation_error)?;
        }
        Ok(updated)
    }

    pub fn begin_payment_settlement(
        &self,
        begin: AdmissionPaymentSettlementBegin<'_>,
    ) -> Result<AdmissionPaymentSettlement, AdmissionPaymentJournalError> {
        let AdmissionPaymentSettlementBegin {
            operation,
            recovery_lease,
            expected,
            transition,
            release_evidence,
            budget_reconcile,
            active_fence,
            trusted_now_unix_ms,
        } = begin;
        if active_fence != &self.serving_owner.fence
            || recovery_lease.store_fence() != active_fence
            || expected.operation_id != operation.binding().operation_id().as_str()
        {
            return Err(AdmissionPaymentJournalError::Fenced);
        }
        validate_payment_reconcile_binding(expected, transition, &budget_reconcile)?;
        let mut connection = self.connection().map_err(map_payment_operation_error)?;
        let transaction = self
            .begin_write(&mut connection, Some(active_fence))
            .map_err(map_payment_operation_error)?;
        verify_payment_write_context(
            &transaction,
            &self.serving_owner,
            operation,
            recovery_lease,
            active_fence,
            trusted_now_unix_ms,
        )?;
        let budget = crate::budget_store::SqliteBudgetStore::open_alongside(
            self.connection.clone(),
            self.serving_owner.clone(),
        );
        let (budget, budget_changed) = budget
            .reconcile_composite_hold_in_transaction(&transaction, &budget_reconcile)
            .map_err(map_payment_budget_error)?;
        let (journal, payment_changed) = match transition {
            Some(transition) => crate::budget_store::advance_payment_journal(
                &transaction,
                expected,
                transition,
                release_evidence,
                trusted_now_unix_ms,
            )
            .map_err(map_payment_budget_error)?,
            None => {
                if release_evidence.is_some() {
                    return Err(AdmissionPaymentJournalError::Invariant(
                        "payment release evidence requires a journal transition".to_owned(),
                    ));
                }
                let stored = crate::budget_store::load_payment_journal(
                    &transaction,
                    operation.binding().operation_id().as_str(),
                )
                .map_err(map_payment_budget_error)?
                .ok_or_else(|| {
                    AdmissionPaymentJournalError::Invariant(
                        "payment settlement journal is absent".to_owned(),
                    )
                })?;
                if stored != *expected {
                    return Err(AdmissionPaymentJournalError::Conflict(
                        "payment settlement journal changed".to_owned(),
                    ));
                }
                (stored, false)
            }
        };
        if payment_changed {
            self.serving_owner
                .append_global_commit(
                    &transaction,
                    "payment_settlement_intent",
                    "payment",
                    &journal.operation_id,
                    journal.journal_version,
                )
                .map_err(map_payment_owner_error)?;
        }
        self.commit_write(transaction)
            .map_err(map_payment_operation_error)?;
        if budget_changed || payment_changed {
            self.sync_after_write(&connection)
                .map_err(map_payment_operation_error)?;
        }
        Ok(AdmissionPaymentSettlement {
            journal,
            budget,
            budget_already_reconciled: !budget_changed,
        })
    }

    pub fn commit_terminal_projection(
        &self,
        projection: &AdmissionTerminalProjection,
    ) -> Result<AdmissionTerminal, AdmissionOperationStoreError> {
        if projection.requires_anchored_economic_commit() {
            return Err(invariant(
                "terminal projection requires an advanced economic anchor",
            ));
        }
        projection.context().validate()?;
        let mut connection = self.connection()?;
        let transaction =
            self.begin_write(&mut connection, Some(&projection.context().store_fence))?;
        let (terminal, changed) =
            self.commit_terminal_projection_in_transaction(&transaction, projection)?;
        self.commit_write(transaction)?;
        if changed {
            self.sync_after_write(&connection)?;
        }
        Ok(terminal)
    }

    pub(super) fn commit_terminal_projection_in_transaction(
        &self,
        transaction: &Transaction<'_>,
        projection: &AdmissionTerminalProjection,
    ) -> Result<(AdmissionTerminal, bool), AdmissionOperationStoreError> {
        let stored = load_by_operation_id_tx(transaction, &projection.context().operation_id)?
            .ok_or(AdmissionOperationStoreError::NotFound)?;
        self.commit_terminal_projection_from_source_in_transaction(transaction, projection, &stored)
    }

    fn commit_terminal_projection_from_source_in_transaction(
        &self,
        transaction: &Transaction<'_>,
        projection: &AdmissionTerminalProjection,
        stored: &StoredOperation,
    ) -> Result<(AdmissionTerminal, bool), AdmissionOperationStoreError> {
        if projection.requires_anchored_economic_commit() {
            return Err(invariant(
                "terminal projection requires an advanced economic anchor",
            ));
        }
        let canonical = projection.canonical_projection()?;
        validate_canonical_projection_size(&canonical)?;
        let context = projection.context();
        context.validate()?;
        verify_trusted_time(transaction, context.trusted_time_unix_ms)?;
        verify_payment_terminal_source(
            transaction,
            &stored.operation,
            context,
            projected_terminal_state(projection),
            canonical.records().iter().filter_map(|record| {
                (record.commitment().kind() == AdmissionProjectionRecordKind::PaymentTerminal)
                    .then_some(record.canonical_bytes())
            }),
        )?;

        if stored.operation.state().is_terminal() {
            let terminal = verify_exact_terminal_replay(
                transaction,
                &stored.operation,
                projection,
                &canonical,
            )?;
            apply_credit_exposure_terminal_tx(
                transaction,
                &stored.operation,
                canonical.projection_digest(),
                projection.pre_dispatch_release_proof(),
                &context.store_fence,
                context.trusted_time_unix_ms,
            )?;
            return Ok((terminal, false));
        }
        if context.request_id != stored.operation.replay_key().request_id
            || context.expected_operation_version != stored.operation.version()
            || context.coordinator_lease_epoch != stored.operation.coordinator_lease_epoch()
            || context.trusted_time_unix_ms < stored.updated_at_unix_ms
        {
            return Err(AdmissionOperationError::TerminalProjectionBindingMismatch.into());
        }
        let recovery_claim = stored
            .recovery_claim
            .as_ref()
            .ok_or(AdmissionOperationStoreError::Fenced)?;
        if recovery_claim.coordinator_lease_id() != &context.coordinator_lease_id
            || recovery_claim.coordinator_lease_epoch() != context.coordinator_lease_epoch
            || recovery_claim.store_fence() != &context.store_fence
        {
            return Err(AdmissionOperationStoreError::Fenced);
        }
        verify_stored_recovery_claim(
            transaction,
            &self.serving_owner,
            stored,
            recovery_claim,
            context.trusted_time_unix_ms,
            &context.store_fence,
        )?;
        let capabilities = full_projection_capabilities();
        let updated = stored
            .operation
            .apply_terminal_projection(projection, &capabilities)?;
        if updated
            .terminal_replay()
            .is_none_or(|replay| replay.projection_digest() != canonical.projection_digest())
        {
            return Err(invariant(
                "terminal operation does not retain its exact projection digest",
            ));
        }
        let encoded = encode_operation(&updated)?;
        let changed = transaction
            .execute(
                r#"
                UPDATE admission_operations
                SET operation_json = ?1, state = ?2, terminal = 1,
                    coordinator_lease_epoch = ?3, version = ?4,
                    updated_at_unix_ms = ?5
                WHERE operation_id = ?6 AND version = ?7 AND terminal = 0
                "#,
                params![
                    &encoded,
                    state_name(updated.state()),
                    sqlite_i64(updated.coordinator_lease_epoch(), "coordinator_lease_epoch")?,
                    sqlite_i64(updated.version(), "terminal_operation_version")?,
                    sqlite_i64(context.trusted_time_unix_ms, "trusted_now_unix_ms")?,
                    context.operation_id.as_str(),
                    sqlite_i64(stored.operation.version(), "expected_operation_version")?,
                ],
            )
            .map_err(sqlite_error)?;
        if changed != 1 {
            return Err(AdmissionOperationStoreError::Fenced);
        }
        insert_terminal_projection(transaction, projection, &canonical, &updated)?;
        apply_credit_exposure_terminal_tx(
            transaction,
            &updated,
            canonical.projection_digest(),
            projection.pre_dispatch_release_proof(),
            &context.store_fence,
            context.trusted_time_unix_ms,
        )?;
        append_operation_commit(
            transaction,
            &updated,
            &encoded,
            Some(recovery_claim),
            "compare_and_swap",
            &self.serving_owner,
            context.trusted_time_unix_ms,
        )?;
        terminal_from_operation(&updated).map(|terminal| (terminal, true))
    }

    pub fn commit_signed_terminal_projection(
        &self,
        envelope: &SignedAdmissionTerminalProjectionV1,
    ) -> Result<AdmissionTerminal, AdmissionOperationStoreError> {
        let verified = envelope.verify()?;
        let context = verified.context();
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, Some(&context.store_fence))?;
        let terminal = self.commit_verified_signed_terminal_projection_in_transaction(
            &transaction,
            &verified,
            context.trusted_time_unix_ms,
            None,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(terminal)
    }
}

fn verify_stored_recovery_claim(
    transaction: &Transaction<'_>,
    owner: &SqliteServingOwner,
    stored: &StoredOperation,
    claim: &UntrustedAdmissionRecoveryClaim,
    trusted_now_unix_ms: u64,
    current_store_fence: &StoreMutationFence,
) -> Result<(), AdmissionOperationStoreError> {
    if trusted_now_unix_ms < stored.updated_at_unix_ms {
        return Err(invariant("trusted operation time regressed"));
    }
    if stored.recovery_claim.as_ref() != Some(claim)
        || stored.operation.binding().operation_id() != claim.operation_id()
        || stored.operation.version() != claim.claimed_version()
        || stored.operation.coordinator_lease_epoch() != claim.coordinator_lease_epoch()
        || claim.store_fence() != current_store_fence
        || current_store_fence != &owner.fence
    {
        return Err(AdmissionOperationStoreError::Fenced);
    }
    let historical_lease_id = coordinator_lease_id_for_epoch(
        transaction,
        owner,
        stored.operation.coordinator_lease_epoch(),
    )?;
    if &historical_lease_id != claim.coordinator_lease_id() {
        return Err(AdmissionOperationStoreError::Fenced);
    }
    if trusted_now_unix_ms >= claim.expires_at_unix_ms() {
        return Err(AdmissionOperationError::LeaseExpired.into());
    }
    Ok(())
}

struct StoredOperation {
    operation: AdmissionOperationV1,
    recovery_claim: Option<UntrustedAdmissionRecoveryClaim>,
    updated_at_unix_ms: u64,
}

pub(crate) enum PreparedAdmissionBeginTxResult {
    Created {
        encoded: Vec<u8>,
    },
    ExactReplay {
        operation: Box<AdmissionOperationV1>,
        terminal_replay: Option<AdmissionTerminalReplay>,
    },
    Conflict {
        existing_operation_id: AdmissionOperationId,
    },
}

struct RawOperationRow {
    operation_id: String,
    request_namespace_digest: String,
    request_id: String,
    operation_json: Vec<u8>,
    state: String,
    terminal: i64,
    coordinator_lease_epoch: i64,
    version: i64,
    created_at_unix_ms: i64,
    updated_at_unix_ms: i64,
    recovery_claimant_id: Option<String>,
    recovery_coordinator_lease_id: Option<String>,
    recovery_coordinator_lease_epoch: Option<i64>,
    recovery_claimed_version: Option<i64>,
    recovery_expires_at_unix_ms: Option<i64>,
    recovery_store_uuid: Option<String>,
    recovery_store_lease_id: Option<String>,
    recovery_store_owner_epoch: Option<i64>,
}

fn read_raw_row(row: &Row<'_>) -> rusqlite::Result<RawOperationRow> {
    Ok(RawOperationRow {
        operation_id: row.get(0)?,
        request_namespace_digest: row.get(1)?,
        request_id: row.get(2)?,
        operation_json: row.get(3)?,
        state: row.get(4)?,
        terminal: row.get(5)?,
        coordinator_lease_epoch: row.get(6)?,
        version: row.get(7)?,
        created_at_unix_ms: row.get(8)?,
        updated_at_unix_ms: row.get(9)?,
        recovery_claimant_id: row.get(10)?,
        recovery_coordinator_lease_id: row.get(11)?,
        recovery_coordinator_lease_epoch: row.get(12)?,
        recovery_claimed_version: row.get(13)?,
        recovery_expires_at_unix_ms: row.get(14)?,
        recovery_store_uuid: row.get(15)?,
        recovery_store_lease_id: row.get(16)?,
        recovery_store_owner_epoch: row.get(17)?,
    })
}

fn decode_row(raw: RawOperationRow) -> Result<StoredOperation, AdmissionOperationStoreError> {
    if raw.operation_json.is_empty() || raw.operation_json.len() > MAX_PERSISTED_OPERATION_BYTES {
        return Err(invariant("persisted admission operation size is invalid"));
    }
    let persisted: PersistedAdmissionOperationV1 = serde_json::from_slice(&raw.operation_json)
        .map_err(|error| invariant(format!("persisted admission operation is invalid: {error}")))?;
    let operation = AdmissionOperationV1::from_persisted(persisted)?;
    let canonical = encode_operation(&operation)?;
    if canonical != raw.operation_json {
        return Err(invariant(
            "persisted admission operation encoding is not canonical",
        ));
    }
    let replay_key = operation.replay_key();
    if operation.binding().operation_id().as_str() != raw.operation_id
        || replay_key.request_namespace_digest.as_str() != raw.request_namespace_digest
        || replay_key.request_id.as_str() != raw.request_id
        || state_name(operation.state()) != raw.state
        || i64::from(operation.state().is_terminal()) != raw.terminal
        || operation.coordinator_lease_epoch()
            != stored_u64(raw.coordinator_lease_epoch, "coordinator_lease_epoch")?
        || operation.version() != stored_u64(raw.version, "version")?
    {
        return Err(invariant(
            "admission operation columns do not match the checked record",
        ));
    }
    let created_at = stored_u64(raw.created_at_unix_ms, "created_at_unix_ms")?;
    let updated_at = stored_u64(raw.updated_at_unix_ms, "updated_at_unix_ms")?;
    validate_trusted_time(created_at, "created_at_unix_ms")?;
    validate_trusted_time(updated_at, "updated_at_unix_ms")?;
    if updated_at < created_at {
        return Err(invariant("admission operation timestamp regressed"));
    }

    let recovery_claim = match (
        raw.recovery_claimant_id,
        raw.recovery_coordinator_lease_id,
        raw.recovery_coordinator_lease_epoch,
        raw.recovery_claimed_version,
        raw.recovery_expires_at_unix_ms,
        raw.recovery_store_uuid,
        raw.recovery_store_lease_id,
        raw.recovery_store_owner_epoch,
    ) {
        (None, None, None, None, None, None, None, None) => None,
        (
            Some(claimant_id),
            Some(coordinator_lease_id),
            Some(coordinator_lease_epoch),
            Some(claimed_version),
            Some(expires_at_unix_ms),
            Some(store_uuid),
            Some(store_lease_id),
            Some(store_owner_epoch),
        ) => {
            let claimed_version = stored_u64(claimed_version, "recovery_claimed_version")?;
            if claimed_version > operation.version()
                || claimed_version
                    .checked_add(1)
                    .is_some_and(|next| next < operation.version())
            {
                return Err(invariant(
                    "recovery claim is not for the current or immediately preceding version",
                ));
            }
            let expires_at_unix_ms = stored_u64(expires_at_unix_ms, "recovery_expires_at_unix_ms")?;
            validate_trusted_time(expires_at_unix_ms, "recovery_expires_at_unix_ms")?;
            Some(UntrustedAdmissionRecoveryClaim::new(
                operation.binding().operation_id().clone(),
                AdmissionIdentifier::try_new("recovery_claimant_id", claimant_id)?,
                AdmissionIdentifier::try_new(
                    "recovery_coordinator_lease_id",
                    coordinator_lease_id,
                )?,
                stored_u64(coordinator_lease_epoch, "recovery_coordinator_lease_epoch")?,
                claimed_version,
                expires_at_unix_ms,
                StoreMutationFence {
                    store_uuid,
                    lease_id: store_lease_id,
                    owner_epoch: stored_u64(store_owner_epoch, "recovery_store_owner_epoch")?,
                },
            )?)
        }
        _ => return Err(invariant("recovery claim tuple is partial")),
    };
    Ok(StoredOperation {
        operation,
        recovery_claim,
        updated_at_unix_ms: updated_at,
    })
}

fn load_by_operation_id_tx(
    transaction: &Transaction<'_>,
    operation_id: &AdmissionOperationId,
) -> Result<Option<StoredOperation>, AdmissionOperationStoreError> {
    let raw = transaction
        .query_row(
            r#"
            SELECT operation_id, request_namespace_digest, request_id,
                   operation_json, state, terminal, coordinator_lease_epoch,
                   version, created_at_unix_ms, updated_at_unix_ms,
                   recovery_claimant_id, recovery_coordinator_lease_id,
                   recovery_coordinator_lease_epoch, recovery_claimed_version,
                   recovery_expires_at_unix_ms, recovery_store_uuid,
                   recovery_store_lease_id, recovery_store_owner_epoch
            FROM admission_operations WHERE operation_id = ?1
            "#,
            [operation_id.as_str()],
            read_raw_row,
        )
        .optional()
        .map_err(sqlite_error)?;
    let stored = raw.map(decode_row).transpose()?;
    if let Some(stored) = &stored {
        verify_latest_commit(transaction, stored)?;
        verify_stored_terminal_projection(transaction, stored)?;
    }
    Ok(stored)
}

pub(crate) fn begin_prepared_operation_tx(
    transaction: &Transaction<'_>,
    operation: &AdmissionOperationV1,
    fence: &StoreMutationFence,
    trusted_now_unix_ms: u64,
) -> Result<PreparedAdmissionBeginTxResult, AdmissionOperationStoreError> {
    operation.validate()?;
    if operation.state() != AdmissionOperationState::Prepared || operation.version() != 1 {
        return Err(invariant("begin requires a version-one Prepared operation"));
    }
    if operation.coordinator_lease_epoch() != fence.owner_epoch {
        return Err(AdmissionOperationStoreError::Fenced);
    }
    verify_trusted_time(transaction, trusted_now_unix_ms)?;
    let encoded = encode_operation(operation)?;
    let replay_key = operation.replay_key();
    if let Some(existing) = load_by_replay_key_tx(transaction, &replay_key)? {
        return Ok(match existing.operation.classify_replay(operation) {
            AdmissionReplayClassification::Exact { terminal_replay } => {
                PreparedAdmissionBeginTxResult::ExactReplay {
                    operation: Box::new(existing.operation),
                    terminal_replay,
                }
            }
            AdmissionReplayClassification::Conflict => PreparedAdmissionBeginTxResult::Conflict {
                existing_operation_id: existing.operation.binding().operation_id().clone(),
            },
        });
    }
    if load_by_operation_id_tx(transaction, operation.binding().operation_id())?.is_some() {
        return Err(invariant(
            "operation id is already bound to a different replay key",
        ));
    }
    let changed = transaction
        .execute(
            r#"
            INSERT INTO admission_operations (
                operation_id, request_namespace_digest, request_id,
                operation_json, state, terminal, coordinator_lease_epoch,
                version, created_at_unix_ms, updated_at_unix_ms
            ) VALUES (?1, ?2, ?3, ?4, ?5, 0, ?6, ?7, ?8, ?8)
            "#,
            params![
                operation.binding().operation_id().as_str(),
                replay_key.request_namespace_digest.as_str(),
                replay_key.request_id.as_str(),
                encoded,
                state_name(operation.state()),
                sqlite_i64(
                    operation.coordinator_lease_epoch(),
                    "coordinator_lease_epoch"
                )?,
                sqlite_i64(operation.version(), "version")?,
                sqlite_i64(trusted_now_unix_ms, "created_at_unix_ms")?,
            ],
        )
        .map_err(sqlite_error)?;
    if changed != 1 {
        return Err(invariant("begin did not insert exactly one operation"));
    }
    Ok(PreparedAdmissionBeginTxResult::Created { encoded })
}

pub(crate) fn load_operation_for_participant_tx(
    transaction: &Transaction<'_>,
    operation_id: &AdmissionOperationId,
) -> Result<Option<AdmissionOperationV1>, AdmissionOperationStoreError> {
    load_by_operation_id_tx(transaction, operation_id)
        .map(|stored| stored.map(|stored| stored.operation))
}

fn load_by_replay_key_tx(
    transaction: &Transaction<'_>,
    replay_key: &AdmissionReplayKey,
) -> Result<Option<StoredOperation>, AdmissionOperationStoreError> {
    let raw = transaction
        .query_row(
            r#"
            SELECT operation_id, request_namespace_digest, request_id,
                   operation_json, state, terminal, coordinator_lease_epoch,
                   version, created_at_unix_ms, updated_at_unix_ms,
                   recovery_claimant_id, recovery_coordinator_lease_id,
                   recovery_coordinator_lease_epoch, recovery_claimed_version,
                   recovery_expires_at_unix_ms, recovery_store_uuid,
                   recovery_store_lease_id, recovery_store_owner_epoch
            FROM admission_operations
            WHERE request_namespace_digest = ?1 AND request_id = ?2
            "#,
            params![
                replay_key.request_namespace_digest.as_str(),
                replay_key.request_id.as_str(),
            ],
            read_raw_row,
        )
        .optional()
        .map_err(sqlite_error)?;
    let stored = raw.map(decode_row).transpose()?;
    if let Some(stored) = &stored {
        verify_latest_commit(transaction, stored)?;
        verify_stored_terminal_projection(transaction, stored)?;
    }
    Ok(stored)
}

fn encode_operation(
    operation: &AdmissionOperationV1,
) -> Result<Vec<u8>, AdmissionOperationStoreError> {
    operation.validate()?;
    let encoded = canonical_json_bytes(&operation.to_persisted())
        .map_err(|error| invariant(format!("admission operation encoding failed: {error}")))?;
    if encoded.is_empty() || encoded.len() > MAX_PERSISTED_OPERATION_BYTES {
        return Err(invariant(
            "persisted admission operation exceeds its size limit",
        ));
    }
    Ok(encoded)
}

fn state_name(state: AdmissionOperationState) -> &'static str {
    match state {
        AdmissionOperationState::Prepared => "prepared",
        AdmissionOperationState::BrokerAttemptRegistered => "broker_attempt_registered",
        AdmissionOperationState::ApprovalRequired => "approval_required",
        AdmissionOperationState::BudgetAuthorized => "budget_authorized",
        AdmissionOperationState::ApprovalReserved => "approval_reserved",
        AdmissionOperationState::ReadyToDispatch => "ready_to_dispatch",
        AdmissionOperationState::CapturePending => "capture_pending",
        AdmissionOperationState::DispatchCommitted => "dispatch_committed",
        AdmissionOperationState::Finalizing => "finalizing",
        AdmissionOperationState::Completed => "completed",
        AdmissionOperationState::CompensatedBeforeDispatch => "compensated_before_dispatch",
        AdmissionOperationState::NotAcceptedAfterDispatchCommit => {
            "not_accepted_after_dispatch_commit"
        }
        AdmissionOperationState::OutcomeUnknownAfterDispatch => "outcome_unknown_after_dispatch",
        AdmissionOperationState::MutationReady => "mutation_ready",
        AdmissionOperationState::MutationSubmitted => "mutation_submitted",
        AdmissionOperationState::EconomicMutationApplied => "economic_mutation_applied",
        AdmissionOperationState::EconomicMutationNotApplied => "economic_mutation_not_applied",
    }
}

fn sqlite_i64(value: u64, field: &'static str) -> Result<i64, AdmissionOperationStoreError> {
    i64::try_from(value).map_err(|_| invariant(format!("{field} exceeds SQLite integer range")))
}

fn stored_u64(value: i64, field: &'static str) -> Result<u64, AdmissionOperationStoreError> {
    u64::try_from(value).map_err(|_| invariant(format!("{field} is negative")))
}

fn invariant(detail: impl Into<String>) -> AdmissionOperationStoreError {
    AdmissionOperationStoreError::Invariant(detail.into())
}

pub(crate) fn receipt_projection_error(error: AdmissionOperationStoreError) -> ReceiptStoreError {
    match error {
        AdmissionOperationStoreError::Unavailable(detail) => ReceiptStoreError::Pool(detail),
        AdmissionOperationStoreError::Fenced => ReceiptStoreError::Fenced,
        AdmissionOperationStoreError::NotFound => {
            ReceiptStoreError::NotFound("admission operation".to_string())
        }
        AdmissionOperationStoreError::Invariant(detail) => ReceiptStoreError::Conflict(detail),
        AdmissionOperationStoreError::OutcomeUnknown(detail) => {
            ReceiptStoreError::OutcomeUnknown(detail)
        }
        AdmissionOperationStoreError::Operation(error) => {
            ReceiptStoreError::Conflict(error.to_string())
        }
    }
}

fn decode_projection_receipt(bytes: Vec<u8>) -> Result<ChioReceipt, ReceiptStoreError> {
    let receipt: ChioReceipt = serde_json::from_slice(&bytes)?;
    if canonical_json_bytes(&receipt)
        .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
        != bytes
        || !receipt
            .verify_signature()
            .map_err(|error| ReceiptStoreError::CryptoDecode(error.to_string()))?
    {
        return Err(ReceiptStoreError::Conflict(
            "persisted admission receipt is invalid".to_string(),
        ));
    }
    Ok(receipt)
}

fn map_owner_error(error: SqliteServingOwnerError) -> AdmissionOperationStoreError {
    match error {
        SqliteServingOwnerError::OutcomeUnknown(detail) => {
            AdmissionOperationStoreError::OutcomeUnknown(detail)
        }
        error => invariant(error.to_string()),
    }
}

fn map_economic_cache_error(
    error: crate::economic_state_cache::EconomicStateCacheError,
) -> AdmissionOperationStoreError {
    match error {
        crate::economic_state_cache::EconomicStateCacheError::Unavailable(detail) => {
            AdmissionOperationStoreError::Unavailable(detail)
        }
        crate::economic_state_cache::EconomicStateCacheError::Fenced => {
            AdmissionOperationStoreError::Fenced
        }
        crate::economic_state_cache::EconomicStateCacheError::NotFound => {
            AdmissionOperationStoreError::NotFound
        }
        crate::economic_state_cache::EconomicStateCacheError::OutcomeUnknown(detail) => {
            AdmissionOperationStoreError::OutcomeUnknown(detail)
        }
        error => invariant(error.to_string()),
    }
}

fn sqlite_error(error: rusqlite::Error) -> AdmissionOperationStoreError {
    match error {
        rusqlite::Error::FromSqlConversionFailure(..)
        | rusqlite::Error::IntegralValueOutOfRange(..)
        | rusqlite::Error::InvalidColumnType(..)
        | rusqlite::Error::Utf8Error(..) => invariant(error.to_string()),
        other => AdmissionOperationStoreError::Unavailable(other.to_string()),
    }
}

impl From<AdmissionOperationStoreError> for SqliteServingOwnerError {
    fn from(error: AdmissionOperationStoreError) -> Self {
        Self::Invalid(error.to_string())
    }
}

#[cfg(test)]
#[path = "admission_operation_store_tests.rs"]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests;