icydb-core 0.213.36

//! 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, StoreSchemaMetadataCapability},
        schema::{
            AcceptedSchemaRevision, CandidateSchemaRevision, ConstraintId, ConstraintValidationJob,
            SchemaApplicationRecordOp, apply_live_schema_checkpoint,
            apply_schema_application_record_op, preflight_live_schema_checkpoint,
        },
    },
    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.
///
/// Stores with live-only schema allocations checkpoint the exact accepted
/// candidate in database-control memory under the same marker. Recovery can
/// therefore restore accepted catalog authority without consulting authored
/// or generated proposal material.
#[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());
    }

    for publication in &publications {
        if publication.store.storage_capabilities().schema_metadata()
            == StoreSchemaMetadataCapability::LiveRebuiltMetadata
        {
            preflight_live_schema_checkpoint(
                publication.store_path,
                publication.expected_revision,
                publication.candidate,
            )?;
        }
    }

    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),
        None,
    )
}

/// 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,
        },
        None,
    )
}

/// Publish one generated-check abort, validation-job removal, and terminal
/// application receipt through the same marker boundary.
pub(in crate::db) fn publish_generated_check_abort_with_application_record(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    entity_tag: EntityTag,
    constraint_id: ConstraintId,
    application_record: SchemaApplicationRecordOp,
) -> Result<(), InternalError> {
    publish_accepted_schema_candidate_with_prepared_domains(
        store_path,
        store,
        expected_revision,
        candidate,
        StagedSchemaDomains::None,
        ConstraintValidationJobChange::Delete {
            entity_tag,
            constraint_id,
        },
        Some(application_record),
    )
}

/// 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,
        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<'_>,
    application_record: Option<SchemaApplicationRecordOp>,
) -> Result<(), InternalError> {
    validate_constraint_validation_job_change(store, candidate, job_change)?;
    match store.storage_capabilities().recovery() {
        StoreRecoveryCapability::None => publish_live_candidate_with_prepared_domains(
            store_path,
            store,
            expected_revision,
            candidate,
            domains,
            job_change,
            application_record,
        ),
        StoreRecoveryCapability::StableBasePlusJournalReplay => publish_journaled_candidate(
            store_path,
            store,
            expected_revision,
            candidate,
            domains,
            job_change,
            application_record,
        ),
    }
}

fn publish_live_candidate_with_prepared_domains(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    domains: StagedSchemaDomains,
    job_change: ConstraintValidationJobChange<'_>,
    application_record: Option<SchemaApplicationRecordOp>,
) -> Result<(), InternalError> {
    if !matches!(job_change, ConstraintValidationJobChange::None)
        || application_record.is_some()
        || store.storage_capabilities().schema_metadata()
            != StoreSchemaMetadataCapability::LiveRebuiltMetadata
    {
        return Err(InternalError::store_unsupported());
    }
    store.with_schema(|schema_store| {
        schema_store.preflight_accepted_schema_candidate(expected_revision, candidate)
    })?;
    preflight_live_schema_checkpoint(store_path, expected_revision, candidate)?;
    let marker_id = generate_commit_id()?;
    let record = JournalRecord::accepted_schema_publish(
        store_path,
        expected_revision,
        candidate.encoded_bundle().to_vec(),
        candidate.encoded_root().to_vec(),
    )?;
    let batch = JournalBatch::new(marker_id, marker_id, JournalSequence::new(0), vec![record])?;
    let marker = CommitMarker::from_parts(marker_id, vec![batch])?;
    let commit = begin_commit(marker)?;

    finish_commit(commit, |_guard| {
        apply_live_schema_checkpoint(store_path, expected_revision, candidate)?;
        store.with_schema_mut(|schema_store| {
            schema_store.publish_accepted_schema_candidate(expected_revision, candidate)
        })?;
        apply_staged_schema_domains(store, domains);
        Ok(())
    })
}

fn publish_journaled_candidate(
    store_path: &'static str,
    store: StoreHandle,
    expected_revision: AcceptedSchemaRevision,
    candidate: &CandidateSchemaRevision,
    domains: StagedSchemaDomains,
    job_change: ConstraintValidationJobChange<'_>,
    application_record: Option<SchemaApplicationRecordOp>,
) -> 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_with_schema_application(
        marker_id,
        vec![batch.clone()],
        application_record.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);
        if let Some(operation) = application_record.as_ref() {
            apply_schema_application_record_op(operation)?;
        }
        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 {
            if publication.store.storage_capabilities().schema_metadata()
                == StoreSchemaMetadataCapability::LiveRebuiltMetadata
            {
                apply_live_schema_checkpoint(
                    publication.store_path,
                    publication.expected_revision,
                    publication.candidate,
                )?;
            }
        }
        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 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);
    }
}

#[cfg(test)]
mod tests {
    use super::{
        AcceptedSchemaPublication, publish_accepted_schema_candidate,
        publish_accepted_schema_candidates_with_application_record,
    };
    use crate::{
        db::{
            Db, EntityRegistration,
            commit::recovery::forget_recovered_domain_for_tests,
            commit::{
                CommitMarker, begin_commit, ensure_recovered, generate_commit_id,
                generate_marker_batch_id,
            },
            data::DataStore,
            index::IndexStore,
            journal::{JournalBatch, JournalRecord, JournalSequence},
            registry::{StoreAllocationIdentities, StoreRegistry, StoreRuntimeStorageCapabilities},
            schema::{
                AcceptedSchemaRevision, CandidateSchemaRevision, SchemaApplicationRecord,
                SchemaApplicationRecordOp, SchemaChangeOutcome, SchemaChangeReceipt, SchemaStore,
                empty_accepted_schema_candidate_for_tests, load_live_schema_checkpoint,
                with_schema_application_store,
            },
        },
        traits::{CanisterKind, Path},
    };
    use icydb_schema::{
        ExpectedAcceptedHead, ExpectedSchemaFingerprint, SchemaProposalDigest, SchemaSubmissionKey,
        TargetDatabaseIdentity,
    };
    use std::cell::RefCell;

    const COMPLETION_STORE_PATH: &str = "schema_publication_tests::CompletionHeap";
    const RECOVERY_STORE_PATH: &str = "schema_publication_tests::RecoveryHeap";

    thread_local! {
        static COMPLETION_DATA: RefCell<DataStore> =
            const { RefCell::new(DataStore::init_heap()) };
        static COMPLETION_INDEX: RefCell<IndexStore> =
            const { RefCell::new(IndexStore::init_heap()) };
        static COMPLETION_SCHEMA: RefCell<SchemaStore> =
            const { RefCell::new(SchemaStore::init_heap()) };
        static COMPLETION_REGISTRY: StoreRegistry = {
            let mut registry = StoreRegistry::new();
            registry.register_store(
                COMPLETION_STORE_PATH,
                &COMPLETION_DATA,
                &COMPLETION_INDEX,
                &COMPLETION_SCHEMA,
                StoreAllocationIdentities::absent(),
                StoreRuntimeStorageCapabilities::heap(),
            ).expect("completion heap store should register");
            registry
        };
        static RECOVERY_DATA: RefCell<DataStore> =
            const { RefCell::new(DataStore::init_heap()) };
        static RECOVERY_INDEX: RefCell<IndexStore> =
            const { RefCell::new(IndexStore::init_heap()) };
        static RECOVERY_SCHEMA: RefCell<SchemaStore> =
            const { RefCell::new(SchemaStore::init_heap()) };
        static RECOVERY_REGISTRY: StoreRegistry = {
            let mut registry = StoreRegistry::new();
            registry.register_store(
                RECOVERY_STORE_PATH,
                &RECOVERY_DATA,
                &RECOVERY_INDEX,
                &RECOVERY_SCHEMA,
                StoreAllocationIdentities::absent(),
                StoreRuntimeStorageCapabilities::heap(),
            ).expect("recovery heap store should register");
            registry
        };
    }

    struct CompletionCanister;

    impl Path for CompletionCanister {
        const PATH: &'static str = "schema_publication_tests::CompletionCanister";
    }

    impl CanisterKind for CompletionCanister {
        const COMMIT_MEMORY_ID: u8 = 240;
        const COMMIT_STABLE_KEY: &'static str =
            "icydb.test.schema-publication.completion.commit.v1";
        const INTEGRITY_PROGRESS_MEMORY_ID: u8 = 243;
        const INTEGRITY_PROGRESS_STABLE_KEY: &'static str =
            "icydb.test.schema-publication.completion.integrity.v1";
    }

    struct RecoveryCanister;

    impl Path for RecoveryCanister {
        const PATH: &'static str = "schema_publication_tests::RecoveryCanister";
    }

    impl CanisterKind for RecoveryCanister {
        const COMMIT_MEMORY_ID: u8 = 241;
        const COMMIT_STABLE_KEY: &'static str = "icydb.test.schema-publication.recovery.commit.v1";
        const INTEGRITY_PROGRESS_MEMORY_ID: u8 = 242;
        const INTEGRITY_PROGRESS_STABLE_KEY: &'static str =
            "icydb.test.schema-publication.recovery.integrity.v1";
    }

    fn applied_record(
        candidate: &CandidateSchemaRevision,
        discriminator: u8,
        submission: &str,
    ) -> SchemaApplicationRecord {
        let accepted_head = ExpectedAcceptedHead::Exact {
            revision: candidate.revision().get(),
            fingerprint: ExpectedSchemaFingerprint::from_bytes(
                candidate.root().fingerprint().as_bytes(),
            ),
        };
        let receipt = SchemaChangeReceipt::new(
            TargetDatabaseIdentity::from_bytes([discriminator; 32]),
            SchemaSubmissionKey::try_new(submission).expect("submission key should admit"),
            SchemaProposalDigest::from_bytes([discriminator.wrapping_add(1); 32]),
            ExpectedAcceptedHead::Empty,
            SchemaChangeOutcome::Applied { accepted_head },
        )
        .expect("applied receipt should admit");
        SchemaApplicationRecord::new(receipt, Vec::new())
            .expect("terminal application record should admit")
    }

    fn assert_candidate_and_record_published(
        store: crate::db::registry::StoreHandle,
        candidate: &CandidateSchemaRevision,
        record: &SchemaApplicationRecord,
    ) {
        let current = store
            .with_schema(SchemaStore::current_accepted_schema_bundle)
            .expect("heap schema should remain readable");
        assert_eq!(
            current.as_ref(),
            Some(candidate.bundle()),
            "accepted heap candidate must publish before success",
        );
        let checkpoint = load_live_schema_checkpoint(candidate.store_path())
            .expect("live accepted checkpoint should remain readable")
            .expect("live accepted checkpoint should exist");
        assert_eq!(checkpoint.encoded_bundle(), candidate.encoded_bundle());
        assert_eq!(checkpoint.encoded_root(), candidate.encoded_root());
        let loaded = with_schema_application_store(|store| {
            store.load(
                record.receipt().database_identity(),
                record.receipt().submission_key(),
            )
        })
        .expect("application record store should remain readable");
        assert_eq!(
            loaded.as_ref(),
            Some(record),
            "application receipt must describe the published candidate",
        );
    }

    #[test]
    fn marker_owned_application_publishes_one_live_only_store_and_receipt() {
        let db = Db::<CompletionCanister>::new_with_registrations(
            &COMPLETION_REGISTRY,
            &[] as &[EntityRegistration<CompletionCanister>],
        );
        ensure_recovered(&db).expect("test database format should initialize");
        let store = db
            .store_handle(COMPLETION_STORE_PATH)
            .expect("completion heap store should resolve");
        let candidate = empty_accepted_schema_candidate_for_tests(
            COMPLETION_STORE_PATH,
            AcceptedSchemaRevision::new(1),
        );
        let record = applied_record(&candidate, 0x51, "single-live-completion");
        let operation =
            SchemaApplicationRecordOp::insert(&record).expect("record insertion should prepare");

        publish_accepted_schema_candidates_with_application_record(
            vec![AcceptedSchemaPublication::new(
                COMPLETION_STORE_PATH,
                store,
                AcceptedSchemaRevision::NONE,
                &candidate,
            )],
            operation,
        )
        .expect("marker-owned live-only application should publish");

        assert_candidate_and_record_published(store, &candidate, &record);

        COMPLETION_DATA.with(|store| *store.borrow_mut() = DataStore::init_heap());
        COMPLETION_INDEX.with(|store| *store.borrow_mut() = IndexStore::init_heap());
        COMPLETION_SCHEMA.with(|store| *store.borrow_mut() = SchemaStore::init_heap());
        forget_recovered_domain_for_tests(&db).expect("upgrade should reset recovery ownership");
        ensure_recovered(&db).expect("checkpoint recovery should restore the live accepted schema");
        assert_candidate_and_record_published(store, &candidate, &record);

        let second = empty_accepted_schema_candidate_for_tests(
            COMPLETION_STORE_PATH,
            AcceptedSchemaRevision::new(2),
        );
        publish_accepted_schema_candidate(
            COMPLETION_STORE_PATH,
            store,
            AcceptedSchemaRevision::new(1),
            &second,
        )
        .expect("plain live-only publication should checkpoint");
        assert_candidate_and_record_published(store, &second, &record);

        COMPLETION_DATA.with(|store| *store.borrow_mut() = DataStore::init_heap());
        COMPLETION_INDEX.with(|store| *store.borrow_mut() = IndexStore::init_heap());
        COMPLETION_SCHEMA.with(|store| *store.borrow_mut() = SchemaStore::init_heap());
        forget_recovered_domain_for_tests(&db).expect("upgrade should reset recovery ownership");
        ensure_recovered(&db).expect("latest plain checkpoint should restore after upgrade");
        assert_candidate_and_record_published(store, &second, &record);
    }

    #[test]
    fn interrupted_live_only_application_recovers_candidate_and_receipt_from_marker() {
        let db = Db::<RecoveryCanister>::new_with_registrations(
            &RECOVERY_REGISTRY,
            &[] as &[EntityRegistration<RecoveryCanister>],
        );
        ensure_recovered(&db).expect("test database format should initialize");
        let store = db
            .store_handle(RECOVERY_STORE_PATH)
            .expect("recovery heap store should resolve");
        let initial = empty_accepted_schema_candidate_for_tests(
            RECOVERY_STORE_PATH,
            AcceptedSchemaRevision::new(1),
        );
        let initial_record = applied_record(&initial, 0x60, "single-live-initial");
        let initial_operation = SchemaApplicationRecordOp::insert(&initial_record)
            .expect("initial record insertion should prepare");
        publish_accepted_schema_candidates_with_application_record(
            vec![AcceptedSchemaPublication::new(
                RECOVERY_STORE_PATH,
                store,
                AcceptedSchemaRevision::NONE,
                &initial,
            )],
            initial_operation,
        )
        .expect("initial live-only application should publish");
        assert_candidate_and_record_published(store, &initial, &initial_record);

        let candidate = empty_accepted_schema_candidate_for_tests(
            RECOVERY_STORE_PATH,
            AcceptedSchemaRevision::new(2),
        );
        let record = applied_record(&candidate, 0x61, "single-live-recovery");
        let operation =
            SchemaApplicationRecordOp::insert(&record).expect("record insertion should prepare");
        let marker_id = generate_commit_id().expect("marker id should generate");
        let batch = JournalBatch::new(
            generate_marker_batch_id(marker_id, 0).expect("batch id should derive"),
            marker_id,
            JournalSequence::new(0),
            vec![
                JournalRecord::accepted_schema_publish(
                    RECOVERY_STORE_PATH,
                    AcceptedSchemaRevision::new(1),
                    candidate.encoded_bundle().to_vec(),
                    candidate.encoded_root().to_vec(),
                )
                .expect("schema journal record should admit"),
            ],
        )
        .expect("schema journal batch should admit");
        let marker = CommitMarker::from_parts_with_schema_application(
            marker_id,
            vec![batch],
            Some(operation),
        )
        .expect("application marker should admit");
        let _interrupted = begin_commit(marker).expect("marker should persist before interruption");

        RECOVERY_DATA.with(|store| *store.borrow_mut() = DataStore::init_heap());
        RECOVERY_INDEX.with(|store| *store.borrow_mut() = IndexStore::init_heap());
        RECOVERY_SCHEMA.with(|store| *store.borrow_mut() = SchemaStore::init_heap());
        assert!(
            store
                .with_schema(SchemaStore::current_accepted_schema_root)
                .expect("heap schema root should remain readable")
                .is_none(),
            "simulated upgrade must clear the live schema projection",
        );
        ensure_recovered(&db).expect("marker recovery should complete the application");

        assert_candidate_and_record_published(store, &candidate, &record);
        ensure_recovered(&db).expect("completed recovery should remain idempotent");
        assert_candidate_and_record_published(store, &candidate, &record);
    }
}