icydb-core 0.213.37

use crate::{
    db::{
        commit::{
            publish_accepted_schema_candidate,
            publish_accepted_schema_candidate_with_constraint_validation_job_removal,
        },
        registry::StoreHandle,
        schema::{
            AcceptedCatalogIdentity, AcceptedSchemaSnapshot, ConstraintActivationKind,
            ConstraintActivationState, ConstraintId, ConstraintOrigin, FieldId,
            PersistedFieldSnapshot, PersistedSchemaSnapshot, SchemaDdlAcceptedSnapshotDerivation,
            SchemaFieldDropTarget, SchemaFieldNullabilityTarget, SchemaFieldRenameTarget,
            SchemaInsertDefaultTarget, SchemaRowLayout,
            derive_sql_ddl_field_nullability_persisted_after,
        },
    },
    error::InternalError,
    types::EntityTag,
};

use super::{
    SqlDdlPublicationEnvelope,
    constraint::{candidate_with_snapshot, current_sql_ddl_bundle},
    require_exact_empty_sql_ddl_entity, validate_sql_ddl_drop_schema_gate,
};

/// Execute one SQL DDL field drop after proving no physical row requires the
/// dense-layout rewrite. Nonempty rewrites belong to the later migration
/// protocol and reject before publication in 0.209.
pub(in crate::db) fn execute_admin_sql_ddl_field_drop(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &'static str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    derivation: &SchemaDdlAcceptedSnapshotDerivation,
) -> Result<usize, InternalError> {
    let envelope = SqlDdlPublicationEnvelope::new(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        accepted_before_identity,
        derivation,
    );
    let Some(target) = derivation.admission().field_drop_target() else {
        return Err(InternalError::store_unsupported());
    };
    validate_sql_ddl_field_drop_metadata_change(
        entity_path,
        envelope.before(),
        envelope.after(),
        target,
    )?;
    validate_sql_ddl_drop_schema_gate(
        store,
        entity_tag,
        entity_path,
        envelope.before(),
        "before row rewrite",
    )?;
    require_exact_empty_sql_ddl_entity(store, entity_tag, entity_path)?;
    envelope.publish()?;

    Ok(0)
}

fn validate_sql_ddl_field_drop_metadata_change(
    _entity_path: &'static str,
    before: &PersistedSchemaSnapshot,
    after: &PersistedSchemaSnapshot,
    target: &SchemaFieldDropTarget,
) -> Result<(), InternalError> {
    if before.entity_path() != after.entity_path()
        || before.entity_name() != after.entity_name()
        || before.fields().len() != after.fields().len().saturating_add(1)
    {
        return Err(InternalError::store_unsupported());
    }

    let before_field = before
        .fields()
        .iter()
        .find(|field| field.id() == target.field_id())
        .ok_or_else(InternalError::store_unsupported)?;
    if before_field.name() != target.name() || before_field.slot() != target.slot() {
        return Err(InternalError::store_unsupported());
    }
    let target_name_remains = after
        .fields()
        .iter()
        .any(|field| field.name() == target.name());
    if target_name_remains {
        return Err(InternalError::store_unsupported());
    }
    if before.row_layout().slot_for_field(target.field_id()) != Some(target.slot()) {
        return Err(InternalError::store_unsupported());
    }

    let retained_fields = before
        .fields()
        .iter()
        .filter(|field| field.id() != target.field_id())
        .collect::<Vec<_>>();
    let identities = retained_fields
        .iter()
        .enumerate()
        .map(|(offset, field)| {
            let id = u32::try_from(offset)
                .ok()
                .and_then(|offset| offset.checked_add(1))
                .map(FieldId::new)
                .ok_or_else(InternalError::store_unsupported)?;
            Ok::<_, InternalError>((
                field.id(),
                id,
                crate::db::schema::SchemaFieldSlot::from_generated_index(offset)
                    .ok_or_else(InternalError::store_unsupported)?,
            ))
        })
        .collect::<Result<Vec<_>, _>>()?;
    let map_field = |field_id: FieldId, _slot| {
        identities
            .iter()
            .find(|(before_id, _, _)| *before_id == field_id)
            .map(|(_, after_id, after_slot)| (*after_id, *after_slot))
    };
    let expected_fields = retained_fields
        .iter()
        .zip(&identities)
        .map(|(field, (_, id, slot))| field.clone_for_full_layout_rewrite(*id, *slot))
        .collect::<Vec<_>>();
    let expected_layout = after.row_layout().clone();
    let expected_primary_key = before
        .primary_key_field_ids()
        .iter()
        .map(|field_id| map_field(*field_id, target.slot()).map(|(id, _)| id))
        .collect::<Option<Vec<_>>>()
        .ok_or_else(InternalError::store_unsupported)?;
    let expected_indexes = before
        .indexes()
        .iter()
        .map(|index| index.clone_with_dense_identities(index.ordinal(), map_field))
        .collect::<Option<Vec<_>>>()
        .ok_or_else(InternalError::store_unsupported)?;
    let expected_relations = before
        .relations()
        .iter()
        .map(|relation| {
            relation.clone_with_mapped_field_ids(|field_id| {
                map_field(field_id, target.slot()).map(|(id, _)| id)
            })
        })
        .collect::<Option<Vec<_>>>()
        .ok_or_else(InternalError::store_unsupported)?;
    if after.fields() != expected_fields
        || after.row_layout() != &expected_layout
        || after.primary_key_field_ids() != expected_primary_key
        || after.indexes() != expected_indexes
        || after.relations() != expected_relations
    {
        return Err(InternalError::store_unsupported());
    }

    Ok(())
}

/// Execute one metadata-only SQL DDL field-default publication.
pub(in crate::db) fn execute_admin_sql_ddl_field_default_change(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &'static str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    derivation: &SchemaDdlAcceptedSnapshotDerivation,
) -> Result<(), InternalError> {
    let envelope = SqlDdlPublicationEnvelope::new(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        accepted_before_identity,
        derivation,
    );
    execute_admin_sql_ddl_checked_field_metadata_publication(
        envelope,
        derivation.admission().field_default_target(),
        validate_sql_ddl_field_default_metadata_change,
    )
}

fn validate_sql_ddl_field_default_metadata_change(
    entity_path: &'static str,
    before: &PersistedSchemaSnapshot,
    after: &PersistedSchemaSnapshot,
    target: &SchemaInsertDefaultTarget,
) -> Result<(), InternalError> {
    validate_sql_ddl_single_field_metadata_change(
        entity_path,
        before,
        after,
        SqlDdlSingleFieldMetadataTarget {
            field_id: target.field_id(),
            before_name: target.name(),
            after_name: target.name(),
        },
        SqlDdlSingleFieldMetadataChange::Default,
    )
}

/// Result of one SQL DDL field-nullability lifecycle operation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum SqlDdlFieldNullabilityOutcome {
    /// A direct metadata publication or activation abort completed.
    Published,
    /// A new-write gate was published for bounded historical validation.
    ActivationPublished { constraint_id: ConstraintId },
}

/// Execute one SQL DDL field-nullability lifecycle operation.
pub(in crate::db) fn execute_admin_sql_ddl_field_nullability_change(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &'static str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    derivation: &SchemaDdlAcceptedSnapshotDerivation,
) -> Result<SqlDdlFieldNullabilityOutcome, InternalError> {
    let envelope = SqlDdlPublicationEnvelope::new(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        accepted_before_identity,
        derivation,
    );
    let Some(target) = derivation.admission().field_nullability_target() else {
        return Err(InternalError::store_unsupported());
    };
    validate_sql_ddl_field_nullability_metadata_change(
        entity_path,
        envelope.before(),
        envelope.after(),
        target,
    )?;

    if target_field_is_required(envelope.after(), target)? {
        return publish_sql_ddl_not_null_activation(
            envelope.store(),
            entity_tag,
            entity_path,
            accepted_before,
            accepted_before_identity,
            envelope.after().version(),
            target,
        );
    }

    envelope.publish()?;

    Ok(SqlDdlFieldNullabilityOutcome::Published)
}

fn publish_sql_ddl_not_null_activation(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &'static str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    next_schema_version: crate::db::schema::SchemaVersion,
    target: &SchemaFieldNullabilityTarget,
) -> Result<SqlDdlFieldNullabilityOutcome, 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 field = before
        .fields()
        .iter()
        .find(|field| field.id() == target.field_id() && field.nullable())
        .ok_or_else(InternalError::store_invariant)?;
    let catalog = before
        .constraint_catalog()
        .clone()
        .with_added_not_null_activation(field, current_fingerprint, activation_epoch)
        .map_err(|_| InternalError::store_unsupported())?;
    let after = before
        .clone()
        .with_constraint_catalog(catalog)
        .with_schema_version(next_schema_version);
    let constraint_id = after
        .constraint_catalog()
        .activations()
        .iter()
        .find(|activation| {
            matches!(
                activation.kind(),
                ConstraintActivationKind::NotNull { field_id } if *field_id == target.field_id()
            )
        })
        .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(SqlDdlFieldNullabilityOutcome::ActivationPublished { constraint_id })
}

#[expect(
    clippy::too_many_arguments,
    reason = "the abort boundary keeps accepted identity and exact activation state explicit"
)]
pub(in crate::db) fn execute_admin_sql_ddl_not_null_activation_abort(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &'static str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    next_schema_version: crate::db::schema::SchemaVersion,
    field_id: FieldId,
    constraint_id: ConstraintId,
) -> 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 = before
        .constraint_catalog()
        .activation(constraint_id)
        .filter(|activation| {
            activation.origin() == ConstraintOrigin::SqlDdl
                && matches!(
                    activation.kind(),
                    ConstraintActivationKind::NotNull {
                        field_id: activation_field_id
                    } if *activation_field_id == field_id
                )
        })
        .ok_or_else(InternalError::store_unsupported)?;
    let state = activation.state();
    let catalog = before
        .constraint_catalog()
        .clone()
        .with_aborted_activation(constraint_id)
        .map_err(|_| InternalError::store_invariant())?;
    let after = before
        .clone()
        .with_constraint_catalog(catalog)
        .with_schema_version(next_schema_version);
    let candidate = candidate_with_snapshot(&current, entity_tag, after)?;
    if state == ConstraintActivationState::Validating {
        publish_accepted_schema_candidate_with_constraint_validation_job_removal(
            accepted_before_identity.store_path(),
            store,
            current_revision,
            &candidate,
            entity_tag,
            constraint_id,
        )
    } else {
        publish_accepted_schema_candidate(
            accepted_before_identity.store_path(),
            store,
            current_revision,
            &candidate,
        )
    }
}

fn validate_sql_ddl_field_nullability_metadata_change(
    _entity_path: &'static str,
    before: &PersistedSchemaSnapshot,
    after: &PersistedSchemaSnapshot,
    target: &SchemaFieldNullabilityTarget,
) -> Result<(), InternalError> {
    let after_field = after
        .fields()
        .iter()
        .find(|field| field.id() == target.field_id())
        .ok_or_else(InternalError::store_unsupported)?;
    if after_field.name() != target.name() {
        return Err(InternalError::store_unsupported());
    }
    let expected = derive_sql_ddl_field_nullability_persisted_after(
        before,
        target.field_id(),
        after_field.nullable(),
        after.version(),
    )
    .map_err(|_| InternalError::store_unsupported())?;
    if &expected != after {
        return Err(InternalError::store_unsupported());
    }

    Ok(())
}

fn target_field_is_required(
    snapshot: &PersistedSchemaSnapshot,
    target: &SchemaFieldNullabilityTarget,
) -> Result<bool, InternalError> {
    snapshot
        .fields()
        .iter()
        .find(|field| field.id() == target.field_id())
        .map(|field| !field.nullable())
        .ok_or_else(InternalError::store_unsupported)
}

/// Execute one metadata-only SQL DDL field-rename publication.
pub(in crate::db) fn execute_admin_sql_ddl_field_rename(
    store: StoreHandle,
    entity_tag: EntityTag,
    entity_path: &'static str,
    accepted_before: &AcceptedSchemaSnapshot,
    accepted_before_identity: AcceptedCatalogIdentity,
    derivation: &SchemaDdlAcceptedSnapshotDerivation,
) -> Result<(), InternalError> {
    let envelope = SqlDdlPublicationEnvelope::new(
        store,
        entity_tag,
        entity_path,
        accepted_before,
        accepted_before_identity,
        derivation,
    );
    execute_admin_sql_ddl_checked_field_metadata_publication(
        envelope,
        derivation.admission().field_rename_target(),
        validate_sql_ddl_field_rename_metadata_change,
    )
}

fn validate_sql_ddl_field_rename_metadata_change(
    entity_path: &'static str,
    before: &PersistedSchemaSnapshot,
    after: &PersistedSchemaSnapshot,
    target: &SchemaFieldRenameTarget,
) -> Result<(), InternalError> {
    validate_sql_ddl_single_field_metadata_change(
        entity_path,
        before,
        after,
        SqlDdlSingleFieldMetadataTarget {
            field_id: target.field_id(),
            before_name: target.old_name(),
            after_name: target.new_name(),
        },
        SqlDdlSingleFieldMetadataChange::Rename,
    )?;

    let expected_indexes = before
        .indexes()
        .iter()
        .map(|index| {
            index.clone_with_renamed_field_path_root(
                target.field_id(),
                target.old_name(),
                target.new_name(),
            )
        })
        .collect::<Vec<_>>();
    if after.indexes() != expected_indexes {
        return Err(InternalError::store_unsupported());
    }

    Ok(())
}

#[derive(Clone, Copy)]
struct SqlDdlSingleFieldMetadataTarget<'a> {
    field_id: FieldId,
    before_name: &'a str,
    after_name: &'a str,
}

#[derive(Clone, Copy)]
enum SqlDdlSingleFieldMetadataChange {
    Default,
    Rename,
}

impl SqlDdlSingleFieldMetadataChange {
    const fn require_unchanged_indexes(self) -> bool {
        !matches!(self, Self::Rename)
    }

    fn target_field_matches_allowed_change(
        self,
        before_field: &PersistedFieldSnapshot,
        after_field: &PersistedFieldSnapshot,
    ) -> bool {
        match self {
            Self::Default => {
                before_field.clone_with_insert_default(after_field.insert_default().clone())
                    == *after_field
            }
            Self::Rename => {
                before_field.clone_with_name(after_field.name().to_string()) == *after_field
            }
        }
    }

    fn target_field_changed(
        self,
        before_field: &PersistedFieldSnapshot,
        after_field: &PersistedFieldSnapshot,
    ) -> bool {
        match self {
            Self::Default => before_field.insert_default() != after_field.insert_default(),
            Self::Rename => before_field.name() != after_field.name(),
        }
    }
}

fn validate_sql_ddl_single_field_metadata_change(
    _entity_path: &'static str,
    before: &PersistedSchemaSnapshot,
    after: &PersistedSchemaSnapshot,
    target: SqlDdlSingleFieldMetadataTarget<'_>,
    change: SqlDdlSingleFieldMetadataChange,
) -> Result<(), InternalError> {
    if before.entity_path() != after.entity_path()
        || before.entity_name() != after.entity_name()
        || before.primary_key_field_ids() != after.primary_key_field_ids()
        || !row_layout_allocation_matches(before.row_layout(), after.row_layout())
        || before.fields().len() != after.fields().len()
        || (change.require_unchanged_indexes() && before.indexes() != after.indexes())
    {
        return Err(InternalError::store_unsupported());
    }

    let mut changed = 0usize;
    for (before_field, after_field) in before.fields().iter().zip(after.fields()) {
        if before_field.id() == target.field_id {
            let field_id_drifted = after_field.id() != target.field_id;
            let before_name_drifted = before_field.name() != target.before_name;
            let after_name_drifted = after_field.name() != target.after_name;
            if field_id_drifted || before_name_drifted || after_name_drifted {
                return Err(InternalError::store_unsupported());
            }
            if !change.target_field_matches_allowed_change(before_field, after_field) {
                return Err(InternalError::store_unsupported());
            }
            if change.target_field_changed(before_field, after_field) {
                changed = changed.saturating_add(1);
            }
            continue;
        }

        if before_field != after_field {
            return Err(InternalError::store_unsupported());
        }
    }

    if changed != 1 {
        return Err(InternalError::store_unsupported());
    }

    Ok(())
}

fn row_layout_allocation_matches(left: &SchemaRowLayout, right: &SchemaRowLayout) -> bool {
    left == right
}

fn execute_admin_sql_ddl_checked_field_metadata_publication<T>(
    envelope: SqlDdlPublicationEnvelope<'_>,
    target: Option<&T>,
    validate: impl FnOnce(
        &'static str,
        &PersistedSchemaSnapshot,
        &PersistedSchemaSnapshot,
        &T,
    ) -> Result<(), InternalError>,
) -> Result<(), InternalError> {
    let Some(target) = target else {
        return Err(InternalError::store_unsupported());
    };
    validate(
        envelope.entity_path(),
        envelope.before(),
        envelope.after(),
        target,
    )?;

    envelope.publish()
}