icydb-core 0.213.33

//! Module: db::commit::schema_publication
//! Responsibility: marker-bound accepted-schema bundle/root publication.
//! Does not own: candidate construction, schema compatibility, or root codecs.
//! Boundary: schema reconciliation -> commit marker/journal -> schema live projection.

#[cfg(any(test, feature = "sql"))]
use crate::db::index::{IndexEntryValue, IndexKey, RawIndexStoreKey};
#[cfg(feature = "sql")]
use crate::db::{
    data::DataStore,
    index::IndexStore,
    schema::{
        StagedUserIndexDomainReplacement, accepted_schema_cache_fingerprint_for_persisted_snapshot,
    },
};
use crate::{
    db::{
        commit::{
            CommitMarker, begin_commit, finish_commit, generate_commit_id, generate_marker_batch_id,
        },
        journal::{JournalBatch, JournalRecord, JournalSequence},
        registry::{StoreHandle, StoreRecoveryCapability},
        schema::{
            AcceptedSchemaRevision, CandidateSchemaRevision, ConstraintId, ConstraintValidationJob,
            SchemaApplicationRecordOp, apply_schema_application_record_op,
        },
    },
    error::InternalError,
    types::EntityTag,
};
#[cfg(feature = "sql")]
use std::collections::BTreeSet;

/// Prepared derived domains that cross the same accepted-schema marker.

enum StagedSchemaDomains {
    None,
    #[cfg(feature = "sql")]
    UserIndexes(Vec<StagedUserIndexDomainReplacement>),
}

/// Exact validation-job mutation paired with one accepted-schema publication.

#[derive(Clone, Copy)]
enum ConstraintValidationJobChange<'a> {
    None,
    Put(&'a ConstraintValidationJob),
    Delete {
        entity_tag: EntityTag,
        constraint_id: ConstraintId,
    },
}

///
/// AcceptedSchemaPublication
///
/// One preconstructed store-local candidate participating in an atomic
/// database-scoped accepted-schema publication.
///

pub(in crate::db) struct AcceptedSchemaPublication<'a> {
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &'a CandidateSchemaRevision,
}

impl<'a> AcceptedSchemaPublication<'a> {
    /// Bind one catalog-native candidate to its registered store authority.
    pub(in crate::db) const fn new(
        store_path: &'static str,
        store: StoreHandle,
        expected_revision: AcceptedSchemaRevision,
        candidate: &'a CandidateSchemaRevision,
    ) -> Self {
        Self {
            store_path,
            store,
            expected_revision,
            candidate,
        }
    }
}

#[cfg(any(test, feature = "sql"))]
pub(in crate::db) fn publish_accepted_schema_candidate(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
) -> Result<(), InternalError> {
    publish_accepted_schema_candidates_atomically(vec![AcceptedSchemaPublication::new(
        store_path,
        store,
        expected_revision,
        candidate,
    )])
}

/// Publish one or more catalog-native accepted candidates through one durable
/// marker boundary.
///
/// Multi-store publication currently requires journaled stores. A volatile
/// heap store has no recovery authority capable of completing an interrupted
/// database-scoped publication, so mixed or multi-heap input rejects before
/// mutation.
#[cfg(any(test, feature = "sql"))]
pub(in crate::db) fn publish_accepted_schema_candidates_atomically(
    publications: Vec<AcceptedSchemaPublication<'_>>,
) -> Result<(), InternalError> {
    publish_accepted_schema_candidates_with_optional_application_record(publications, None)
}

/// Publish zero or more accepted candidates and one terminal application
/// receipt through the same database-wide marker boundary.
pub(in crate::db) fn publish_accepted_schema_candidates_with_application_record(
    publications: Vec<AcceptedSchemaPublication<'_>>,
    application_record: SchemaApplicationRecordOp,
) -> Result<(), InternalError> {
    publish_accepted_schema_candidates_with_optional_application_record(
        publications,
        Some(application_record),
    )
}

fn publish_accepted_schema_candidates_with_optional_application_record(
    mut publications: Vec<AcceptedSchemaPublication<'_>>,
    application_record: Option<SchemaApplicationRecordOp>,
) -> Result<(), InternalError> {
    if publications.is_empty() {
        let application_record = application_record.ok_or_else(InternalError::store_invariant)?;
        return publish_application_record_atomically(application_record);
    }
    publications.sort_by_key(|publication| publication.store_path);
    if publications
        .windows(2)
        .any(|pair| pair[0].store_path == pair[1].store_path)
        || publications
            .iter()
            .any(|publication| publication.candidate.store_path() != publication.store_path)
    {
        return Err(InternalError::store_invariant());
    }
    for publication in &publications {
        validate_constraint_validation_job_change(
            publication.store,
            publication.candidate,
            ConstraintValidationJobChange::None,
        )?;
    }

    let already_current = publications
        .iter()
        .map(|publication| {
            publication.store.with_schema(|schema_store| {
                schema_store.preflight_accepted_schema_candidate(
                    publication.expected_revision,
                    publication.candidate,
                )
            })
        })
        .collect::<Result<Vec<_>, _>>()?;
    if already_current.iter().all(|current| *current) {
        return if application_record.is_none() {
            Ok(())
        } else {
            Err(InternalError::store_invariant())
        };
    }
    if already_current.iter().any(|current| *current) {
        return Err(InternalError::store_invariant());
    }

    if publications.len() == 1
        && publications[0].store.storage_capabilities().recovery() == StoreRecoveryCapability::None
    {
        if application_record.is_some() {
            return Err(InternalError::store_unsupported());
        }
        let publication = &publications[0];
        return publication.store.with_schema_mut(|schema_store| {
            schema_store.publish_accepted_schema_candidate(
                publication.expected_revision,
                publication.candidate,
            )
        });
    }
    if application_record.is_none()
        && publications.iter().any(|publication| {
            publication.store.storage_capabilities().recovery() == StoreRecoveryCapability::None
        })
    {
        return Err(InternalError::store_unsupported());
    }

    publish_candidates_atomically(publications.as_slice(), application_record)
}

/// Publish one accepted candidate and the exact validation job its new
/// `Validating` activation requires through one marker boundary.
pub(in crate::db) fn publish_accepted_schema_candidate_with_constraint_validation_job(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    job: &ConstraintValidationJob,
) -> Result<(), InternalError> {
    publish_accepted_schema_candidate_with_prepared_domains(
        store_path,
        store,
        expected_revision,
        candidate,
        StagedSchemaDomains::None,
        ConstraintValidationJobChange::Put(job),
    )
}

/// Publish one promotion or abort and remove its validation job through the
/// same marker boundary.
pub(in crate::db) fn publish_accepted_schema_candidate_with_constraint_validation_job_removal(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    entity_tag: EntityTag,
    constraint_id: ConstraintId,
) -> Result<(), InternalError> {
    publish_accepted_schema_candidate_with_prepared_domains(
        store_path,
        store,
        expected_revision,
        candidate,
        StagedSchemaDomains::None,
        ConstraintValidationJobChange::Delete {
            entity_tag,
            constraint_id,
        },
    )
}

/// Advance one validation job without changing accepted constraint meaning.
pub(in crate::db) fn publish_constraint_validation_job(
    store_path: &'static str,
    store: StoreHandle,
    job: &ConstraintValidationJob,
) -> Result<(), InternalError> {
    let bundle = store
        .with_schema(crate::db::schema::SchemaStore::current_accepted_schema_bundle)?
        .ok_or_else(InternalError::store_corruption)?;
    if bundle.store_path() != store_path {
        return Err(InternalError::store_corruption());
    }
    store.with_schema(|schema_store| {
        schema_store.validate_constraint_validation_job_closure_with_change(
            &bundle,
            Some(job),
            None,
        )
    })?;

    match store.storage_capabilities().recovery() {
        StoreRecoveryCapability::None => {
            store.with_schema_mut(|schema_store| schema_store.apply_constraint_validation_job(job))
        }
        StoreRecoveryCapability::StableBasePlusJournalReplay => {
            publish_journaled_constraint_validation_job(store_path, store, job)
        }
    }
}

/// Advance one unique-index validation page and its isolated candidate writes
/// through the same marker-owned checkpoint boundary.
#[cfg(any(test, feature = "sql"))]
pub(in crate::db) fn publish_constraint_validation_job_with_candidate_index_entries(
    store_path: &'static str,
    store: StoreHandle,
    job: &ConstraintValidationJob,
    entries: Vec<RawIndexStoreKey>,
) -> Result<(), InternalError> {
    let bundle = store
        .with_schema(crate::db::schema::SchemaStore::current_accepted_schema_bundle)?
        .ok_or_else(InternalError::store_corruption)?;
    if bundle.store_path() != store_path {
        return Err(InternalError::store_corruption());
    }
    store.with_schema(|schema_store| {
        schema_store.validate_constraint_validation_job_closure_with_change(
            &bundle,
            Some(job),
            None,
        )
    })?;
    validate_candidate_index_entries(&bundle, job, entries.as_slice())?;
    if store.storage_capabilities().recovery()
        != StoreRecoveryCapability::StableBasePlusJournalReplay
    {
        return Err(InternalError::store_unsupported());
    }

    publish_journaled_constraint_validation_job_with_candidate_index_entries(
        store_path, store, job, entries,
    )
}

/// Publish one accepted-schema candidate and its prevalidated per-entity
/// user-index domains through the same marker window.
#[cfg(feature = "sql")]
pub(in crate::db) fn publish_accepted_schema_candidate_with_user_index_domains(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    replacements: Vec<StagedUserIndexDomainReplacement>,
) -> Result<(), InternalError> {
    validate_user_index_domain_candidates(
        store_path,
        store,
        expected_revision,
        candidate,
        replacements.as_slice(),
    )?;
    publish_accepted_schema_candidate_with_prepared_domains(
        store_path,
        store,
        expected_revision,
        candidate,
        StagedSchemaDomains::UserIndexes(replacements),
        ConstraintValidationJobChange::None,
    )
}

fn publish_accepted_schema_candidate_with_prepared_domains(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    domains: StagedSchemaDomains,
    job_change: ConstraintValidationJobChange<'_>,
) -> Result<(), InternalError> {
    validate_constraint_validation_job_change(store, candidate, job_change)?;
    match store.storage_capabilities().recovery() {
        StoreRecoveryCapability::None => {
            publish_heap_candidate_with_constraint_validation_job_change(
                store,
                expected_revision,
                candidate,
                job_change,
            )?;
            apply_staged_schema_domains(store, domains);
            Ok(())
        }
        StoreRecoveryCapability::StableBasePlusJournalReplay => publish_journaled_candidate(
            store_path,
            store,
            expected_revision,
            candidate,
            domains,
            job_change,
        ),
    }
}

fn publish_journaled_candidate(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    domains: StagedSchemaDomains,
    job_change: ConstraintValidationJobChange<'_>,
) -> Result<(), InternalError> {
    let journal_store = store
        .journal_tail_store()
        .ok_or_else(InternalError::store_invariant)?;
    let marker_id = generate_commit_id()?;
    let sequence = journal_store
        .with_borrow(crate::db::journal::JournalTailStore::next_mutation_append_sequence)?;
    let schema_record = JournalRecord::accepted_schema_publish(
        store_path,
        expected_revision,
        candidate.encoded_bundle().to_vec(),
        candidate.encoded_root().to_vec(),
    )?;
    let mut records = vec![schema_record];
    if let Some(record) = constraint_validation_job_journal_record(store_path, job_change)? {
        records.push(record);
    }
    let batch = JournalBatch::new(marker_id, marker_id, sequence, records)?;
    let marker = CommitMarker::from_parts(marker_id, vec![batch.clone()])?;
    let commit = begin_commit(marker)?;

    finish_commit(commit, |_guard| {
        journal_store.with_borrow_mut(|journal| journal.append_batch(&batch))?;
        store.with_schema_mut(|schema_store| {
            schema_store.apply_journaled_accepted_schema_candidate(expected_revision, candidate)
        })?;
        apply_constraint_validation_job_change(store, job_change)?;
        apply_staged_schema_domains(store, domains);
        Ok(())
    })
}

fn publish_candidates_atomically(
    publications: &[AcceptedSchemaPublication<'_>],
    application_record: Option<SchemaApplicationRecordOp>,
) -> Result<(), InternalError> {
    let marker_id = generate_commit_id()?;
    let mut batches = Vec::with_capacity(publications.len());
    for (ordinal, publication) in publications.iter().enumerate() {
        let sequence = match (
            publication.expected_revision,
            publication.store.journal_tail_store(),
        ) {
            (AcceptedSchemaRevision::NONE, _) | (_, None) => JournalSequence::new(0),
            (_, Some(journal_store)) => journal_store
                .with_borrow(crate::db::journal::JournalTailStore::next_mutation_append_sequence)?,
        };
        let record = JournalRecord::accepted_schema_publish(
            publication.store_path,
            publication.expected_revision,
            publication.candidate.encoded_bundle().to_vec(),
            publication.candidate.encoded_root().to_vec(),
        )?;
        let batch_id = generate_marker_batch_id(marker_id, ordinal)?;
        batches.push(JournalBatch::new(
            batch_id,
            marker_id,
            sequence,
            vec![record],
        )?);
    }
    let marker = CommitMarker::from_parts_with_schema_application(
        marker_id,
        batches.clone(),
        application_record.clone(),
    )?;
    let commit = begin_commit(marker)?;

    finish_commit(commit, |_guard| {
        for (publication, batch) in publications.iter().zip(&batches) {
            if batch.journal_sequence() != JournalSequence::new(0)
                && let Some(journal_store) = publication.store.journal_tail_store()
            {
                journal_store.with_borrow_mut(|journal| journal.append_batch(batch))?;
            }
        }
        for publication in publications {
            publication.store.with_schema_mut(|schema_store| {
                match (
                    publication.expected_revision,
                    publication.store.storage_capabilities().recovery(),
                ) {
                    (AcceptedSchemaRevision::NONE, _) | (_, StoreRecoveryCapability::None) => {
                        schema_store.publish_accepted_schema_candidate(
                            publication.expected_revision,
                            publication.candidate,
                        )
                    }
                    (_, StoreRecoveryCapability::StableBasePlusJournalReplay) => schema_store
                        .apply_journaled_accepted_schema_candidate(
                            publication.expected_revision,
                            publication.candidate,
                        ),
                }
            })?;
        }
        if let Some(operation) = application_record.as_ref() {
            apply_schema_application_record_op(operation)?;
        }
        Ok(())
    })
}

fn publish_application_record_atomically(
    application_record: SchemaApplicationRecordOp,
) -> Result<(), InternalError> {
    let marker_id = generate_commit_id()?;
    let marker = CommitMarker::from_parts_with_schema_application(
        marker_id,
        Vec::new(),
        Some(application_record.clone()),
    )?;
    let commit = begin_commit(marker)?;
    finish_commit(commit, |_guard| {
        apply_schema_application_record_op(&application_record)
    })
}

fn publish_journaled_constraint_validation_job(
    store_path: &'static str,
    store: StoreHandle,
    job: &ConstraintValidationJob,
) -> Result<(), InternalError> {
    let journal_store = store
        .journal_tail_store()
        .ok_or_else(InternalError::store_invariant)?;
    let marker_id = generate_commit_id()?;
    let sequence = journal_store
        .with_borrow(crate::db::journal::JournalTailStore::next_mutation_append_sequence)?;
    let record = JournalRecord::constraint_validation_job_put(store_path, job)?;
    let batch = JournalBatch::new(marker_id, marker_id, sequence, vec![record])?;
    let marker = CommitMarker::from_parts(marker_id, vec![batch.clone()])?;
    let commit = begin_commit(marker)?;

    finish_commit(commit, |_guard| {
        journal_store.with_borrow_mut(|journal| journal.append_batch(&batch))?;
        store.with_schema_mut(|schema_store| schema_store.apply_constraint_validation_job(job))
    })
}

#[cfg(any(test, feature = "sql"))]
fn publish_journaled_constraint_validation_job_with_candidate_index_entries(
    store_path: &'static str,
    store: StoreHandle,
    job: &ConstraintValidationJob,
    entries: Vec<RawIndexStoreKey>,
) -> Result<(), InternalError> {
    let journal_store = store
        .journal_tail_store()
        .ok_or_else(InternalError::store_invariant)?;
    let marker_id = generate_commit_id()?;
    let sequence = journal_store
        .with_borrow(crate::db::journal::JournalTailStore::next_mutation_append_sequence)?;
    let record = JournalRecord::constraint_validation_job_put(store_path, job)?;
    let batch = JournalBatch::new(marker_id, marker_id, sequence, vec![record])?;
    let marker = CommitMarker::from_parts(marker_id, vec![batch.clone()])?;
    let commit = begin_commit(marker)?;

    finish_commit(commit, |_guard| {
        journal_store.with_borrow_mut(|journal| journal.append_batch(&batch))?;
        store.with_index_mut(|index_store| {
            for key in entries {
                index_store.insert(key, IndexEntryValue::presence());
            }
        });
        store.with_schema_mut(|schema_store| schema_store.apply_constraint_validation_job(job))
    })
}

#[cfg(any(test, feature = "sql"))]
fn validate_candidate_index_entries(
    bundle: &crate::db::schema::AcceptedSchemaRevisionBundle,
    job: &ConstraintValidationJob,
    entries: &[RawIndexStoreKey],
) -> Result<(), InternalError> {
    let snapshot = bundle
        .entity_snapshots()
        .get(&job.entity_tag())
        .ok_or_else(InternalError::store_corruption)?;
    let activation = snapshot
        .constraint_catalog()
        .activation(job.constraint_id())
        .ok_or_else(InternalError::store_corruption)?;
    let crate::db::schema::ConstraintActivationKind::Unique { index_id } = activation.kind() else {
        return Err(InternalError::store_corruption());
    };
    let candidate = snapshot
        .candidate_indexes()
        .iter()
        .find(|index| index.schema_id() == *index_id)
        .ok_or_else(InternalError::store_corruption)?;
    let expected = crate::db::index::IndexId::new_with_generation(
        job.entity_tag(),
        candidate.ordinal(),
        candidate.physical_generation(),
    );
    if job.staged_generation() != Some(candidate.physical_generation())
        || entries.windows(2).any(|pair| pair[0] >= pair[1])
        || entries.iter().any(|raw| {
            IndexKey::try_from_raw(raw).map_or(true, |key| {
                key.key_kind() != crate::db::index::IndexKeyKind::User
                    || *key.index_id() != expected
            })
        })
    {
        return Err(InternalError::store_corruption());
    }
    Ok(())
}

fn validate_constraint_validation_job_change(
    store: StoreHandle,
    candidate: &CandidateSchemaRevision,
    change: ConstraintValidationJobChange<'_>,
) -> Result<(), InternalError> {
    store.with_schema(|schema_store| {
        schema_store.validate_live_activation_transition(candidate.bundle())?;
        match change {
            ConstraintValidationJobChange::None => {
                schema_store.validate_constraint_validation_job_closure(candidate.bundle())
            }
            ConstraintValidationJobChange::Put(job) => schema_store
                .validate_constraint_validation_job_closure_with_change(
                    candidate.bundle(),
                    Some(job),
                    None,
                ),
            ConstraintValidationJobChange::Delete {
                entity_tag,
                constraint_id,
            } => schema_store.validate_constraint_validation_job_closure_with_change(
                candidate.bundle(),
                None,
                Some((entity_tag, constraint_id)),
            ),
        }
    })
}

fn publish_heap_candidate_with_constraint_validation_job_change(
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    change: ConstraintValidationJobChange<'_>,
) -> Result<(), InternalError> {
    store.with_schema_mut(|schema_store| match change {
        ConstraintValidationJobChange::None => {
            schema_store.publish_accepted_schema_candidate(expected_revision, candidate)
        }
        ConstraintValidationJobChange::Put(job) => {
            schema_store.apply_constraint_validation_job(job)?;
            if let Err(error) =
                schema_store.publish_accepted_schema_candidate(expected_revision, candidate)
            {
                schema_store.apply_constraint_validation_job_removal(
                    job.entity_tag(),
                    job.constraint_id(),
                )?;
                return Err(error);
            }
            Ok(())
        }
        ConstraintValidationJobChange::Delete {
            entity_tag,
            constraint_id,
        } => {
            schema_store.publish_accepted_schema_candidate(expected_revision, candidate)?;
            schema_store.apply_constraint_validation_job_removal(entity_tag, constraint_id)
        }
    })
}

fn constraint_validation_job_journal_record(
    store_path: &'static str,
    change: ConstraintValidationJobChange<'_>,
) -> Result<Option<JournalRecord>, InternalError> {
    match change {
        ConstraintValidationJobChange::None => Ok(None),
        ConstraintValidationJobChange::Put(job) => {
            JournalRecord::constraint_validation_job_put(store_path, job).map(Some)
        }
        ConstraintValidationJobChange::Delete {
            entity_tag,
            constraint_id,
        } => JournalRecord::constraint_validation_job_delete(store_path, entity_tag, constraint_id)
            .map(Some),
    }
}

fn apply_constraint_validation_job_change(
    store: StoreHandle,
    change: ConstraintValidationJobChange<'_>,
) -> Result<(), InternalError> {
    store.with_schema_mut(|schema_store| match change {
        ConstraintValidationJobChange::None => Ok(()),
        ConstraintValidationJobChange::Put(job) => {
            schema_store.apply_constraint_validation_job(job)
        }
        ConstraintValidationJobChange::Delete {
            entity_tag,
            constraint_id,
        } => schema_store.apply_constraint_validation_job_removal(entity_tag, constraint_id),
    })
}

#[cfg(feature = "sql")]
fn validate_user_index_domain_candidates(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    replacements: &[StagedUserIndexDomainReplacement],
) -> Result<(), InternalError> {
    if replacements.is_empty() || candidate.store_path() != store_path {
        return Err(InternalError::store_invariant());
    }
    let mut entities = BTreeSet::new();
    for replacement in replacements {
        validate_user_index_domain_candidate(
            store_path,
            store,
            expected_revision,
            candidate,
            replacement,
            &mut entities,
        )?;
    }
    if store.index_state() != crate::db::index::IndexState::Ready {
        return Err(InternalError::store_unsupported());
    }

    Ok(())
}

#[cfg(feature = "sql")]
fn validate_user_index_domain_candidate(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    replacement: &StagedUserIndexDomainReplacement,
    entities: &mut BTreeSet<crate::types::EntityTag>,
) -> Result<(), InternalError> {
    let accepted_before_identity = replacement.accepted_before_identity();
    if !entities.insert(replacement.entity_tag())
        || replacement.store_path() != store_path
        || accepted_before_identity.store_path() != store_path
        || accepted_before_identity.accepted_schema_revision() != expected_revision
    {
        return Err(InternalError::store_invariant());
    }
    let current_identity = store
        .with_schema(|schema_store| {
            schema_store.current_accepted_catalog_selection(
                replacement.entity_tag(),
                accepted_before_identity.entity_path(),
                store_path,
            )
        })?
        .ok_or_else(InternalError::store_corruption)?
        .identity();
    if current_identity != accepted_before_identity {
        return Err(InternalError::store_invariant());
    }
    let accepted_after = candidate
        .bundle()
        .entity_snapshots()
        .get(&replacement.entity_tag())
        .ok_or_else(InternalError::store_corruption)?;
    let accepted_after_fingerprint =
        accepted_schema_cache_fingerprint_for_persisted_snapshot(accepted_after)?;
    let entity_path_matches =
        accepted_after.entity_path() == accepted_before_identity.entity_path();
    let schema_version_matches = accepted_after.version() == replacement.accepted_after_version();
    let schema_fingerprint_matches =
        accepted_after_fingerprint == replacement.accepted_after_fingerprint();
    if !(entity_path_matches && schema_version_matches && schema_fingerprint_matches) {
        return Err(InternalError::store_invariant());
    }

    Ok(())
}

#[cfg(not(feature = "sql"))]
const fn apply_staged_schema_domains(_store: StoreHandle, domains: StagedSchemaDomains) {
    match domains {
        StagedSchemaDomains::None => {}
    }
}

#[cfg(feature = "sql")]
fn apply_staged_schema_domains(store: StoreHandle, domains: StagedSchemaDomains) {
    match domains {
        StagedSchemaDomains::None => {}
        StagedSchemaDomains::UserIndexes(replacements) => {
            apply_user_index_domain_replacements(store, replacements);
        }
    }
}

#[cfg(feature = "sql")]
fn apply_user_index_domain_replacements(
    store: StoreHandle,
    replacements: Vec<StagedUserIndexDomainReplacement>,
) {
    let data_generation = store.with_data(DataStore::generation);
    store.with_index_mut(|index_store| {
        index_store.mark_building();
        for replacement in replacements {
            apply_user_index_domain_replacement(index_store, replacement);
        }
        index_store.mark_prefix_cardinality_data_generation(data_generation);
        index_store.mark_ready();
    });
}

#[cfg(feature = "sql")]
fn apply_user_index_domain_replacement(
    index_store: &mut IndexStore,
    replacement: StagedUserIndexDomainReplacement,
) {
    let (deletion_keys, final_entries) = replacement.into_apply_parts();
    for key in deletion_keys {
        index_store.remove(&key);
    }
    for entry in final_entries {
        let (key, value) = entry.into_parts();
        index_store.insert(key, value);
    }
}