icydb-core 0.271.0

//! Module: schema::publication::constraint
//! Responsibility: SQL DDL constraint candidate construction and publication.
//! Does not own: SQL syntax binding or catalog persistence mechanics.
//! Boundary: applies bound constraint requests through accepted-schema transitions.

use crate::{
    db::{
        commit::publish_accepted_schema_candidate,
        registry::StoreHandle,
        schema::{
            AcceptedCatalogIdentity, AcceptedCheckExprV1, AcceptedSchemaRevision,
            AcceptedSchemaRevisionBundle, AcceptedSchemaSnapshot, ConstraintId, ConstraintOrigin,
            PersistedIndexSnapshot, SchemaVersion,
            publication::{candidate_with_snapshot, publish_sql_ddl_constraint_removal},
            validate_unpublished_check_candidate_exact,
        },
    },
    error::InternalError,
    types::EntityTag,
};

/// Accepted-catalog-bound `ADD CONSTRAINT ... CHECK` request.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct SchemaAddCheckRequest {
    pub(in crate::db) entity_name: String,
    pub(in crate::db) constraint_name: String,
    pub(in crate::db) expression: AcceptedCheckExprV1,
    pub(in crate::db) not_valid: bool,
}

impl SchemaAddCheckRequest {
    #[must_use]
    pub(in crate::db) const fn entity_name(&self) -> &str {
        self.entity_name.as_str()
    }

    #[must_use]
    pub(in crate::db) const fn constraint_name(&self) -> &str {
        self.constraint_name.as_str()
    }

    #[must_use]
    pub(in crate::db) const fn expression(&self) -> &AcceptedCheckExprV1 {
        &self.expression
    }

    #[must_use]
    pub(in crate::db) const fn not_valid(&self) -> bool {
        self.not_valid
    }
}

/// Accepted-catalog-bound check removal request.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct SchemaDropConstraintRequest {
    pub(in crate::db) entity_name: String,
    pub(in crate::db) constraint_name: String,
    pub(in crate::db) constraint_id: ConstraintId,
    pub(in crate::db) activation: bool,
}

impl SchemaDropConstraintRequest {
    #[must_use]
    pub(in crate::db) const fn entity_name(&self) -> &str {
        self.entity_name.as_str()
    }

    #[must_use]
    pub(in crate::db) const fn constraint_name(&self) -> &str {
        self.constraint_name.as_str()
    }

    #[must_use]
    pub(in crate::db) const fn constraint_id(&self) -> ConstraintId {
        self.constraint_id
    }

    #[must_use]
    pub(in crate::db) const fn is_activation(&self) -> bool {
        self.activation
    }
}

/// Publish one SQL-DDL check as either a write-gated activation or an exactly
/// validated accepted constraint.
pub(in crate::db) fn execute_schema_check_addition(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    request: &SchemaAddCheckRequest,
    next_schema_version: SchemaVersion,
) -> Result<(usize, crate::db::schema::ConstraintId), InternalError> {
    let (current_revision, current_fingerprint, current) = current_sql_ddl_bundle(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        &accepted_before_identity,
    )?;
    let activation_epoch = current_revision
        .checked_next()
        .ok_or_else(InternalError::store_unsupported)?
        .get();
    let before = current
        .entity_snapshots()
        .get(&entity_tag)
        .ok_or_else(InternalError::store_corruption)?;
    let catalog = before
        .constraint_catalog()
        .clone()
        .with_added_check_activation(
            request.constraint_name().to_string(),
            ConstraintOrigin::SqlDdl,
            request.expression().clone(),
            current_fingerprint,
            activation_epoch,
        )
        .map_err(|_| InternalError::store_unsupported())?;
    let activation_snapshot = before
        .clone()
        .with_constraint_catalog(catalog)
        .with_schema_version(next_schema_version);
    let activation_id = activation_snapshot
        .constraint_catalog()
        .activations()
        .iter()
        .find(|activation| activation.name() == request.constraint_name())
        .map(crate::db::schema::ConstraintActivationSnapshot::id)
        .ok_or_else(InternalError::store_invariant)?;
    let activation_candidate = candidate_with_snapshot(&current, entity_tag, &activation_snapshot)?;

    if request.not_valid() {
        publish_accepted_schema_candidate(
            accepted_before_identity.store_path(),
            store,
            current_revision,
            &activation_candidate,
        )?;
        return Ok((0, activation_id));
    }

    let rows_scanned = validate_unpublished_check_candidate_exact(
        store,
        accepted_before_identity.store_path(),
        entity_tag,
        entity_path,
        &activation_candidate,
        activation_id,
    )?;
    let validated_catalog = activation_snapshot
        .constraint_catalog()
        .clone()
        .with_directly_validated_activation(activation_id)
        .map_err(|_| InternalError::store_invariant())?;
    let validated = activation_snapshot.with_constraint_catalog(validated_catalog);
    let validated_candidate = candidate_with_snapshot(&current, entity_tag, &validated)?;
    publish_accepted_schema_candidate(
        accepted_before_identity.store_path(),
        store,
        current_revision,
        &validated_candidate,
    )?;
    Ok((rows_scanned, activation_id))
}

/// Reserve one SQL-DDL unique index as a planner-invisible activation.
pub(in crate::db) fn execute_schema_unique_index_activation(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    candidate_index: &PersistedIndexSnapshot,
    next_schema_version: SchemaVersion,
) -> Result<crate::db::schema::ConstraintId, InternalError> {
    if !candidate_index.unique() {
        return Err(InternalError::store_invariant());
    }
    let (current_revision, current_fingerprint, current) = current_sql_ddl_bundle(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        &accepted_before_identity,
    )?;
    let activation_epoch = current_revision
        .checked_next()
        .ok_or_else(InternalError::store_unsupported)?
        .get();
    let before = current
        .entity_snapshots()
        .get(&entity_tag)
        .ok_or_else(InternalError::store_corruption)?;
    let candidate = candidate_index.clone_with_schema_identity(
        candidate_index.schema_id(),
        candidate_index.ordinal(),
        activation_epoch,
    );
    let after = before
        .clone()
        .with_added_unique_activation(candidate, current_fingerprint, activation_epoch)
        .map_err(|_| InternalError::store_unsupported())?
        .with_schema_version(next_schema_version);
    let activation_id = after
        .constraint_catalog()
        .activations()
        .iter()
        .find(|activation| activation.name() == candidate_index.name())
        .map(crate::db::schema::ConstraintActivationSnapshot::id)
        .ok_or_else(InternalError::store_invariant)?;
    let candidate = candidate_with_snapshot(&current, entity_tag, &after)?;
    publish_accepted_schema_candidate(
        accepted_before_identity.store_path(),
        store,
        current_revision,
        &candidate,
    )?;
    Ok(activation_id)
}

/// Abort one SQL-DDL unique-index activation and remove its isolated owner.
pub(in crate::db) fn execute_schema_unique_index_activation_abort(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    removal: (ConstraintId, &PersistedIndexSnapshot),
    next_schema_version: SchemaVersion,
) -> Result<(), InternalError> {
    let (constraint_id, dropped_index) = removal;
    let (current_revision, _current_fingerprint, current) = current_sql_ddl_bundle(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        &accepted_before_identity,
    )?;
    let before = current
        .entity_snapshots()
        .get(&entity_tag)
        .ok_or_else(InternalError::store_corruption)?;
    if !before.candidate_indexes().contains(dropped_index) {
        return Err(InternalError::schema_ddl_publication_race_lost());
    }
    let activation_state = before
        .constraint_catalog()
        .activation(constraint_id)
        .map(crate::db::schema::ConstraintActivationSnapshot::state)
        .ok_or_else(InternalError::schema_ddl_publication_race_lost)?;
    let after = before
        .with_aborted_unique_activation(constraint_id, next_schema_version)
        .map_err(|_| InternalError::store_invariant())?;
    let candidate = candidate_with_snapshot(&current, entity_tag, &after)?;
    publish_sql_ddl_constraint_removal(
        store,
        &accepted_before_identity,
        current_revision,
        &candidate,
        entity_tag,
        constraint_id,
        Some(activation_state),
    )
}

/// Drop or abort one SQL-DDL-owned check through the same marker-owned schema
/// publication boundary as activation progress.
pub(in crate::db) fn execute_schema_check_drop(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    request: &SchemaDropConstraintRequest,
    next_schema_version: SchemaVersion,
) -> Result<(), InternalError> {
    let (current_revision, _current_fingerprint, current) = current_sql_ddl_bundle(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        &accepted_before_identity,
    )?;
    let before = current
        .entity_snapshots()
        .get(&entity_tag)
        .ok_or_else(InternalError::store_corruption)?;
    let activation_state = before
        .constraint_catalog()
        .activation(request.constraint_id())
        .map(crate::db::schema::ConstraintActivationSnapshot::state);
    if request.is_activation() != activation_state.is_some() {
        return Err(InternalError::schema_ddl_publication_race_lost());
    }
    let catalog = if request.is_activation() {
        before
            .constraint_catalog()
            .clone()
            .with_aborted_activation(request.constraint_id())
    } else {
        before
            .constraint_catalog()
            .clone()
            .with_removed_sql_ddl_check(request.constraint_id())
    }
    .map_err(|_| InternalError::store_unsupported())?;
    let after = before
        .clone()
        .with_constraint_catalog(catalog)
        .with_schema_version(next_schema_version);
    let candidate = candidate_with_snapshot(&current, entity_tag, &after)?;
    publish_sql_ddl_constraint_removal(
        store,
        &accepted_before_identity,
        current_revision,
        &candidate,
        entity_tag,
        request.constraint_id(),
        activation_state,
    )
}

pub(super) fn current_sql_ddl_bundle(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &str,
    accepted_before: &AcceptedSchemaSnapshot,
    expected_identity: &AcceptedCatalogIdentity,
) -> Result<
    (
        AcceptedSchemaRevision,
        crate::db::schema::AcceptedSchemaFingerprint,
        AcceptedSchemaRevisionBundle,
    ),
    InternalError,
> {
    let (root, selection, bundle) = store.with_schema(|schema_store| {
        Ok::<_, InternalError>((
            schema_store
                .current_accepted_schema_root()?
                .ok_or_else(InternalError::store_corruption)?,
            schema_store
                .current_accepted_catalog_selection(
                    entity_tag,
                    entity_path,
                    expected_identity.store_path(),
                )?
                .ok_or_else(InternalError::store_corruption)?,
            schema_store
                .current_accepted_schema_bundle()?
                .ok_or_else(InternalError::store_corruption)?,
        ))
    })?;
    if selection.identity() != *expected_identity
        || bundle.store_path() != expected_identity.store_path()
        || bundle.entity_snapshots().get(&entity_tag) != Some(accepted_before.persisted_snapshot())
    {
        return Err(InternalError::schema_ddl_publication_race_lost());
    }
    Ok((root.root().revision(), root.root().fingerprint(), bundle))
}