frame-core 0.2.0

//! Linearized component registration and lifecycle operations.

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

use beamr::atom::Atom;
use beamr::namespace::NamespaceId;
use beamr::scheduler::Scheduler;

use crate::capability::{
    Capability, CapabilityCheckError, CapabilityChecker, CapabilityMutationError, CapabilityTable,
    CheckVerdict, HostCapabilityFacade,
};
use crate::component::{ComponentId, ComponentMeta};
use crate::error::{FailureReason, RegistryError};
use crate::event::{EventHub, LifecycleState, LifecycleSubscription};
use crate::status::ComponentStatus;
use crate::supervision::{LifecycleConfig, SharedStatus, StatusState, TreeHandle, start_tree};

/// Result of an idempotent stop operation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StopOutcome {
    /// A live process tree completed the ordered drain.
    Stopped,
    /// No process tree was live, so no transition was needed.
    AlreadyStopped,
}

/// Result of an idempotent remove operation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RemoveOutcome {
    /// Code and registry state were removed after an orderly drain.
    Removed,
    /// Code and registry state were removed after host-forced failed-tree cleanup.
    RemovedAfterForcedCleanup(ForcedCleanupReport),
    /// No registration remained for the identity.
    AlreadyRemoved,
}

/// Registry for component definitions and their native process trees.
pub struct ComponentRegistry {
    scheduler: Arc<Scheduler>,
    config: LifecycleConfig,
    records: Mutex<HashMap<ComponentId, Arc<ComponentRecord>>>,
    capabilities: CapabilityTable,
    events: Arc<EventHub>,
}

impl ComponentRegistry {
    /// Creates a registry over an embedding-owned scheduler.
    ///
    /// The scheduler is never shut down by the registry. The caller must remove
    /// or stop every component before shutting it down.
    #[must_use]
    pub fn new(scheduler: Arc<Scheduler>, config: LifecycleConfig) -> Self {
        Self {
            scheduler,
            config,
            records: Mutex::new(HashMap::new()),
            capabilities: CapabilityTable::new(),
            events: Arc::new(EventHub::default()),
        }
    }

    /// Adds a bounded lifecycle subscriber.
    ///
    /// Each subscriber drops its own oldest event on overflow and increments
    /// its lag counter; publication never waits for a consumer.
    ///
    /// # Errors
    ///
    /// Returns a synchronization failure if a prior panic poisoned the stream.
    pub fn subscribe(
        &self,
        capacity: NonZeroUsize,
    ) -> Result<LifecycleSubscription, RegistryError> {
        self.events
            .subscribe(capacity)
            .map_err(|_| RegistryError::SynchronizationPoisoned)
    }

    /// Borrows the embedding host's sole grant mutation and inspection facade.
    ///
    /// This Rust host value has no BEAM representation and is never passed to a
    /// component entrypoint; component-facing code receives check-only paths.
    #[must_use]
    pub fn host_capabilities(&self) -> HostCapabilityFacade<'_> {
        HostCapabilityFacade::new(&self.capabilities, Arc::clone(&self.events))
    }

    /// Creates a retained check-only path for one registered component.
    ///
    /// # Errors
    ///
    /// Refuses a missing component or poisoned capability table.
    pub fn capability_checker(
        &self,
        component_id: ComponentId,
    ) -> Result<CapabilityChecker, CapabilityMutationError> {
        self.host_capabilities().checker(component_id)
    }

    /// Performs a fresh capability check for one host-mediated consuming act.
    ///
    /// Denials are returned as typed verdicts and emitted exactly once on the
    /// lifecycle stream. No verdict or authority token is retained.
    ///
    /// # Errors
    ///
    /// Refuses a missing component or poisoned table/event synchronization.
    pub fn check_capability(
        &self,
        component_id: ComponentId,
        capability: &Capability,
    ) -> Result<CheckVerdict, CapabilityCheckError> {
        let checker = self
            .capability_checker(component_id)
            .map_err(|error| match error {
                CapabilityMutationError::NotRegistered { component_id } => {
                    CapabilityCheckError::NotRegistered { component_id }
                }
                CapabilityMutationError::UndeclaredNeed { .. }
                | CapabilityMutationError::AlreadyGranted { .. }
                | CapabilityMutationError::SynchronizationPoisoned => {
                    CapabilityCheckError::SynchronizationPoisoned
                }
            })?;
        checker.check(capability)
    }

    /// Validates and atomically registers metadata with its BEAM bytecode.
    ///
    /// # Errors
    ///
    /// Refuses duplicate identities, missing/cyclic dependency edges,
    /// capability collisions, duplicate child keys, and absent/invalid policy.
    pub fn register(&self, meta: ComponentMeta, bytecode: Vec<u8>) -> Result<(), RegistryError> {
        let mut records = self
            .records
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        validate_registration(&records, &meta)?;
        let id = meta.id;
        let needs = meta.needs.clone();
        let record = Arc::new(ComponentRecord::new(meta, bytecode));
        records.insert(id, record);
        if self.capabilities.register_component(id, needs).is_err() {
            records.remove(&id);
            return Err(RegistryError::SynchronizationPoisoned);
        }
        if self
            .events
            .publish_transition(id, None, LifecycleState::Registered)
            .is_err()
        {
            self.capabilities
                .remove_component(id)
                .map_err(|_| RegistryError::SynchronizationPoisoned)?;
            records.remove(&id);
            return Err(RegistryError::SynchronizationPoisoned);
        }
        Ok(())
    }

    /// Starts one component after checking every dependency is Running.
    ///
    /// Successful return means every child answered its declared mailbox probe.
    ///
    /// # Errors
    ///
    /// Returns typed state, dependency, loader, spawn, liveness, or protocol
    /// failures. A runtime failure leaves an observable Failed registration.
    pub fn start(&self, id: ComponentId) -> Result<(), RegistryError> {
        let record = self.record(id)?;
        claim_operation(&record, id, Operation::Start)?;
        if let Err(error) = self.check_running_dependencies(&record.meta) {
            clear_operation(&record)?;
            return Err(error);
        }
        transition(&record, &self.events, id, LifecycleState::Starting, None)?;
        let loaded = match self.scheduler.hot_load_module(&record.bytecode) {
            Ok(loaded) => loaded,
            Err(error) => {
                let reason = FailureReason {
                    child: None,
                    tombstone: None,
                    detail: error.to_string(),
                };
                transition(
                    &record,
                    &self.events,
                    id,
                    LifecycleState::Failed,
                    Some(reason),
                )?;
                clear_operation(&record)?;
                return Err(RegistryError::ModuleLoad {
                    id,
                    detail: error.to_string(),
                });
            }
        };
        record
            .modules
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .push(loaded.module_name);
        if let Err(error) = self.refuse_deferred_bif_imports(id, loaded.module_name) {
            transition(
                &record,
                &self.events,
                id,
                LifecycleState::Failed,
                Some(failure_from_error(&error)),
            )?;
            clear_operation(&record)?;
            return Err(error);
        }
        let policy = record
            .meta
            .supervision
            .clone()
            .ok_or(RegistryError::UndeclaredSupervision { id })?;
        match start_tree(
            id,
            Arc::clone(&self.scheduler),
            &record.meta.children,
            policy,
            self.config,
            &record.status,
            Arc::clone(&self.events),
        ) {
            Ok(tree) => {
                *record
                    .tree
                    .lock()
                    .map_err(|_| RegistryError::SynchronizationPoisoned)? = Some(tree);
                transition(&record, &self.events, id, LifecycleState::Running, None)?;
                clear_operation(&record)
            }
            Err(error) => {
                let reason = failure_from_error(&error);
                transition(
                    &record,
                    &self.events,
                    id,
                    LifecycleState::Failed,
                    Some(reason),
                )?;
                clear_operation(&record)?;
                Err(error)
            }
        }
    }

    /// Stops a running component by draining children in declaration order and
    /// then observing the host supervisor's Normal tombstone.
    ///
    /// # Errors
    ///
    /// Returns typed in-flight, protocol, timeout, or non-normal exit failures.
    pub fn stop(&self, id: ComponentId) -> Result<StopOutcome, RegistryError> {
        let record = self.record(id)?;
        claim_operation(&record, id, Operation::Stop)?;
        let current = state(&record)?;
        if matches!(
            current,
            LifecycleState::Registered | LifecycleState::Stopped | LifecycleState::Failed
        ) {
            clear_operation(&record)?;
            return Ok(StopOutcome::AlreadyStopped);
        }
        transition(&record, &self.events, id, LifecycleState::Stopping, None)?;
        let tree = record
            .tree
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .take()
            .ok_or(RegistryError::MonitorTerminated { id })?;
        match tree.stop(id) {
            Ok(()) => {
                clear_runtime(&record)?;
                transition(&record, &self.events, id, LifecycleState::Stopped, None)?;
                clear_operation(&record)?;
                Ok(StopOutcome::Stopped)
            }
            Err(error) => {
                transition(
                    &record,
                    &self.events,
                    id,
                    LifecycleState::Failed,
                    Some(failure_from_error(&error)),
                )?;
                clear_operation(&record)?;
                Err(error)
            }
        }
    }

    /// Stops if needed, unloads every loaded generation, and unregisters.
    ///
    /// # Errors
    ///
    /// Refuses concurrent operations, running dependents, abnormal drain, unsafe
    /// purge, failed deletion, or post-delete lookup residue.
    pub fn remove(&self, id: ComponentId) -> Result<RemoveOutcome, RegistryError> {
        let Some(record) = self.optional_record(id)? else {
            return Ok(RemoveOutcome::AlreadyRemoved);
        };
        claim_operation(&record, id, Operation::Remove)?;
        if let Err(error) = self.check_running_dependents(id) {
            clear_operation(&record)?;
            return Err(error);
        }
        let current = state(&record)?;
        let mut forced_cleanup = None;
        if current == LifecycleState::Running {
            transition(&record, &self.events, id, LifecycleState::Stopping, None)?;
            let tree = record
                .tree
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?
                .take()
                .ok_or(RegistryError::MonitorTerminated { id })?;
            match tree.stop(id) {
                Ok(()) => {
                    clear_runtime(&record)?;
                    transition(&record, &self.events, id, LifecycleState::Stopped, None)?;
                }
                Err(error) => {
                    transition(
                        &record,
                        &self.events,
                        id,
                        LifecycleState::Failed,
                        Some(failure_from_error(&error)),
                    )?;
                    clear_operation(&record)?;
                    return Err(error);
                }
            }
        } else if current == LifecycleState::Failed {
            forced_cleanup = self.recover_failed_tree(id, &record)?;
        }
        self.unload_modules(id, &record)?;
        let from = state(&record)?;
        self.capabilities
            .remove_component(id)
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        self.events
            .publish_transition(id, Some(from), LifecycleState::Removed)
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        self.records
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .remove(&id);
        Ok(forced_cleanup.map_or(
            RemoveOutcome::Removed,
            RemoveOutcome::RemovedAfterForcedCleanup,
        ))
    }

    /// Sends an integer mailbox command to a named live child.
    ///
    /// # Errors
    ///
    /// Refuses non-running components, unknown children, and monitor failures.
    pub fn send_child(
        &self,
        id: ComponentId,
        child: impl Into<String>,
        message: i64,
    ) -> Result<(), RegistryError> {
        let record = self.record(id)?;
        require_running(&record, id)?;
        record
            .tree
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .as_ref()
            .ok_or(RegistryError::MonitorTerminated { id })?
            .send(id, child.into(), message)
    }

    /// Runs the child's declared liveness mailbox round-trip and returns its
    /// integer state reply.
    ///
    /// # Errors
    ///
    /// Refuses non-running components, unknown children, timeout, or failure.
    pub fn probe_child(
        &self,
        id: ComponentId,
        child: impl Into<String>,
    ) -> Result<i64, RegistryError> {
        let record = self.record(id)?;
        require_running(&record, id)?;
        record
            .tree
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .as_ref()
            .ok_or(RegistryError::MonitorTerminated { id })?
            .probe(id, child.into())
    }

    /// Returns the current status snapshot, or `None` when unregistered.
    ///
    /// # Errors
    ///
    /// Returns a synchronization failure if registry state was poisoned.
    pub fn status(&self, id: ComponentId) -> Result<Option<ComponentStatus>, RegistryError> {
        self.optional_record(id)?
            .map(|record| snapshot(id, &record))
            .transpose()
    }

    /// Returns the number of registrations, including Failed components.
    ///
    /// # Errors
    ///
    /// Returns a synchronization failure if registry state was poisoned.
    pub fn len(&self) -> Result<usize, RegistryError> {
        self.records
            .lock()
            .map(|records| records.len())
            .map_err(|_| RegistryError::SynchronizationPoisoned)
    }

    /// Returns whether the registry has no component definitions.
    ///
    /// # Errors
    ///
    /// Returns a synchronization failure if registry state was poisoned.
    pub fn is_empty(&self) -> Result<bool, RegistryError> {
        self.len().map(|length| length == 0)
    }

    fn record(&self, id: ComponentId) -> Result<Arc<ComponentRecord>, RegistryError> {
        self.optional_record(id)?
            .ok_or(RegistryError::NotRegistered { id })
    }

    fn optional_record(
        &self,
        id: ComponentId,
    ) -> Result<Option<Arc<ComponentRecord>>, RegistryError> {
        self.records
            .lock()
            .map(|records| records.get(&id).cloned())
            .map_err(|_| RegistryError::SynchronizationPoisoned)
    }

    fn check_running_dependencies(&self, meta: &ComponentMeta) -> Result<(), RegistryError> {
        for required in &meta.requires {
            let record = self.record(*required)?;
            let required_state = state(&record)?;
            if required_state != LifecycleState::Running {
                return Err(RegistryError::DependencyNotRunning {
                    component: meta.id,
                    required: *required,
                    state: required_state,
                });
            }
        }
        Ok(())
    }

    fn check_running_dependents(&self, id: ComponentId) -> Result<(), RegistryError> {
        let records = self
            .records
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        for (dependent, record) in records.iter() {
            if record.meta.requires.contains(&id) && state(record)? == LifecycleState::Running {
                return Err(RegistryError::RunningDependent {
                    dependent: *dependent,
                    required: id,
                });
            }
        }
        Ok(())
    }

    fn unload_modules(
        &self,
        id: ComponentId,
        record: &ComponentRecord,
    ) -> Result<(), RegistryError> {
        let modules = std::mem::take(
            &mut *record
                .modules
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?,
        );
        for module in unique_modules(modules) {
            let name = self.module_name(module);
            if self.scheduler.check_old_code(module) {
                self.scheduler.purge_module(module).map_err(|error| {
                    RegistryError::ModulePurge {
                        id,
                        module: name.clone(),
                        detail: error.to_string(),
                    }
                })?;
            }
            if !self.scheduler.delete_module(module) {
                return Err(RegistryError::ModuleDelete { id, module: name });
            }
            if self
                .scheduler
                .lookup_module_in(NamespaceId::DEFAULT, module)
                .is_some()
            {
                return Err(RegistryError::ModuleStillLoaded { id, module: name });
            }
        }
        Ok(())
    }

    fn module_name(&self, module: Atom) -> String {
        crate::composition::resolve_atom(&self.scheduler, module)
    }

    /// Refuses a freshly hot-loaded module whose committed import table still
    /// defers any `erlang:*` entry (the scan itself, shared with the runtime
    /// wrapper's support-module path, is
    /// [`crate::composition::deferred_erlang_imports`] — see its docs for why
    /// a deferred built-in is process-fatal at first dispatch).
    fn refuse_deferred_bif_imports(
        &self,
        id: ComponentId,
        module: Atom,
    ) -> Result<(), RegistryError> {
        let Some(deferred) = crate::composition::deferred_erlang_imports(&self.scheduler, module)
        else {
            return Err(RegistryError::ModuleLoad {
                id,
                detail: "committed module was absent immediately after hot load".to_owned(),
            });
        };
        if deferred.is_empty() {
            return Ok(());
        }
        Err(RegistryError::DeferredBifImports {
            id,
            module: self.module_name(module),
            imports: deferred.join(", "),
        })
    }
}

mod recovery;
mod support;

pub use recovery::{ChildCleanup, ForcedCleanupReport, ProcessCleanup};
use support::{
    ComponentRecord, Operation, claim_operation, clear_operation, clear_runtime,
    failure_from_error, require_running, snapshot, state, transition, unique_modules,
    validate_registration,
};