icydb-core 0.98.1

//! Module: db::executor::tests::mutation_save
//! Covers save and mutation execution behavior in the executor layer.
//! Does not own: cross-module orchestration outside this module.
//! Boundary: exposes this module API while keeping implementation details internal.

use crate::{
    db::{
        Db, DbSession, EntityRuntimeHooks,
        codec::serialize_row_payload,
        commit::{
            CommitRowOp, commit_marker_present, ensure_recovered, init_commit_store_for_tests,
            prepare_row_commit_for_entity_with_structural_readers,
        },
        data::{
            CanonicalRow, DataKey, DataStore, RawRow, UpdatePatch,
            decode_persisted_custom_many_slot_payload, decode_persisted_scalar_slot_payload,
            encode_persisted_custom_many_slot_payload, encode_persisted_scalar_slot_payload,
        },
        executor::{
            DeleteExecutor, SaveExecutor,
            mutation::commit_window::{
                OpenCommitWindow, apply_prepared_row_ops, open_commit_window,
            },
        },
        index::IndexStore,
        predicate::MissingRowPolicy,
        query::intent::Query,
        registry::StoreRegistry,
        relation::{validate_delete_strong_relations_for_source, validate_save_strong_relations},
        schema::commit_schema_fingerprint_for_entity,
    },
    error::{ErrorClass, ErrorOrigin},
    metrics::{metrics_report, metrics_reset_all},
    model::{
        field::{FieldKind, FieldStorageDecode, RelationStrength},
        index::IndexModel,
    },
    testing::test_memory,
    traits::{EntityKind, EntitySchema, FieldValue, FieldValueKind, Path},
    types::{Account, Decimal, EntityTag, Id, Ulid},
    value::Value,
};
use icydb_derive::{FieldProjection, PersistedRow};
use serde::Deserialize;
use std::{
    cell::RefCell,
    collections::{BTreeMap, BTreeSet},
};

// TestCanister

crate::test_canister! {
    ident = TestCanister,
    commit_memory_id = crate::testing::test_commit_memory_id(),
}

// SourceStore

crate::test_store! {
    ident = SourceStore,
    canister = TestCanister,
}

// TargetStore

crate::test_store! {
    ident = TargetStore,
    canister = TestCanister,
}

const UNIQUE_INDEX_STORE_PATH: &str = SourceStore::PATH;

thread_local! {
    static SOURCE_DATA_STORE: RefCell<DataStore> =
        RefCell::new(DataStore::init(test_memory(0)));
    static TARGET_DATA_STORE: RefCell<DataStore> =
        RefCell::new(DataStore::init(test_memory(1)));
    static UNIQUE_INDEX_STORE: RefCell<IndexStore> =
        RefCell::new(IndexStore::init(test_memory(2)));
    static TARGET_INDEX_STORE: RefCell<IndexStore> =
        RefCell::new(IndexStore::init(test_memory(3)));
    static STORE_REGISTRY: StoreRegistry = {
        let mut reg = StoreRegistry::new();
        reg.register_store(SourceStore::PATH, &SOURCE_DATA_STORE, &UNIQUE_INDEX_STORE)
            .expect("source store registration should succeed");
        reg.register_store(TargetStore::PATH, &TARGET_DATA_STORE, &TARGET_INDEX_STORE)
            .expect("target store registration should succeed");
        reg
    };
}

fn with_data_store<R>(path: &'static str, f: impl FnOnce(&DataStore) -> R) -> R {
    DB.with_store_registry(|reg| reg.try_get_store(path).map(|store| store.with_data(f)))
        .expect("data store access should succeed")
}

fn with_data_store_mut<R>(path: &'static str, f: impl FnOnce(&mut DataStore) -> R) -> R {
    DB.with_store_registry(|reg| reg.try_get_store(path).map(|store| store.with_data_mut(f)))
        .expect("data store access should succeed")
}

fn with_index_store_mut<R>(path: &'static str, f: impl FnOnce(&mut IndexStore) -> R) -> R {
    DB.with_store_registry(|reg| reg.try_get_store(path).map(|store| store.with_index_mut(f)))
        .expect("index store access should succeed")
}

fn with_index_store<R>(path: &'static str, f: impl FnOnce(&IndexStore) -> R) -> R {
    DB.with_store_registry(|reg| reg.try_get_store(path).map(|store| store.with_index(f)))
        .expect("index store access should succeed")
}

// Clear test stores and ensure recovery has completed before each test mutation.
fn reset_store() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    ensure_recovered(&DB).expect("write-side recovery should succeed");
    with_data_store_mut(SourceStore::PATH, DataStore::clear);
    with_data_store_mut(TargetStore::PATH, DataStore::clear);
    with_index_store_mut(UNIQUE_INDEX_STORE_PATH, IndexStore::clear);
    with_index_store_mut(TargetStore::PATH, IndexStore::clear);
}

///
/// TargetEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct TargetEntity {
    id: Ulid,
}

crate::test_entity_schema! {
    ident = TargetEntity,
    id = Ulid,
    id_field = id,
    entity_name = "TargetEntity",
    entity_tag = crate::testing::TARGET_ENTITY_TAG,
    pk_index = 0,
    fields = [("id", FieldKind::Ulid)],
    indexes = [],
    store = TargetStore,
    canister = TestCanister,
}

///
/// SourceEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct SourceEntity {
    id: Ulid,
    target: Ulid,
}

crate::test_entity_schema! {
    ident = SourceEntity,
    id = Ulid,
    id_field = id,
    entity_name = "SourceEntity",
    entity_tag = crate::testing::SOURCE_ENTITY_TAG,
    pk_index = 0,
    fields = [
        ("id", FieldKind::Ulid),
        (
            "target",
            FieldKind::Relation {
                target_path: TargetEntity::PATH,
                target_entity_name:
                    <TargetEntity as crate::traits::EntitySchema>::MODEL.name(),
                target_entity_tag: TargetEntity::ENTITY_TAG,
                target_store_path: TargetStore::PATH,
                key_kind: &FieldKind::Ulid,
                strength: RelationStrength::Strong,
            }
        ),
    ],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

///
/// InvalidRelationMetadataEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct InvalidRelationMetadataEntity {
    id: Ulid,
    target: Ulid,
}

crate::test_entity_schema! {
    ident = InvalidRelationMetadataEntity,
    id = Ulid,
    id_field = id,
    entity_name = "InvalidRelationMetadataEntity",
    entity_tag = crate::testing::INVALID_RELATION_METADATA_ENTITY_TAG,
    pk_index = 0,
    fields = [
        ("id", FieldKind::Ulid),
        (
            "target",
            FieldKind::Relation {
                target_path: TargetEntity::PATH,
                target_entity_name: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                target_entity_tag: TargetEntity::ENTITY_TAG,
                target_store_path: TargetStore::PATH,
                key_kind: &FieldKind::Ulid,
                strength: RelationStrength::Strong,
            }
        ),
    ],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

///
/// SourceSetEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct SourceSetEntity {
    id: Ulid,
    targets: Vec<Ulid>,
}

static SOURCE_SET_TARGET_KIND: FieldKind = FieldKind::Relation {
    target_path: TargetEntity::PATH,
    target_entity_name: <TargetEntity as crate::traits::EntitySchema>::MODEL.name(),
    target_entity_tag: TargetEntity::ENTITY_TAG,
    target_store_path: TargetStore::PATH,
    key_kind: &FieldKind::Ulid,
    strength: RelationStrength::Strong,
};

crate::test_entity_schema! {
    ident = SourceSetEntity,
    id = Ulid,
    id_field = id,
    entity_name = "SourceSetEntity",
    entity_tag = crate::testing::SOURCE_SET_ENTITY_TAG,
    pk_index = 0,
    fields = [
        ("id", FieldKind::Ulid),
        ("targets", FieldKind::Set(&SOURCE_SET_TARGET_KIND)),
    ],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

///
/// SaveSelectedPart
///
/// SaveSelectedPart mirrors one nested record payload stored inside a queryable
/// collection field so save preflight can prove it accepts structured leaves
/// during typed writes instead of treating them as predicate literals.
///

#[derive(Clone, Debug, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd)]
struct SaveSelectedPart {
    layer_id: Ulid,
    part_id: Ulid,
}

impl FieldValue for SaveSelectedPart {
    fn kind() -> FieldValueKind {
        FieldValueKind::Structured { queryable: false }
    }

    fn to_value(&self) -> Value {
        Value::from_map(vec![
            (
                Value::Text("layer_id".to_string()),
                Value::Ulid(self.layer_id),
            ),
            (
                Value::Text("part_id".to_string()),
                Value::Ulid(self.part_id),
            ),
        ])
        .expect("selected part map should be canonical")
    }

    fn from_value(value: &Value) -> Option<Self> {
        let Value::Map(entries) = value else {
            return None;
        };
        let normalized = Value::normalize_map_entries(entries.clone()).ok()?;
        if normalized.len() != 2 {
            return None;
        }

        let layer_id = normalized
            .iter()
            .find_map(|(entry_key, entry_value)| match entry_key {
                Value::Text(entry_key) if entry_key == "layer_id" => Some(entry_value),
                _ => None,
            })?;
        let part_id = normalized
            .iter()
            .find_map(|(entry_key, entry_value)| match entry_key {
                Value::Text(entry_key) if entry_key == "part_id" => Some(entry_value),
                _ => None,
            })?;

        Some(Self {
            layer_id: Ulid::from_value(layer_id)?,
            part_id: Ulid::from_value(part_id)?,
        })
    }
}

///
/// StructuredSelectionEntity
///
/// StructuredSelectionEntity keeps one repeated record field on the executor
/// test surface so typed save preflight can exercise the same `Vec<Record>`
/// contract that application entities use for `selected_parts`.
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq)]
struct StructuredSelectionEntity {
    id: Ulid,
    selected_parts: Vec<SaveSelectedPart>,
}

const STRUCTURED_SELECTION_ENTITY_TAG: EntityTag = EntityTag::new(0x1034);

static STRUCTURED_SELECTED_PART_KIND: FieldKind = FieldKind::Structured { queryable: false };

crate::test_entity_schema! {
    ident = StructuredSelectionEntity,
    id = Ulid,
    id_field = id,
    entity_name = "StructuredSelectionEntity",
    entity_tag = STRUCTURED_SELECTION_ENTITY_TAG,
    pk_index = 0,
    fields = [
        ("id", FieldKind::Ulid),
        (
            "selected_parts",
            FieldKind::List(&STRUCTURED_SELECTED_PART_KIND),
            FieldStorageDecode::Value
        ),
    ],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

impl crate::db::PersistedRow for StructuredSelectionEntity {
    fn materialize_from_slots(
        slots: &mut dyn crate::db::SlotReader,
    ) -> Result<Self, crate::error::InternalError> {
        Ok(Self {
            id: match slots.get_bytes(0) {
                Some(bytes) => decode_persisted_scalar_slot_payload::<Ulid>(bytes, "id")?,
                None => return Err(crate::error::InternalError::missing_persisted_slot("id")),
            },
            selected_parts: match slots.get_bytes(1) {
                Some(bytes) => decode_persisted_custom_many_slot_payload::<SaveSelectedPart>(
                    bytes,
                    "selected_parts",
                )?,
                None => {
                    return Err(crate::error::InternalError::missing_persisted_slot(
                        "selected_parts",
                    ));
                }
            },
        })
    }

    fn write_slots(
        &self,
        out: &mut dyn crate::db::SlotWriter,
    ) -> Result<(), crate::error::InternalError> {
        let id_payload = encode_persisted_scalar_slot_payload(&self.id, "id")?;
        out.write_slot(0, Some(id_payload.as_slice()))?;

        let selected_parts_payload =
            encode_persisted_custom_many_slot_payload(&self.selected_parts, "selected_parts")?;
        out.write_slot(1, Some(selected_parts_payload.as_slice()))?;

        Ok(())
    }
}

fn load_structured_selection_entity(id: Ulid) -> Option<StructuredSelectionEntity> {
    let data_key = DataKey::try_new::<StructuredSelectionEntity>(id)
        .expect("structured selection data key should build")
        .to_raw()
        .expect("structured selection data key should encode");

    with_data_store(SourceStore::PATH, |data_store| {
        data_store.get(&data_key).map(|row| {
            row.try_decode::<StructuredSelectionEntity>()
                .expect("structured selection row decode should succeed")
        })
    })
}

///
/// SaveSelectedPartSet
///
/// SaveSelectedPartSet mirrors one generated set wrapper type so executor
/// tests can prove non-empty `Set<Record>` writes persist and reload through
/// the same typed save boundary as application entities.
///

#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq)]
struct SaveSelectedPartSet(BTreeSet<SaveSelectedPart>);

impl FieldValue for SaveSelectedPartSet {
    fn kind() -> FieldValueKind {
        FieldValueKind::Structured { queryable: true }
    }

    fn to_value(&self) -> Value {
        Value::List(self.0.iter().map(FieldValue::to_value).collect())
    }

    fn from_value(value: &Value) -> Option<Self> {
        let Value::List(values) = value else {
            return None;
        };

        let mut out = BTreeSet::new();
        for value in values {
            if !out.insert(SaveSelectedPart::from_value(value)?) {
                return None;
            }
        }

        Some(Self(out))
    }
}

///
/// StructuredSelectionSetEntity
///
/// StructuredSelectionSetEntity keeps one set-valued structured field on the
/// executor test surface so save coverage now includes `Set<Record>` as well
/// as `List<Record>` and `Map<..., Record>`.
///

#[derive(Clone, Debug, Default, Deserialize, Eq, FieldProjection, PartialEq)]
struct StructuredSelectionSetEntity {
    id: Ulid,
    selected_parts: SaveSelectedPartSet,
}

const STRUCTURED_SELECTION_SET_ENTITY_TAG: EntityTag = EntityTag::new(0x1036);

static STRUCTURED_SELECTION_SET_KIND: FieldKind = FieldKind::Set(&STRUCTURED_SELECTED_PART_KIND);

crate::test_entity_schema! {
    ident = StructuredSelectionSetEntity,
    id = Ulid,
    id_field = id,
    entity_name = "StructuredSelectionSetEntity",
    entity_tag = STRUCTURED_SELECTION_SET_ENTITY_TAG,
    pk_index = 0,
    fields = [
        ("id", FieldKind::Ulid),
        (
            "selected_parts",
            STRUCTURED_SELECTION_SET_KIND,
            FieldStorageDecode::Value
        ),
    ],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

impl crate::db::PersistedRow for StructuredSelectionSetEntity {
    fn materialize_from_slots(
        slots: &mut dyn crate::db::SlotReader,
    ) -> Result<Self, crate::error::InternalError> {
        Ok(Self {
            id: match slots.get_bytes(0) {
                Some(bytes) => decode_persisted_scalar_slot_payload::<Ulid>(bytes, "id")?,
                None => return Err(crate::error::InternalError::missing_persisted_slot("id")),
            },
            selected_parts: match slots.get_bytes(1) {
                Some(bytes) => crate::db::decode_persisted_custom_slot_payload::<
                    SaveSelectedPartSet,
                >(bytes, "selected_parts")?,
                None => {
                    return Err(crate::error::InternalError::missing_persisted_slot(
                        "selected_parts",
                    ));
                }
            },
        })
    }

    fn write_slots(
        &self,
        out: &mut dyn crate::db::SlotWriter,
    ) -> Result<(), crate::error::InternalError> {
        let id_payload = encode_persisted_scalar_slot_payload(&self.id, "id")?;
        out.write_slot(0, Some(id_payload.as_slice()))?;

        let selected_parts_payload = crate::db::encode_persisted_custom_slot_payload(
            &self.selected_parts,
            "selected_parts",
        )?;
        out.write_slot(1, Some(selected_parts_payload.as_slice()))?;

        Ok(())
    }
}

fn load_structured_selection_set_entity(id: Ulid) -> Option<StructuredSelectionSetEntity> {
    let data_key = DataKey::try_new::<StructuredSelectionSetEntity>(id)
        .expect("structured selection set data key should build")
        .to_raw()
        .expect("structured selection set data key should encode");

    with_data_store(SourceStore::PATH, |data_store| {
        data_store.get(&data_key).map(|row| {
            row.try_decode::<StructuredSelectionSetEntity>()
                .expect("structured selection set row decode should succeed")
        })
    })
}

///
/// SaveSelectedPartMap
///
/// SaveSelectedPartMap mirrors one generated map wrapper type so executor tests
/// can prove map-valued typed writes survive the full save path and persist
/// with the expected canonical `Value::Map` payload shape.
///

#[derive(Clone, Debug, Default, Deserialize, PartialEq)]
struct SaveSelectedPartMap(BTreeMap<Ulid, SaveSelectedPart>);

impl FieldValue for SaveSelectedPartMap {
    fn kind() -> FieldValueKind {
        FieldValueKind::Structured { queryable: false }
    }

    fn to_value(&self) -> Value {
        let mut entries = self
            .0
            .iter()
            .map(|(key, value)| (Value::Ulid(*key), value.to_value()))
            .collect::<Vec<_>>();

        Value::sort_map_entries_in_place(entries.as_mut_slice());

        Value::Map(entries)
    }

    fn from_value(value: &Value) -> Option<Self> {
        let Value::Map(entries) = value else {
            return None;
        };

        let normalized = Value::normalize_map_entries(entries.clone()).ok()?;
        let mut out = BTreeMap::new();
        for (entry_key, entry_value) in normalized {
            out.insert(
                Ulid::from_value(&entry_key)?,
                SaveSelectedPart::from_value(&entry_value)?,
            );
        }

        Some(Self(out))
    }
}

///
/// StructuredSelectionMapEntity
///
/// StructuredSelectionMapEntity keeps one map-valued field on the executor
/// test surface so save tests cover the actual write boundary used for
/// persisted `Map<..., ...>` fields.
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq)]
struct StructuredSelectionMapEntity {
    id: Ulid,
    selected_parts_by_layer: SaveSelectedPartMap,
}

const STRUCTURED_SELECTION_MAP_ENTITY_TAG: EntityTag = EntityTag::new(0x1035);

static STRUCTURED_SELECTION_MAP_KIND: FieldKind = FieldKind::Map {
    key: &FieldKind::Ulid,
    value: &STRUCTURED_SELECTED_PART_KIND,
};

crate::test_entity_schema! {
    ident = StructuredSelectionMapEntity,
    id = Ulid,
    id_field = id,
    entity_name = "StructuredSelectionMapEntity",
    entity_tag = STRUCTURED_SELECTION_MAP_ENTITY_TAG,
    pk_index = 0,
    fields = [
        ("id", FieldKind::Ulid),
        (
            "selected_parts_by_layer",
            STRUCTURED_SELECTION_MAP_KIND,
            FieldStorageDecode::Value
        ),
    ],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

impl crate::db::PersistedRow for StructuredSelectionMapEntity {
    fn materialize_from_slots(
        slots: &mut dyn crate::db::SlotReader,
    ) -> Result<Self, crate::error::InternalError> {
        Ok(Self {
            id: match slots.get_bytes(0) {
                Some(bytes) => decode_persisted_scalar_slot_payload::<Ulid>(bytes, "id")?,
                None => return Err(crate::error::InternalError::missing_persisted_slot("id")),
            },
            selected_parts_by_layer: match slots.get_bytes(1) {
                Some(bytes) => crate::db::decode_persisted_custom_slot_payload::<
                    SaveSelectedPartMap,
                >(bytes, "selected_parts_by_layer")?,
                None => {
                    return Err(crate::error::InternalError::missing_persisted_slot(
                        "selected_parts_by_layer",
                    ));
                }
            },
        })
    }

    fn write_slots(
        &self,
        out: &mut dyn crate::db::SlotWriter,
    ) -> Result<(), crate::error::InternalError> {
        let id_payload = encode_persisted_scalar_slot_payload(&self.id, "id")?;
        out.write_slot(0, Some(id_payload.as_slice()))?;

        let selected_parts_by_layer_payload = crate::db::encode_persisted_custom_slot_payload(
            &self.selected_parts_by_layer,
            "selected_parts_by_layer",
        )?;
        out.write_slot(1, Some(selected_parts_by_layer_payload.as_slice()))?;

        Ok(())
    }
}

fn load_structured_selection_map_entity(id: Ulid) -> Option<StructuredSelectionMapEntity> {
    let data_key = DataKey::try_new::<StructuredSelectionMapEntity>(id)
        .expect("structured selection map data key should build")
        .to_raw()
        .expect("structured selection map data key should encode");

    with_data_store(SourceStore::PATH, |data_store| {
        data_store.get(&data_key).map(|row| {
            row.try_decode::<StructuredSelectionMapEntity>()
                .expect("structured selection map row decode should succeed")
        })
    })
}

///
/// UniqueEmailEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct UniqueEmailEntity {
    id: Ulid,
    email: String,
}

static UNIQUE_EMAIL_INDEX_FIELDS: [&str; 1] = ["email"];
static UNIQUE_EMAIL_INDEX: IndexModel = IndexModel::generated(
    "save_tests::UniqueEmailEntity::email",
    UNIQUE_INDEX_STORE_PATH,
    &UNIQUE_EMAIL_INDEX_FIELDS,
    true,
);

crate::test_entity_schema! {
    ident = UniqueEmailEntity,
    id = Ulid,
    id_field = id,
    entity_name = "UniqueEmailEntity",
    entity_tag = crate::testing::UNIQUE_EMAIL_ENTITY_TAG,
    pk_index = 0,
    fields = [("id", FieldKind::Ulid), ("email", FieldKind::Text)],
    indexes = [&UNIQUE_EMAIL_INDEX],
    store = SourceStore,
    canister = TestCanister,
}

fn load_unique_email_entity(id: Ulid) -> Option<UniqueEmailEntity> {
    let data_key = DataKey::try_new::<UniqueEmailEntity>(id)
        .expect("unique email data key should build")
        .to_raw()
        .expect("unique email data key should encode");

    with_data_store(SourceStore::PATH, |data_store| {
        data_store.get(&data_key).map(|row| {
            row.try_decode::<UniqueEmailEntity>()
                .expect("unique email row decode should succeed")
        })
    })
}

fn unique_email_patch(id: Ulid, email: &str) -> UpdatePatch {
    UpdatePatch::new()
        .set_field(UniqueEmailEntity::MODEL, "id", Value::Ulid(id))
        .expect("resolve id slot")
        .set_field(
            UniqueEmailEntity::MODEL,
            "email",
            Value::Text(email.to_string()),
        )
        .expect("resolve email slot")
}

fn load_source_set_entity(id: Ulid) -> Option<SourceSetEntity> {
    let data_key = DataKey::try_new::<SourceSetEntity>(id)
        .expect("source-set data key should build")
        .to_raw()
        .expect("source-set data key should encode");

    with_data_store(SourceStore::PATH, |data_store| {
        data_store.get(&data_key).map(|row| {
            row.try_decode::<SourceSetEntity>()
                .expect("source-set row decode should succeed")
        })
    })
}

fn load_nullable_account_event_entity(id: Ulid) -> Option<NullableAccountEventEntity> {
    let data_key = DataKey::try_new::<NullableAccountEventEntity>(id)
        .expect("nullable account event data key should build")
        .to_raw()
        .expect("nullable account event data key should encode");

    with_data_store(SourceStore::PATH, |data_store| {
        data_store.get(&data_key).map(|row| {
            row.try_decode::<NullableAccountEventEntity>()
                .expect("nullable account event row decode should succeed")
        })
    })
}

///
/// MismatchedPkEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct MismatchedPkEntity {
    id: Ulid,
    actual_id: Ulid,
}

crate::test_entity_schema! {
    ident = MismatchedPkEntity,
    id = Ulid,
    id_field = actual_id,
    entity_name = "MismatchedPkEntity",
    entity_tag = crate::testing::MISMATCHED_PK_ENTITY_TAG,
    pk_index = 0,
    fields = [("id", FieldKind::Ulid), ("actual_id", FieldKind::Ulid)],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

///
/// DecimalScaleEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct DecimalScaleEntity {
    id: Ulid,
    #[icydb(scale = 2)]
    amount: Decimal,
}

crate::test_entity_schema! {
    ident = DecimalScaleEntity,
    id = Ulid,
    id_field = id,
    entity_name = "DecimalScaleEntity",
    entity_tag = crate::testing::DECIMAL_SCALE_ENTITY_TAG,
    pk_index = 0,
    fields = [
        ("id", FieldKind::Ulid),
        ("amount", FieldKind::Decimal { scale: 2 }),
    ],
    indexes = [],
    store = SourceStore,
    canister = TestCanister,
}

///
/// NullableAccountEventEntity
///

#[derive(Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow)]
struct NullableAccountEventEntity {
    id: Ulid,
    from: Option<Account>,
    to: Option<Account>,
}

crate::impl_test_entity_markers!(NullableAccountEventEntity);

crate::impl_test_entity_model_storage!(
    NullableAccountEventEntity,
    "NullableAccountEventEntity",
    0,
    fields = [
        crate::model::field::FieldModel::generated("id", FieldKind::Ulid),
        crate::model::field::FieldModel::generated_with_storage_decode_and_nullability(
            "from",
            FieldKind::Account,
            crate::model::field::FieldStorageDecode::ByKind,
            true,
        ),
        crate::model::field::FieldModel::generated_with_storage_decode_and_nullability(
            "to",
            FieldKind::Account,
            crate::model::field::FieldStorageDecode::ByKind,
            true,
        )
    ],
    indexes = [],
);

crate::impl_test_entity_runtime_surface!(
    NullableAccountEventEntity,
    Ulid,
    "NullableAccountEventEntity",
    MODEL_DEF
);

impl crate::traits::EntityPlacement for NullableAccountEventEntity {
    type Store = SourceStore;
    type Canister = TestCanister;
}

impl EntityKind for NullableAccountEventEntity {
    const ENTITY_TAG: EntityTag = crate::testing::NULLABLE_ACCOUNT_EVENT_ENTITY_TAG;
}

impl crate::traits::EntityValue for NullableAccountEventEntity {
    fn id(&self) -> Id<Self> {
        Id::from_key(self.id)
    }
}

static ENTITY_RUNTIME_HOOKS: &[EntityRuntimeHooks<TestCanister>] = &[
    EntityRuntimeHooks::new(
        TargetEntity::ENTITY_TAG,
        <TargetEntity as crate::traits::EntitySchema>::MODEL,
        TargetEntity::PATH,
        TargetStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<TargetEntity>,
        validate_delete_strong_relations_for_source::<TargetEntity>,
    ),
    EntityRuntimeHooks::new(
        SourceEntity::ENTITY_TAG,
        <SourceEntity as crate::traits::EntitySchema>::MODEL,
        SourceEntity::PATH,
        SourceStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<SourceEntity>,
        validate_delete_strong_relations_for_source::<SourceEntity>,
    ),
    EntityRuntimeHooks::new(
        InvalidRelationMetadataEntity::ENTITY_TAG,
        <InvalidRelationMetadataEntity as crate::traits::EntitySchema>::MODEL,
        InvalidRelationMetadataEntity::PATH,
        SourceStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<InvalidRelationMetadataEntity>,
        validate_delete_strong_relations_for_source::<InvalidRelationMetadataEntity>,
    ),
    EntityRuntimeHooks::new(
        SourceSetEntity::ENTITY_TAG,
        <SourceSetEntity as crate::traits::EntitySchema>::MODEL,
        SourceSetEntity::PATH,
        SourceStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<SourceSetEntity>,
        validate_delete_strong_relations_for_source::<SourceSetEntity>,
    ),
    EntityRuntimeHooks::new(
        UniqueEmailEntity::ENTITY_TAG,
        <UniqueEmailEntity as crate::traits::EntitySchema>::MODEL,
        UniqueEmailEntity::PATH,
        SourceStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<UniqueEmailEntity>,
        validate_delete_strong_relations_for_source::<UniqueEmailEntity>,
    ),
    EntityRuntimeHooks::new(
        MismatchedPkEntity::ENTITY_TAG,
        <MismatchedPkEntity as crate::traits::EntitySchema>::MODEL,
        MismatchedPkEntity::PATH,
        SourceStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<MismatchedPkEntity>,
        validate_delete_strong_relations_for_source::<MismatchedPkEntity>,
    ),
    EntityRuntimeHooks::new(
        DecimalScaleEntity::ENTITY_TAG,
        <DecimalScaleEntity as crate::traits::EntitySchema>::MODEL,
        DecimalScaleEntity::PATH,
        SourceStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<DecimalScaleEntity>,
        validate_delete_strong_relations_for_source::<DecimalScaleEntity>,
    ),
    EntityRuntimeHooks::new(
        NullableAccountEventEntity::ENTITY_TAG,
        <NullableAccountEventEntity as crate::traits::EntitySchema>::MODEL,
        NullableAccountEventEntity::PATH,
        SourceStore::PATH,
        prepare_row_commit_for_entity_with_structural_readers::<NullableAccountEventEntity>,
        validate_delete_strong_relations_for_source::<NullableAccountEventEntity>,
    ),
];

static DB: Db<TestCanister> = Db::new_with_hooks(&STORE_REGISTRY, ENTITY_RUNTIME_HOOKS);

#[test]
fn strong_relation_missing_fails_preflight() {
    let entity = SourceEntity {
        id: Ulid::generate(),
        target: Ulid::generate(), // non-existent target
    };

    let err = validate_save_strong_relations::<SourceEntity>(&DB, &entity)
        .expect_err("expected missing strong relation to fail");

    assert_eq!(
        err.class,
        ErrorClass::Unsupported,
        "missing strong relation should classify as unsupported",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "missing strong relation should originate from executor validation",
    );
    assert!(
        err.message.contains("strong relation missing"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn strong_relation_invalid_metadata_fails_internal() {
    let entity = InvalidRelationMetadataEntity {
        id: Ulid::generate(),
        target: Ulid::generate(),
    };

    let err = validate_save_strong_relations::<InvalidRelationMetadataEntity>(&DB, &entity)
        .expect_err("invalid relation metadata should fail deterministic preflight");
    assert_eq!(
        err.class,
        ErrorClass::Internal,
        "invalid relation metadata should classify as internal",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "invalid relation metadata should originate from executor boundary",
    );
    assert!(
        err.message.contains("strong relation target name invalid"),
        "unexpected error: {err:?}",
    );
}

#[test]
fn strong_set_relation_missing_key_fails_save() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let executor = SaveExecutor::<SourceSetEntity>::new(DB, false);
    let missing = Ulid::generate();
    let entity = SourceSetEntity {
        id: Ulid::generate(),
        targets: vec![missing],
    };

    let err = executor
        .insert(entity)
        .expect_err("missing set relation should fail");
    assert_eq!(
        err.class,
        ErrorClass::Unsupported,
        "missing set relation should classify as unsupported",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "missing set relation should originate from executor validation",
    );
    assert!(
        err.message.contains("strong relation missing"),
        "unexpected error: {err:?}"
    );

    let source_empty = with_data_store(SourceStore::PATH, |data_store| {
        data_store.iter().next().is_none()
    });
    assert!(
        source_empty,
        "source store must remain empty after failed save"
    );
}

#[test]
fn strong_set_relation_all_present_save_succeeds() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let target_save = SaveExecutor::<TargetEntity>::new(DB, false);
    let target_a = Ulid::generate();
    let target_b = Ulid::generate();
    target_save
        .insert(TargetEntity { id: target_a })
        .expect("target A save should succeed");
    target_save
        .insert(TargetEntity { id: target_b })
        .expect("target B save should succeed");

    let source_save = SaveExecutor::<SourceSetEntity>::new(DB, false);
    let saved = source_save
        .insert(SourceSetEntity {
            id: Ulid::generate(),
            targets: vec![target_a, target_b],
        })
        .expect("source save should succeed when all targets exist");

    assert!(saved.targets.contains(&target_a));
    assert!(saved.targets.contains(&target_b));
}

#[test]
fn save_accepts_non_empty_queryable_collection_of_structured_values() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let entity = StructuredSelectionEntity {
        id: Ulid::from_u128(70),
        selected_parts: vec![
            SaveSelectedPart {
                layer_id: Ulid::from_u128(701),
                part_id: Ulid::from_u128(702),
            },
            SaveSelectedPart {
                layer_id: Ulid::from_u128(703),
                part_id: Ulid::from_u128(704),
            },
        ],
    };

    let save = SaveExecutor::<StructuredSelectionEntity>::new(DB, false);
    let saved = save
        .insert(entity.clone())
        .expect("structured collection save should succeed");

    assert_eq!(saved, entity);
    assert_eq!(
        load_structured_selection_entity(entity.id),
        Some(entity),
        "structured collection save should persist the typed after-image",
    );
}

#[test]
fn save_accepts_set_with_structured_values() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let entity = StructuredSelectionSetEntity {
        id: Ulid::from_u128(75),
        selected_parts: SaveSelectedPartSet(
            [
                SaveSelectedPart {
                    layer_id: Ulid::from_u128(751),
                    part_id: Ulid::from_u128(752),
                },
                SaveSelectedPart {
                    layer_id: Ulid::from_u128(753),
                    part_id: Ulid::from_u128(754),
                },
            ]
            .into_iter()
            .collect(),
        ),
    };

    let save = SaveExecutor::<StructuredSelectionSetEntity>::new(DB, false);
    let saved = save
        .insert(entity.clone())
        .expect("structured set save should succeed");

    assert_eq!(saved, entity);
    assert_eq!(
        load_structured_selection_set_entity(entity.id),
        Some(entity),
        "structured set save should persist the typed after-image",
    );
}

#[test]
fn save_accepts_map_with_structured_values() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let mut selected_parts_by_layer = BTreeMap::new();
    selected_parts_by_layer.insert(
        Ulid::from_u128(801),
        SaveSelectedPart {
            layer_id: Ulid::from_u128(801),
            part_id: Ulid::from_u128(802),
        },
    );
    selected_parts_by_layer.insert(
        Ulid::from_u128(803),
        SaveSelectedPart {
            layer_id: Ulid::from_u128(803),
            part_id: Ulid::from_u128(804),
        },
    );

    let entity = StructuredSelectionMapEntity {
        id: Ulid::from_u128(80),
        selected_parts_by_layer: SaveSelectedPartMap(selected_parts_by_layer),
    };

    let save = SaveExecutor::<StructuredSelectionMapEntity>::new(DB, false);
    let saved = save
        .insert(entity.clone())
        .expect("structured map save should succeed");

    assert_eq!(saved, entity);
    assert_eq!(
        load_structured_selection_map_entity(entity.id),
        Some(entity),
        "structured map save should persist the typed after-image",
    );
}

#[test]
fn strong_set_relation_mixed_valid_invalid_fails_atomically() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let target_save = SaveExecutor::<TargetEntity>::new(DB, false);
    let valid = Ulid::generate();
    target_save
        .insert(TargetEntity { id: valid })
        .expect("valid target save should succeed");

    let invalid = Ulid::generate();
    let source_save = SaveExecutor::<SourceSetEntity>::new(DB, false);
    let err = source_save
        .insert(SourceSetEntity {
            id: Ulid::generate(),
            targets: vec![valid, invalid],
        })
        .expect_err("mixed valid/invalid set relation should fail");
    assert_eq!(
        err.class,
        ErrorClass::Unsupported,
        "missing strong relation in set should classify as unsupported",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "missing strong relation in set should originate from executor validation",
    );
    assert!(
        err.message.contains("strong relation missing"),
        "unexpected error: {err:?}"
    );

    let source_empty = with_data_store(SourceStore::PATH, |data_store| {
        data_store.iter().next().is_none()
    });
    assert!(
        source_empty,
        "source save must be atomic: failed save must not persist partial rows"
    );
}

#[test]
fn insert_many_atomic_rejects_partial_commit_on_late_failure() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<TargetEntity>::new(DB, false);
    let existing = Ulid::from_u128(41);
    save.insert(TargetEntity { id: existing })
        .expect("seed row insert should succeed");

    let new_id = Ulid::from_u128(42);
    let err = save
        .insert_many_atomic(vec![
            TargetEntity { id: new_id },
            TargetEntity { id: existing },
        ])
        .expect_err("atomic insert batch should fail on duplicate key");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "duplicate key should classify as conflict",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Store,
        "duplicate key should originate from store checks",
    );

    let rows = with_data_store(TargetStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(
        rows, 1,
        "atomic insert batch must not persist earlier rows when a later row fails"
    );
}

#[test]
fn insert_many_atomic_rejects_duplicate_keys_in_request() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<TargetEntity>::new(DB, false);
    let dup = Ulid::from_u128(47);
    let err = save
        .insert_many_atomic(vec![TargetEntity { id: dup }, TargetEntity { id: dup }])
        .expect_err("atomic insert batch should reject duplicate keys in one request");
    assert_eq!(
        err.class,
        ErrorClass::Unsupported,
        "duplicate key request should fail deterministic pre-commit validation",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "duplicate key request should fail at executor boundary",
    );
    assert!(
        err.message.contains("duplicate key"),
        "unexpected error: {err:?}",
    );

    let rows = with_data_store(TargetStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(
        rows, 0,
        "duplicate-key atomic batch must not persist any row"
    );
}

#[test]
fn insert_many_non_atomic_commits_prefix_before_late_failure() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<TargetEntity>::new(DB, false);
    let existing = Ulid::from_u128(51);
    save.insert(TargetEntity { id: existing })
        .expect("seed row insert should succeed");

    let new_id = Ulid::from_u128(52);
    let err = save
        .insert_many_non_atomic(vec![
            TargetEntity { id: new_id },
            TargetEntity { id: existing },
        ])
        .expect_err("non-atomic insert batch should fail on duplicate key");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "duplicate key should classify as conflict",
    );

    let rows = with_data_store(TargetStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(
        rows, 2,
        "non-atomic insert batch must preserve earlier committed rows before failure"
    );
}

#[test]
fn insert_many_empty_batch_is_noop_for_atomic_and_non_atomic_lanes() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<TargetEntity>::new(DB, false);
    let atomic = save
        .insert_many_atomic(Vec::<TargetEntity>::new())
        .expect("atomic empty batch should succeed");
    let non_atomic = save
        .insert_many_non_atomic(Vec::<TargetEntity>::new())
        .expect("non-atomic empty batch should succeed");

    assert!(
        atomic.is_empty(),
        "atomic empty batch should return no rows"
    );
    assert!(
        non_atomic.is_empty(),
        "non-atomic empty batch should return no rows",
    );

    let rows = with_data_store(TargetStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(rows, 0, "empty batches must not persist rows");
}

#[test]
fn commit_window_preflight_does_not_mutate_real_stores_before_apply() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let entity = UniqueEmailEntity {
        id: Ulid::from_u128(89),
        email: "preflight@example.com".to_string(),
    };
    let data_key = DataKey::try_new::<UniqueEmailEntity>(entity.id)
        .expect("data key should build for preflight test")
        .to_raw()
        .expect("data key should encode for preflight test");
    let row = CanonicalRow::from_entity(&entity)
        .expect("row encoding should succeed for preflight test")
        .into_raw_row();
    let row_op = CommitRowOp::new(
        UniqueEmailEntity::PATH,
        data_key,
        None,
        Some(row.as_bytes().to_vec()),
        commit_schema_fingerprint_for_entity::<UniqueEmailEntity>(),
    );

    let baseline_index_generation =
        with_index_store(UNIQUE_INDEX_STORE_PATH, IndexStore::generation);
    let baseline_index_len = with_index_store(UNIQUE_INDEX_STORE_PATH, IndexStore::len);

    let OpenCommitWindow {
        commit,
        prepared_row_ops,
        index_store_guards,
        ..
    } = open_commit_window::<UniqueEmailEntity>(&DB, vec![row_op])
        .expect("commit window open should succeed");

    assert!(
        commit_marker_present().expect("commit marker probe should succeed"),
        "open commit window should persist commit marker before apply",
    );
    assert!(
        load_unique_email_entity(entity.id).is_none(),
        "preflight must not persist data rows before apply",
    );
    assert_eq!(
        with_index_store(UNIQUE_INDEX_STORE_PATH, IndexStore::len),
        baseline_index_len,
        "preflight must not persist index rows before apply",
    );
    assert_eq!(
        with_index_store(UNIQUE_INDEX_STORE_PATH, IndexStore::generation),
        baseline_index_generation,
        "preflight must not mutate index generation before apply",
    );

    apply_prepared_row_ops(
        commit,
        "save_row_apply_preflight_purity_test",
        prepared_row_ops,
        index_store_guards,
        || {},
        || {},
    )
    .expect("apply should persist prepared row ops");

    assert!(
        load_unique_email_entity(entity.id).is_some(),
        "apply should persist the prepared row",
    );
    assert!(
        with_index_store(UNIQUE_INDEX_STORE_PATH, IndexStore::len) > baseline_index_len,
        "apply should persist index entry after preflight-only open",
    );
}

#[test]
fn commit_window_rejects_apply_when_index_store_generation_changes() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let entity = UniqueEmailEntity {
        id: Ulid::from_u128(90),
        email: "guard@example.com".to_string(),
    };
    let data_key = DataKey::try_new::<UniqueEmailEntity>(entity.id)
        .expect("data key should build for generation guard test")
        .to_raw()
        .expect("data key should encode for generation guard test");
    let row = CanonicalRow::from_entity(&entity)
        .expect("row encoding should succeed for generation guard test")
        .into_raw_row();
    let row_op = CommitRowOp::new(
        UniqueEmailEntity::PATH,
        data_key,
        None,
        Some(row.as_bytes().to_vec()),
        commit_schema_fingerprint_for_entity::<UniqueEmailEntity>(),
    );

    let OpenCommitWindow {
        commit,
        prepared_row_ops,
        index_store_guards,
        ..
    } = open_commit_window::<UniqueEmailEntity>(&DB, vec![row_op])
        .expect("commit window open should succeed");

    // Simulate cross-phase drift: preflight saw one generation, apply sees another.
    with_index_store_mut(UNIQUE_INDEX_STORE_PATH, IndexStore::clear);

    let err = apply_prepared_row_ops(
        commit,
        "save_row_apply_generation_guard_test",
        prepared_row_ops,
        index_store_guards,
        || {},
        || {},
    )
    .expect_err("generation mismatch must fail before apply");
    assert_eq!(
        err.class,
        ErrorClass::InvariantViolation,
        "generation mismatch should classify as invariant violation",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "generation mismatch should originate from executor apply invariants",
    );
    assert!(
        err.message
            .contains("index store generation changed between preflight and apply"),
        "unexpected error: {err:?}",
    );

    let persisted = load_unique_email_entity(entity.id);
    assert!(
        persisted.is_none(),
        "generation guard failure must prevent row persistence"
    );
}

#[test]
fn update_many_atomic_rejects_partial_commit_on_late_conflict() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let first = Ulid::from_u128(60);
    let second = Ulid::from_u128(61);
    save.insert(UniqueEmailEntity {
        id: first,
        email: "a@example.com".to_string(),
    })
    .expect("first seed row should save");
    save.insert(UniqueEmailEntity {
        id: second,
        email: "b@example.com".to_string(),
    })
    .expect("second seed row should save");

    let err = save
        .update_many_atomic(vec![
            UniqueEmailEntity {
                id: first,
                email: "carol@example.com".to_string(),
            },
            UniqueEmailEntity {
                id: second,
                email: "carol@example.com".to_string(),
            },
        ])
        .expect_err("atomic update batch should fail on unique index conflict");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "expected conflict error class",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Index,
        "expected index error origin",
    );

    let first_row = load_unique_email_entity(first).expect("first row should remain");
    let second_row = load_unique_email_entity(second).expect("second row should remain");
    assert_eq!(
        first_row.email, "a@example.com",
        "atomic update batch failure must not persist earlier updates",
    );
    assert_eq!(
        second_row.email, "b@example.com",
        "atomic update batch failure must not persist later updates",
    );
}

#[test]
fn update_many_non_atomic_commits_prefix_before_late_conflict() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let first = Ulid::from_u128(62);
    let second = Ulid::from_u128(63);
    save.insert(UniqueEmailEntity {
        id: first,
        email: "a@example.com".to_string(),
    })
    .expect("first seed row should save");
    save.insert(UniqueEmailEntity {
        id: second,
        email: "b@example.com".to_string(),
    })
    .expect("second seed row should save");

    let err = save
        .update_many_non_atomic(vec![
            UniqueEmailEntity {
                id: first,
                email: "carol@example.com".to_string(),
            },
            UniqueEmailEntity {
                id: second,
                email: "carol@example.com".to_string(),
            },
        ])
        .expect_err("non-atomic update batch should fail on unique index conflict");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "expected conflict error class",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Index,
        "expected index error origin",
    );

    let first_row = load_unique_email_entity(first).expect("first row should remain");
    let second_row = load_unique_email_entity(second).expect("second row should remain");
    assert_eq!(
        first_row.email, "carol@example.com",
        "non-atomic update batch should keep earlier committed updates",
    );
    assert_eq!(
        second_row.email, "b@example.com",
        "non-atomic update batch should leave later row unchanged on failure",
    );
}

#[test]
fn replace_many_atomic_mixed_existing_missing_rejects_partial_commit_on_conflict() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let existing = Ulid::from_u128(70);
    let missing = Ulid::from_u128(72);
    save.insert(UniqueEmailEntity {
        id: existing,
        email: "a@example.com".to_string(),
    })
    .expect("existing seed row should save");

    let err = save
        .replace_many_atomic(vec![
            UniqueEmailEntity {
                id: existing,
                email: "carol@example.com".to_string(),
            },
            UniqueEmailEntity {
                id: missing,
                email: "carol@example.com".to_string(),
            },
        ])
        .expect_err("atomic replace batch should fail on unique index conflict");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "expected conflict error class",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Index,
        "expected index error origin",
    );

    let existing_row = load_unique_email_entity(existing).expect("existing row should remain");
    assert_eq!(
        existing_row.email, "a@example.com",
        "atomic replace failure must not persist earlier replacements",
    );
    let missing_row = load_unique_email_entity(missing);
    assert!(
        missing_row.is_none(),
        "atomic replace failure must not insert missing-row replacement",
    );
}

#[test]
fn replace_many_non_atomic_mixed_existing_missing_commits_prefix_before_conflict() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let existing = Ulid::from_u128(73);
    let missing = Ulid::from_u128(74);
    save.insert(UniqueEmailEntity {
        id: existing,
        email: "a@example.com".to_string(),
    })
    .expect("existing seed row should save");

    let err = save
        .replace_many_non_atomic(vec![
            UniqueEmailEntity {
                id: existing,
                email: "carol@example.com".to_string(),
            },
            UniqueEmailEntity {
                id: missing,
                email: "carol@example.com".to_string(),
            },
        ])
        .expect_err("non-atomic replace batch should fail on unique index conflict");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "expected conflict error class",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Index,
        "expected index error origin",
    );

    let existing_row = load_unique_email_entity(existing).expect("existing row should remain");
    assert_eq!(
        existing_row.email, "carol@example.com",
        "non-atomic replace batch should keep earlier committed replacements",
    );
    let missing_row = load_unique_email_entity(missing);
    assert!(
        missing_row.is_none(),
        "failed non-atomic replacement should not persist the failing item",
    );
}

#[test]
fn insert_many_atomic_with_strong_relations_mixed_valid_invalid_fails_atomically() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let target_save = SaveExecutor::<TargetEntity>::new(DB, false);
    let valid_target = Ulid::from_u128(80);
    target_save
        .insert(TargetEntity { id: valid_target })
        .expect("valid target should save");

    let missing_target = Ulid::from_u128(81);
    let source_save = SaveExecutor::<SourceSetEntity>::new(DB, false);
    let err = source_save
        .insert_many_atomic(vec![
            SourceSetEntity {
                id: Ulid::from_u128(82),
                targets: vec![valid_target],
            },
            SourceSetEntity {
                id: Ulid::from_u128(83),
                targets: vec![valid_target, missing_target],
            },
        ])
        .expect_err("atomic relation batch should fail when one item has missing strong relation");
    assert_eq!(
        err.class,
        ErrorClass::Unsupported,
        "missing strong relation should classify as unsupported",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "missing strong relation should originate from executor validation",
    );
    assert!(
        err.message.contains("strong relation missing"),
        "unexpected error: {err:?}",
    );

    let source_rows = with_data_store(SourceStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(
        source_rows, 0,
        "atomic relation batch failure must not persist any source row",
    );
}

#[test]
fn update_many_atomic_with_strong_relations_mixed_valid_invalid_fails_atomically() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let target_save = SaveExecutor::<TargetEntity>::new(DB, false);
    let valid_a = Ulid::from_u128(84);
    let valid_b = Ulid::from_u128(85);
    target_save
        .insert(TargetEntity { id: valid_a })
        .expect("valid target A should save");
    target_save
        .insert(TargetEntity { id: valid_b })
        .expect("valid target B should save");

    let source_save = SaveExecutor::<SourceSetEntity>::new(DB, false);
    let first_id = Ulid::from_u128(86);
    let second_id = Ulid::from_u128(87);
    source_save
        .insert(SourceSetEntity {
            id: first_id,
            targets: vec![valid_a],
        })
        .expect("first source seed row should save");
    source_save
        .insert(SourceSetEntity {
            id: second_id,
            targets: vec![valid_a],
        })
        .expect("second source seed row should save");

    let missing_target = Ulid::from_u128(88);
    let err = source_save
        .update_many_atomic(vec![
            SourceSetEntity {
                id: first_id,
                targets: vec![valid_b],
            },
            SourceSetEntity {
                id: second_id,
                targets: vec![valid_b, missing_target],
            },
        ])
        .expect_err("atomic relation update batch should fail when one item has missing relation");
    assert_eq!(
        err.class,
        ErrorClass::Unsupported,
        "missing strong relation should classify as unsupported",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "missing strong relation should originate from executor validation",
    );

    let first_row = load_source_set_entity(first_id).expect("first source row should remain");
    let second_row = load_source_set_entity(second_id).expect("second source row should remain");
    assert_eq!(
        first_row.targets,
        vec![valid_a],
        "atomic relation update failure must not persist earlier updates",
    );
    assert_eq!(
        second_row.targets,
        vec![valid_a],
        "atomic relation update failure must not persist later updates",
    );
}

#[test]
fn replace_many_atomic_with_strong_relations_mixed_valid_invalid_fails_atomically() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let target_save = SaveExecutor::<TargetEntity>::new(DB, false);
    let valid_target = Ulid::from_u128(89);
    target_save
        .insert(TargetEntity { id: valid_target })
        .expect("valid target should save");

    let source_save = SaveExecutor::<SourceSetEntity>::new(DB, false);
    let existing_id = Ulid::from_u128(90);
    source_save
        .insert(SourceSetEntity {
            id: existing_id,
            targets: vec![valid_target],
        })
        .expect("existing source row should save");

    let missing_target = Ulid::from_u128(91);
    let inserted_id = Ulid::from_u128(92);
    let err = source_save
        .replace_many_atomic(vec![
            SourceSetEntity {
                id: existing_id,
                targets: vec![valid_target],
            },
            SourceSetEntity {
                id: inserted_id,
                targets: vec![valid_target, missing_target],
            },
        ])
        .expect_err("atomic relation replace batch should fail when one item has missing relation");
    assert_eq!(
        err.class,
        ErrorClass::Unsupported,
        "missing strong relation should classify as unsupported",
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Executor,
        "missing strong relation should originate from executor validation",
    );

    let existing_row =
        load_source_set_entity(existing_id).expect("existing source row should remain");
    assert_eq!(
        existing_row.targets,
        vec![valid_target],
        "atomic relation replace failure must not persist earlier replacements",
    );
    let inserted_row = load_source_set_entity(inserted_id);
    assert!(
        inserted_row.is_none(),
        "atomic relation replace failure must not insert later rows",
    );
}

#[test]
fn batch_lane_metrics_atomic_success_failure_and_non_atomic_partial_are_distinct() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();
    metrics_reset_all();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);

    // Atomic success: both rows commit, so both index inserts are counted.
    save.insert_many_atomic(vec![
        UniqueEmailEntity {
            id: Ulid::from_u128(93),
            email: "x@example.com".to_string(),
        },
        UniqueEmailEntity {
            id: Ulid::from_u128(94),
            email: "y@example.com".to_string(),
        },
    ])
    .expect("atomic insert batch should succeed");
    let after_atomic_success_report = metrics_report(None);
    let after_atomic_success = after_atomic_success_report
        .counters()
        .expect("metrics counters should exist after atomic success");
    assert_eq!(
        after_atomic_success.ops.save_calls, 1,
        "atomic batch success should count as one save execution call",
    );
    assert_eq!(
        after_atomic_success.ops.index_inserts, 2,
        "atomic success should emit index inserts for all committed rows",
    );

    // Atomic pre-commit failure: no index delta should be emitted.
    metrics_reset_all();
    let err = save
        .insert_many_atomic(vec![
            UniqueEmailEntity {
                id: Ulid::from_u128(95),
                email: "z@example.com".to_string(),
            },
            UniqueEmailEntity {
                id: Ulid::from_u128(95),
                email: "z@example.com".to_string(),
            },
        ])
        .expect_err("atomic duplicate-key batch should fail pre-commit");
    assert_eq!(err.class, ErrorClass::Unsupported);
    let after_atomic_failure_report = metrics_report(None);
    let after_atomic_failure = after_atomic_failure_report
        .counters()
        .expect("metrics counters should exist after atomic failure");
    assert_eq!(
        after_atomic_failure.ops.save_calls, 1,
        "atomic batch failure should still count as one attempted save execution call",
    );
    assert_eq!(
        after_atomic_failure.ops.index_inserts, 0,
        "atomic pre-commit failure must not emit index insert deltas",
    );
    assert_eq!(
        after_atomic_failure.ops.index_removes, 0,
        "atomic pre-commit failure must not emit index remove deltas",
    );

    // Non-atomic partial failure: successful prefix should emit index delta.
    metrics_reset_all();
    let existing = Ulid::from_u128(96);
    save.insert(UniqueEmailEntity {
        id: existing,
        email: "base@example.com".to_string(),
    })
    .expect("seed row should save");
    save.insert_many_non_atomic(vec![
        UniqueEmailEntity {
            id: Ulid::from_u128(97),
            email: "partial@example.com".to_string(),
        },
        UniqueEmailEntity {
            id: existing,
            email: "base@example.com".to_string(),
        },
    ])
    .expect_err("non-atomic batch should fail after prefix commit");
    let after_non_atomic_partial_report = metrics_report(None);
    let after_non_atomic_partial = after_non_atomic_partial_report
        .counters()
        .expect("metrics counters should exist after non-atomic partial failure");
    assert_eq!(
        after_non_atomic_partial.ops.save_calls, 2,
        "non-atomic batch should count one seed save plus one batch save call",
    );
    assert_eq!(
        after_non_atomic_partial.ops.index_inserts, 2,
        "non-atomic path should count seed insert + committed prefix insert",
    );
}

#[test]
fn set_field_encoding_requires_canonical_order_and_uniqueness() {
    let kind = FieldKind::Set(&FieldKind::Ulid);
    let lower = Value::Ulid(Ulid::from_u128(1));
    let higher = Value::Ulid(Ulid::from_u128(2));

    let err = SaveExecutor::<SourceSetEntity>::validate_deterministic_field_value(
        "targets",
        &kind,
        &Value::List(vec![higher, lower]),
    )
    .expect_err("unordered set encoding must fail");
    assert!(
        err.message
            .contains("set field must be strictly ordered and deduplicated"),
        "unexpected error: {err:?}"
    );

    let dup = Value::Ulid(Ulid::from_u128(7));
    let err = SaveExecutor::<SourceSetEntity>::validate_deterministic_field_value(
        "targets",
        &kind,
        &Value::List(vec![dup.clone(), dup]),
    )
    .expect_err("duplicate set entries must fail");
    assert!(
        err.message
            .contains("set field must be strictly ordered and deduplicated"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn map_field_encoding_requires_canonical_entry_order() {
    let kind = FieldKind::Map {
        key: &FieldKind::Text,
        value: &FieldKind::Uint,
    };
    let unordered = Value::Map(vec![
        (Value::Text("z".to_string()), Value::Uint(9u64)),
        (Value::Text("a".to_string()), Value::Uint(1u64)),
    ]);

    let err = SaveExecutor::<SourceSetEntity>::validate_deterministic_field_value(
        "settings", &kind, &unordered,
    )
    .expect_err("unordered map entries must fail");
    assert!(
        err.message
            .contains("map field entries are not in canonical deterministic order"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn save_rejects_primary_key_field_and_identity_mismatch() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let executor = SaveExecutor::<MismatchedPkEntity>::new(DB, false);
    let entity = MismatchedPkEntity {
        id: Ulid::from_u128(10),
        actual_id: Ulid::from_u128(20),
    };

    let err = executor
        .insert(entity)
        .expect_err("mismatched primary key identity should fail save");
    assert!(
        err.message.contains("entity primary key mismatch"),
        "unexpected error: {err:?}"
    );

    let source_empty = with_data_store(SourceStore::PATH, |data_store| {
        data_store.iter().next().is_none()
    });
    assert!(
        source_empty,
        "failed invariant checks must not persist rows"
    );
}

#[test]
fn unique_index_violation_rejected_on_insert() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    save.insert(UniqueEmailEntity {
        id: Ulid::from_u128(10),
        email: "alice@example.com".to_string(),
    })
    .expect("first unique insert should succeed");

    let err = save
        .insert(UniqueEmailEntity {
            id: Ulid::from_u128(11),
            email: "alice@example.com".to_string(),
        })
        .expect_err("duplicate unique index value should fail");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "expected conflict error class"
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Index,
        "expected index error origin"
    );
    assert!(
        err.message.contains("index constraint violation"),
        "unexpected error: {err:?}"
    );

    let rows = with_data_store(SourceStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(rows, 1, "conflicting insert must not persist");
}

#[test]
fn unique_index_row_key_mismatch_surfaces_store_invariant_violation() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let existing_id = Ulid::from_u128(510);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    save.insert(UniqueEmailEntity {
        id: existing_id,
        email: "alice@example.com".to_string(),
    })
    .expect("seed unique row should save");

    let raw_key = DataKey::try_new::<UniqueEmailEntity>(existing_id)
        .expect("existing key should build")
        .to_raw()
        .expect("existing key should encode");
    let mismatched_row = UniqueEmailEntity {
        id: Ulid::from_u128(511),
        email: "alice@example.com".to_string(),
    };
    let raw_row = CanonicalRow::from_entity(&mismatched_row)
        .expect("mismatched row should encode")
        .into_raw_row();
    with_data_store_mut(SourceStore::PATH, |data_store| {
        data_store.insert_raw_for_test(raw_key, raw_row);
    });

    let err = save
        .insert(UniqueEmailEntity {
            id: Ulid::from_u128(512),
            email: "alice@example.com".to_string(),
        })
        .expect_err("row-key mismatch should fail unique validation");
    assert_eq!(err.class, ErrorClass::Corruption);
    assert_eq!(err.origin, ErrorOrigin::Serialize);
    assert!(
        err.message
            .contains("failed to decode structural primary-key slot"),
        "row-key mismatch should fail through the canonical unique-validation structural decode lane: {err:?}"
    );
    assert!(
        err.message.contains("row key mismatch"),
        "row-key mismatch details should remain visible in the wrapped corruption: {err:?}"
    );
}

#[test]
fn decimal_scale_mixed_writes_reject_noncanonical_scale() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<DecimalScaleEntity>::new(DB, false);
    save.insert(DecimalScaleEntity {
        id: Ulid::from_u128(8101),
        amount: Decimal::new(123, 2),
    })
    .expect("canonical decimal scale should save");

    let err = save
        .insert(DecimalScaleEntity {
            id: Ulid::from_u128(8102),
            amount: Decimal::new(1234, 3),
        })
        .expect_err("mixed decimal scale write must be rejected");
    assert_eq!(err.class, ErrorClass::Unsupported);
    assert_eq!(err.origin, ErrorOrigin::Executor);
    assert!(
        err.message.contains("decimal field scale mismatch"),
        "unexpected error: {err:?}"
    );

    let rows = with_data_store(SourceStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(rows, 1, "rejected mixed-scale write must not persist");
}

#[test]
fn save_update_rejects_persisted_row_with_decimal_scale_drift() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let id = Ulid::from_u128(8201);
    let data_key = DataKey::try_new::<DecimalScaleEntity>(id)
        .expect("decimal entity key should build")
        .to_raw()
        .expect("decimal entity raw key should encode");
    let id_payload =
        encode_persisted_scalar_slot_payload(&id, "id").expect("id slot payload should encode");
    let amount_payload = crate::db::data::encode_structural_field_by_kind_bytes(
        crate::model::field::FieldKind::Decimal { scale: 3 },
        &crate::value::Value::Decimal(Decimal::new(1234, 3)),
        "amount",
    )
    .expect("amount slot payload should encode");
    let slot_payloads = [id_payload, amount_payload];
    let mut row_payload = Vec::new();
    let slot_count = u16::try_from(slot_payloads.len()).expect("slot count should fit in u16");
    row_payload.extend_from_slice(&slot_count.to_be_bytes());
    let mut payload_start = 0u32;
    for payload in &slot_payloads {
        let payload_len = u32::try_from(payload.len()).expect("payload length should fit in u32");
        row_payload.extend_from_slice(&payload_start.to_be_bytes());
        row_payload.extend_from_slice(&payload_len.to_be_bytes());
        payload_start = payload_start.saturating_add(payload_len);
    }
    for payload in &slot_payloads {
        row_payload.extend_from_slice(payload);
    }
    let raw_row =
        RawRow::try_new(serialize_row_payload(row_payload).expect("row payload should serialize"))
            .expect("malformed row bytes should satisfy row bound");
    with_data_store_mut(SourceStore::PATH, |data_store| {
        data_store.insert_raw_for_test(data_key, raw_row);
    });

    let save = SaveExecutor::<DecimalScaleEntity>::new(DB, false);
    let err = save
        .update(DecimalScaleEntity {
            id,
            amount: Decimal::new(123, 2),
        })
        .expect_err("decode path must reject persisted decimal scale drift");
    assert_eq!(err.class, ErrorClass::Corruption);
    assert_eq!(err.origin, ErrorOrigin::Store);
    assert!(
        err.message.contains("persisted row invariant violation"),
        "unexpected error: {err:?}"
    );
    assert!(
        err.message.contains("decimal field scale mismatch"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn unique_index_violation_rejected_on_update() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    save.insert(UniqueEmailEntity {
        id: Ulid::from_u128(20),
        email: "alice@example.com".to_string(),
    })
    .expect("first unique row should save");
    save.insert(UniqueEmailEntity {
        id: Ulid::from_u128(21),
        email: "bob@example.com".to_string(),
    })
    .expect("second unique row should save");

    let err = save
        .update(UniqueEmailEntity {
            id: Ulid::from_u128(21),
            email: "alice@example.com".to_string(),
        })
        .expect_err("update that collides with unique index should fail");
    assert_eq!(
        err.class,
        ErrorClass::Conflict,
        "expected conflict error class"
    );
    assert_eq!(
        err.origin,
        ErrorOrigin::Index,
        "expected index error origin"
    );
    assert!(
        err.message.contains("index constraint violation"),
        "unexpected error: {err:?}"
    );

    let rows = with_data_store(SourceStore::PATH, |data_store| data_store.iter().count());
    assert_eq!(rows, 2, "failed update must not remove persisted rows");
}

#[test]
fn structural_update_applies_patch_to_existing_row() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let id = Ulid::from_u128(22);
    save.insert(UniqueEmailEntity {
        id,
        email: "alice@example.com".to_string(),
    })
    .expect("seed unique row should save");

    let patched: UniqueEmailEntity = session
        .update_structural::<UniqueEmailEntity>(
            id,
            UpdatePatch::new()
                .set_field(
                    UniqueEmailEntity::MODEL,
                    "email",
                    Value::Text("grace@example.com".to_string()),
                )
                .expect("resolve email slot"),
        )
        .expect("structural update should succeed");

    assert_eq!(patched.id, id);
    assert_eq!(patched.email, "grace@example.com");

    let persisted = load_unique_email_entity(id).expect("row should remain after patch update");
    assert_eq!(persisted.email, "grace@example.com");
}

#[test]
fn structural_update_rejects_primary_key_mismatch() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let id = Ulid::from_u128(23);
    save.insert(UniqueEmailEntity {
        id,
        email: "alice@example.com".to_string(),
    })
    .expect("seed unique row should save");

    let err = session
        .update_structural::<UniqueEmailEntity>(
            id,
            UpdatePatch::new()
                .set_field(
                    UniqueEmailEntity::MODEL,
                    "id",
                    Value::Ulid(Ulid::from_u128(24)),
                )
                .expect("resolve id slot"),
        )
        .expect_err("primary key mismatch patch must fail");

    assert_eq!(err.class, ErrorClass::InvariantViolation);
    assert_eq!(err.origin, ErrorOrigin::Executor);
    assert!(
        err.message.contains("entity primary key mismatch"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn structural_update_builder_rejects_unknown_field_names() {
    let err = UpdatePatch::new()
        .set_field(
            UniqueEmailEntity::MODEL,
            "missing_email",
            Value::Text("grace@example.com".to_string()),
        )
        .expect_err("unknown structural field names must fail");

    assert_eq!(err.class, ErrorClass::InvariantViolation);
    assert_eq!(err.origin, ErrorOrigin::Executor);
    assert!(
        err.message.contains("mutation field not found"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn structural_insert_rejects_missing_required_fields_after_sparse_materialization() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let id = Ulid::from_u128(24);
    let err = session
        .insert_structural::<UniqueEmailEntity>(
            id,
            UpdatePatch::new()
                .set_field(UniqueEmailEntity::MODEL, "id", Value::Ulid(id))
                .expect("resolve id slot"),
        )
        .expect_err("structural insert without required fields must still fail");

    assert_eq!(err.class, ErrorClass::InvariantViolation);
    assert_eq!(err.origin, ErrorOrigin::Executor);
    assert!(
        err.message.contains("missing required field 'email'"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn structural_update_reuses_typed_unique_index_conflicts() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let first_id = Ulid::from_u128(25);
    let second_id = Ulid::from_u128(26);
    save.insert(UniqueEmailEntity {
        id: first_id,
        email: "alice@example.com".to_string(),
    })
    .expect("first unique row should save");
    save.insert(UniqueEmailEntity {
        id: second_id,
        email: "bob@example.com".to_string(),
    })
    .expect("second unique row should save");

    let err = session
        .update_structural::<UniqueEmailEntity>(
            first_id,
            UpdatePatch::new()
                .set_field(
                    UniqueEmailEntity::MODEL,
                    "email",
                    Value::Text("bob@example.com".to_string()),
                )
                .expect("resolve email slot"),
        )
        .expect_err("structural update that collides with unique index must fail");

    assert_eq!(err.class, ErrorClass::Conflict);
    assert_eq!(err.origin, ErrorOrigin::Index);
    assert!(
        err.message.contains("index constraint violation"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn structural_replace_rejects_missing_required_fields_after_sparse_materialization() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let id = Ulid::from_u128(27);
    save.insert(UniqueEmailEntity {
        id,
        email: "alice@example.com".to_string(),
    })
    .expect("seed unique row should save");

    let err = session
        .replace_structural::<UniqueEmailEntity>(
            id,
            UpdatePatch::new()
                .set_field(UniqueEmailEntity::MODEL, "id", Value::Ulid(id))
                .expect("resolve id slot"),
        )
        .expect_err("structural replace without required fields must still fail");

    assert_eq!(err.class, ErrorClass::InvariantViolation);
    assert_eq!(err.origin, ErrorOrigin::Executor);
    assert!(
        err.message.contains("missing required field 'email'"),
        "unexpected error: {err:?}"
    );
}

#[test]
fn structural_insert_matches_typed_insert_outcome() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let id = Ulid::from_u128(28);
    let inserted = session
        .insert_structural(
            id,
            UpdatePatch::new()
                .set_field(UniqueEmailEntity::MODEL, "id", Value::Ulid(id))
                .expect("resolve id slot")
                .set_field(
                    UniqueEmailEntity::MODEL,
                    "email",
                    Value::Text("grace@example.com".to_string()),
                )
                .expect("resolve email slot"),
        )
        .expect("structural insert should succeed");

    assert_eq!(
        inserted,
        UniqueEmailEntity {
            id,
            email: "grace@example.com".to_string(),
        }
    );

    let persisted = load_unique_email_entity(id).expect("inserted row should persist");
    assert_eq!(persisted, inserted);
}

#[test]
fn structural_insert_matches_typed_insert_parity() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let typed_id = Ulid::from_u128(280);
    let structural_id = Ulid::from_u128(281);
    let typed_email = "grace@example.com";
    let structural_email = "heidi@example.com";

    let typed = SaveExecutor::<UniqueEmailEntity>::new(DB, false)
        .insert(UniqueEmailEntity {
            id: typed_id,
            email: typed_email.to_string(),
        })
        .expect("typed create should succeed");
    let structural: UniqueEmailEntity = session
        .insert_structural(
            structural_id,
            unique_email_patch(structural_id, structural_email),
        )
        .expect("structural insert should succeed");

    assert_eq!(typed.email, typed_email);
    assert_eq!(structural.email, structural_email);
    assert_eq!(
        load_unique_email_entity(typed_id)
            .expect("typed row should persist")
            .email,
        typed_email
    );
    assert_eq!(
        load_unique_email_entity(structural_id)
            .expect("structural row should persist")
            .email,
        structural_email
    );
}

#[test]
fn structural_replace_matches_typed_replace_for_existing_rows() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let id = Ulid::from_u128(29);
    save.insert(UniqueEmailEntity {
        id,
        email: "alice@example.com".to_string(),
    })
    .expect("seed unique row should save");

    let replaced = session
        .replace_structural(
            id,
            UpdatePatch::new()
                .set_field(UniqueEmailEntity::MODEL, "id", Value::Ulid(id))
                .expect("resolve id slot")
                .set_field(
                    UniqueEmailEntity::MODEL,
                    "email",
                    Value::Text("grace@example.com".to_string()),
                )
                .expect("resolve email slot"),
        )
        .expect("structural replace should succeed");

    assert_eq!(
        replaced,
        UniqueEmailEntity {
            id,
            email: "grace@example.com".to_string(),
        }
    );

    let persisted = load_unique_email_entity(id).expect("replaced row should persist");
    assert_eq!(persisted, replaced);
}

#[test]
fn structural_update_matches_typed_update_parity() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let typed_id = Ulid::from_u128(290);
    let structural_id = Ulid::from_u128(291);
    let typed_email = "grace@example.com";
    let structural_email = "heidi@example.com";
    save.insert(UniqueEmailEntity {
        id: typed_id,
        email: "alice@example.com".to_string(),
    })
    .expect("typed seed row should save");
    save.insert(UniqueEmailEntity {
        id: structural_id,
        email: "bob@example.com".to_string(),
    })
    .expect("structural seed row should save");

    let typed = save
        .update(UniqueEmailEntity {
            id: typed_id,
            email: typed_email.to_string(),
        })
        .expect("typed update should succeed");
    let structural: UniqueEmailEntity = session
        .update_structural(
            structural_id,
            UpdatePatch::new()
                .set_field(
                    UniqueEmailEntity::MODEL,
                    "email",
                    Value::Text(structural_email.to_string()),
                )
                .expect("resolve email slot"),
        )
        .expect("structural update should succeed");

    assert_eq!(typed.email, typed_email);
    assert_eq!(structural.email, structural_email);
    assert_eq!(
        load_unique_email_entity(typed_id)
            .expect("typed row should persist")
            .email,
        typed_email
    );
    assert_eq!(
        load_unique_email_entity(structural_id)
            .expect("structural row should persist")
            .email,
        structural_email
    );
}

#[test]
fn structural_replace_matches_typed_replace_parity() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let typed_id = Ulid::from_u128(292);
    let structural_id = Ulid::from_u128(293);
    let typed_email = "grace@example.com";
    let structural_email = "heidi@example.com";
    save.insert(UniqueEmailEntity {
        id: typed_id,
        email: "alice@example.com".to_string(),
    })
    .expect("typed seed row should save");
    save.insert(UniqueEmailEntity {
        id: structural_id,
        email: "bob@example.com".to_string(),
    })
    .expect("structural seed row should save");

    let typed = save
        .replace(UniqueEmailEntity {
            id: typed_id,
            email: typed_email.to_string(),
        })
        .expect("typed replace should succeed");
    let structural: UniqueEmailEntity = session
        .replace_structural::<UniqueEmailEntity>(
            structural_id,
            unique_email_patch(structural_id, structural_email),
        )
        .expect("structural replace should succeed");

    assert_eq!(typed.email, typed_email);
    assert_eq!(structural.email, structural_email);
    assert_eq!(
        load_unique_email_entity(typed_id)
            .expect("typed row should persist")
            .email,
        typed_email
    );
    assert_eq!(
        load_unique_email_entity(structural_id)
            .expect("structural row should persist")
            .email,
        structural_email
    );
}

#[test]
fn structural_replace_inserts_missing_rows_with_sparse_after_image_when_required_fields_present() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let session = DbSession::new(DB);
    let id = Ulid::from_u128(30);
    let replaced = session
        .replace_structural(
            id,
            UpdatePatch::new()
                .set_field(UniqueEmailEntity::MODEL, "id", Value::Ulid(id))
                .expect("resolve id slot")
                .set_field(
                    UniqueEmailEntity::MODEL,
                    "email",
                    Value::Text("grace@example.com".to_string()),
                )
                .expect("resolve email slot"),
        )
        .expect("structural replace should insert missing rows");

    assert_eq!(
        replaced,
        UniqueEmailEntity {
            id,
            email: "grace@example.com".to_string(),
        }
    );

    let persisted = load_unique_email_entity(id).expect("replaced row should persist");
    assert_eq!(persisted, replaced);
}

#[test]
fn unique_index_update_same_pk_same_components_is_allowed() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let id = Ulid::from_u128(31);
    save.insert(UniqueEmailEntity {
        id,
        email: "alice@example.com".to_string(),
    })
    .expect("seed unique row should save");

    let updated = save
        .update(UniqueEmailEntity {
            id,
            email: "alice@example.com".to_string(),
        })
        .expect("update with same pk and identical unique components should succeed");
    assert_eq!(updated.id, id);
    assert_eq!(updated.email, "alice@example.com");

    let persisted = load_unique_email_entity(id).expect("row should remain after no-op update");
    assert_eq!(persisted.email, "alice@example.com");
}

#[test]
fn unique_index_delete_then_insert_same_value_succeeds() {
    init_commit_store_for_tests().expect("commit store init should succeed");
    reset_store();

    let save = SaveExecutor::<UniqueEmailEntity>::new(DB, false);
    let delete = DeleteExecutor::<UniqueEmailEntity>::new(DB);

    let original = Ulid::from_u128(40);
    save.insert(UniqueEmailEntity {
        id: original,
        email: "alice@example.com".to_string(),
    })
    .expect("seed unique row should save");

    let delete_plan = Query::<UniqueEmailEntity>::new(MissingRowPolicy::Ignore)
        .delete()
        .by_id(original)
        .plan()
        .map(crate::db::executor::PreparedExecutionPlan::from)
        .expect("delete plan should build");
    let deleted = delete
        .execute(delete_plan)
        .expect("delete should clear existing unique row");
    assert_eq!(deleted.len(), 1);

    let replacement = Ulid::from_u128(41);
    save.insert(UniqueEmailEntity {
        id: replacement,
        email: "alice@example.com".to_string(),
    })
    .expect("reinsert after delete should succeed for same unique value");

    let original_row = load_unique_email_entity(original);
    let replacement_row = load_unique_email_entity(replacement);
    assert!(original_row.is_none(), "deleted row should remain removed");
    assert!(
        replacement_row.is_some(),
        "replacement row should persist with reclaimed unique value"
    );
}

#[test]
fn save_executor_insert_allows_nullable_account_event_with_missing_from() {
    reset_store();

    let save = SaveExecutor::<NullableAccountEventEntity>::new(DB, false);
    let id = Ulid::from_u128(400);
    let account = Account::dummy(7);
    let saved = save
        .insert(NullableAccountEventEntity {
            id,
            from: None,
            to: Some(account),
        })
        .expect("mint-style nullable account event should save");

    assert_eq!(saved.from, None);
    assert_eq!(saved.to, Some(account));

    let persisted = load_nullable_account_event_entity(id)
        .expect("mint-style nullable account event should persist");
    assert_eq!(persisted, saved);
}

#[test]
fn save_executor_insert_allows_nullable_account_event_with_missing_to() {
    reset_store();

    let save = SaveExecutor::<NullableAccountEventEntity>::new(DB, false);
    let id = Ulid::from_u128(401);
    let account = Account::dummy(9);
    let saved = save
        .insert(NullableAccountEventEntity {
            id,
            from: Some(account),
            to: None,
        })
        .expect("burn-style nullable account event should save");

    assert_eq!(saved.from, Some(account));
    assert_eq!(saved.to, None);

    let persisted = load_nullable_account_event_entity(id)
        .expect("burn-style nullable account event should persist");
    assert_eq!(persisted, saved);
}