chio_store_sqlite/

fiscal_store.rs

use std::collections::HashSet;
use std::sync::{Arc, Mutex, MutexGuard};

use chio_core::canonical::canonical_json_bytes;
use chio_core::{sha256_hex, Keypair, StoreMutationFence};
use chio_fiscal::{
    fee_schedule::SignedOpenMarketFeeSchedule, FiscalActivationTarget, FiscalAdmissionAuthority,
    FiscalAdmissionTrustRegistry, FiscalAuthorityState, FiscalCharterRegistry, FiscalDomain,
    FiscalGenesisPolicy, FiscalParams, FiscalProposalAdmissionBuilder,
    FiscalProposalAdmissionState, FiscalProposalAdmissionStatus, FiscalRuntimeAdapterRegistry,
    FiscalScheduleHead, FiscalStagedTransition, SignedFiscalActivation, SignedFiscalProposal,
    SignedFiscalSchedule, VerifiedFiscalActivation, VerifiedFiscalApproval, VerifiedFiscalCharter,
    VerifiedFiscalContinuityAdvance, VerifiedFiscalContinuityCheckpoint, VerifiedFiscalProposal,
    VerifiedFiscalProposalAdmission, VerifiedFiscalRuntimeReadiness, VerifiedFiscalSchedule,
};
use rusqlite::{params, Connection, OptionalExtension, Transaction, TransactionBehavior};
use serde::Serialize;

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

const FISCAL_STORE_SCHEMA_KEY: &str = "fiscal";
pub(crate) const FISCAL_STORE_SUPPORTED_SCHEMA_VERSION: i32 = 6;
const FISCAL_STORE_SCHEMA: &str = include_str!("fiscal_store.sql");
const FISCAL_GENESIS_PROJECTION_KEY: &str = "authority";
const ZERO_DIGEST: &str = "0000000000000000000000000000000000000000000000000000000000000000";

#[derive(Debug, thiserror::Error)]
pub enum FiscalStoreError {
    #[error("fiscal store is unavailable: {0}")]
    Unavailable(String),
    #[error("fiscal store mutation was fenced")]
    Fenced,
    #[error("fiscal store record was not found")]
    NotFound,
    #[error("fiscal store record conflicts with retained state")]
    Conflict,
    #[error("fiscal store invariant failed: {0}")]
    Invariant(String),
    #[error("fiscal store durable outcome is unknown: {0}")]
    OutcomeUnknown(String),
    #[error(transparent)]
    Fiscal(#[from] chio_fiscal::FiscalError),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FiscalStageStatus {
    DbStaged,
    FiscalAnchorAdvanced,
    DbFinalized,
    Discarded,
}

impl FiscalStageStatus {
    const fn as_str(self) -> &'static str {
        match self {
            Self::DbStaged => "db_staged",
            Self::FiscalAnchorAdvanced => "fiscal_anchor_advanced",
            Self::DbFinalized => "db_finalized",
            Self::Discarded => "discarded",
        }
    }

    fn parse(value: &str) -> Result<Self, FiscalStoreError> {
        match value {
            "db_staged" => Ok(Self::DbStaged),
            "fiscal_anchor_advanced" => Ok(Self::FiscalAnchorAdvanced),
            "db_finalized" => Ok(Self::DbFinalized),
            "discarded" => Ok(Self::Discarded),
            _ => Err(invariant("fiscal stage status is invalid")),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FiscalStagedTransitionRecord {
    pub transition_id: String,
    pub current_checkpoint_digest: String,
    pub next_checkpoint_digest: String,
    pub proof_json: Vec<u8>,
    pub status: FiscalStageStatus,
    pub stage_version: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FiscalLegacyFeeScheduleBindingRecord {
    pub legacy_schedule_id: String,
    pub fiscal_schedule_id: String,
    pub fiscal_schedule_digest: String,
    pub legacy_envelope_digest: String,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct FiscalProjectionCommit<'a> {
    format: &'static str,
    projection_key: &'a str,
    projection_sequence: u64,
    mutation_kind: &'a str,
    snapshot_digest: &'a str,
    previous_commit_digest: &'a str,
    store_uuid: &'a str,
    store_lease_id: &'a str,
    store_owner_epoch: u64,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct FiscalGenesisSnapshot<'a> {
    policy: &'a FiscalGenesisPolicy,
    authority: &'a FiscalAuthorityState,
    charter_digest: &'a str,
    readiness_digest: &'a str,
    checkpoint_digest: &'a str,
}

struct PreparedFiscalActivationMutation {
    transition_id: String,
    activation_id: String,
    activation_digest: String,
    admission_id: String,
    admission_digest: String,
    expected_admission_version: u64,
    expected_admission_json: Vec<u8>,
    activated_admission_version: u64,
    activated_admission_json: Vec<u8>,
    candidate_schedule_id: String,
    candidate_schedule_digest: String,
    predecessor_schedule_id: Option<String>,
    predecessor_schedule_digest: Option<String>,
}

struct PreparedFiscalRotationScheduleMutation {
    domain: FiscalDomain,
    candidate_schedule_id: String,
    candidate_schedule_digest: String,
    predecessor_schedule_id: String,
    predecessor_schedule_digest: String,
}

struct PreparedFiscalRotationMutation {
    transition_id: String,
    activation_id: String,
    activation_digest: String,
    admission_id: String,
    admission_digest: String,
    expected_admission_version: u64,
    expected_admission_json: Vec<u8>,
    activated_admission_version: u64,
    activated_admission_json: Vec<u8>,
    successor_charter_id: String,
    successor_charter_digest: String,
    predecessor_charter_id: String,
    predecessor_charter_digest: String,
    schedules: Vec<PreparedFiscalRotationScheduleMutation>,
}

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

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

    fn connection(&self) -> Result<MutexGuard<'_, Connection>, FiscalStoreError> {
        self.connection
            .lock()
            .map_err(|_| invariant("fiscal store lock is poisoned"))
    }

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

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

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

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

    pub fn initialize_genesis(
        &self,
        policy: &FiscalGenesisPolicy,
        authority: &FiscalAuthorityState,
        charter: &VerifiedFiscalCharter,
        readiness: &VerifiedFiscalRuntimeReadiness,
        checkpoint: &VerifiedFiscalContinuityCheckpoint,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        policy.validate(charter)?;
        authority.validate()?;
        if checkpoint.body().continuity_sequence != 0
            || authority.finalized_checkpoint_digest != checkpoint.digest()
            || authority.genesis_policy_id != policy.policy_id
            || authority.genesis_policy_digest != policy.digest()?
            || authority.current_charter_id != charter.body().charter_id
            || authority.current_charter_digest != charter.digest()
            || checkpoint.body().runtime_readiness_digest != readiness.digest()
        {
            return Err(invariant("fiscal genesis bindings do not match"));
        }
        let policy_json = canonical_json_bytes(policy).map_err(canonical_error)?;
        let authority_json = canonical_json_bytes(authority).map_err(canonical_error)?;
        let charter_json = charter.canonical_bytes()?;
        let readiness_json = readiness.canonical_bytes()?;
        let registry_json = readiness.runtime_registry().canonical_bytes()?;
        let checkpoint_json = checkpoint.canonical_bytes()?;
        let snapshot_digest = canonical_digest(&FiscalGenesisSnapshot {
            policy,
            authority,
            charter_digest: charter.digest(),
            readiness_digest: readiness.digest(),
            checkpoint_digest: checkpoint.digest(),
        })?;

        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        if transaction
            .query_row(
                "SELECT state_json FROM fiscal_authority_state WHERE singleton = 1",
                [],
                |row| row.get::<_, Vec<u8>>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .is_some()
        {
            verify_exact_genesis(
                &transaction,
                &policy_json,
                &authority_json,
                &charter_json,
                &readiness_json,
                &registry_json,
                &checkpoint_json,
                policy,
                charter,
                readiness,
                checkpoint,
            )?;
            transaction.commit().map_err(sqlite_error)?;
            return Ok(());
        }

        transaction
            .execute(
                "INSERT INTO fiscal_genesis_policies (policy_id, policy_digest, policy_json) VALUES (?1, ?2, ?3)",
                params![&policy.policy_id, policy.digest()?, &policy_json],
            )
            .map_err(sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO fiscal_charters (charter_id, charter_digest, charter_sequence, lifecycle_state, signed_json) VALUES (?1, ?2, ?3, 'pinned', ?4)",
                params![
                    &charter.body().charter_id,
                    charter.digest(),
                    sqlite_i64(charter.body().sequence, "charter sequence")?,
                    &charter_json,
                ],
            )
            .map_err(sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO fiscal_runtime_readiness (readiness_id, readiness_digest, readiness_sequence, registry_json, signed_json) VALUES (?1, ?2, ?3, ?4, ?5)",
                params![
                    &readiness.body().readiness_id,
                    readiness.digest(),
                    sqlite_i64(readiness.body().readiness_sequence, "readiness sequence")?,
                    &registry_json,
                    &readiness_json,
                ],
            )
            .map_err(sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO fiscal_continuity_checkpoints (checkpoint_digest, continuity_sequence, status, signed_json) VALUES (?1, 0, 'finalized', ?2)",
                params![checkpoint.digest(), &checkpoint_json],
            )
            .map_err(sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO fiscal_authority_state (singleton, state_json, finalized_checkpoint_digest, state_version) VALUES (1, ?1, ?2, 1)",
                params![&authority_json, checkpoint.digest()],
            )
            .map_err(sqlite_error)?;
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            FISCAL_GENESIS_PROJECTION_KEY,
            1,
            "initialize_genesis",
            &snapshot_digest,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)
    }

    pub fn persist_proposal(
        &self,
        proposal: &VerifiedFiscalProposal,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let json = canonical_json_bytes(proposal.signed()).map_err(canonical_error)?;
        self.persist_immutable_artifact(
            "fiscal_proposals",
            "proposal_id",
            &proposal.body().proposal_id,
            "proposal_digest",
            proposal.digest(),
            &json,
            "persist_proposal",
            fence,
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub fn admit_proposal(
        &self,
        proposal: &VerifiedFiscalProposal,
        current_charter: &VerifiedFiscalCharter,
        admission_authority_id: &str,
        signer_key_epoch: u64,
        signer: &Keypair,
        admitted_at: u64,
        fence: &StoreMutationFence,
    ) -> Result<VerifiedFiscalProposalAdmission, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        let proposal_present = transaction
            .query_row(
                "SELECT proposal_digest = ?1 AND signed_json = ?2 FROM fiscal_proposals WHERE proposal_id = ?3",
                params![
                    proposal.digest(),
                    canonical_json_bytes(proposal.signed()).map_err(canonical_error)?,
                    &proposal.body().proposal_id,
                ],
                |row| row.get::<_, bool>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .unwrap_or(false);
        let charter_present = transaction
            .query_row(
                "SELECT charter_digest = ?1 AND signed_json = ?2 AND lifecycle_state IN ('pinned', 'active') FROM fiscal_charters WHERE charter_id = ?3",
                params![
                    current_charter.digest(),
                    current_charter.canonical_bytes()?,
                    &current_charter.body().charter_id,
                ],
                |row| row.get::<_, bool>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .unwrap_or(false);
        let proposal_already_admitted = transaction
            .query_row(
                "SELECT EXISTS(SELECT 1 FROM fiscal_proposal_admissions WHERE json_extract(state_json, '$.signedAdmission.body.proposalId') = ?1)",
                [&proposal.body().proposal_id],
                |row| row.get::<_, bool>(0),
            )
            .map_err(sqlite_error)?;
        if !proposal_present || !charter_present || proposal_already_admitted {
            return Err(FiscalStoreError::Conflict);
        }
        let current_sequence = transaction
            .query_row(
                "SELECT current_sequence FROM fiscal_admission_sequence WHERE singleton = 1",
                [],
                |row| row.get::<_, i64>(0),
            )
            .map_err(sqlite_error)?;
        let admission_sequence = u64::try_from(current_sequence)
            .map_err(|_| invariant("fiscal admission sequence is negative"))?
            .checked_add(1)
            .ok_or_else(|| invariant("fiscal admission sequence overflow"))?;
        let signed = FiscalProposalAdmissionBuilder {
            admission_sequence,
            admitted_at,
            admission_authority_id: admission_authority_id.to_owned(),
            signer_key_epoch,
        }
        .sign(proposal, current_charter, signer)?;
        let trust = FiscalAdmissionTrustRegistry::new(vec![FiscalAdmissionAuthority::new(
            current_charter.body().governing_operator_id.clone(),
            admission_authority_id.to_owned(),
            signer_key_epoch,
            signer.public_key(),
        )?])?;
        let admission = VerifiedFiscalProposalAdmission::verify(
            signed,
            proposal,
            current_charter,
            &trust,
            admitted_at,
        )?;
        let state = FiscalProposalAdmissionState::admitted(&admission);
        let state_json = canonical_json_bytes(&state).map_err(canonical_error)?;
        let sequence_updated = transaction
            .execute(
                "UPDATE fiscal_admission_sequence SET current_sequence = ?1 WHERE singleton = 1 AND current_sequence = ?2",
                params![
                    sqlite_i64(admission_sequence, "admission sequence")?,
                    current_sequence,
                ],
            )
            .map_err(sqlite_error)?;
        let admission_inserted = transaction
            .execute(
                "INSERT INTO fiscal_proposal_admissions (admission_id, admission_digest, admitted_at, status, state_version, state_json) VALUES (?1, ?2, ?3, 'admitted', 1, ?4)",
                params![
                    &admission.body().admission_id,
                    admission.digest(),
                    sqlite_i64(admitted_at, "admitted_at")?,
                    &state_json,
                ],
            )
            .map_err(sqlite_error)?;
        if sequence_updated != 1 || admission_inserted != 1 {
            return Err(FiscalStoreError::Conflict);
        }
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &format!("admission:{}", admission.body().admission_id),
            1,
            "admit_fiscal_proposal",
            &sha256_hex(&state_json),
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(admission)
    }

    pub fn persist_charter(
        &self,
        charter: &VerifiedFiscalCharter,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let json = charter.canonical_bytes()?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        if let Some((digest, stored)) = transaction
            .query_row(
                "SELECT charter_digest, signed_json FROM fiscal_charters WHERE charter_id = ?1",
                [&charter.body().charter_id],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?)),
            )
            .optional()
            .map_err(sqlite_error)?
        {
            if digest != charter.digest() || stored != json {
                return Err(FiscalStoreError::Conflict);
            }
            transaction.commit().map_err(sqlite_error)?;
            return Ok(());
        }
        transaction
            .execute(
                "INSERT INTO fiscal_charters (charter_id, charter_digest, charter_sequence, lifecycle_state, signed_json) VALUES (?1, ?2, ?3, 'proposed', ?4)",
                params![
                    &charter.body().charter_id,
                    charter.digest(),
                    sqlite_i64(charter.body().sequence, "charter sequence")?,
                    &json,
                ],
            )
            .map_err(sqlite_error)?;
        let projection_key = format!("charter:{}", charter.body().charter_id);
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            1,
            "persist_fiscal_charter",
            charter.digest(),
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)
    }

    pub fn persist_schedule(
        &self,
        schedule: &VerifiedFiscalSchedule,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let json = schedule.canonical_bytes()?;
        let digest = sha256_hex(&json);
        let domain_json = String::from_utf8(
            canonical_json_bytes(&schedule.body().domain).map_err(canonical_error)?,
        )
        .map_err(|error| invariant(format!("fiscal domain encoding is not UTF-8: {error}")))?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        if let Some((stored_digest, stored)) = transaction
            .query_row(
                "SELECT schedule_digest, signed_json FROM fiscal_schedules WHERE schedule_id = ?1",
                [&schedule.body().schedule_id],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?)),
            )
            .optional()
            .map_err(sqlite_error)?
        {
            if stored_digest != digest || stored != json {
                return Err(FiscalStoreError::Conflict);
            }
            transaction.commit().map_err(sqlite_error)?;
            return Ok(());
        }
        transaction
            .execute(
                "INSERT INTO fiscal_schedules (schedule_id, schedule_digest, domain_json, schedule_sequence, lifecycle_state, signed_json) VALUES (?1, ?2, ?3, ?4, 'staged', ?5)",
                params![
                    &schedule.body().schedule_id,
                    &digest,
                    &domain_json,
                    sqlite_i64(schedule.body().sequence, "schedule sequence")?,
                    &json,
                ],
            )
            .map_err(sqlite_error)?;
        let projection_key = format!("schedule:{}", schedule.body().schedule_id);
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            1,
            "persist_fiscal_schedule",
            &digest,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)
    }

    pub fn persist_runtime_readiness(
        &self,
        readiness: &VerifiedFiscalRuntimeReadiness,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let signed_json = readiness.canonical_bytes()?;
        let registry_json = readiness.runtime_registry().canonical_bytes()?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        if let Some((digest, stored_registry, stored_signed)) = transaction
            .query_row(
                "SELECT readiness_digest, registry_json, signed_json FROM fiscal_runtime_readiness WHERE readiness_id = ?1",
                [&readiness.body().readiness_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, Vec<u8>>(1)?,
                        row.get::<_, Vec<u8>>(2)?,
                    ))
                },
            )
            .optional()
            .map_err(sqlite_error)?
        {
            if digest != readiness.digest()
                || stored_registry != registry_json
                || stored_signed != signed_json
            {
                return Err(FiscalStoreError::Conflict);
            }
            transaction.commit().map_err(sqlite_error)?;
            return Ok(());
        }
        transaction
            .execute(
                "INSERT INTO fiscal_runtime_readiness (readiness_id, readiness_digest, readiness_sequence, registry_json, signed_json) VALUES (?1, ?2, ?3, ?4, ?5)",
                params![
                    &readiness.body().readiness_id,
                    readiness.digest(),
                    sqlite_i64(readiness.body().readiness_sequence, "readiness sequence")?,
                    &registry_json,
                    &signed_json,
                ],
            )
            .map_err(sqlite_error)?;
        let projection_key = format!("readiness:{}", readiness.body().readiness_id);
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            1,
            "persist_fiscal_readiness",
            readiness.digest(),
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)
    }

    pub fn load_genesis_policy(&self) -> Result<FiscalGenesisPolicy, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let (policy_id, policy_digest, policy_json) = transaction
            .query_row(
                "SELECT policy_id, policy_digest, policy_json FROM fiscal_genesis_policies",
                [],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, Vec<u8>>(2)?,
                    ))
                },
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or(FiscalStoreError::NotFound)?;
        let policy: FiscalGenesisPolicy =
            serde_json::from_slice(&policy_json).map_err(|error| {
                invariant(format!("stored fiscal genesis policy is invalid: {error}"))
            })?;
        if canonical_json_bytes(&policy).map_err(canonical_error)? != policy_json
            || policy.policy_id != policy_id
            || policy.digest()? != policy_digest
        {
            return Err(invariant("stored fiscal genesis policy binding is invalid"));
        }
        transaction.commit().map_err(sqlite_error)?;
        Ok(policy)
    }

    pub fn load_charter_registry(&self) -> Result<FiscalCharterRegistry, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let mut statement = transaction
            .prepare("SELECT signed_json FROM fiscal_charters ORDER BY charter_sequence")
            .map_err(sqlite_error)?;
        let signed = statement
            .query_map([], |row| row.get::<_, Vec<u8>>(0))
            .map_err(sqlite_error)?
            .map(|row| {
                let bytes = row.map_err(sqlite_error)?;
                Ok(VerifiedFiscalCharter::from_canonical_bytes(&bytes)?
                    .signed()
                    .clone())
            })
            .collect::<Result<Vec<_>, FiscalStoreError>>()?;
        drop(statement);
        transaction.commit().map_err(sqlite_error)?;
        Ok(FiscalCharterRegistry::new(signed)?)
    }

    pub fn load_runtime_readiness(
        &self,
        readiness_digest: &str,
        policy: &FiscalGenesisPolicy,
    ) -> Result<VerifiedFiscalRuntimeReadiness, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let (registry_json, signed_json) = transaction
            .query_row(
                "SELECT registry_json, signed_json FROM fiscal_runtime_readiness WHERE readiness_digest = ?1",
                [readiness_digest],
                |row| Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, Vec<u8>>(1)?)),
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or(FiscalStoreError::NotFound)?;
        let registry = FiscalRuntimeAdapterRegistry::from_canonical_bytes(&registry_json)?;
        let readiness =
            VerifiedFiscalRuntimeReadiness::from_canonical_bytes(&signed_json, policy, registry)?;
        if readiness.digest() != readiness_digest {
            return Err(invariant(
                "stored fiscal runtime readiness digest is invalid",
            ));
        }
        transaction.commit().map_err(sqlite_error)?;
        Ok(readiness)
    }

    pub fn load_verified_schedule(
        &self,
        schedule_id: &str,
        charters: &FiscalCharterRegistry,
    ) -> Result<VerifiedFiscalSchedule, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let mut lineage = Vec::new();
        let mut next_id = Some(schedule_id.to_owned());
        let mut seen = HashSet::new();
        while let Some(id) = next_id {
            if lineage.len() >= 4096 || !seen.insert(id.clone()) {
                return Err(invariant(
                    "stored fiscal schedule lineage is cyclic or too deep",
                ));
            }
            let bytes = transaction
                .query_row(
                    "SELECT signed_json FROM fiscal_schedules WHERE schedule_id = ?1",
                    [&id],
                    |row| row.get::<_, Vec<u8>>(0),
                )
                .optional()
                .map_err(sqlite_error)?
                .ok_or(FiscalStoreError::NotFound)?;
            let signed: chio_fiscal::SignedFiscalSchedule = serde_json::from_slice(&bytes)
                .map_err(|error| {
                    invariant(format!("stored fiscal schedule is invalid: {error}"))
                })?;
            if canonical_json_bytes(&signed).map_err(canonical_error)? != bytes
                || signed.body.schedule_id != id
            {
                return Err(invariant(
                    "stored fiscal schedule canonical binding is invalid",
                ));
            }
            next_id = signed.body.supersedes_schedule_id.clone();
            lineage.push(signed);
        }
        lineage.reverse();
        let mut verified: Option<VerifiedFiscalSchedule> = None;
        for signed in lineage {
            let charter = charters.resolve(&signed.body.charter_id, &signed.body.charter_digest)?;
            let next = match verified.as_ref() {
                Some(predecessor)
                    if predecessor.body().charter_digest != signed.body.charter_digest =>
                {
                    VerifiedFiscalSchedule::verify_rotation_replacement(
                        signed,
                        &charter,
                        predecessor,
                    )?
                }
                predecessor => VerifiedFiscalSchedule::verify(signed, &charter, predecessor)?,
            };
            verified = Some(next);
        }
        let verified = verified.ok_or(FiscalStoreError::NotFound)?;
        transaction.commit().map_err(sqlite_error)?;
        Ok(verified)
    }

    pub fn load_signed_schedules(&self) -> Result<Vec<SignedFiscalSchedule>, FiscalStoreError> {
        self.load_signed_artifacts(
            "SELECT signed_json FROM fiscal_schedules ORDER BY domain_json, schedule_sequence",
            "stored fiscal schedule",
        )
    }

    pub fn load_signed_proposals(&self) -> Result<Vec<SignedFiscalProposal>, FiscalStoreError> {
        self.load_signed_artifacts(
            "SELECT signed_json FROM fiscal_proposals ORDER BY proposal_id",
            "stored fiscal proposal",
        )
    }

    pub fn load_signed_activations(&self) -> Result<Vec<SignedFiscalActivation>, FiscalStoreError> {
        self.load_signed_artifacts(
            "SELECT activation.signed_json FROM fiscal_activations AS activation WHERE EXISTS(SELECT 1 FROM fiscal_staged_activation_mutations AS mutation JOIN fiscal_staged_transitions AS stage ON stage.transition_id = mutation.transition_id WHERE mutation.activation_id = activation.activation_id AND mutation.activation_digest = activation.activation_digest AND stage.status = 'db_finalized') OR EXISTS(SELECT 1 FROM fiscal_staged_rotation_mutations AS mutation JOIN fiscal_staged_transitions AS stage ON stage.transition_id = mutation.transition_id WHERE mutation.activation_id = activation.activation_id AND mutation.activation_digest = activation.activation_digest AND stage.status = 'db_finalized') ORDER BY activation.activation_id",
            "stored fiscal activation",
        )
    }

    fn load_signed_artifacts<T: serde::de::DeserializeOwned + Serialize>(
        &self,
        sql: &str,
        label: &str,
    ) -> Result<Vec<T>, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let mut statement = transaction.prepare(sql).map_err(sqlite_error)?;
        let artifacts = statement
            .query_map([], |row| row.get::<_, Vec<u8>>(0))
            .map_err(sqlite_error)?
            .map(|row| {
                let bytes = row.map_err(sqlite_error)?;
                let artifact: T = serde_json::from_slice(&bytes)
                    .map_err(|error| invariant(format!("{label} is invalid: {error}")))?;
                if canonical_json_bytes(&artifact).map_err(canonical_error)? != bytes {
                    return Err(invariant(format!("{label} is not canonical")));
                }
                Ok(artifact)
            })
            .collect::<Result<Vec<_>, FiscalStoreError>>()?;
        drop(statement);
        transaction.commit().map_err(sqlite_error)?;
        Ok(artifacts)
    }

    pub fn persist_approval(
        &self,
        approval: &VerifiedFiscalApproval,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let json = canonical_json_bytes(approval.signed()).map_err(canonical_error)?;
        self.persist_immutable_artifact(
            "fiscal_approvals",
            "approval_id",
            &approval.body().approval_id,
            "approval_digest",
            approval.digest(),
            &json,
            "persist_approval",
            fence,
        )
    }

    pub fn require_approval(
        &self,
        approval: &VerifiedFiscalApproval,
    ) -> Result<(), FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let exact = transaction
            .query_row(
                "SELECT approval_digest = ?1 AND signed_json = ?2 FROM fiscal_approvals WHERE approval_id = ?3",
                params![
                    approval.digest(),
                    canonical_json_bytes(approval.signed()).map_err(canonical_error)?,
                    &approval.body().approval_id,
                ],
                |row| row.get::<_, bool>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .unwrap_or(false);
        if !exact {
            return Err(FiscalStoreError::Conflict);
        }
        transaction.commit().map_err(sqlite_error)
    }

    pub fn load_admission_state(
        &self,
        admission_id: &str,
    ) -> Result<FiscalProposalAdmissionState, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let (digest, status, version, json) = transaction
            .query_row(
                "SELECT admission_digest, status, state_version, state_json FROM fiscal_proposal_admissions WHERE admission_id = ?1",
                [admission_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, i64>(2)?,
                        row.get::<_, Vec<u8>>(3)?,
                    ))
                },
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or(FiscalStoreError::NotFound)?;
        let state: FiscalProposalAdmissionState =
            serde_json::from_slice(&json).map_err(|error| {
                invariant(format!("stored fiscal admission state is invalid: {error}"))
            })?;
        let expected_status = match state.status {
            FiscalProposalAdmissionStatus::Admitted => "admitted",
            FiscalProposalAdmissionStatus::Activated => "activated",
        };
        if canonical_json_bytes(&state).map_err(canonical_error)? != json
            || state.signed_admission.body.admission_id != admission_id
            || state.admission_digest != digest
            || expected_status != status
            || sqlite_i64(state.version, "admission state version")? != version
        {
            return Err(invariant("stored fiscal admission binding is invalid"));
        }
        transaction.commit().map_err(sqlite_error)?;
        Ok(state)
    }

    pub fn persist_activation(
        &self,
        activation: &VerifiedFiscalActivation,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let json = canonical_json_bytes(activation.signed()).map_err(canonical_error)?;
        self.persist_immutable_artifact(
            "fiscal_activations",
            "activation_id",
            &activation.body().activation_id,
            "activation_digest",
            activation.digest(),
            &json,
            "persist_activation",
            fence,
        )
    }

    pub fn bind_legacy_fee_schedule(
        &self,
        legacy_schedule: &SignedOpenMarketFeeSchedule,
        schedule: &VerifiedFiscalSchedule,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        legacy_schedule
            .body
            .validate()
            .map_err(|error| invariant(format!("legacy fee schedule is invalid: {error}")))?;
        if !legacy_schedule
            .verify_signature()
            .map_err(|error| invariant(format!("legacy fee schedule signature failed: {error}")))?
        {
            return Err(invariant("legacy fee schedule signature is invalid"));
        }
        let FiscalParams::OpenMarketFeeAndBondSchedule { legacy_body } = &schedule.body().params
        else {
            return Err(FiscalStoreError::Conflict);
        };
        if legacy_body.as_ref() != &legacy_schedule.body {
            return Err(FiscalStoreError::Conflict);
        }
        let legacy_schedule_id = &legacy_schedule.body.fee_schedule_id;
        let legacy_envelope_json =
            canonical_json_bytes(legacy_schedule).map_err(canonical_error)?;
        let legacy_envelope_digest = sha256_hex(&legacy_envelope_json);
        let schedule_json = schedule.canonical_bytes()?;
        let schedule_digest = sha256_hex(&schedule_json);
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        let retained = transaction
            .query_row(
                "SELECT fiscal_schedule_id, fiscal_schedule_digest, legacy_envelope_digest FROM fiscal_legacy_fee_schedule_bindings WHERE legacy_schedule_id = ?1",
                [legacy_schedule_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                },
            )
            .optional()
            .map_err(sqlite_error)?;
        if let Some((schedule_id, digest, envelope_digest)) = retained {
            if schedule_id != schedule.body().schedule_id
                || digest != schedule_digest
                || envelope_digest != legacy_envelope_digest
            {
                return Err(FiscalStoreError::Conflict);
            }
            transaction.commit().map_err(sqlite_error)?;
            return Ok(());
        }
        let exact_schedule = transaction
            .query_row(
                "SELECT schedule_digest = ?1 AND signed_json = ?2 FROM fiscal_schedules WHERE schedule_id = ?3",
                params![&schedule_digest, &schedule_json, &schedule.body().schedule_id],
                |row| row.get::<_, bool>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .unwrap_or(false);
        if !exact_schedule {
            return Err(FiscalStoreError::Conflict);
        }
        transaction
            .execute(
                "INSERT INTO fiscal_legacy_fee_schedule_bindings (legacy_schedule_id, fiscal_schedule_id, fiscal_schedule_digest, legacy_envelope_digest) VALUES (?1, ?2, ?3, ?4)",
                params![
                    legacy_schedule_id,
                    &schedule.body().schedule_id,
                    &schedule_digest,
                    &legacy_envelope_digest,
                ],
            )
            .map_err(sqlite_error)?;
        let projection_key = format!("legacy-fee:{legacy_schedule_id}");
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            1,
            "bind_legacy_fee_schedule",
            &schedule_digest,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)
    }

    pub fn load_legacy_fee_schedule_binding(
        &self,
        fiscal_schedule_id: &str,
    ) -> Result<FiscalLegacyFeeScheduleBindingRecord, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let record = transaction
            .query_row(
                "SELECT legacy_schedule_id, fiscal_schedule_id, fiscal_schedule_digest, legacy_envelope_digest FROM fiscal_legacy_fee_schedule_bindings WHERE fiscal_schedule_id = ?1",
                [fiscal_schedule_id],
                |row| {
                    Ok(FiscalLegacyFeeScheduleBindingRecord {
                        legacy_schedule_id: row.get(0)?,
                        fiscal_schedule_id: row.get(1)?,
                        fiscal_schedule_digest: row.get(2)?,
                        legacy_envelope_digest: row.get(3)?,
                    })
                },
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or(FiscalStoreError::NotFound)?;
        transaction.commit().map_err(sqlite_error)?;
        Ok(record)
    }

    #[allow(clippy::too_many_arguments)]
    fn persist_immutable_artifact(
        &self,
        table: &str,
        id_column: &str,
        id: &str,
        digest_column: &str,
        digest: &str,
        json: &[u8],
        mutation_kind: &str,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        let sql =
            format!("SELECT {digest_column}, signed_json FROM {table} WHERE {id_column} = ?1");
        if let Some((stored_digest, stored_json)) = transaction
            .query_row(&sql, [id], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
            })
            .optional()
            .map_err(sqlite_error)?
        {
            if stored_digest != digest || stored_json != json {
                return Err(FiscalStoreError::Conflict);
            }
            transaction.commit().map_err(sqlite_error)?;
            return Ok(());
        }
        let insert = format!(
            "INSERT INTO {table} ({id_column}, {digest_column}, signed_json) VALUES (?1, ?2, ?3)"
        );
        transaction
            .execute(&insert, params![id, digest, json])
            .map_err(sqlite_error)?;
        let projection_key = format!("artifact:{id}");
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            1,
            mutation_kind,
            digest,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)
    }

    pub fn persist_admission_state(
        &self,
        state: &FiscalProposalAdmissionState,
        expected_version: Option<u64>,
        fence: &StoreMutationFence,
    ) -> Result<(), FiscalStoreError> {
        let state_json = canonical_json_bytes(state).map_err(canonical_error)?;
        let id = &state.signed_admission.body.admission_id;
        let status = match state.status {
            FiscalProposalAdmissionStatus::Admitted => "admitted",
            FiscalProposalAdmissionStatus::Activated => "activated",
        };
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        if expected_version.is_none() {
            let current_sequence = transaction
                .query_row(
                    "SELECT current_sequence FROM fiscal_admission_sequence WHERE singleton = 1",
                    [],
                    |row| row.get::<_, i64>(0),
                )
                .map_err(sqlite_error)?;
            let expected_sequence = u64::try_from(current_sequence)
                .map_err(|_| invariant("fiscal admission sequence is negative"))?
                .checked_add(1)
                .ok_or_else(|| invariant("fiscal admission sequence overflow"))?;
            if state.signed_admission.body.admission_sequence != expected_sequence
                || transaction
                    .execute(
                        "UPDATE fiscal_admission_sequence SET current_sequence = ?1 WHERE singleton = 1 AND current_sequence = ?2",
                        params![
                            sqlite_i64(expected_sequence, "admission sequence")?,
                            current_sequence,
                        ],
                    )
                    .map_err(sqlite_error)?
                    != 1
            {
                return Err(FiscalStoreError::Conflict);
            }
        }
        let changed = match expected_version {
            None => transaction.execute(
                "INSERT INTO fiscal_proposal_admissions (admission_id, admission_digest, admitted_at, status, state_version, state_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    id,
                    &state.admission_digest,
                    sqlite_i64(state.signed_admission.body.admitted_at, "admitted_at")?,
                    status,
                    sqlite_i64(state.version, "admission state version")?,
                    &state_json,
                ],
            ),
            Some(version) => transaction.execute(
                "UPDATE fiscal_proposal_admissions SET status = ?1, state_version = ?2, state_json = ?3 WHERE admission_id = ?4 AND admission_digest = ?5 AND state_version = ?6",
                params![
                    status,
                    sqlite_i64(state.version, "admission state version")?,
                    &state_json,
                    id,
                    &state.admission_digest,
                    sqlite_i64(version, "expected admission state version")?,
                ],
            ),
        }
        .map_err(sqlite_error)?;
        if changed != 1 {
            return Err(FiscalStoreError::Conflict);
        }
        let projection_key = format!("admission:{id}");
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            state.version,
            "persist_admission_state",
            &sha256_hex(&state_json),
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)
    }

    pub fn stage_advance(
        &self,
        advance: &VerifiedFiscalContinuityAdvance,
        next_authority: &FiscalAuthorityState,
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        self.stage_advance_inner(advance, next_authority, None, None, fence)
    }

    #[allow(clippy::too_many_arguments)]
    pub fn stage_activation_advance(
        &self,
        advance: &VerifiedFiscalContinuityAdvance,
        next_authority: &FiscalAuthorityState,
        activation: &VerifiedFiscalActivation,
        activated_admission: &FiscalProposalAdmissionState,
        candidate: &VerifiedFiscalSchedule,
        predecessor: Option<&VerifiedFiscalSchedule>,
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        let mutation = prepare_activation_mutation(
            advance,
            activation,
            activated_admission,
            candidate,
            predecessor,
        )?;
        self.stage_advance_inner(advance, next_authority, Some(&mutation), None, fence)
    }

    #[allow(clippy::too_many_arguments)]
    pub fn stage_charter_rotation_advance(
        &self,
        advance: &VerifiedFiscalContinuityAdvance,
        next_authority: &FiscalAuthorityState,
        activation: &VerifiedFiscalActivation,
        activated_admission: &FiscalProposalAdmissionState,
        successor_charter: &VerifiedFiscalCharter,
        successor_schedules: &[VerifiedFiscalSchedule],
        predecessor_charter: &VerifiedFiscalCharter,
        predecessor_schedules: &[VerifiedFiscalSchedule],
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        let mutation = prepare_rotation_mutation(
            advance,
            activation,
            activated_admission,
            successor_charter,
            successor_schedules,
            predecessor_charter,
            predecessor_schedules,
        )?;
        self.stage_advance_inner(advance, next_authority, None, Some(&mutation), fence)
    }

    fn stage_advance_inner(
        &self,
        advance: &VerifiedFiscalContinuityAdvance,
        next_authority: &FiscalAuthorityState,
        activation_mutation: Option<&PreparedFiscalActivationMutation>,
        rotation_mutation: Option<&PreparedFiscalRotationMutation>,
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        if activation_mutation.is_some() && rotation_mutation.is_some() {
            return Err(FiscalStoreError::Conflict);
        }
        next_authority.validate()?;
        if next_authority.finalized_checkpoint_digest != advance.next().digest() {
            return Err(invariant("staged authority does not match next checkpoint"));
        }
        let transition_id = advance
            .next()
            .body()
            .staged_transition
            .as_ref()
            .map(|transition| transition.transition_id.clone())
            .unwrap_or_else(|| sha256_hex(advance.canonical_proof_bytes()));
        let record = FiscalStagedTransitionRecord {
            transition_id,
            current_checkpoint_digest: advance.current().digest().to_owned(),
            next_checkpoint_digest: advance.next().digest().to_owned(),
            proof_json: advance.canonical_proof_bytes().to_vec(),
            status: FiscalStageStatus::DbStaged,
            stage_version: 1,
        };
        let checkpoint_json = advance.next().canonical_bytes()?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        match load_transition(&transaction, &record.transition_id) {
            Ok(existing) => {
                let stored_checkpoint = transaction
                    .query_row(
                        "SELECT signed_json FROM fiscal_continuity_checkpoints WHERE checkpoint_digest = ?1",
                        [&record.next_checkpoint_digest],
                        |row| row.get::<_, Vec<u8>>(0),
                    )
                    .optional()
                    .map_err(sqlite_error)?;
                if existing.current_checkpoint_digest != record.current_checkpoint_digest
                    || existing.next_checkpoint_digest != record.next_checkpoint_digest
                    || existing.proof_json != record.proof_json
                    || stored_checkpoint.as_deref() != Some(checkpoint_json.as_slice())
                {
                    return Err(FiscalStoreError::Conflict);
                }
                verify_exact_activation_mutation(
                    &transaction,
                    &record.transition_id,
                    activation_mutation,
                )?;
                verify_exact_rotation_mutation(
                    &transaction,
                    &record.transition_id,
                    rotation_mutation,
                )?;
                transaction.commit().map_err(sqlite_error)?;
                return Ok(existing);
            }
            Err(FiscalStoreError::NotFound) => {}
            Err(error) => return Err(error),
        }
        let current: String = transaction
            .query_row(
                "SELECT finalized_checkpoint_digest FROM fiscal_authority_state WHERE singleton = 1",
                [],
                |row| row.get(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or(FiscalStoreError::NotFound)?;
        if current != record.current_checkpoint_digest {
            return Err(FiscalStoreError::Conflict);
        }
        let readiness_present = transaction
            .query_row(
                "SELECT EXISTS(SELECT 1 FROM fiscal_runtime_readiness WHERE readiness_digest = ?1)",
                [&advance.next().body().runtime_readiness_digest],
                |row| row.get::<_, bool>(0),
            )
            .map_err(sqlite_error)?;
        let charter_present = transaction
            .query_row(
                "SELECT EXISTS(SELECT 1 FROM fiscal_charters WHERE charter_id = ?1 AND charter_digest = ?2)",
                params![
                    &advance.next().body().pinned_charter_id,
                    &advance.next().body().pinned_charter_digest,
                ],
                |row| row.get::<_, bool>(0),
            )
            .map_err(sqlite_error)?;
        if !readiness_present || !charter_present {
            return Err(invariant(
                "staged fiscal checkpoint references an unavailable charter or readiness record",
            ));
        }
        for head in advance
            .next()
            .body()
            .domains
            .iter()
            .flat_map(|state| [state.active.as_ref(), state.last_known_good.as_ref()])
            .flatten()
        {
            let present = transaction
                .query_row(
                    "SELECT EXISTS(SELECT 1 FROM fiscal_schedules WHERE schedule_id = ?1 AND schedule_digest = ?2 AND schedule_sequence = ?3)",
                    params![
                        &head.schedule_id,
                        &head.schedule_digest,
                        sqlite_i64(head.sequence, "schedule head sequence")?,
                    ],
                    |row| row.get::<_, bool>(0),
                )
                .map_err(sqlite_error)?;
            if !present {
                return Err(invariant(
                    "staged fiscal checkpoint references an unavailable schedule",
                ));
            }
        }
        transaction
            .execute(
                "INSERT INTO fiscal_continuity_checkpoints (checkpoint_digest, continuity_sequence, status, signed_json) VALUES (?1, ?2, 'staged', ?3)",
                params![
                    &record.next_checkpoint_digest,
                    sqlite_i64(advance.next().body().continuity_sequence, "continuity sequence")?,
                    &checkpoint_json,
                ],
            )
            .map_err(sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO fiscal_staged_transitions (transition_id, current_checkpoint_digest, next_checkpoint_digest, proof_json, status, stage_version) VALUES (?1, ?2, ?3, ?4, ?5, 1)",
                params![
                    &record.transition_id,
                    &record.current_checkpoint_digest,
                    &record.next_checkpoint_digest,
                    &record.proof_json,
                    record.status.as_str(),
                ],
            )
            .map_err(sqlite_error)?;
        if let Some(mutation) = activation_mutation {
            insert_activation_mutation(&transaction, mutation)?;
        }
        if let Some(mutation) = rotation_mutation {
            insert_rotation_mutation(&transaction, mutation)?;
        }
        let projection_key = format!("transition:{}", record.transition_id);
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            1,
            "stage_fiscal_advance",
            &sha256_hex(&record.proof_json),
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(record)
    }

    pub fn mark_anchor_advanced(
        &self,
        transition_id: &str,
        acknowledged: &VerifiedFiscalContinuityCheckpoint,
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        self.transition_status_update(
            transition_id,
            FiscalStageStatus::DbStaged,
            FiscalStageStatus::FiscalAnchorAdvanced,
            Some(acknowledged),
            fence,
        )
    }

    pub fn discard_unanchored_stage(
        &self,
        transition_id: &str,
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        self.transition_status_update(
            transition_id,
            FiscalStageStatus::DbStaged,
            FiscalStageStatus::Discarded,
            None,
            fence,
        )
    }

    pub fn finalize_advance(
        &self,
        transition_id: &str,
        acknowledged: &VerifiedFiscalContinuityCheckpoint,
        authority: &FiscalAuthorityState,
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        authority.validate()?;
        if authority.finalized_checkpoint_digest != acknowledged.digest() {
            return Err(invariant(
                "finalized authority does not match anchor acknowledgement",
            ));
        }
        let acknowledged_json = acknowledged.canonical_bytes()?;
        let authority_json = canonical_json_bytes(authority).map_err(canonical_error)?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        let mut record = load_transition(&transaction, transition_id)?;
        verify_acknowledgement(&transaction, &record, acknowledged, &acknowledged_json)?;
        if record.status == FiscalStageStatus::DbFinalized {
            let stored: Vec<u8> = transaction
                .query_row(
                    "SELECT state_json FROM fiscal_authority_state WHERE singleton = 1",
                    [],
                    |row| row.get(0),
                )
                .map_err(sqlite_error)?;
            if stored != authority_json {
                return Err(FiscalStoreError::Conflict);
            }
            transaction.commit().map_err(sqlite_error)?;
            return Ok(record);
        }
        if record.status != FiscalStageStatus::FiscalAnchorAdvanced {
            return Err(FiscalStoreError::Conflict);
        }
        let next_version = record
            .stage_version
            .checked_add(1)
            .ok_or_else(|| invariant("fiscal stage version overflowed"))?;
        let updated = transaction
            .execute(
                "UPDATE fiscal_staged_transitions SET status = 'db_finalized', stage_version = ?1 WHERE transition_id = ?2 AND status = 'fiscal_anchor_advanced' AND stage_version = ?3",
                params![
                    sqlite_i64(next_version, "fiscal stage version")?,
                    transition_id,
                    sqlite_i64(record.stage_version, "fiscal stage version")?,
                ],
            )
            .map_err(sqlite_error)?;
        let checkpoint_updated = transaction
            .execute(
                "UPDATE fiscal_continuity_checkpoints SET status = 'finalized' WHERE checkpoint_digest = ?1 AND status = 'staged' AND signed_json = ?2",
                params![acknowledged.digest(), &acknowledged_json],
            )
            .map_err(sqlite_error)?;
        let authority_updated = transaction
            .execute(
                "UPDATE fiscal_authority_state SET state_json = ?1, finalized_checkpoint_digest = ?2, state_version = state_version + 1 WHERE singleton = 1 AND finalized_checkpoint_digest = ?3 AND state_version < 9223372036854775807",
                params![
                    &authority_json,
                    acknowledged.digest(),
                    &record.current_checkpoint_digest,
                ],
            )
            .map_err(sqlite_error)?;
        if updated != 1 || checkpoint_updated != 1 || authority_updated != 1 {
            return Err(FiscalStoreError::Conflict);
        }
        finalize_activation_mutation(&transaction, &self.serving_owner, transition_id)?;
        finalize_rotation_mutation(&transaction, &self.serving_owner, transition_id)?;
        record.status = FiscalStageStatus::DbFinalized;
        record.stage_version = next_version;
        let projection_key = format!("transition:{transition_id}");
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            next_version,
            "finalize_fiscal_advance",
            acknowledged.digest(),
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(record)
    }

    pub fn load_authority_state(&self) -> Result<FiscalAuthorityState, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let json = transaction
            .query_row(
                "SELECT state_json FROM fiscal_authority_state WHERE singleton = 1",
                [],
                |row| row.get::<_, Vec<u8>>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or(FiscalStoreError::NotFound)?;
        let state: FiscalAuthorityState = serde_json::from_slice(&json)
            .map_err(|error| invariant(format!("stored fiscal authority is invalid: {error}")))?;
        state.validate()?;
        if canonical_json_bytes(&state).map_err(canonical_error)? != json {
            return Err(invariant("stored fiscal authority is not canonical"));
        }
        transaction.commit().map_err(sqlite_error)?;
        Ok(state)
    }

    pub fn load_transition(
        &self,
        transition_id: &str,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let record = load_transition(&transaction, transition_id)?;
        transaction.commit().map_err(sqlite_error)?;
        Ok(record)
    }

    pub fn load_open_transition(
        &self,
    ) -> Result<Option<FiscalStagedTransitionRecord>, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let transition_id = transaction
            .query_row(
                "SELECT transition_id FROM fiscal_staged_transitions WHERE status IN ('db_staged', 'fiscal_anchor_advanced')",
                [],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(sqlite_error)?;
        let record = transition_id
            .as_deref()
            .map(|id| load_transition(&transaction, id))
            .transpose()?;
        transaction.commit().map_err(sqlite_error)?;
        Ok(record)
    }

    pub fn load_checkpoint(
        &self,
        checkpoint_digest: &str,
        policy: &FiscalGenesisPolicy,
        charters: &FiscalCharterRegistry,
    ) -> Result<VerifiedFiscalContinuityCheckpoint, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let json = transaction
            .query_row(
                "SELECT signed_json FROM fiscal_continuity_checkpoints WHERE checkpoint_digest = ?1",
                [checkpoint_digest],
                |row| row.get::<_, Vec<u8>>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or(FiscalStoreError::NotFound)?;
        let checkpoint =
            VerifiedFiscalContinuityCheckpoint::from_canonical_bytes(&json, policy, charters)?;
        if checkpoint.digest() != checkpoint_digest {
            return Err(invariant("stored fiscal checkpoint digest is inconsistent"));
        }
        transaction.commit().map_err(sqlite_error)?;
        Ok(checkpoint)
    }

    pub fn load_finalized_checkpoints(
        &self,
        policy: &FiscalGenesisPolicy,
        charters: &FiscalCharterRegistry,
    ) -> Result<Vec<VerifiedFiscalContinuityCheckpoint>, FiscalStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let mut statement = transaction
            .prepare(
                "SELECT signed_json FROM fiscal_continuity_checkpoints WHERE status = 'finalized' ORDER BY continuity_sequence",
            )
            .map_err(sqlite_error)?;
        let checkpoints = statement
            .query_map([], |row| row.get::<_, Vec<u8>>(0))
            .map_err(sqlite_error)?
            .map(|row| {
                let bytes = row.map_err(sqlite_error)?;
                VerifiedFiscalContinuityCheckpoint::from_canonical_bytes(&bytes, policy, charters)
                    .map_err(FiscalStoreError::from)
            })
            .collect::<Result<Vec<_>, FiscalStoreError>>()?;
        drop(statement);
        transaction.commit().map_err(sqlite_error)?;
        Ok(checkpoints)
    }

    fn transition_status_update(
        &self,
        transition_id: &str,
        expected: FiscalStageStatus,
        next: FiscalStageStatus,
        acknowledged: Option<&VerifiedFiscalContinuityCheckpoint>,
        fence: &StoreMutationFence,
    ) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
        let acknowledged_json = acknowledged
            .map(VerifiedFiscalContinuityCheckpoint::canonical_bytes)
            .transpose()?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, fence)?;
        let mut record = load_transition(&transaction, transition_id)?;
        if let (Some(checkpoint), Some(json)) = (acknowledged, acknowledged_json.as_deref()) {
            verify_acknowledgement(&transaction, &record, checkpoint, json)?;
        }
        if record.status == next {
            transaction.commit().map_err(sqlite_error)?;
            return Ok(record);
        }
        if record.status != expected {
            return Err(FiscalStoreError::Conflict);
        }
        let next_version = record
            .stage_version
            .checked_add(1)
            .ok_or_else(|| invariant("fiscal stage version overflowed"))?;
        let updated = transaction
            .execute(
                "UPDATE fiscal_staged_transitions SET status = ?1, stage_version = ?2 WHERE transition_id = ?3 AND status = ?4 AND stage_version = ?5",
                params![
                    next.as_str(),
                    sqlite_i64(next_version, "fiscal stage version")?,
                    transition_id,
                    expected.as_str(),
                    sqlite_i64(record.stage_version, "fiscal stage version")?,
                ],
            )
            .map_err(sqlite_error)?;
        if updated != 1 {
            return Err(FiscalStoreError::Conflict);
        }
        record.status = next;
        record.stage_version = next_version;
        let projection_key = format!("transition:{transition_id}");
        append_projection_commit(
            &transaction,
            &self.serving_owner,
            &projection_key,
            next_version,
            "advance_fiscal_stage",
            &record.next_checkpoint_digest,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(record)
    }
}

fn prepare_activation_mutation(
    advance: &VerifiedFiscalContinuityAdvance,
    activation: &VerifiedFiscalActivation,
    activated_admission: &FiscalProposalAdmissionState,
    candidate: &VerifiedFiscalSchedule,
    predecessor: Option<&VerifiedFiscalSchedule>,
) -> Result<PreparedFiscalActivationMutation, FiscalStoreError> {
    let FiscalActivationTarget::Schedule {
        schedule_id,
        supersedes_schedule_id,
    } = &activation.body().target
    else {
        return Err(FiscalStoreError::Conflict);
    };
    let transition = FiscalStagedTransition::new(
        activation.body().activation_id.clone(),
        activation.digest().to_owned(),
    )?;
    let candidate_head = FiscalScheduleHead::from_signed(candidate.signed())?;
    let predecessor_head = predecessor
        .map(|schedule| FiscalScheduleHead::from_signed(schedule.signed()))
        .transpose()?;
    let next_domain = advance
        .next()
        .body()
        .domains
        .iter()
        .find(|state| state.domain == candidate.body().domain)
        .ok_or(FiscalStoreError::Conflict)?;
    let expected_admission_version = activated_admission
        .version
        .checked_sub(1)
        .ok_or(FiscalStoreError::Conflict)?;
    if advance.next().body().staged_transition.as_ref() != Some(&transition)
        || schedule_id != &candidate.body().schedule_id
        || next_domain.active.as_ref() != Some(&candidate_head)
        || next_domain.last_known_good.as_ref() != Some(&candidate_head)
        || activated_admission.status != FiscalProposalAdmissionStatus::Activated
        || activated_admission.signed_admission.body.admission_id != activation.body().admission_id
        || activated_admission.admission_digest != activation.body().admission_digest
        || activated_admission.activation_digest.as_deref() != Some(activation.digest())
        || activated_admission.activated_sequence != Some(candidate.body().sequence)
        || expected_admission_version == 0
        || supersedes_schedule_id.as_deref()
            != predecessor.map(|schedule| schedule.body().schedule_id.as_str())
        || candidate.body().supersedes_schedule_id.as_deref()
            != predecessor.map(|schedule| schedule.body().schedule_id.as_str())
    {
        return Err(FiscalStoreError::Conflict);
    }
    let expected_admission = FiscalProposalAdmissionState {
        signed_admission: activated_admission.signed_admission.clone(),
        admission_digest: activated_admission.admission_digest.clone(),
        version: expected_admission_version,
        status: FiscalProposalAdmissionStatus::Admitted,
        activation_digest: None,
        activated_sequence: None,
    };
    Ok(PreparedFiscalActivationMutation {
        transition_id: transition.transition_id,
        activation_id: activation.body().activation_id.clone(),
        activation_digest: activation.digest().to_owned(),
        admission_id: activated_admission
            .signed_admission
            .body
            .admission_id
            .clone(),
        admission_digest: activated_admission.admission_digest.clone(),
        expected_admission_version,
        expected_admission_json: canonical_json_bytes(&expected_admission)
            .map_err(canonical_error)?,
        activated_admission_version: activated_admission.version,
        activated_admission_json: canonical_json_bytes(activated_admission)
            .map_err(canonical_error)?,
        candidate_schedule_id: candidate.body().schedule_id.clone(),
        candidate_schedule_digest: candidate_head.schedule_digest,
        predecessor_schedule_id: predecessor.map(|schedule| schedule.body().schedule_id.clone()),
        predecessor_schedule_digest: predecessor_head.map(|head| head.schedule_digest),
    })
}

#[allow(clippy::too_many_arguments)]
fn prepare_rotation_mutation(
    advance: &VerifiedFiscalContinuityAdvance,
    activation: &VerifiedFiscalActivation,
    activated_admission: &FiscalProposalAdmissionState,
    successor_charter: &VerifiedFiscalCharter,
    successor_schedules: &[VerifiedFiscalSchedule],
    predecessor_charter: &VerifiedFiscalCharter,
    predecessor_schedules: &[VerifiedFiscalSchedule],
) -> Result<PreparedFiscalRotationMutation, FiscalStoreError> {
    let FiscalActivationTarget::CharterRotation {
        successor_charter_digest,
        predecessor_charter_digest,
        successor_schedules: signed_successors,
    } = &activation.body().target
    else {
        return Err(FiscalStoreError::Conflict);
    };
    let transition = FiscalStagedTransition::new(
        activation.body().activation_id.clone(),
        activation.digest().to_owned(),
    )?;
    let expected_admission_version = activated_admission
        .version
        .checked_sub(1)
        .ok_or(FiscalStoreError::Conflict)?;
    let activated_sequence = successor_schedules
        .iter()
        .map(|schedule| schedule.body().sequence)
        .max()
        .ok_or(FiscalStoreError::Conflict)?;
    if advance.next().body().staged_transition.as_ref() != Some(&transition)
        || advance.next().body().pinned_charter_id != successor_charter.body().charter_id
        || advance.next().body().pinned_charter_digest != successor_charter.digest()
        || advance.next().body().pinned_charter_sequence != successor_charter.body().sequence
        || successor_charter_digest != successor_charter.digest()
        || predecessor_charter_digest != predecessor_charter.digest()
        || successor_charter
            .body()
            .predecessor_charter_digest
            .as_deref()
            != Some(predecessor_charter.digest())
        || successor_schedules.len() != predecessor_schedules.len()
        || signed_successors.len() != successor_schedules.len()
        || signed_successors
            .iter()
            .zip(successor_schedules)
            .any(|(signed, verified)| signed != verified.signed())
        || activated_admission.status != FiscalProposalAdmissionStatus::Activated
        || activated_admission.signed_admission.body.admission_id != activation.body().admission_id
        || activated_admission.admission_digest != activation.body().admission_digest
        || activated_admission.activation_digest.as_deref() != Some(activation.digest())
        || activated_admission.activated_sequence != Some(activated_sequence)
        || expected_admission_version == 0
    {
        return Err(FiscalStoreError::Conflict);
    }
    let mut schedules = Vec::with_capacity(successor_schedules.len());
    for (candidate, predecessor) in successor_schedules.iter().zip(predecessor_schedules) {
        let candidate_head = FiscalScheduleHead::from_signed(candidate.signed())?;
        let predecessor_head = FiscalScheduleHead::from_signed(predecessor.signed())?;
        let next_domain = advance
            .next()
            .body()
            .domains
            .iter()
            .find(|state| state.domain == candidate.body().domain)
            .ok_or(FiscalStoreError::Conflict)?;
        if candidate.body().domain != predecessor.body().domain
            || candidate.body().supersedes_schedule_id.as_deref()
                != Some(predecessor.body().schedule_id.as_str())
            || next_domain.active.as_ref() != Some(&candidate_head)
            || next_domain.last_known_good.as_ref() != Some(&candidate_head)
        {
            return Err(FiscalStoreError::Conflict);
        }
        schedules.push(PreparedFiscalRotationScheduleMutation {
            domain: candidate.body().domain,
            candidate_schedule_id: candidate.body().schedule_id.clone(),
            candidate_schedule_digest: candidate_head.schedule_digest,
            predecessor_schedule_id: predecessor.body().schedule_id.clone(),
            predecessor_schedule_digest: predecessor_head.schedule_digest,
        });
    }
    if schedules
        .windows(2)
        .any(|pair| pair[0].domain >= pair[1].domain)
    {
        return Err(FiscalStoreError::Conflict);
    }
    let expected_admission = FiscalProposalAdmissionState {
        signed_admission: activated_admission.signed_admission.clone(),
        admission_digest: activated_admission.admission_digest.clone(),
        version: expected_admission_version,
        status: FiscalProposalAdmissionStatus::Admitted,
        activation_digest: None,
        activated_sequence: None,
    };
    Ok(PreparedFiscalRotationMutation {
        transition_id: transition.transition_id,
        activation_id: activation.body().activation_id.clone(),
        activation_digest: activation.digest().to_owned(),
        admission_id: activated_admission
            .signed_admission
            .body
            .admission_id
            .clone(),
        admission_digest: activated_admission.admission_digest.clone(),
        expected_admission_version,
        expected_admission_json: canonical_json_bytes(&expected_admission)
            .map_err(canonical_error)?,
        activated_admission_version: activated_admission.version,
        activated_admission_json: canonical_json_bytes(activated_admission)
            .map_err(canonical_error)?,
        successor_charter_id: successor_charter.body().charter_id.clone(),
        successor_charter_digest: successor_charter.digest().to_owned(),
        predecessor_charter_id: predecessor_charter.body().charter_id.clone(),
        predecessor_charter_digest: predecessor_charter.digest().to_owned(),
        schedules,
    })
}

fn insert_activation_mutation(
    transaction: &Transaction<'_>,
    mutation: &PreparedFiscalActivationMutation,
) -> Result<(), FiscalStoreError> {
    let activation_exact = transaction
        .query_row(
            "SELECT activation_digest = ?1 FROM fiscal_activations WHERE activation_id = ?2",
            params![&mutation.activation_digest, &mutation.activation_id],
            |row| row.get::<_, bool>(0),
        )
        .optional()
        .map_err(sqlite_error)?
        .unwrap_or(false);
    let admission_exact = transaction
        .query_row(
            "SELECT admission_digest = ?1 AND state_version = ?2 AND status = 'admitted' AND state_json = ?3 FROM fiscal_proposal_admissions WHERE admission_id = ?4",
            params![
                &mutation.admission_digest,
                sqlite_i64(mutation.expected_admission_version, "expected admission version")?,
                &mutation.expected_admission_json,
                &mutation.admission_id,
            ],
            |row| row.get::<_, bool>(0),
        )
        .optional()
        .map_err(sqlite_error)?
        .unwrap_or(false);
    let candidate_exact = schedule_has_state(
        transaction,
        &mutation.candidate_schedule_id,
        &mutation.candidate_schedule_digest,
        "staged",
    )?;
    let predecessor_exact = match (
        &mutation.predecessor_schedule_id,
        &mutation.predecessor_schedule_digest,
    ) {
        (Some(id), Some(digest)) => schedule_has_state(transaction, id, digest, "active")?,
        (None, None) => true,
        _ => false,
    };
    if !activation_exact || !admission_exact || !candidate_exact || !predecessor_exact {
        return Err(FiscalStoreError::Conflict);
    }
    transaction
        .execute(
            "INSERT INTO fiscal_staged_activation_mutations (transition_id, activation_id, activation_digest, admission_id, admission_digest, expected_admission_version, activated_admission_version, activated_admission_json, candidate_schedule_id, candidate_schedule_digest, predecessor_schedule_id, predecessor_schedule_digest) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
            params![
                &mutation.transition_id,
                &mutation.activation_id,
                &mutation.activation_digest,
                &mutation.admission_id,
                &mutation.admission_digest,
                sqlite_i64(mutation.expected_admission_version, "expected admission version")?,
                sqlite_i64(mutation.activated_admission_version, "activated admission version")?,
                &mutation.activated_admission_json,
                &mutation.candidate_schedule_id,
                &mutation.candidate_schedule_digest,
                &mutation.predecessor_schedule_id,
                &mutation.predecessor_schedule_digest,
            ],
        )
        .map_err(|error| invariant(format!("staged activation mutation insert failed: {error}")))?;
    Ok(())
}

fn verify_exact_activation_mutation(
    transaction: &Transaction<'_>,
    transition_id: &str,
    expected: Option<&PreparedFiscalActivationMutation>,
) -> Result<(), FiscalStoreError> {
    let retained = transaction
        .query_row(
            "SELECT activation_id, activation_digest, admission_id, admission_digest, expected_admission_version, activated_admission_version, activated_admission_json, candidate_schedule_id, candidate_schedule_digest, predecessor_schedule_id, predecessor_schedule_digest FROM fiscal_staged_activation_mutations WHERE transition_id = ?1",
            [transition_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, i64>(4)?,
                    row.get::<_, i64>(5)?,
                    row.get::<_, Vec<u8>>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, String>(8)?,
                    row.get::<_, Option<String>>(9)?,
                    row.get::<_, Option<String>>(10)?,
                ))
            },
        )
        .optional()
        .map_err(sqlite_error)?;
    match (retained, expected) {
        (None, None) => Ok(()),
        (Some(row), Some(expected))
            if row.0 == expected.activation_id
                && row.1 == expected.activation_digest
                && row.2 == expected.admission_id
                && row.3 == expected.admission_digest
                && read_u64(row.4, "expected admission version")?
                    == expected.expected_admission_version
                && read_u64(row.5, "activated admission version")?
                    == expected.activated_admission_version
                && row.6 == expected.activated_admission_json
                && row.7 == expected.candidate_schedule_id
                && row.8 == expected.candidate_schedule_digest
                && row.9 == expected.predecessor_schedule_id
                && row.10 == expected.predecessor_schedule_digest =>
        {
            Ok(())
        }
        _ => Err(FiscalStoreError::Conflict),
    }
}

fn insert_rotation_mutation(
    transaction: &Transaction<'_>,
    mutation: &PreparedFiscalRotationMutation,
) -> Result<(), FiscalStoreError> {
    let activation_exact = transaction
        .query_row(
            "SELECT activation_digest = ?1 FROM fiscal_activations WHERE activation_id = ?2",
            params![&mutation.activation_digest, &mutation.activation_id],
            |row| row.get::<_, bool>(0),
        )
        .optional()
        .map_err(sqlite_error)?
        .unwrap_or(false);
    let admission_exact = transaction
        .query_row(
            "SELECT admission_digest = ?1 AND state_version = ?2 AND status = 'admitted' AND state_json = ?3 FROM fiscal_proposal_admissions WHERE admission_id = ?4",
            params![
                &mutation.admission_digest,
                sqlite_i64(mutation.expected_admission_version, "expected admission version")?,
                &mutation.expected_admission_json,
                &mutation.admission_id,
            ],
            |row| row.get::<_, bool>(0),
        )
        .optional()
        .map_err(sqlite_error)?
        .unwrap_or(false);
    let successor_exact = charter_has_state(
        transaction,
        &mutation.successor_charter_id,
        &mutation.successor_charter_digest,
        "proposed",
    )?;
    let predecessor_exact = charter_has_current_state(
        transaction,
        &mutation.predecessor_charter_id,
        &mutation.predecessor_charter_digest,
    )?;
    if !activation_exact || !admission_exact || !successor_exact || !predecessor_exact {
        return Err(FiscalStoreError::Conflict);
    }
    for schedule in &mutation.schedules {
        if !schedule_has_state(
            transaction,
            &schedule.candidate_schedule_id,
            &schedule.candidate_schedule_digest,
            "staged",
        )? || !schedule_has_state(
            transaction,
            &schedule.predecessor_schedule_id,
            &schedule.predecessor_schedule_digest,
            "active",
        )? {
            return Err(FiscalStoreError::Conflict);
        }
    }
    transaction
        .execute(
            "INSERT INTO fiscal_staged_rotation_mutations (transition_id, activation_id, activation_digest, admission_id, admission_digest, expected_admission_version, activated_admission_version, activated_admission_json, successor_charter_id, successor_charter_digest, predecessor_charter_id, predecessor_charter_digest) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
            params![
                &mutation.transition_id,
                &mutation.activation_id,
                &mutation.activation_digest,
                &mutation.admission_id,
                &mutation.admission_digest,
                sqlite_i64(mutation.expected_admission_version, "expected admission version")?,
                sqlite_i64(mutation.activated_admission_version, "activated admission version")?,
                &mutation.activated_admission_json,
                &mutation.successor_charter_id,
                &mutation.successor_charter_digest,
                &mutation.predecessor_charter_id,
                &mutation.predecessor_charter_digest,
            ],
        )
        .map_err(sqlite_error)?;
    for schedule in &mutation.schedules {
        let domain_json =
            String::from_utf8(canonical_json_bytes(&schedule.domain).map_err(canonical_error)?)
                .map_err(|error| {
                    invariant(format!("fiscal domain encoding is not UTF-8: {error}"))
                })?;
        transaction
            .execute(
                "INSERT INTO fiscal_staged_rotation_schedules (transition_id, domain_json, candidate_schedule_id, candidate_schedule_digest, predecessor_schedule_id, predecessor_schedule_digest) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    &mutation.transition_id,
                    domain_json,
                    &schedule.candidate_schedule_id,
                    &schedule.candidate_schedule_digest,
                    &schedule.predecessor_schedule_id,
                    &schedule.predecessor_schedule_digest,
                ],
            )
            .map_err(sqlite_error)?;
    }
    Ok(())
}

fn verify_exact_rotation_mutation(
    transaction: &Transaction<'_>,
    transition_id: &str,
    expected: Option<&PreparedFiscalRotationMutation>,
) -> Result<(), FiscalStoreError> {
    let retained = transaction
        .query_row(
            "SELECT activation_id, activation_digest, admission_id, admission_digest, expected_admission_version, activated_admission_version, activated_admission_json, successor_charter_id, successor_charter_digest, predecessor_charter_id, predecessor_charter_digest FROM fiscal_staged_rotation_mutations WHERE transition_id = ?1",
            [transition_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, i64>(4)?,
                    row.get::<_, i64>(5)?,
                    row.get::<_, Vec<u8>>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, String>(8)?,
                    row.get::<_, String>(9)?,
                    row.get::<_, String>(10)?,
                ))
            },
        )
        .optional()
        .map_err(sqlite_error)?;
    match (retained, expected) {
        (None, None) => Ok(()),
        (Some(row), Some(expected))
            if row.0 == expected.activation_id
                && row.1 == expected.activation_digest
                && row.2 == expected.admission_id
                && row.3 == expected.admission_digest
                && read_u64(row.4, "expected admission version")?
                    == expected.expected_admission_version
                && read_u64(row.5, "activated admission version")?
                    == expected.activated_admission_version
                && row.6 == expected.activated_admission_json
                && row.7 == expected.successor_charter_id
                && row.8 == expected.successor_charter_digest
                && row.9 == expected.predecessor_charter_id
                && row.10 == expected.predecessor_charter_digest =>
        {
            let mut statement = transaction
                .prepare(
                    "SELECT domain_json, candidate_schedule_id, candidate_schedule_digest, predecessor_schedule_id, predecessor_schedule_digest FROM fiscal_staged_rotation_schedules WHERE transition_id = ?1 ORDER BY rowid",
                )
                .map_err(sqlite_error)?;
            let retained_schedules = statement
                .query_map([transition_id], |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                    ))
                })
                .map_err(sqlite_error)?
                .collect::<Result<Vec<_>, _>>()
                .map_err(sqlite_error)?;
            let expected_schedules = expected
                .schedules
                .iter()
                .map(|schedule| {
                    let domain = canonical_json_bytes(&schedule.domain).map_err(canonical_error)?;
                    let domain = String::from_utf8(domain).map_err(|error| {
                        invariant(format!("fiscal domain encoding is not UTF-8: {error}"))
                    })?;
                    Ok((
                        domain,
                        schedule.candidate_schedule_id.clone(),
                        schedule.candidate_schedule_digest.clone(),
                        schedule.predecessor_schedule_id.clone(),
                        schedule.predecessor_schedule_digest.clone(),
                    ))
                })
                .collect::<Result<Vec<_>, FiscalStoreError>>()?;
            if retained_schedules == expected_schedules {
                Ok(())
            } else {
                Err(FiscalStoreError::Conflict)
            }
        }
        _ => Err(FiscalStoreError::Conflict),
    }
}

fn charter_has_state(
    transaction: &Transaction<'_>,
    charter_id: &str,
    charter_digest: &str,
    state: &str,
) -> Result<bool, FiscalStoreError> {
    transaction
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM fiscal_charters WHERE charter_id = ?1 AND charter_digest = ?2 AND lifecycle_state = ?3)",
            params![charter_id, charter_digest, state],
            |row| row.get(0),
        )
        .map_err(sqlite_error)
}

fn charter_has_current_state(
    transaction: &Transaction<'_>,
    charter_id: &str,
    charter_digest: &str,
) -> Result<bool, FiscalStoreError> {
    transaction
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM fiscal_charters WHERE charter_id = ?1 AND charter_digest = ?2 AND lifecycle_state IN ('pinned', 'active'))",
            params![charter_id, charter_digest],
            |row| row.get(0),
        )
        .map_err(sqlite_error)
}

fn schedule_has_state(
    transaction: &Transaction<'_>,
    schedule_id: &str,
    schedule_digest: &str,
    state: &str,
) -> Result<bool, FiscalStoreError> {
    transaction
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM fiscal_schedules WHERE schedule_id = ?1 AND schedule_digest = ?2 AND lifecycle_state = ?3)",
            params![schedule_id, schedule_digest, state],
            |row| row.get(0),
        )
        .map_err(sqlite_error)
}

fn finalize_activation_mutation(
    transaction: &Transaction<'_>,
    owner: &SqliteServingOwner,
    transition_id: &str,
) -> Result<(), FiscalStoreError> {
    let mutation = transaction
        .query_row(
            "SELECT admission_id, admission_digest, expected_admission_version, activated_admission_version, activated_admission_json, candidate_schedule_id, candidate_schedule_digest, predecessor_schedule_id, predecessor_schedule_digest FROM fiscal_staged_activation_mutations WHERE transition_id = ?1",
            [transition_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, i64>(3)?,
                    row.get::<_, Vec<u8>>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, Option<String>>(7)?,
                    row.get::<_, Option<String>>(8)?,
                ))
            },
        )
        .optional()
        .map_err(sqlite_error)?;
    let Some(mutation) = mutation else {
        return Ok(());
    };
    let expected_version = read_u64(mutation.2, "expected admission version")?;
    let activated_version = read_u64(mutation.3, "activated admission version")?;
    let admission_updated = transaction
        .execute(
            "UPDATE fiscal_proposal_admissions SET status = 'activated', state_version = ?1, state_json = ?2 WHERE admission_id = ?3 AND admission_digest = ?4 AND state_version = ?5 AND status = 'admitted'",
            params![
                sqlite_i64(activated_version, "activated admission version")?,
                &mutation.4,
                &mutation.0,
                &mutation.1,
                sqlite_i64(expected_version, "expected admission version")?,
            ],
        )
        .map_err(sqlite_error)?;
    let candidate_updated = transaction
        .execute(
            "UPDATE fiscal_schedules SET lifecycle_state = 'active' WHERE schedule_id = ?1 AND schedule_digest = ?2 AND lifecycle_state = 'staged'",
            params![&mutation.5, &mutation.6],
        )
        .map_err(sqlite_error)?;
    let predecessor_updated = match (&mutation.7, &mutation.8) {
        (Some(id), Some(digest)) => transaction
            .execute(
                "UPDATE fiscal_schedules SET lifecycle_state = 'superseded' WHERE schedule_id = ?1 AND schedule_digest = ?2 AND lifecycle_state = 'active'",
                params![id, digest],
            )
            .map_err(sqlite_error)?,
        (None, None) => 1,
        _ => 0,
    };
    if admission_updated != 1 || candidate_updated != 1 || predecessor_updated != 1 {
        return Err(FiscalStoreError::Conflict);
    }
    append_projection_commit(
        transaction,
        owner,
        &format!("admission:{}", mutation.0),
        activated_version,
        "activate_fiscal_admission",
        &sha256_hex(&mutation.4),
    )
}

fn finalize_rotation_mutation(
    transaction: &Transaction<'_>,
    owner: &SqliteServingOwner,
    transition_id: &str,
) -> Result<(), FiscalStoreError> {
    let mutation = transaction
        .query_row(
            "SELECT admission_id, admission_digest, expected_admission_version, activated_admission_version, activated_admission_json, successor_charter_id, successor_charter_digest, predecessor_charter_id, predecessor_charter_digest FROM fiscal_staged_rotation_mutations WHERE transition_id = ?1",
            [transition_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, i64>(3)?,
                    row.get::<_, Vec<u8>>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, String>(8)?,
                ))
            },
        )
        .optional()
        .map_err(sqlite_error)?;
    let Some(mutation) = mutation else {
        return Ok(());
    };
    let expected_version = read_u64(mutation.2, "expected admission version")?;
    let activated_version = read_u64(mutation.3, "activated admission version")?;
    let admission_updated = transaction
        .execute(
            "UPDATE fiscal_proposal_admissions SET status = 'activated', state_version = ?1, state_json = ?2 WHERE admission_id = ?3 AND admission_digest = ?4 AND state_version = ?5 AND status = 'admitted'",
            params![
                sqlite_i64(activated_version, "activated admission version")?,
                &mutation.4,
                &mutation.0,
                &mutation.1,
                sqlite_i64(expected_version, "expected admission version")?,
            ],
        )
        .map_err(sqlite_error)?;
    let successor_updated = transaction
        .execute(
            "UPDATE fiscal_charters SET lifecycle_state = 'active' WHERE charter_id = ?1 AND charter_digest = ?2 AND lifecycle_state = 'proposed'",
            params![&mutation.5, &mutation.6],
        )
        .map_err(sqlite_error)?;
    let predecessor_updated = transaction
        .execute(
            "UPDATE fiscal_charters SET lifecycle_state = 'superseded' WHERE charter_id = ?1 AND charter_digest = ?2 AND lifecycle_state IN ('pinned', 'active')",
            params![&mutation.7, &mutation.8],
        )
        .map_err(sqlite_error)?;
    let schedule_count: i64 = transaction
        .query_row(
            "SELECT COUNT(*) FROM fiscal_staged_rotation_schedules WHERE transition_id = ?1",
            [transition_id],
            |row| row.get(0),
        )
        .map_err(sqlite_error)?;
    let candidates_updated = transaction
        .execute(
            "UPDATE fiscal_schedules SET lifecycle_state = 'active' WHERE lifecycle_state = 'staged' AND schedule_id IN (SELECT candidate_schedule_id FROM fiscal_staged_rotation_schedules WHERE transition_id = ?1)",
            [transition_id],
        )
        .map_err(sqlite_error)?;
    let predecessors_updated = transaction
        .execute(
            "UPDATE fiscal_schedules SET lifecycle_state = 'superseded' WHERE lifecycle_state = 'active' AND schedule_id IN (SELECT predecessor_schedule_id FROM fiscal_staged_rotation_schedules WHERE transition_id = ?1)",
            [transition_id],
        )
        .map_err(sqlite_error)?;
    if admission_updated != 1
        || successor_updated != 1
        || predecessor_updated != 1
        || i64::try_from(candidates_updated).map_err(|_| invariant("candidate update overflow"))?
            != schedule_count
        || i64::try_from(predecessors_updated)
            .map_err(|_| invariant("predecessor update overflow"))?
            != schedule_count
    {
        return Err(FiscalStoreError::Conflict);
    }
    append_projection_commit(
        transaction,
        owner,
        &format!("admission:{}", mutation.0),
        activated_version,
        "activate_fiscal_admission",
        &sha256_hex(&mutation.4),
    )
}

pub(crate) fn initialize_fiscal_schema(
    connection: &mut Connection,
) -> Result<(), FiscalStoreError> {
    let on_disk = crate::check_schema_version(
        connection,
        FISCAL_STORE_SCHEMA_KEY,
        FISCAL_STORE_SUPPORTED_SCHEMA_VERSION,
        &["fiscal_authority_state", "capability_grant_budgets"],
    )
    .map_err(|error| invariant(error.to_string()))?;
    if on_disk == FISCAL_STORE_SUPPORTED_SCHEMA_VERSION {
        return verify_fiscal_sql_invariants(connection);
    }
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(sqlite_error)?;
    ensure_legacy_envelope_digest_column(&transaction)?;
    transaction
        .execute_batch(FISCAL_STORE_SCHEMA)
        .map_err(sqlite_error)?;
    crate::stamp_schema_version(
        &transaction,
        FISCAL_STORE_SCHEMA_KEY,
        FISCAL_STORE_SUPPORTED_SCHEMA_VERSION,
    )
    .map_err(|error| invariant(error.to_string()))?;
    verify_fiscal_sql_invariants(&transaction)?;
    transaction.commit().map_err(sqlite_error)
}

pub(crate) fn verify_fiscal_sql_invariants(
    connection: &Connection,
) -> Result<(), FiscalStoreError> {
    let invalid = connection
        .query_row(
            r#"
            SELECT
                EXISTS(
                    SELECT 1 FROM fiscal_authority_state AS authority
                    WHERE NOT EXISTS(
                        SELECT 1 FROM fiscal_continuity_checkpoints AS checkpoint
                        WHERE checkpoint.checkpoint_digest = authority.finalized_checkpoint_digest
                          AND checkpoint.status = 'finalized'
                    )
                )
                OR EXISTS(
                    SELECT 1 FROM fiscal_staged_transitions AS stage
                    WHERE NOT EXISTS(
                        SELECT 1 FROM fiscal_projection_commits AS commit_record
                        WHERE commit_record.projection_key = 'transition:' || stage.transition_id
                          AND commit_record.projection_sequence = stage.stage_version
                    )
                      OR NOT EXISTS(
                        SELECT 1 FROM fiscal_continuity_checkpoints AS checkpoint
                        WHERE checkpoint.checkpoint_digest = stage.next_checkpoint_digest
                          AND (
                            (stage.status = 'db_finalized' AND checkpoint.status = 'finalized')
                            OR (stage.status <> 'db_finalized' AND checkpoint.status = 'staged')
                          )
                    )
                      OR (
                        stage.status <> 'db_finalized'
                        AND NOT EXISTS(
                            SELECT 1 FROM fiscal_authority_state AS authority
                            WHERE authority.finalized_checkpoint_digest = stage.current_checkpoint_digest
                        )
                    )
                )
                OR EXISTS(
                    SELECT 1 FROM fiscal_proposal_admissions AS admission
                    WHERE NOT EXISTS(
                        SELECT 1 FROM fiscal_projection_commits AS commit_record
                        WHERE commit_record.projection_key = 'admission:' || admission.admission_id
                          AND commit_record.projection_sequence = admission.state_version
                    )
                )
                OR EXISTS(
                    SELECT 1 FROM fiscal_legacy_fee_schedule_bindings AS binding
                    WHERE binding.legacy_envelope_digest IS NULL
                       OR length(binding.legacy_envelope_digest) <> 64
                       OR binding.legacy_envelope_digest GLOB '*[^0-9a-f]*'
                       OR NOT EXISTS(
                           SELECT 1 FROM fiscal_schedules AS schedule
                           WHERE schedule.schedule_id = binding.fiscal_schedule_id
                             AND schedule.schedule_digest = binding.fiscal_schedule_digest
                       )
                )
                OR EXISTS(
                    SELECT 1
                    FROM fiscal_staged_activation_mutations AS mutation
                    JOIN fiscal_staged_transitions AS stage
                      ON stage.transition_id = mutation.transition_id
                    WHERE NOT EXISTS(
                              SELECT 1 FROM fiscal_activations AS activation
                              WHERE activation.activation_id = mutation.activation_id
                                AND activation.activation_digest = mutation.activation_digest
                          )
                       OR (
                           stage.status = 'db_finalized'
                           AND (
                               NOT EXISTS(
                                   SELECT 1 FROM fiscal_proposal_admissions AS admission
                                   WHERE admission.admission_id = mutation.admission_id
                                     AND admission.admission_digest = mutation.admission_digest
                                     AND admission.status = 'activated'
                                     AND admission.state_version = mutation.activated_admission_version
                                     AND admission.state_json = mutation.activated_admission_json
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_schedules AS candidate
                                   WHERE candidate.schedule_id = mutation.candidate_schedule_id
                                     AND candidate.schedule_digest = mutation.candidate_schedule_digest
                                     AND candidate.lifecycle_state IN ('active', 'superseded')
                               )
                               OR (
                                   mutation.predecessor_schedule_id IS NOT NULL
                                   AND NOT EXISTS(
                                       SELECT 1 FROM fiscal_schedules AS predecessor
                                       WHERE predecessor.schedule_id = mutation.predecessor_schedule_id
                                         AND predecessor.schedule_digest = mutation.predecessor_schedule_digest
                                         AND predecessor.lifecycle_state = 'superseded'
                                   )
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_projection_commits AS commit_record
                                   WHERE commit_record.projection_key = 'admission:' || mutation.admission_id
                                     AND commit_record.projection_sequence = mutation.activated_admission_version
                               )
                           )
                       )
                       OR (
                           stage.status <> 'db_finalized'
                           AND (
                               NOT EXISTS(
                                   SELECT 1 FROM fiscal_proposal_admissions AS admission
                                   WHERE admission.admission_id = mutation.admission_id
                                     AND admission.admission_digest = mutation.admission_digest
                                     AND admission.status = 'admitted'
                                     AND admission.state_version = mutation.expected_admission_version
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_schedules AS candidate
                                   WHERE candidate.schedule_id = mutation.candidate_schedule_id
                                     AND candidate.schedule_digest = mutation.candidate_schedule_digest
                                     AND candidate.lifecycle_state = 'staged'
                               )
                               OR (
                                   mutation.predecessor_schedule_id IS NOT NULL
                                   AND NOT EXISTS(
                                       SELECT 1 FROM fiscal_schedules AS predecessor
                                       WHERE predecessor.schedule_id = mutation.predecessor_schedule_id
                                         AND predecessor.schedule_digest = mutation.predecessor_schedule_digest
                                         AND predecessor.lifecycle_state = 'active'
                                   )
                               )
                           )
                       )
                )
                OR EXISTS(
                    SELECT 1
                    FROM fiscal_staged_rotation_mutations AS mutation
                    JOIN fiscal_staged_transitions AS stage
                      ON stage.transition_id = mutation.transition_id
                    WHERE NOT EXISTS(
                              SELECT 1 FROM fiscal_activations AS activation
                              WHERE activation.activation_id = mutation.activation_id
                                AND activation.activation_digest = mutation.activation_digest
                          )
                       OR NOT EXISTS(
                              SELECT 1 FROM fiscal_staged_rotation_schedules AS replacement
                              WHERE replacement.transition_id = mutation.transition_id
                          )
                       OR (
                           stage.status = 'db_finalized'
                           AND (
                               NOT EXISTS(
                                   SELECT 1 FROM fiscal_proposal_admissions AS admission
                                   WHERE admission.admission_id = mutation.admission_id
                                     AND admission.admission_digest = mutation.admission_digest
                                     AND admission.status = 'activated'
                                     AND admission.state_version = mutation.activated_admission_version
                                     AND admission.state_json = mutation.activated_admission_json
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_charters AS successor
                                   WHERE successor.charter_id = mutation.successor_charter_id
                                     AND successor.charter_digest = mutation.successor_charter_digest
                                     AND successor.lifecycle_state IN ('active', 'superseded')
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_charters AS predecessor
                                   WHERE predecessor.charter_id = mutation.predecessor_charter_id
                                     AND predecessor.charter_digest = mutation.predecessor_charter_digest
                                     AND predecessor.lifecycle_state = 'superseded'
                               )
                               OR EXISTS(
                                   SELECT 1 FROM fiscal_staged_rotation_schedules AS replacement
                                   WHERE replacement.transition_id = mutation.transition_id
                                     AND (
                                         NOT EXISTS(
                                             SELECT 1 FROM fiscal_schedules AS candidate
                                             WHERE candidate.schedule_id = replacement.candidate_schedule_id
                                               AND candidate.schedule_digest = replacement.candidate_schedule_digest
                                               AND candidate.lifecycle_state IN ('active', 'superseded')
                                         )
                                         OR NOT EXISTS(
                                             SELECT 1 FROM fiscal_schedules AS predecessor
                                             WHERE predecessor.schedule_id = replacement.predecessor_schedule_id
                                               AND predecessor.schedule_digest = replacement.predecessor_schedule_digest
                                               AND predecessor.lifecycle_state = 'superseded'
                                         )
                                     )
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_projection_commits AS commit_record
                                   WHERE commit_record.projection_key = 'admission:' || mutation.admission_id
                                     AND commit_record.projection_sequence = mutation.activated_admission_version
                               )
                           )
                       )
                       OR (
                           stage.status <> 'db_finalized'
                           AND (
                               NOT EXISTS(
                                   SELECT 1 FROM fiscal_proposal_admissions AS admission
                                   WHERE admission.admission_id = mutation.admission_id
                                     AND admission.admission_digest = mutation.admission_digest
                                     AND admission.status = 'admitted'
                                     AND admission.state_version = mutation.expected_admission_version
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_charters AS successor
                                   WHERE successor.charter_id = mutation.successor_charter_id
                                     AND successor.charter_digest = mutation.successor_charter_digest
                                     AND successor.lifecycle_state = 'proposed'
                               )
                               OR NOT EXISTS(
                                   SELECT 1 FROM fiscal_charters AS predecessor
                                   WHERE predecessor.charter_id = mutation.predecessor_charter_id
                                     AND predecessor.charter_digest = mutation.predecessor_charter_digest
                                     AND predecessor.lifecycle_state IN ('pinned', 'active')
                               )
                               OR EXISTS(
                                   SELECT 1 FROM fiscal_staged_rotation_schedules AS replacement
                                   WHERE replacement.transition_id = mutation.transition_id
                                     AND (
                                         NOT EXISTS(
                                             SELECT 1 FROM fiscal_schedules AS candidate
                                             WHERE candidate.schedule_id = replacement.candidate_schedule_id
                                               AND candidate.schedule_digest = replacement.candidate_schedule_digest
                                               AND candidate.lifecycle_state = 'staged'
                                         )
                                         OR NOT EXISTS(
                                             SELECT 1 FROM fiscal_schedules AS predecessor
                                             WHERE predecessor.schedule_id = replacement.predecessor_schedule_id
                                               AND predecessor.schedule_digest = replacement.predecessor_schedule_digest
                                               AND predecessor.lifecycle_state = 'active'
                                         )
                                     )
                               )
                           )
                       )
                )
                OR EXISTS(
                    SELECT 1 FROM fiscal_projection_commits AS commit_record
                    WHERE commit_record.projection_sequence > 1
                      AND NOT EXISTS(
                        SELECT 1 FROM fiscal_projection_commits AS previous
                        WHERE previous.projection_key = commit_record.projection_key
                          AND previous.projection_sequence = commit_record.projection_sequence - 1
                          AND previous.commit_digest = commit_record.previous_commit_digest
                    )
                )
                OR EXISTS(
                    SELECT 1 FROM fiscal_projection_commits
                    WHERE projection_sequence = 1 AND previous_commit_digest <> ?1
                )
            "#,
            [ZERO_DIGEST],
            |row| row.get::<_, bool>(0),
        )
        .map_err(sqlite_error)?;
    if invalid {
        Err(invariant("fiscal store projection is inconsistent"))
    } else {
        Ok(())
    }
}

fn ensure_legacy_envelope_digest_column(
    transaction: &Transaction<'_>,
) -> Result<(), FiscalStoreError> {
    let table_present = transaction
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = 'fiscal_legacy_fee_schedule_bindings')",
            [],
            |row| row.get::<_, bool>(0),
        )
        .map_err(sqlite_error)?;
    if !table_present {
        return Ok(());
    }
    let mut statement = transaction
        .prepare("PRAGMA table_info(fiscal_legacy_fee_schedule_bindings)")
        .map_err(sqlite_error)?;
    let columns = statement
        .query_map([], |row| row.get::<_, String>(1))
        .map_err(sqlite_error)?
        .collect::<Result<Vec<_>, _>>()
        .map_err(sqlite_error)?;
    if !columns
        .iter()
        .any(|column| column == "legacy_envelope_digest")
    {
        transaction
            .execute(
                "ALTER TABLE fiscal_legacy_fee_schedule_bindings ADD COLUMN legacy_envelope_digest TEXT",
                [],
            )
            .map_err(sqlite_error)?;
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn verify_exact_genesis(
    transaction: &Transaction<'_>,
    policy_json: &[u8],
    authority_json: &[u8],
    charter_json: &[u8],
    readiness_json: &[u8],
    registry_json: &[u8],
    checkpoint_json: &[u8],
    policy: &FiscalGenesisPolicy,
    charter: &VerifiedFiscalCharter,
    readiness: &VerifiedFiscalRuntimeReadiness,
    checkpoint: &VerifiedFiscalContinuityCheckpoint,
) -> Result<(), FiscalStoreError> {
    let exact = transaction
        .query_row(
            r#"
            SELECT
                (SELECT policy_json FROM fiscal_genesis_policies WHERE policy_id = ?1) = ?2
                AND (SELECT state_json FROM fiscal_authority_state WHERE singleton = 1) = ?3
                AND (SELECT signed_json FROM fiscal_charters WHERE charter_id = ?4) = ?5
                AND (SELECT signed_json FROM fiscal_runtime_readiness WHERE readiness_id = ?6) = ?7
                AND (SELECT registry_json FROM fiscal_runtime_readiness WHERE readiness_id = ?6) = ?8
                AND (SELECT signed_json FROM fiscal_continuity_checkpoints WHERE checkpoint_digest = ?9) = ?10
            "#,
            params![
                &policy.policy_id,
                policy_json,
                authority_json,
                &charter.body().charter_id,
                charter_json,
                &readiness.body().readiness_id,
                readiness_json,
                registry_json,
                checkpoint.digest(),
                checkpoint_json,
            ],
            |row| row.get::<_, bool>(0),
        )
        .map_err(sqlite_error)?;
    if exact {
        Ok(())
    } else {
        Err(FiscalStoreError::Conflict)
    }
}

fn verify_acknowledgement(
    transaction: &Transaction<'_>,
    record: &FiscalStagedTransitionRecord,
    checkpoint: &VerifiedFiscalContinuityCheckpoint,
    canonical: &[u8],
) -> Result<(), FiscalStoreError> {
    if checkpoint.digest() != record.next_checkpoint_digest {
        return Err(FiscalStoreError::Conflict);
    }
    let stored = transaction
        .query_row(
            "SELECT signed_json FROM fiscal_continuity_checkpoints WHERE checkpoint_digest = ?1",
            [&record.next_checkpoint_digest],
            |row| row.get::<_, Vec<u8>>(0),
        )
        .optional()
        .map_err(sqlite_error)?
        .ok_or(FiscalStoreError::NotFound)?;
    if stored != canonical {
        return Err(FiscalStoreError::Conflict);
    }
    Ok(())
}

fn load_transition(
    transaction: &Transaction<'_>,
    transition_id: &str,
) -> Result<FiscalStagedTransitionRecord, FiscalStoreError> {
    transaction
        .query_row(
            "SELECT transition_id, current_checkpoint_digest, next_checkpoint_digest, proof_json, status, stage_version FROM fiscal_staged_transitions WHERE transition_id = ?1",
            [transition_id],
            |row| {
                let status = row.get::<_, String>(4)?;
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, Vec<u8>>(3)?,
                    status,
                    row.get::<_, i64>(5)?,
                ))
            },
        )
        .optional()
        .map_err(sqlite_error)?
        .ok_or(FiscalStoreError::NotFound)
        .and_then(|row| {
            Ok(FiscalStagedTransitionRecord {
                transition_id: row.0,
                current_checkpoint_digest: row.1,
                next_checkpoint_digest: row.2,
                proof_json: row.3,
                status: FiscalStageStatus::parse(&row.4)?,
                stage_version: read_u64(row.5, "fiscal stage version")?,
            })
        })
}

fn append_projection_commit(
    transaction: &Transaction<'_>,
    owner: &SqliteServingOwner,
    projection_key: &str,
    projection_sequence: u64,
    mutation_kind: &str,
    snapshot_digest: &str,
) -> Result<(), FiscalStoreError> {
    let previous = if projection_sequence == 1 {
        ZERO_DIGEST.to_owned()
    } else {
        let previous_sequence = projection_sequence
            .checked_sub(1)
            .ok_or_else(|| invariant("fiscal projection sequence underflowed"))?;
        transaction
            .query_row(
                "SELECT commit_digest FROM fiscal_projection_commits WHERE projection_key = ?1 AND projection_sequence = ?2",
                params![projection_key, sqlite_i64(previous_sequence, "previous fiscal projection sequence")?],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(sqlite_error)?
            .ok_or_else(|| invariant("previous fiscal projection commit is absent"))?
    };
    let commit_digest = canonical_digest(&FiscalProjectionCommit {
        format: "chio.sqlite-fiscal-projection-commit.v1",
        projection_key,
        projection_sequence,
        mutation_kind,
        snapshot_digest,
        previous_commit_digest: &previous,
        store_uuid: &owner.fence.store_uuid,
        store_lease_id: &owner.fence.lease_id,
        store_owner_epoch: owner.fence.owner_epoch,
    })?;
    transaction
        .execute(
            "INSERT INTO fiscal_projection_commits (projection_key, projection_sequence, mutation_kind, snapshot_digest, previous_commit_digest, commit_digest, store_uuid, store_lease_id, store_owner_epoch) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                projection_key,
                sqlite_i64(projection_sequence, "fiscal projection sequence")?,
                mutation_kind,
                snapshot_digest,
                &previous,
                &commit_digest,
                &owner.fence.store_uuid,
                &owner.fence.lease_id,
                sqlite_i64(owner.fence.owner_epoch, "store owner epoch")?,
            ],
        )
        .map_err(sqlite_error)?;
    owner
        .append_global_commit(
            transaction,
            mutation_kind,
            "fiscal",
            projection_key,
            projection_sequence,
        )
        .map_err(map_owner_error)
}

fn verify_owner(
    transaction: &Transaction<'_>,
    owner: &SqliteServingOwner,
    fence: Option<&StoreMutationFence>,
) -> Result<(), FiscalStoreError> {
    crate::admission_operation_store::verify_active_owner(transaction, owner, fence).map_err(
        |error| match error {
            chio_kernel::admission_operation::AdmissionOperationStoreError::Fenced => {
                FiscalStoreError::Fenced
            }
            other => FiscalStoreError::Unavailable(other.to_string()),
        },
    )
}

fn canonical_digest(value: &impl Serialize) -> Result<String, FiscalStoreError> {
    canonical_json_bytes(value)
        .map(|bytes| sha256_hex(&bytes))
        .map_err(canonical_error)
}

fn canonical_error(error: impl std::fmt::Display) -> FiscalStoreError {
    invariant(format!("canonical fiscal encoding failed: {error}"))
}

fn sqlite_error(error: rusqlite::Error) -> FiscalStoreError {
    FiscalStoreError::Unavailable(error.to_string())
}

fn map_owner_error(error: SqliteServingOwnerError) -> FiscalStoreError {
    match error {
        SqliteServingOwnerError::OutcomeUnknown(detail) => FiscalStoreError::OutcomeUnknown(detail),
        other => FiscalStoreError::Unavailable(other.to_string()),
    }
}

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

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

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

#[cfg(test)]
#[path = "fiscal_store_tests.rs"]
mod tests;