myko 6.5.1

//! Cell-based `RelationshipManager` for handling entity relationship cascades.
//!
//! This module handles cascade operations based on relationships registered via
//! `#[belongs_to]`, `#[owns_many]`, and `#[ensure_for]` attribute macros.
//!
//! Uses `MykoServerContext` for queries and event publishing, keeping this module
//! decoupled from direct store and event processor access.
//!
//! # Relationship Types
//!
//! ## `BelongsTo` (Foreign Key)
//!
//! A child entity has a foreign key pointing to a parent. When the parent is deleted,
//! all children with matching foreign keys are cascade-deleted.
//!
//! ```text
//! use myko::prelude::*;
//! use std::sync::Arc;
//!
//! #[myko_item]
//! pub struct Scene {
//!     pub name: String,
//! }
//!
//! #[myko_item]
//! pub struct Binding {
//!     #[belongs_to(Scene)]
//!     pub scope_id: Arc<str>,
//! }
//! ```
//!
//! ## `OwnsMany` (Parent has array of child IDs)
//!
//! A parent entity owns an array of child IDs. Deleting the parent deletes all children.
//! Deleting a child removes its ID from the parent's array.
//!
//! ```text
//! use myko::prelude::*;
//! use std::sync::Arc;
//!
//! #[myko_item]
//! pub struct BindingNode {
//!     pub name: String,
//! }
//!
//! #[myko_item]
//! pub struct Scene {
//!     #[owns_many(BindingNode)]
//!     pub node_ids: Vec<Arc<str>>,
//! }
//! ```
//!
//! ## `EnsureFor` (Auto-create for combinations)
//!
//! Automatically create one entity for each combination of dependency entities.
//!
//! ```text
//! use myko::prelude::*;
//! use std::sync::Arc;
//!
//! #[myko_item]
//! pub struct Session {
//!     pub name: String,
//! }
//!
//! #[myko_item]
//! pub struct Bundle {
//!     pub name: String,
//! }
//!
//! #[myko_item]
//! pub struct BundleStatus {
//!     #[ensure_for(Session)]
//!     pub session_id: Arc<str>,
//!     #[ensure_for(Bundle)]
//!     pub bundle_id: Arc<str>,
//! }
//! ```

use std::{
    collections::{BTreeSet, HashMap, HashSet},
    sync::{Arc, Mutex},
};

use dashmap::DashMap;
use hyphae::{Gettable, Materialize};
use tracing::{debug, info, trace};

use super::{MykoServerContext, persister::PersistError};
use crate::{
    core::item::AnyItem,
    relationship::{
        ArrayExtractor, ArrayRemover, EnsureForDependency, EntityFactory, FkExtractor, Relation,
        iter_relations,
    },
};

type EnsureForReservation = (&'static str, Vec<Arc<str>>);

/// Lookup info for `BelongsTo` cascades
#[derive(Clone)]
struct BelongsToLookup {
    id: u64,
    local_type: &'static str,
    foreign_type: &'static str,
    extract_fk: FkExtractor,
}

/// Lookup info for `OwnsMany` cascades
#[derive(Clone)]
struct OwnsManyLookup {
    local_type: &'static str,
    foreign_type: &'static str,
    extract_ids: ArrayExtractor,
    remove_id: ArrayRemover,
}

/// Lookup info for `EnsureFor` cascades
#[derive(Clone)]
struct EnsureForLookup {
    local_type: &'static str,
    dependencies: Vec<EnsureForDependency>,
    make_entity: EntityFactory,
}

#[derive(Default)]
struct RelationshipTables {
    belongs_to_by_foreign: HashMap<&'static str, Vec<BelongsToLookup>>,
    belongs_to_by_local: HashMap<&'static str, Vec<BelongsToLookup>>,
    owns_many_by_local: HashMap<&'static str, Vec<OwnsManyLookup>>,
    owns_many_by_foreign: HashMap<&'static str, Vec<OwnsManyLookup>>,
    ensure_for_by_dependency: HashMap<&'static str, Vec<EnsureForLookup>>,
    next_belongs_to_id: u64,
}

impl RelationshipTables {
    fn register(&mut self, relation: &Relation) {
        match relation {
            Relation::BelongsTo {
                local_type,
                foreign_type,
                extract_fk,
                ..
            } => self.register_belongs_to(local_type, foreign_type, *extract_fk),
            Relation::OwnsMany {
                local_type,
                foreign_type,
                extract_ids,
                remove_id,
                ..
            } => self.register_owns_many(local_type, foreign_type, *extract_ids, *remove_id),
            Relation::EnsureFor {
                local_type,
                dependencies,
                make_entity,
                ..
            } => self.register_ensure_for(local_type, dependencies, *make_entity),
        }
    }

    fn ensure_for_cycle(&self) -> Option<Vec<&'static str>> {
        fn visit(
            node: &'static str,
            graph: &HashMap<&'static str, Vec<&'static str>>,
            states: &mut HashMap<&'static str, u8>,
            stack: &mut Vec<&'static str>,
        ) -> Option<Vec<&'static str>> {
            match states.get(node).copied() {
                Some(1) => {
                    let start = stack.iter().position(|entry| *entry == node).unwrap_or(0);
                    let mut cycle = stack.get(start..).unwrap_or_default().to_vec();
                    cycle.push(node);
                    return Some(cycle);
                }
                Some(2) => return None,
                _ => {}
            }
            states.insert(node, 1);
            stack.push(node);
            if let Some(next) = graph.get(node) {
                for dependent in next {
                    if let Some(cycle) = visit(dependent, graph, states, stack) {
                        return Some(cycle);
                    }
                }
            }
            stack.pop();
            states.insert(node, 2);
            None
        }

        let mut graph: HashMap<&'static str, Vec<&'static str>> = HashMap::new();
        for (dependency, lookups) in &self.ensure_for_by_dependency {
            for lookup in lookups {
                let dependents = graph.entry(*dependency).or_default();
                if !dependents.contains(&lookup.local_type) {
                    dependents.push(lookup.local_type);
                }
            }
        }
        let mut states = HashMap::new();
        for node in graph.keys() {
            if let Some(cycle) = visit(node, &graph, &mut states, &mut Vec::new()) {
                return Some(cycle);
            }
        }
        None
    }

    fn register_belongs_to(
        &mut self,
        local_type: &'static str,
        foreign_type: &'static str,
        extract_fk: FkExtractor,
    ) {
        trace!("RelationshipManager: Registered BelongsTo {local_type} -> {foreign_type}");
        self.next_belongs_to_id = self.next_belongs_to_id.saturating_add(1);
        let lookup = BelongsToLookup {
            id: self.next_belongs_to_id,
            local_type,
            foreign_type,
            extract_fk,
        };
        self.belongs_to_by_foreign
            .entry(foreign_type)
            .or_default()
            .push(lookup.clone());
        self.belongs_to_by_local
            .entry(local_type)
            .or_default()
            .push(lookup);
    }

    fn register_owns_many(
        &mut self,
        local_type: &'static str,
        foreign_type: &'static str,
        extract_ids: ArrayExtractor,
        remove_id: ArrayRemover,
    ) {
        trace!("RelationshipManager: Registered OwnsMany {local_type} ->> {foreign_type}");
        let lookup = OwnsManyLookup {
            local_type,
            foreign_type,
            extract_ids,
            remove_id,
        };
        self.owns_many_by_local
            .entry(local_type)
            .or_default()
            .push(lookup.clone());
        self.owns_many_by_foreign
            .entry(foreign_type)
            .or_default()
            .push(lookup);
    }

    fn register_ensure_for(
        &mut self,
        local_type: &'static str,
        dependencies: &'static [EnsureForDependency],
        make_entity: EntityFactory,
    ) {
        trace!(
            "RelationshipManager: Registered EnsureFor {} for {:?}",
            local_type,
            dependencies
                .iter()
                .map(|dep| dep.foreign_type)
                .collect::<Vec<_>>()
        );
        let dependencies = dependencies.to_vec();
        let mut indexed_types = HashSet::new();
        for dependency in &dependencies {
            if !indexed_types.insert(dependency.foreign_type) {
                continue;
            }
            self.ensure_for_by_dependency
                .entry(dependency.foreign_type)
                .or_default()
                .push(EnsureForLookup {
                    local_type,
                    dependencies: dependencies.clone(),
                    make_entity,
                });
        }
    }
}

/// Cell-based `RelationshipManager` for handling entity relationship cascades.
///
/// This manager discovers relationships via [`inventory`] at initialization,
/// builds lookup indexes for efficient cascade processing, and provides
/// methods for processing events and establishing relations on startup.
///
/// Unlike the actor-based version, this implementation uses `MykoServerContext`
/// for queries and event publishing, keeping it decoupled from direct
/// store and event processor access.
pub struct RelationshipManager {
    /// `BelongsTo` relations indexed by `foreign_type` (the parent type)
    /// When a parent is deleted, look up children to cascade delete
    belongs_to_by_foreign: HashMap<&'static str, Vec<BelongsToLookup>>,

    /// `BelongsTo` relations indexed by `local_type` (the child type)
    /// Used for orphan cleanup on startup
    belongs_to_by_local: HashMap<&'static str, Vec<BelongsToLookup>>,

    /// `OwnsMany` relations indexed by `local_type` (the parent type)
    /// When a parent is deleted, delete all owned children
    owns_many_by_local: HashMap<&'static str, Vec<OwnsManyLookup>>,

    /// `OwnsMany` relations indexed by `foreign_type` (the child type)
    /// When a child is deleted, update parent arrays
    owns_many_by_foreign: HashMap<&'static str, Vec<OwnsManyLookup>>,

    /// `EnsureFor` relations indexed by their dependency types
    /// When a dependency entity is created, ensure derived entities exist
    ensure_for_by_dependency: HashMap<&'static str, Vec<EnsureForLookup>>,

    /// Reverse `belongs_to` index: `lookup_id` -> `parent_id` -> `child_ids`
    belongs_to_children_by_parent: DashMap<u64, DashMap<Arc<str>, BTreeSet<Arc<str>>>>,

    /// Reverse `belongs_to` index: `lookup_id` -> `child_id` -> `parent_id`
    belongs_to_parent_by_child: DashMap<u64, DashMap<Arc<str>, Arc<str>>>,

    /// Combination reservations make ensure-for check-and-create atomic without
    /// holding a lock across recursive cascade publication.
    ensure_for_in_flight: Mutex<HashSet<EnsureForReservation>>,
}

impl RelationshipManager {
    /// Create a new `RelationshipManager` with lookup tables built from inventory.
    pub fn new() -> Self {
        trace!("RelationshipManager: Initializing from inventory");
        let mut tables = RelationshipTables::default();
        for registration in iter_relations() {
            tables.register(&registration.relation);
        }
        if let Some(cycle) = tables.ensure_for_cycle() {
            // A recursive creation cycle cannot reach a fixed point because
            // ensured entities receive fresh IDs. Reject the schema up front.
            #[allow(clippy::panic)]
            std::panic::panic_any(format!(
                "cyclic ensure_for relationship graph: {}",
                cycle.join(" -> ")
            ));
        }

        let relation_count = tables
            .belongs_to_by_foreign
            .len()
            .saturating_add(tables.owns_many_by_local.len())
            .saturating_add(tables.ensure_for_by_dependency.len());
        trace!(
            "RelationshipManager: {} relation types indexed",
            relation_count
        );

        Self {
            belongs_to_by_foreign: tables.belongs_to_by_foreign,
            belongs_to_by_local: tables.belongs_to_by_local,
            owns_many_by_local: tables.owns_many_by_local,
            owns_many_by_foreign: tables.owns_many_by_foreign,
            ensure_for_by_dependency: tables.ensure_for_by_dependency,
            belongs_to_children_by_parent: DashMap::new(),
            belongs_to_parent_by_child: DashMap::new(),
            ensure_for_in_flight: Mutex::new(HashSet::new()),
        }
    }

    /// Whether this item type can synchronously derive relationship work.
    pub(crate) fn coordinates(&self, entity_type: &str, change: crate::wire::MEventType) -> bool {
        match change {
            crate::wire::MEventType::SET => {
                self.belongs_to_by_local.contains_key(entity_type)
                    || self.ensure_for_by_dependency.contains_key(entity_type)
            }
            crate::wire::MEventType::DEL => {
                self.belongs_to_by_foreign.contains_key(entity_type)
                    || self.belongs_to_by_local.contains_key(entity_type)
                    || self.owns_many_by_local.contains_key(entity_type)
                    || self.owns_many_by_foreign.contains_key(entity_type)
                    || self.ensure_for_by_dependency.contains_key(entity_type)
            }
        }
    }

    /// Release an ensure-for reservation when its derived entity reaches the
    /// store. Derived writes may be queued on the active causal transaction,
    /// so releasing the reservation when the write is enqueued is too early.
    pub(crate) fn release_ensure_reservation_for_item(&self, item: &dyn AnyItem) {
        let reservations: Vec<EnsureForReservation> = self
            .ensure_for_by_dependency
            .values()
            .flatten()
            .filter(|lookup| lookup.local_type == item.entity_type())
            .filter_map(|lookup| {
                let combo = lookup
                    .dependencies
                    .iter()
                    .map(|dependency| (dependency.extract_fk)(item.as_any()))
                    .collect::<Option<Vec<_>>>()?;
                Some((lookup.local_type, combo))
            })
            .collect();

        let mut in_flight = self
            .ensure_for_in_flight
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        for reservation in reservations {
            in_flight.remove(&reservation);
        }
    }

    /// Forward a SET event for relationship processing.
    ///
    /// Handles `EnsureFor`: when a dependency entity is created, ensures
    /// all derived entities exist for all combinations.
    ///
    /// # Errors
    ///
    /// Returns an error when the requested operation cannot be completed.
    pub fn forward_set(
        &self,
        item: Arc<dyn AnyItem>,
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let result = self.forward_set_batch(std::slice::from_ref(&item), ctx);
        drop(item);
        result
    }

    /// Forward a same-type batch of SET events for relationship processing.
    ///
    /// `ensure_for` operates on the fully-settled dependency stores, so running
    /// it once per item repeats the same Cartesian-product reconciliation for
    /// every member of a large import. Reconcile once for the affected type.
    ///
    /// # Errors
    ///
    /// Returns an error when an ensured relationship mutation cannot be persisted.
    pub fn forward_set_batch(
        &self,
        items: &[Arc<dyn AnyItem>],
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let Some(first) = items.first() else {
            return Ok(());
        };
        let item_type = first.entity_type();

        if let Some(lookups) = self.belongs_to_by_local.get(item_type) {
            for item in items {
                for lookup in lookups {
                    self.index_belongs_to_child(lookup, item);
                }
            }
        }

        if self.ensure_for_by_dependency.contains_key(item_type) {
            self.handle_ensure_for_type(item_type, ctx)?;
        }

        Ok(())
    }

    /// Forward a DEL event for relationship processing.
    ///
    /// Handles:
    /// - `BelongsTo` cascade deletes (parent deleted → delete children)
    /// - `OwnsMany` parent deletes (parent deleted → delete owned children)
    /// - `OwnsMany` child deletes (child deleted → update parent arrays)
    /// - `EnsureFor` cascade deletes (dependency deleted → delete derived entities)
    ///
    /// # Errors
    ///
    /// Returns an error when the requested operation cannot be completed.
    pub fn forward_del(
        &self,
        item: Arc<dyn AnyItem>,
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        // Handle BelongsTo cascades (parent deleted → delete children)
        self.handle_belongs_to_cascade(&item, ctx)?;

        // Handle OwnsMany parent deleted → delete owned children
        self.handle_owns_many_parent_delete(&item, ctx)?;

        // Handle OwnsMany child deleted → update parent arrays
        self.handle_owns_many_child_delete(&item, ctx)?;

        // Handle EnsureFor dependency deleted → delete derived entities
        self.handle_ensure_for_delete(&item, ctx)?;

        if let Some(lookups) = self.belongs_to_by_local.get(item.entity_type()) {
            for lookup in lookups {
                self.remove_belongs_to_child(lookup, &item.id());
            }
        }

        drop(item);
        Ok(())
    }

    /// Forward a batch of DEL events for relationship processing.
    ///
    /// Items should all be the same entity type. This keeps cascade deletes grouped
    /// so downstream stores and views can process one wider delete wave instead of
    /// thousands of tiny per-parent cascades.
    ///
    /// # Errors
    ///
    /// Returns an error when the requested operation cannot be completed.
    pub fn forward_del_batch(
        &self,
        items: &[Arc<dyn AnyItem>],
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        if items.is_empty() {
            return Ok(());
        }

        self.handle_belongs_to_cascade_batch(items, ctx)?;
        self.handle_owns_many_parent_delete_batch(items, ctx)?;
        self.handle_ensure_for_delete_batch(items, ctx)?;

        for item in items {
            self.handle_owns_many_child_delete(item, ctx)?;

            if let Some(lookups) = self.belongs_to_by_local.get(item.entity_type()) {
                for lookup in lookups {
                    self.remove_belongs_to_child(lookup, &item.id());
                }
            }
        }

        Ok(())
    }

    /// Establish relations on startup (called after durable backend catchup).
    ///
    /// This performs:
    /// 1. `BelongsTo` orphan cleanup: Delete children pointing to non-existent parents
    /// 2. `OwnsMany` orphan cleanup: Delete children not referenced by any parent
    /// 3. `EnsureFor` initialization: Create missing entities for all dependency combinations
    ///
    /// # Errors
    ///
    /// Returns an error when the requested operation cannot be completed.
    pub fn establish_relations(&self, ctx: &MykoServerContext) -> Result<(), PersistError> {
        info!("RelationshipManager: Establishing relations on startup");
        trace!(
            "RelationshipManager: BelongsTo relations by local: {:?}",
            self.belongs_to_by_local.keys().collect::<Vec<_>>()
        );
        debug!(
            "RelationshipManager: OwnsMany relations by local: {:?}",
            self.owns_many_by_local.keys().collect::<Vec<_>>()
        );

        // 1. Orphan cleanup for BelongsTo relationships
        self.cleanup_belongs_to_orphans(ctx)?;

        // 2. Orphan cleanup for OwnsMany relationships
        self.cleanup_owns_many_orphans(ctx)?;

        // 3. EnsureFor initialization
        self.initialize_ensure_for(ctx)?;

        info!("RelationshipManager: Relations established");
        Ok(())
    }

    // ─────────────────────────────────────────────────────────────────────────────
    // Cascade handlers
    // ─────────────────────────────────────────────────────────────────────────────

    /// Handle `BelongsTo` cascades: when a parent is deleted, delete all children
    fn handle_belongs_to_cascade(
        &self,
        item: &Arc<dyn AnyItem>,
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let item_type = item.entity_type();
        let Some(lookups) = self.belongs_to_by_foreign.get(item_type) else {
            return Ok(());
        };

        let parent_id = item.id();

        for lookup in lookups {
            // Find children whose FK matches the deleted parent ID using extractor
            let children = self.find_children_by_fk(ctx, lookup, &parent_id);
            if children.is_empty() {
                continue;
            }

            trace!(
                "RelationshipManager: Cascade delete batch {} count={} (parent {} deleted)",
                lookup.local_type,
                children.len(),
                parent_id
            );
            Self::publish_del_cascade_batch(ctx, &children)?;
        }

        Ok(())
    }

    fn handle_belongs_to_cascade_batch(
        &self,
        items: &[Arc<dyn AnyItem>],
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let Some(first) = items.first() else {
            return Ok(());
        };
        let item_type = first.entity_type();
        let Some(lookups) = self.belongs_to_by_foreign.get(item_type) else {
            return Ok(());
        };

        let parent_ids: Vec<Arc<str>> = items.iter().map(|item| item.id()).collect();

        for lookup in lookups {
            let mut children_by_id: HashMap<Arc<str>, Arc<dyn AnyItem>> = HashMap::new();
            for parent_id in &parent_ids {
                for child in self.find_children_by_fk(ctx, lookup, parent_id) {
                    children_by_id.entry(child.id()).or_insert(child);
                }
            }

            if children_by_id.is_empty() {
                continue;
            }

            let children: Vec<_> = children_by_id.into_values().collect();
            trace!(
                "RelationshipManager: Cascade delete batch {} count={} ({} parents deleted)",
                lookup.local_type,
                children.len(),
                parent_ids.len()
            );
            Self::publish_del_cascade_batch(ctx, &children)?;
        }

        Ok(())
    }

    /// Find children whose FK matches a given parent ID
    fn find_children_by_fk(
        &self,
        ctx: &MykoServerContext,
        lookup: &BelongsToLookup,
        parent_id: &str,
    ) -> Vec<Arc<dyn AnyItem>> {
        self.ensure_belongs_to_index_loaded(ctx, lookup);
        if let Some(parent_map) = self.belongs_to_children_by_parent.get(&lookup.id) {
            let store = ctx.registry.get_or_create(lookup.local_type);
            let Some(child_ids) = parent_map.get(parent_id) else {
                return Vec::new();
            };
            return child_ids
                .iter()
                .filter_map(|child_id| store.get_value(child_id))
                .collect();
        }

        let store = ctx.registry.get_or_create(lookup.local_type);
        store
            .entries()
            .materialize()
            .get()
            .into_iter()
            .filter(|(_, item)| {
                (lookup.extract_fk)(item.as_any()).is_some_and(|fk| fk.as_ref() == parent_id)
            })
            .map(|(_, item)| item)
            .collect()
    }

    fn index_belongs_to_child(&self, lookup: &BelongsToLookup, item: &Arc<dyn AnyItem>) {
        let child_id = item.id();
        self.remove_belongs_to_child(lookup, &child_id);

        let Some(parent_id) = (lookup.extract_fk)(item.as_any()) else {
            return;
        };

        self.belongs_to_parent_by_child
            .entry(lookup.id)
            .or_default()
            .insert(child_id.clone(), parent_id.clone());
        self.belongs_to_children_by_parent
            .entry(lookup.id)
            .or_default()
            .entry(parent_id)
            .or_default()
            .insert(child_id);
    }

    fn remove_belongs_to_child(&self, lookup: &BelongsToLookup, child_id: &Arc<str>) {
        let Some(parent_map) = self.belongs_to_parent_by_child.get(&lookup.id) else {
            return;
        };
        let Some((_, parent_id)) = parent_map.remove(child_id) else {
            return;
        };

        let Some(children_by_parent) = self.belongs_to_children_by_parent.get(&lookup.id) else {
            return;
        };
        let should_remove_parent =
            children_by_parent
                .get_mut(parent_id.as_ref())
                .is_some_and(|mut child_ids| {
                    child_ids.remove(child_id);
                    child_ids.is_empty()
                });

        if should_remove_parent {
            children_by_parent.remove(parent_id.as_ref());
        }
    }

    fn ensure_belongs_to_index_loaded(&self, ctx: &MykoServerContext, lookup: &BelongsToLookup) {
        if self.belongs_to_parent_by_child.contains_key(&lookup.id) {
            return;
        }

        let child_index = DashMap::<Arc<str>, Arc<str>>::new();
        let parent_index = DashMap::<Arc<str>, BTreeSet<Arc<str>>>::new();
        let store = ctx.registry.get_or_create(lookup.local_type);

        for (_, item) in store.snapshot() {
            let Some(parent_id) = (lookup.extract_fk)(item.as_any()) else {
                continue;
            };
            let child_id = item.id();
            child_index.insert(child_id.clone(), parent_id.clone());
            parent_index.entry(parent_id).or_default().insert(child_id);
        }

        let _ = self
            .belongs_to_parent_by_child
            .insert(lookup.id, child_index);
        let _ = self
            .belongs_to_children_by_parent
            .insert(lookup.id, parent_index);
    }

    /// Handle `OwnsMany` parent delete: delete all owned children
    fn handle_owns_many_parent_delete(
        &self,
        item: &Arc<dyn AnyItem>,
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let item_type = item.entity_type();
        let Some(lookups) = self.owns_many_by_local.get(item_type) else {
            return Ok(());
        };

        for lookup in lookups {
            // Extract child IDs using the typed extractor
            let Some(child_ids) = (lookup.extract_ids)(item.as_any()) else {
                continue;
            };

            if child_ids.is_empty() {
                continue;
            }

            let mut children = Vec::new();
            for child_id in &child_ids {
                if Self::get_by_id(ctx, lookup.foreign_type, child_id).is_some()
                    && let Some(child) = Self::get_by_id(ctx, lookup.foreign_type, child_id)
                {
                    children.push(child);
                }
            }

            if children.is_empty() {
                continue;
            }

            trace!(
                "RelationshipManager: Cascade delete owned batch {} count={}",
                lookup.foreign_type,
                children.len()
            );
            Self::publish_del_cascade_batch(ctx, &children)?;
        }

        Ok(())
    }

    fn handle_owns_many_parent_delete_batch(
        &self,
        items: &[Arc<dyn AnyItem>],
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let Some(first) = items.first() else {
            return Ok(());
        };
        let item_type = first.entity_type();
        let Some(lookups) = self.owns_many_by_local.get(item_type) else {
            return Ok(());
        };

        for lookup in lookups {
            let mut child_ids = BTreeSet::new();
            for item in items {
                if let Some(ids) = (lookup.extract_ids)(item.as_any()) {
                    child_ids.extend(ids);
                }
            }

            if child_ids.is_empty() {
                continue;
            }

            let mut children = Vec::new();
            for child_id in &child_ids {
                if let Some(child) = Self::get_by_id(ctx, lookup.foreign_type, child_id) {
                    children.push(child);
                }
            }

            if children.is_empty() {
                continue;
            }

            trace!(
                "RelationshipManager: Cascade delete owned batch {} count={} ({} parents deleted)",
                lookup.foreign_type,
                children.len(),
                items.len()
            );
            Self::publish_del_cascade_batch(ctx, &children)?;
        }

        Ok(())
    }

    /// Handle `OwnsMany` child delete: remove child ID from parent arrays
    fn handle_owns_many_child_delete(
        &self,
        item: &Arc<dyn AnyItem>,
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let item_type = item.entity_type();
        let Some(lookups) = self.owns_many_by_foreign.get(item_type) else {
            return Ok(());
        };

        let child_id = item.id();

        for lookup in lookups {
            // Find parents that contain this child ID using extract_ids
            let parents = Self::find_parents_containing(ctx, lookup, &child_id);
            let mut updates = Vec::new();

            for parent_item in parents {
                // Use the remove_id extractor to get updated parent as Value
                if let Some(updated_parent) = (lookup.remove_id)(parent_item.as_any(), &child_id) {
                    trace!(
                        "RelationshipManager: Updating {} {} to remove child {}",
                        lookup.local_type,
                        parent_item.id(),
                        child_id
                    );
                    updates.push(updated_parent);
                }
            }

            if !updates.is_empty() {
                Self::publish_relationship_fixup_batch(ctx, &updates)?;
            }
        }

        Ok(())
    }

    /// Find parents whose owned array contains a given child ID
    fn find_parents_containing(
        ctx: &MykoServerContext,
        lookup: &OwnsManyLookup,
        child_id: &str,
    ) -> Vec<Arc<dyn AnyItem>> {
        let store = ctx.registry.get_or_create(lookup.local_type);
        store
            .entries()
            .materialize()
            .get()
            .into_iter()
            .filter(|(_, item)| {
                (lookup.extract_ids)(item.as_any())
                    .is_some_and(|ids| ids.iter().any(|id| id.as_ref() == child_id))
            })
            .map(|(_, item)| item)
            .collect()
    }

    /// Handle `EnsureFor` once after a dependency type's store has settled.
    fn handle_ensure_for_type(
        &self,
        item_type: &str,
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let Some(lookups) = self.ensure_for_by_dependency.get(item_type) else {
            return Ok(());
        };

        for lookup in lookups {
            let combinations = Self::get_dependency_combinations(ctx, &lookup.dependencies);
            let store = ctx.registry.get_or_create(lookup.local_type);
            let existing_items = store.snapshot();
            let mut existing_combinations: HashSet<Vec<Arc<str>>> = existing_items
                .iter()
                .filter_map(|(_, item)| {
                    lookup
                        .dependencies
                        .iter()
                        .map(|dependency| (dependency.extract_fk)(item.as_any()))
                        .collect()
                })
                .collect();

            for combo in combinations {
                if !existing_combinations.contains(&combo)
                    && self.ensure_combination(lookup, &combo, ctx)?
                {
                    existing_combinations.insert(combo);
                }
            }
        }

        Ok(())
    }

    /// Atomically reserve and create one missing dependency combination.
    /// Publication is synchronous, but the reservation lock is released first
    /// so relationships on the created entity can propagate recursively.
    fn ensure_combination(
        &self,
        lookup: &EnsureForLookup,
        combo: &[Arc<str>],
        ctx: &MykoServerContext,
    ) -> Result<bool, PersistError> {
        let reservation = (lookup.local_type, combo.to_vec());
        {
            let mut in_flight = self
                .ensure_for_in_flight
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if !in_flight.insert(reservation.clone()) {
                return Ok(false);
            }
        }

        // Recheck after reserving: another publisher may have completed between
        // the caller's snapshot and this reservation.
        let store = ctx.registry.get_or_create(lookup.local_type);
        let exists =
            Self::find_ensure_for_entity_in(&store.snapshot(), &lookup.dependencies, combo)
                .is_some();
        if exists {
            self.ensure_for_in_flight
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .remove(&reservation);
            Ok(false)
        } else {
            let entity = (lookup.make_entity)(combo);
            trace!(
                "RelationshipManager: Creating ensured {} for {:?}",
                lookup.local_type, combo
            );
            let result = Self::publish_set_cascade(ctx, lookup.local_type, entity);
            if result.is_err() {
                self.ensure_for_in_flight
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .remove(&reservation);
            }
            result.map(|()| true)
        }
    }

    /// Handle `EnsureFor`: when a dependency entity is deleted, delete the
    /// entities that were auto-created for it — symmetric with
    /// `handle_ensure_for`'s create-if-missing on the SET side. Without
    /// this, `#[ensure_for(X)]`-created entities are orphaned forever once
    /// `X` is deleted (they're never revisited by any other cascade path).
    fn handle_ensure_for_delete(
        &self,
        item: &Arc<dyn AnyItem>,
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        self.handle_ensure_for_delete_batch(std::slice::from_ref(item), ctx)
    }

    fn handle_ensure_for_delete_batch(
        &self,
        items: &[Arc<dyn AnyItem>],
        ctx: &MykoServerContext,
    ) -> Result<(), PersistError> {
        let Some(first) = items.first() else {
            return Ok(());
        };
        let item_type = first.entity_type();
        let Some(lookups) = self.ensure_for_by_dependency.get(item_type) else {
            return Ok(());
        };

        let dep_ids: HashSet<Arc<str>> = items.iter().map(|item| item.id()).collect();

        for lookup in lookups {
            // A lookup can contain several fields targeting the same entity
            // type. Every matching extractor participates: deleting a dependency
            // must remove a Cartesian product when that ID appears on any axis.
            let dependencies: Vec<_> = lookup
                .dependencies
                .iter()
                .filter(|dep| dep.foreign_type == item_type)
                .collect();
            if dependencies.is_empty() {
                continue;
            }

            let orphaned = Self::find_ensure_for_children_by_dependencies(
                ctx,
                lookup,
                &dependencies,
                &dep_ids,
            );
            if orphaned.is_empty() {
                continue;
            }

            trace!(
                "RelationshipManager: EnsureFor cascade delete {} count={} ({} {} dependencies deleted)",
                lookup.local_type,
                orphaned.len(),
                dep_ids.len(),
                item_type
            );
            Self::publish_del_cascade_batch(ctx, &orphaned)?;
        }

        Ok(())
    }

    /// Scan `lookup.local_type`'s store for ensure_for-derived entities
    /// whose `dep`-extracted FK is one of `dep_ids`. Unlike `belongs_to`,
    /// there is no lazily-built reverse index for `ensure_for` — the
    /// created entity's id is a random UUID (`make_entity` in
    /// `handle_ensure_for`), not derivable from the dependency id, so a
    /// full-store scan is the only option here (same as `belongs_to`'s own
    /// fallback path when its index isn't loaded yet).
    fn find_ensure_for_children_by_dependencies(
        ctx: &MykoServerContext,
        lookup: &EnsureForLookup,
        dependencies: &[&EnsureForDependency],
        dep_ids: &HashSet<Arc<str>>,
    ) -> Vec<Arc<dyn AnyItem>> {
        let store = ctx.registry.get_or_create(lookup.local_type);
        store
            .entries()
            .materialize()
            .get()
            .into_iter()
            .filter(|(_, item)| {
                dependencies.iter().any(|dep| {
                    (dep.extract_fk)(item.as_any()).is_some_and(|fk| dep_ids.contains(&fk))
                })
            })
            .map(|(_, item)| item)
            .collect()
    }

    // ─────────────────────────────────────────────────────────────────────────────
    // Orphan cleanup
    // ─────────────────────────────────────────────────────────────────────────────

    /// Cleanup orphaned children for `BelongsTo` relationships
    /// Boot-time **backstop** sweep for `belongs_to` orphans (children whose FK
    /// points at a parent that no longer exists).
    ///
    /// Runtime orphaning is handled by the transitive DEL cascade
    /// (`Origin::Cascade` + DEL descends — see `MykoServerContext::apply_effects`),
    /// so deleting a parent removes its whole subtree without a restart. This
    /// sweep remains only for the "child written with an FK to a never-existent
    /// parent" case. We deliberately do **not** delete such orphans eagerly on
    /// the child's SET: under out-of-order / eventually-consistent ingestion a
    /// child can legitimately arrive before its parent, so eager deletion would
    /// be data loss. The sweep runs at boot, once ordering has settled.
    fn cleanup_belongs_to_orphans(&self, ctx: &MykoServerContext) -> Result<(), PersistError> {
        trace!(
            "RelationshipManager: cleanup_belongs_to_orphans - checking {} child types",
            self.belongs_to_by_local.len()
        );

        for (child_type, lookups) in &self.belongs_to_by_local {
            trace!(
                "RelationshipManager: Checking BelongsTo orphans for child type '{}' ({} lookups)",
                child_type,
                lookups.len()
            );

            for lookup in lookups {
                // Get all parent IDs that exist
                let parents = Self::get_all_items(ctx, lookup.foreign_type);
                let parent_ids: HashSet<Arc<str>> = parents.iter().map(|p| p.id()).collect();

                trace!(
                    "RelationshipManager: {} -> {}: Found {} parents in store",
                    child_type,
                    lookup.foreign_type,
                    parents.len()
                );

                // Get all children and find orphans using typed extractor
                let children = Self::get_all_items(ctx, child_type);
                trace!(
                    "RelationshipManager: {} -> {}: Found {} children in store",
                    child_type,
                    lookup.foreign_type,
                    children.len()
                );

                let mut orphan_count = 0_u64;
                let mut valid_count = 0_u64;
                let mut no_fk_count = 0_u64;

                for child in &children {
                    // Use the typed extractor to get the FK value
                    if let Some(fk_value) = (lookup.extract_fk)(child.as_any()) {
                        if parent_ids.contains(&fk_value) {
                            valid_count = valid_count.saturating_add(1);
                        } else {
                            debug!(
                                "RelationshipManager: ORPHAN {} {} has FK '{}' but parent {} not found (have {} parent IDs)",
                                child_type,
                                child.id(),
                                fk_value,
                                lookup.foreign_type,
                                parent_ids.len()
                            );
                            Self::publish_del_cascade(ctx, child_type, &child.id())?;
                            orphan_count = orphan_count.saturating_add(1);
                        }
                    } else {
                        trace!(
                            "RelationshipManager: {} {} - extract_fk returned None",
                            child_type,
                            child.id()
                        );
                        no_fk_count = no_fk_count.saturating_add(1);
                    }
                }

                trace!(
                    "RelationshipManager: {} -> {}: {} orphans deleted, {} valid, {} no FK",
                    child_type, lookup.foreign_type, orphan_count, valid_count, no_fk_count
                );
            }
        }

        Ok(())
    }

    /// Cleanup orphaned children for `OwnsMany` relationships
    fn cleanup_owns_many_orphans(&self, ctx: &MykoServerContext) -> Result<(), PersistError> {
        trace!(
            "RelationshipManager: cleanup_owns_many_orphans - checking {} parent types",
            self.owns_many_by_local.len()
        );

        for (parent_type, lookups) in &self.owns_many_by_local {
            trace!(
                "RelationshipManager: Checking OwnsMany orphans for parent type '{}' ({} lookups)",
                parent_type,
                lookups.len()
            );

            for lookup in lookups {
                // Get all child IDs referenced by parents using typed extractors
                let parents = Self::get_all_items(ctx, parent_type);
                let mut referenced_ids: HashSet<Arc<str>> = HashSet::new();

                trace!(
                    "RelationshipManager: {} ->> {}: Found {} parents in store",
                    parent_type,
                    lookup.foreign_type,
                    parents.len()
                );

                let mut parents_with_ids = 0_u64;
                let mut parents_no_ids = 0_u64;
                for parent in &parents {
                    if let Some(ids) = (lookup.extract_ids)(parent.as_any()) {
                        if !ids.is_empty() {
                            parents_with_ids = parents_with_ids.saturating_add(1);
                        }
                        referenced_ids.extend(ids);
                    } else {
                        parents_no_ids = parents_no_ids.saturating_add(1);
                    }
                }

                trace!(
                    "RelationshipManager: {} ->> {}: {} parents have child IDs, {} have no IDs, {} total referenced child IDs",
                    parent_type,
                    lookup.foreign_type,
                    parents_with_ids,
                    parents_no_ids,
                    referenced_ids.len()
                );

                // Get all children and find orphans
                let children = Self::get_all_items(ctx, lookup.foreign_type);
                trace!(
                    "RelationshipManager: {} ->> {}: Found {} children in store",
                    parent_type,
                    lookup.foreign_type,
                    children.len()
                );

                let mut orphan_count = 0_u64;
                let mut valid_count = 0_u64;

                for child in children {
                    let child_id = child.id();
                    if referenced_ids.contains(&child_id) {
                        valid_count = valid_count.saturating_add(1);
                    } else {
                        debug!(
                            "RelationshipManager: ORPHAN {} {} not referenced by any {} (have {} referenced IDs)",
                            lookup.foreign_type,
                            child_id,
                            parent_type,
                            referenced_ids.len()
                        );
                        Self::publish_del_cascade(ctx, lookup.foreign_type, &child_id)?;
                        orphan_count = orphan_count.saturating_add(1);
                    }
                }

                if orphan_count > 0 {
                    info!(
                        "RelationshipManager: {} ->> {}: {} orphans deleted, {} valid",
                        parent_type, lookup.foreign_type, orphan_count, valid_count
                    );
                } else {
                    trace!(
                        "RelationshipManager: {} ->> {}: {} orphans deleted, {} valid",
                        parent_type, lookup.foreign_type, orphan_count, valid_count
                    );
                }
            }
        }

        Ok(())
    }

    /// Initialize `EnsureFor` relationships (create missing derived entities)
    fn initialize_ensure_for(&self, ctx: &MykoServerContext) -> Result<(), PersistError> {
        // Track which local_types we've processed to avoid duplicates
        let mut processed: HashSet<&'static str> = HashSet::new();

        for lookups in self.ensure_for_by_dependency.values() {
            for lookup in lookups {
                if processed.contains(lookup.local_type) {
                    continue;
                }
                processed.insert(lookup.local_type);

                // Get all combinations of dependency entities
                let combinations = Self::get_dependency_combinations(ctx, &lookup.dependencies);

                // Snapshot once outside the combo loop
                let store = ctx.registry.get_or_create(lookup.local_type);
                let existing_items = store.snapshot();

                let mut created_count = 0_u64;

                for combo in combinations {
                    // Check if derived entity already exists
                    let existing = Self::find_ensure_for_entity_in(
                        &existing_items,
                        &lookup.dependencies,
                        &combo,
                    );

                    if existing.is_none() && self.ensure_combination(lookup, &combo, ctx)? {
                        created_count = created_count.saturating_add(1);
                    }
                }

                if created_count > 0 {
                    info!(
                        "RelationshipManager: Created {} {} entities via EnsureFor",
                        created_count, lookup.local_type
                    );
                }
            }
        }

        Ok(())
    }

    // ─────────────────────────────────────────────────────────────────────────────
    // Query helpers (using MykoServerContext)
    // ─────────────────────────────────────────────────────────────────────────────

    /// Get an entity by ID
    fn get_by_id(ctx: &MykoServerContext, entity_type: &str, id: &str) -> Option<Arc<dyn AnyItem>> {
        let store = ctx.registry.get_or_create(entity_type);
        store.get_value(&id.into())
    }

    /// Get all entities of a type
    fn get_all_items(ctx: &MykoServerContext, entity_type: &str) -> Vec<Arc<dyn AnyItem>> {
        let store = ctx.registry.get_or_create(entity_type);
        store.snapshot().into_iter().map(|(_, item)| item).collect()
    }

    /// Get all combinations of dependency entity IDs for `EnsureFor`
    fn get_dependency_combinations(
        ctx: &MykoServerContext,
        dependencies: &[EnsureForDependency],
    ) -> Vec<Vec<Arc<str>>> {
        if dependencies.is_empty() {
            return vec![];
        }

        // Get IDs for each dependency type
        let mut dep_ids: Vec<Vec<Arc<str>>> = Vec::new();

        for dep in dependencies {
            let items = Self::get_all_items(ctx, dep.foreign_type);
            let ids: Vec<Arc<str>> = items.iter().map(|item| item.id()).collect();
            dep_ids.push(ids);
        }

        // Compute Cartesian product
        Self::cartesian_product(&dep_ids)
    }

    /// Compute Cartesian product of multiple ID sets
    fn cartesian_product(sets: &[Vec<Arc<str>>]) -> Vec<Vec<Arc<str>>> {
        if sets.is_empty() {
            return vec![];
        }

        let mut result = vec![vec![]];

        for set in sets {
            let mut new_result = Vec::new();
            for existing in &result {
                for item in set {
                    let mut new_combo = existing.clone();
                    new_combo.push(item.clone());
                    new_result.push(new_combo);
                }
            }
            result = new_result;
        }

        result
    }

    /// Find an `EnsureFor` entity matching the given dependency IDs
    /// from a pre-computed snapshot of existing items.
    fn find_ensure_for_entity_in(
        items: &[(Arc<str>, Arc<dyn AnyItem>)],
        dependencies: &[EnsureForDependency],
        combo: &[Arc<str>],
    ) -> Option<Arc<dyn AnyItem>> {
        if dependencies.is_empty() || combo.is_empty() {
            return None;
        }

        items.iter().find_map(|(_, item)| {
            // Check if all dependency FKs match the combo values
            let all_match = dependencies
                .iter()
                .zip(combo.iter())
                .all(|(dep, expected_id)| {
                    (dep.extract_fk)(item.as_any()).is_some_and(|fk| fk == *expected_id)
                });

            if all_match { Some(item.clone()) } else { None }
        })
    }

    // ─────────────────────────────────────────────────────────────────────────────
    // Publishing helpers (using MykoServerContext)
    // ─────────────────────────────────────────────────────────────────────────────

    /// Publish a structural SET and continue enforcing relationships on the
    /// created entity. This is required for transitive `ensure_for` chains.
    fn publish_set_cascade(
        ctx: &MykoServerContext,
        _entity_type: &str,
        item: Arc<dyn AnyItem>,
    ) -> Result<(), PersistError> {
        // If the item has an empty #[server_owned] field, bake in the current server's ID
        let item = if item.server_owner().is_none() {
            item.bake_server_owner(&ctx.host_id.to_string())
                .unwrap_or(item)
        } else {
            item
        };

        ctx.set_dyn_with_origin(item, super::Origin::Cascade)
    }

    fn publish_relationship_fixup_batch(
        ctx: &MykoServerContext,
        items: &[Arc<dyn AnyItem>],
    ) -> Result<(), PersistError> {
        ctx.batch_set_dyn_with_origin(items, super::Origin::RelationshipFixup)
    }

    /// Publish a structural DEL and continue enforcing delete cascades.
    fn publish_del_cascade(
        ctx: &MykoServerContext,
        entity_type: &str,
        id: &str,
    ) -> Result<(), PersistError> {
        // Get the entity from the store
        let id_arc: Arc<str> = id.into();
        if let Some(item) = ctx.registry.get_or_create(entity_type).get_value(&id_arc) {
            debug!(
                "RelationshipManager: publish_del_cascade {} {} - entity found, deleting",
                entity_type, id
            );
            ctx.del_dyn_with_origin(item, super::Origin::Cascade)?;
        } else {
            trace!(
                "RelationshipManager: publish_del_cascade {} {} - entity NOT found in store",
                entity_type, id
            );
        }

        Ok(())
    }

    fn publish_del_cascade_batch(
        ctx: &MykoServerContext,
        items: &[Arc<dyn AnyItem>],
    ) -> Result<(), PersistError> {
        if items.is_empty() {
            return Ok(());
        }

        ctx.batch_del_dyn_with_origin(items, super::Origin::Cascade)
    }
}

impl Default for RelationshipManager {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn unused_entity_factory(_: &[Arc<str>]) -> Arc<dyn AnyItem> {
        std::process::abort()
    }

    #[test]
    fn cyclic_ensure_for_graph_is_rejected() {
        let mut tables = RelationshipTables::default();
        tables
            .ensure_for_by_dependency
            .entry("CycleA")
            .or_default()
            .push(EnsureForLookup {
                local_type: "CycleB",
                dependencies: Vec::new(),
                make_entity: unused_entity_factory,
            });
        tables
            .ensure_for_by_dependency
            .entry("CycleB")
            .or_default()
            .push(EnsureForLookup {
                local_type: "CycleA",
                dependencies: Vec::new(),
                make_entity: unused_entity_factory,
            });

        let cycle = tables.ensure_for_cycle().unwrap_or_default();
        assert_eq!(cycle.first(), cycle.last());
        assert!(cycle.contains(&"CycleA"));
        assert!(cycle.contains(&"CycleB"));
    }

    #[test]
    fn test_relationship_manager_creation() {
        let manager = RelationshipManager::new();

        // Should have built lookup tables from inventory
        // (actual counts depend on entities linked in test binary)
        // Just verify the manager initializes without panic
        let _ = manager.belongs_to_by_foreign.len();
        let _ = manager.owns_many_by_local.len();
    }

    #[test]
    fn test_cartesian_product() {
        let sets = vec![
            vec![Arc::from("a"), Arc::from("b")],
            vec![Arc::from("1"), Arc::from("2")],
        ];

        let product = RelationshipManager::cartesian_product(&sets);

        assert_eq!(product.len(), 4);
        assert!(product.contains(&vec![Arc::from("a"), Arc::from("1")]));
        assert!(product.contains(&vec![Arc::from("a"), Arc::from("2")]));
        assert!(product.contains(&vec![Arc::from("b"), Arc::from("1")]));
        assert!(product.contains(&vec![Arc::from("b"), Arc::from("2")]));
    }

    #[test]
    fn test_cartesian_product_empty() {
        let sets: Vec<Vec<Arc<str>>> = vec![];
        let product = RelationshipManager::cartesian_product(&sets);
        assert!(product.is_empty());
    }
}

#[cfg(test)]
mod cascade_tests {
    //! Transitive relationship cascade (Event Bus Unification, Fix #1).
    //!
    //! Deleting a parent must remove its children, grandchildren, … at runtime
    //! (previously grandchildren survived until the boot-time orphan sweep
    //! because the cascade product's `prevent_relationship_updates` flag was
    //! read as "do not cascade at all"). A cyclic schema must converge, not loop.

    use std::sync::Arc;

    use uuid::Uuid;

    use self::node::CascadeNode;
    use crate::{
        hyphae::{Gettable, Materialize},
        search::SearchIndex,
        server::{
            CausalLimits, HandlerRegistry, MykoServerContext, RelationshipManager,
            persister::PersisterRouter,
        },
        store::StoreRegistry,
        test_util::scheduler_test_serial,
    };

    // `#[myko_item]` re-imports hyphae traits at module scope, so the entity
    // lives in its own module (mirrors `bench_entities::tree`).
    mod node {
        use crate::prelude::*;

        /// Self-referential entity: a node `belongs_to` another node of the same
        /// type, so one type expresses both a multi-level chain and a cycle.
        #[myko_item]
        pub struct CascadeNode {
            #[belongs_to(CascadeNode)]
            pub parent_id: CascadeNodeId,
            pub name: String,
        }
    }

    fn make_ctx() -> (MykoServerContext, Arc<StoreRegistry>) {
        let registry = Arc::new(StoreRegistry::new());
        let ctx = MykoServerContext::new(
            Uuid::new_v4(),
            registry.clone(),
            Arc::new(HandlerRegistry::new()),
            Arc::new(RelationshipManager::new()),
            Arc::new(PersisterRouter::default()),
            Arc::new(SearchIndex::new()),
            crate::server::MykoServerRuntime {
                peer_clients: Arc::new(dashmap::DashMap::new()),
                event_sink: None,
                history_replay: None,
            },
        );
        (ctx, registry)
    }

    fn make_node(id: &str, parent_id: &str) -> CascadeNode {
        CascadeNode {
            id: id.into(),
            parent_id: parent_id.into(),
            name: format!("node-{id}"),
        }
    }

    fn exists(registry: &StoreRegistry, id: &str) -> bool {
        registry
            .get("CascadeNode")
            .and_then(|store| store.get(&Arc::<str>::from(id)).materialize().get())
            .is_some()
    }

    /// A 3-level `belongs_to` chain: deleting the root removes the child *and*
    /// the grandchild at runtime. The grandchild regressed before Fix #1.
    #[test]
    fn del_cascade_descends_to_grandchildren() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();

        // root <- branch <- leaf
        assert!(ctx.set(&make_node("root", "")).is_ok());
        assert!(ctx.set(&make_node("branch", "root")).is_ok());
        assert!(ctx.set(&make_node("leaf", "branch")).is_ok());

        assert!(exists(&registry, "root"));
        assert!(exists(&registry, "branch"));
        assert!(exists(&registry, "leaf"));

        assert!(ctx.del(&make_node("root", "")).is_ok());

        assert!(!exists(&registry, "root"), "root deleted");
        assert!(!exists(&registry, "branch"), "direct child deleted");
        assert!(
            !exists(&registry, "leaf"),
            "grandchild deleted at runtime (Fix #1)"
        );
    }

    /// Regression test: causal cascade work is queued rather than recursively
    /// joining the root's still-open Hyphae batch. Every level must retain its
    /// own observable diff on the same store.
    #[test]
    fn del_cascade_recursion_does_not_drop_earlier_diffs_in_same_store() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();

        assert!(ctx.set(&make_node("root", "")).is_ok());
        assert!(ctx.set(&make_node("branch", "root")).is_ok());
        assert!(ctx.set(&make_node("leaf", "branch")).is_ok());
        assert!(ctx.set(&make_node("island", "")).is_ok());

        let store = registry.get_or_create("CascadeNode");
        let seen: Arc<std::sync::Mutex<Vec<String>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen_for_closure = seen.clone();
        let _guard = store.subscribe_diffs(move |diff| {
            seen_for_closure
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .push(format!("{diff:?}"));
        });
        // subscribe_diffs replays the current snapshot synchronously on
        // subscribe -- drop that so only diffs from the batch below count.
        seen.lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clear();

        // One wire batch: an unrelated standalone delete (island) alongside
        // root's delete, which cascades to branch then leaf -- three
        // distinct mutations to the *same* CascadeNode store triggered by
        // one top-level call.
        let root_item: Arc<dyn crate::core::item::AnyItem> = Arc::new(make_node("root", ""));
        let island_item: Arc<dyn crate::core::item::AnyItem> = Arc::new(make_node("island", ""));
        assert!(ctx.batch_del_dyn(&[root_item, island_item]).is_ok());

        assert!(!exists(&registry, "root"));
        assert!(!exists(&registry, "branch"), "direct child deleted");
        assert!(!exists(&registry, "leaf"), "grandchild deleted");
        assert!(!exists(&registry, "island"));

        let causal = ctx.causal_diagnostics();
        assert!(causal.derived_mutations >= 2);
        assert!(causal.max_observed_depth >= 2);

        let seen = seen
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        assert_eq!(
            seen.len(),
            3,
            "expected 3 separate diffs (root+island reduce, branch cascade, leaf cascade), not coalesced: {:?}",
            *seen
        );
        drop(seen);
    }

    /// A 2-cycle (a.parent = b, b.parent = a): the cascade must converge. The
    /// store-as-visited-set guarantees it — the second visit finds nothing.
    #[test]
    fn del_cascade_terminates_on_cycle() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();

        assert!(ctx.set(&make_node("a", "b")).is_ok());
        assert!(ctx.set(&make_node("b", "a")).is_ok());

        assert!(ctx.del(&make_node("a", "b")).is_ok());

        assert!(!exists(&registry, "a"), "a deleted");
        assert!(
            !exists(&registry, "b"),
            "b deleted via the cycle, then terminated"
        );
        let causal = ctx.causal_diagnostics();
        assert!(causal.derived_mutations >= 1);
        assert_eq!(causal.budget_exhaustions, 0);
    }

    #[test]
    fn causal_depth_budget_stops_a_partial_cascade_without_rollback() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();
        assert!(ctx.set(&make_node("root", "")).is_ok());
        assert!(ctx.set(&make_node("branch", "root")).is_ok());
        assert!(ctx.set(&make_node("leaf", "branch")).is_ok());
        ctx.set_causal_limits(CausalLimits {
            max_depth: 1,
            max_derived_mutations: 100,
        });

        let result = ctx.del(&make_node("root", ""));
        assert!(result.is_err(), "depth-two leaf cascade must be bounded");
        let Err(error) = result else { return };
        assert!(error.message.contains("exceeded max depth 1"));
        assert!(!exists(&registry, "root"), "accepted root DEL is retained");
        assert!(
            !exists(&registry, "branch"),
            "accepted depth-one DEL is retained"
        );
        assert!(
            exists(&registry, "leaf"),
            "unscheduled depth-two DEL remains"
        );
        assert_eq!(ctx.causal_diagnostics().budget_exhaustions, 1);
    }
}

#[cfg(test)]
mod ensure_for_cascade_tests {
    //! `#[ensure_for(X)]` delete-side cleanup. Regression coverage for the
    //! orphan-accumulation bug reported against 4.24.2 (rship bead
    //! rship-e3f): `RelationshipManager::forward_del` handled `belongs_to`
    //! cascades and `owns_many` cleanup, but nothing ever revisited
    //! `ensure_for`-created entities when their dependency was deleted —
    //! `DeleteBindingNode` cascaded the node's `belongs_to` value correctly
    //! but left its `#[ensure_for(BindingNode)]` position behind forever
    //! (4,762 orphaned `BindingNodePosition`s accumulated on the rack, 516
    //! on sandbox, before this fix).

    use std::sync::{Arc, Barrier};

    use uuid::Uuid;

    use self::fixtures::{EnsuredDetail, EnsuredStatus, Node, NodePair, Parent};
    use crate::{
        core::item::AnyItem,
        search::SearchIndex,
        server::{
            HandlerRegistry, MykoServerContext, RelationshipManager, persister::PersisterRouter,
        },
        store::StoreRegistry,
        test_util::scheduler_test_serial,
    };

    // `#[myko_item]` re-imports hyphae traits at module scope, and two
    // invocations in the same module collide — each entity gets its own
    // submodule (mirrors `bench_entities::tree`/`compound_a`/`compound_b`).
    mod fixtures {
        pub use parent::{Parent, ParentId};
        mod parent {
            use crate::prelude::*;

            #[myko_item]
            pub struct Parent {
                pub name: String,
            }
        }

        pub use ensured_status::{EnsuredStatus, EnsuredStatusId};
        mod ensured_status {
            use super::{Parent, ParentId};
            use crate::prelude::*;

            #[myko_item]
            pub struct EnsuredStatus {
                #[ensure_for(Parent)]
                pub parent_id: ParentId,
            }
        }

        pub use ensured_detail::EnsuredDetail;
        mod ensured_detail {
            use super::{EnsuredStatus, EnsuredStatusId};
            use crate::prelude::*;

            #[myko_item]
            pub struct EnsuredDetail {
                #[ensure_for(EnsuredStatus)]
                pub status_id: EnsuredStatusId,
            }
        }

        pub use node::{Node, NodeId};
        mod node {
            use crate::prelude::*;

            #[myko_item]
            pub struct Node {
                pub name: String,
            }
        }

        pub use node_pair::NodePair;
        mod node_pair {
            use super::{Node, NodeId};
            use crate::prelude::*;

            #[myko_item]
            pub struct NodePair {
                #[ensure_for(Node)]
                pub left_id: NodeId,
                #[ensure_for(Node)]
                pub right_id: NodeId,
            }
        }
    }

    fn make_ctx() -> (MykoServerContext, Arc<StoreRegistry>) {
        let registry = Arc::new(StoreRegistry::new());
        let ctx = MykoServerContext::new(
            Uuid::new_v4(),
            registry.clone(),
            Arc::new(HandlerRegistry::new()),
            Arc::new(RelationshipManager::new()),
            Arc::new(PersisterRouter::default()),
            Arc::new(SearchIndex::new()),
            crate::server::MykoServerRuntime {
                peer_clients: Arc::new(dashmap::DashMap::new()),
                event_sink: None,
                history_replay: None,
            },
        );
        (ctx, registry)
    }

    fn make_parent(id: &str) -> Parent {
        Parent {
            id: id.into(),
            name: format!("parent-{id}"),
        }
    }

    /// Every `EnsuredStatus` row currently in the store whose
    /// `#[ensure_for(Parent)]` field points at `parent_id`. The created
    /// row's own id is a random UUID (not derivable from `parent_id`), so
    /// this scans rather than looking up by a known id — same reason
    /// `RelationshipManager`'s own delete-side handler has to scan.
    fn ensured_statuses_for(registry: &StoreRegistry, parent_id: &str) -> Vec<Arc<str>> {
        let Some(store) = registry.get("EnsuredStatus") else {
            return Vec::new();
        };
        store
            .snapshot()
            .into_iter()
            .filter(|(_, item)| {
                item.as_any()
                    .downcast_ref::<EnsuredStatus>()
                    .is_some_and(|status| status.parent_id.as_ref() == parent_id)
            })
            .map(|(id, _)| id)
            .collect()
    }

    fn ensured_details_for(registry: &StoreRegistry, status_id: &str) -> Vec<Arc<str>> {
        let Some(store) = registry.get("EnsuredDetail") else {
            return Vec::new();
        };
        store
            .snapshot()
            .into_iter()
            .filter(|(_, item)| {
                item.as_any()
                    .downcast_ref::<EnsuredDetail>()
                    .is_some_and(|detail| detail.status_id.as_ref() == status_id)
            })
            .map(|(id, _)| id)
            .collect()
    }

    fn node_pairs(registry: &StoreRegistry) -> Vec<Arc<NodePair>> {
        registry
            .get("NodePair")
            .map(|store| {
                store
                    .snapshot()
                    .into_iter()
                    .filter_map(|(_, item)| item.as_any().downcast_ref::<NodePair>().cloned())
                    .map(Arc::new)
                    .collect()
            })
            .unwrap_or_default()
    }

    #[test]
    fn startup_initialization_enforces_recursive_ensure_chain() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();
        let parent: Arc<dyn AnyItem> = Arc::new(make_parent("p1"));
        registry.get_or_create("Parent").insert("p1".into(), parent);

        assert!(RelationshipManager::new().establish_relations(&ctx).is_ok());
        let statuses = ensured_statuses_for(&registry, "p1");
        assert_eq!(statuses.len(), 1);
        let status_id = statuses.first().cloned().unwrap_or_default();
        assert_eq!(ensured_details_for(&registry, &status_id).len(), 1);
    }

    #[test]
    fn concurrent_dependency_sets_create_one_recursive_ensure_chain() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();
        let barrier = Arc::new(Barrier::new(8));

        std::thread::scope(|scope| {
            for _ in 0..8 {
                let barrier = barrier.clone();
                let ctx = &ctx;
                scope.spawn(move || {
                    barrier.wait();
                    assert!(ctx.set(&make_parent("p1")).is_ok());
                });
            }
        });

        let statuses = ensured_statuses_for(&registry, "p1");
        assert_eq!(statuses.len(), 1);
        let status_id = statuses.first().cloned().unwrap_or_default();
        assert_eq!(ensured_details_for(&registry, &status_id).len(), 1);
    }

    #[test]
    fn repeated_dependency_type_enforces_every_cartesian_axis() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();
        let node = |id: &str| Node {
            id: id.into(),
            name: id.to_owned(),
        };

        assert!(ctx.set(&node("n1")).is_ok());
        assert!(ctx.set(&node("n2")).is_ok());
        assert_eq!(node_pairs(&registry).len(), 4);

        assert!(ctx.del(&node("n2")).is_ok());
        let remaining = node_pairs(&registry);
        assert_eq!(remaining.len(), 1);
        assert!(
            remaining
                .iter()
                .all(|pair| { pair.left_id.as_ref() == "n1" && pair.right_id.as_ref() == "n1" })
        );
    }

    #[test]
    fn ensure_for_relationships_are_enforced_transitively() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();

        assert!(ctx.set(&make_parent("p1")).is_ok());
        let statuses = ensured_statuses_for(&registry, "p1");
        assert_eq!(statuses.len(), 1, "first ensure_for level must exist");
        let status_id = statuses.first().cloned().unwrap_or_default();
        assert_eq!(
            ensured_details_for(&registry, &status_id).len(),
            1,
            "an ensured entity must trigger ensure_for relationships that depend on it"
        );

        assert!(ctx.del(&make_parent("p1")).is_ok());
        assert!(ensured_statuses_for(&registry, "p1").is_empty());
        assert!(
            ensured_details_for(&registry, &status_id).is_empty(),
            "recursive enforcement must also remove entities ensured for a cascade-deleted dependency"
        );
    }

    #[test]
    fn del_of_dependency_deletes_its_ensured_entity() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();

        assert!(ctx.set(&make_parent("p1")).is_ok());
        assert_eq!(
            ensured_statuses_for(&registry, "p1").len(),
            1,
            "ensure_for auto-created exactly one EnsuredStatus for p1"
        );

        assert!(ctx.del(&make_parent("p1")).is_ok());

        assert!(
            ensured_statuses_for(&registry, "p1").is_empty(),
            "EnsuredStatus for a deleted dependency must not be orphaned"
        );
    }

    #[test]
    fn batch_del_of_dependencies_deletes_their_ensured_entities() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();

        assert!(ctx.set(&make_parent("p1")).is_ok());
        assert!(ctx.set(&make_parent("p2")).is_ok());
        assert_eq!(ensured_statuses_for(&registry, "p1").len(), 1);
        assert_eq!(ensured_statuses_for(&registry, "p2").len(), 1);

        let p1: Arc<dyn AnyItem> = Arc::new(make_parent("p1"));
        let p2: Arc<dyn AnyItem> = Arc::new(make_parent("p2"));
        assert!(ctx.batch_del_dyn(&[p1, p2]).is_ok());

        assert!(ensured_statuses_for(&registry, "p1").is_empty());
        assert!(ensured_statuses_for(&registry, "p2").is_empty());
    }

    /// A dependency unrelated to `parent_id` must not lose its own
    /// `EnsuredStatus` — the delete-side cleanup must match by FK, not
    /// wipe every `EnsuredStatus` whenever any `Parent` is deleted.
    #[test]
    fn del_of_one_dependency_does_not_orphan_unrelated_ensured_entities() {
        let _serial = scheduler_test_serial();
        let (ctx, registry) = make_ctx();

        assert!(ctx.set(&make_parent("p1")).is_ok());
        assert!(ctx.set(&make_parent("p2")).is_ok());
        assert_eq!(ensured_statuses_for(&registry, "p1").len(), 1);
        assert_eq!(ensured_statuses_for(&registry, "p2").len(), 1);

        assert!(ctx.del(&make_parent("p1")).is_ok());

        assert!(ensured_statuses_for(&registry, "p1").is_empty());
        assert_eq!(
            ensured_statuses_for(&registry, "p2").len(),
            1,
            "p2's EnsuredStatus must survive p1's deletion"
        );
    }
}