aion-rs 0.30.0

//! Runtime-local child NIF adapters for AT child services.

use std::sync::Arc;

use aion_core::{Event, WorkflowId};
use aion_store::EventStore;
use aion_store::visibility::VisibilityStore;
use tokio::runtime::Handle;

use crate::EngineError;
use crate::durability::DurabilityError;
use crate::engine_seam::{
    ChildWorkflowSpawnRequest, ChildWorkflowSpawnResult, EngineHandle, EngineSeamError,
    TimerWheelEntry, WorkflowMailboxMessage, WorkflowProcessHandle, WorkflowResidency,
};
use crate::lifecycle::start::{
    StartWorkflowContext, StartWorkflowOptions, start_workflow_with_options,
};
use crate::loader::WorkflowCatalog;
use crate::registry::{HandleResidency, Registry, WorkflowHandle};
use crate::runtime::engine_tasks::EngineTaskRuntime;
use crate::runtime::{RuntimeHandle, SignalDeliveryConfig};
use crate::signal::SignalResumeHandoff;
use crate::supervision::SupervisionTree;

/// Engine-owned context for child workflow NIF calls.
pub(crate) struct ChildNifBridge {
    store: Arc<dyn EventStore>,
    visibility_store: Arc<dyn VisibilityStore>,
    runtime: Arc<RuntimeHandle>,
    catalog: Arc<WorkflowCatalog>,
    registry: Arc<Registry>,
    supervision: Arc<SupervisionTree>,
    signal_handoff: Arc<SignalResumeHandoff>,
    search_attribute_schema: Arc<aion_core::SearchAttributeSchema>,
    tokio_handle: Handle,
    /// Builder-supplied backoff policy for watcher registry-miss windows,
    /// transient record retries, and spawn-recovery retries.
    watch_backoff: SignalDeliveryConfig,
}

/// Constructor dependencies for [`ChildNifBridge`].
pub(crate) struct ChildNifBridgeParts {
    pub(crate) store: Arc<dyn EventStore>,
    pub(crate) visibility_store: Arc<dyn VisibilityStore>,
    pub(crate) runtime: Arc<RuntimeHandle>,
    pub(crate) catalog: Arc<WorkflowCatalog>,
    pub(crate) registry: Arc<Registry>,
    pub(crate) supervision: Arc<SupervisionTree>,
    pub(crate) signal_handoff: Arc<SignalResumeHandoff>,
    pub(crate) search_attribute_schema: Arc<aion_core::SearchAttributeSchema>,
    pub(crate) tokio_handle: Handle,
    pub(crate) watch_backoff: SignalDeliveryConfig,
}

impl ChildNifBridge {
    /// Creates a bridge from engine components.
    ///
    /// Infallible: the bridge now borrows the engine task executor from the
    /// runtime handle rather than starting one of its own, so there is nothing
    /// left here that can fail.
    pub(crate) fn new(parts: ChildNifBridgeParts) -> Self {
        let ChildNifBridgeParts {
            store,
            visibility_store,
            runtime,
            catalog,
            registry,
            supervision,
            signal_handoff,
            search_attribute_schema,
            tokio_handle,
            watch_backoff,
        } = parts;
        Self {
            store,
            visibility_store,
            runtime,
            catalog,
            registry,
            supervision,
            signal_handoff,
            search_attribute_schema,
            tokio_handle,
            watch_backoff,
        }
    }

    pub(crate) fn registry(&self) -> &Registry {
        self.registry.as_ref()
    }

    pub(crate) fn registry_arc(&self) -> Arc<Registry> {
        Arc::clone(&self.registry)
    }

    pub(crate) fn store(&self) -> Arc<dyn EventStore> {
        Arc::clone(&self.store)
    }

    pub(crate) fn runtime(&self) -> Arc<RuntimeHandle> {
        Arc::clone(&self.runtime)
    }

    pub(crate) fn tokio_handle(&self) -> Handle {
        self.tokio_handle.clone()
    }

    /// The engine-wide task executor, borrowed from the runtime handle.
    ///
    /// The bridge does not own one: an optional bridge must not be the only
    /// route to an executor that a non-optional lifecycle path depends on.
    ///
    /// Named `engine_tasks`, not `child_tasks`: the executor it returns also
    /// carries the process-exit completion retry, which is not a child concern
    /// at all. A borrowed name that still says "child" is how the next reader
    /// re-derives the assumption this method exists to correct.
    pub(crate) fn engine_tasks(&self) -> Arc<EngineTaskRuntime> {
        self.runtime.engine_tasks()
    }

    pub(crate) fn watch_backoff(&self) -> SignalDeliveryConfig {
        self.watch_backoff
    }

    /// Resolves a same-package child at the parent's exact package version,
    /// falling back to the routed version for an explicitly separate child.
    ///
    /// Same-archive entries share one content hash, so this exact lookup keeps
    /// child starts pinned across a redeploy of either route.
    pub(crate) fn package_version_for_child(
        &self,
        workflow_type: &str,
        parent_version: &aion_package::ContentHash,
    ) -> Result<Option<aion_core::PackageVersion>, EngineError> {
        if let Some(workflow) = self.catalog.get(workflow_type, parent_version)? {
            return Ok(Some(crate::loader::package_version_of(workflow.version())));
        }
        self.catalog.routed_version(workflow_type)
    }

    /// Abort every child-terminal watcher armed by an exited parent pid.
    pub(crate) fn abort_child_terminal_watches_for_parent(&self, parent_pid: u64) {
        self.engine_tasks().abort_watches_for_parent(parent_pid);
    }

    /// Close the engine-task epoch: gate new arms, abort every task, and await
    /// each to quiescence (F4).
    ///
    /// Idempotent, and no longer the only route to that close — `RuntimeHandle`
    /// owns the executor and closes it from its own `shutdown`. This call
    /// remains because a bridge that installed watchers should be able to say
    /// so, not because anything depends on it happening here.
    pub(crate) fn shutdown_engine_tasks(&self) {
        self.engine_tasks().shutdown();
    }

    /// Start the child under its parent-recorded identity, inheriting the
    /// parent's current search attributes and namespace.
    ///
    /// Shared by the synchronous spawn path and the background
    /// spawn-recovery retry (F3): both must start exactly the recorded
    /// identity, and both inherit visibility metadata from the parent's
    /// recorded history at start time.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError`] when the parent history cannot be read or the
    /// start path fails.
    pub(super) async fn start_child_under_recorded_id(
        &self,
        parent_workflow_id: &WorkflowId,
        parent_namespace: &str,
        request: ChildWorkflowSpawnRequest,
    ) -> Result<WorkflowHandle, EngineError> {
        // Children inherit the parent's current search attributes so
        // visibility metadata (such as a server-assigned tenancy attribute)
        // follows the execution tree.
        let parent_history = self.store.read_history(parent_workflow_id).await?;
        let inherited = aion_core::search_attributes_from_events(&parent_history);
        // The child wears its parent (aion#77): the parent's workflow id and
        // its CURRENT run id (the latest recorded start in the history this
        // spawn already reads) travel into the child's own WorkflowStarted,
        // so a handoff chain is traversable backward from any run.
        let loaded_version =
            crate::loader::parse_package_version(&request.workflow_type, &request.package_version)?;
        start_workflow_with_options(
            StartWorkflowContext {
                store: Arc::clone(&self.store),
                visibility_store: Arc::clone(&self.visibility_store),
                catalog: Arc::clone(&self.catalog),
                runtime: Arc::clone(&self.runtime),
                supervision: Arc::clone(&self.supervision),
                registry: Arc::clone(&self.registry),
                signal_handoff: Some(Arc::clone(&self.signal_handoff)),
                search_attribute_schema: Arc::clone(&self.search_attribute_schema),
                // Epoch-stable: the host runtime's handle, never the
                // child-task runtime polling a spawn-recovery attempt.
                monitor_tokio_handle: self.tokio_handle.clone(),
            },
            &request.workflow_type,
            request.input,
            StartWorkflowOptions {
                // Record-then-spawn (#56): the parent already recorded
                // ChildWorkflowStarted under this pre-allocated id, so the
                // child must start under exactly this identity — and on
                // exactly the recorded package version (D1).
                workflow_id: Some(request.child_workflow_id),
                loaded_version: Some(loaded_version),
                search_attributes: inherited,
                namespace: Some(parent_namespace.to_owned()),
                // The spawn parent is carried by `parent_workflow_id` ALONE.
                // `parent_run_id` is the continue-as-new predecessor link and
                // stays None on spawn — the store's run-chain root rule
                // (aion-store run_chain.rs) depends on that single meaning.
                parent_workflow_id: Some(request.parent_workflow_id.clone()),
                ..StartWorkflowOptions::default()
            },
        )
        .await
    }

    /// Start a DETACHED hatch under its caller-recorded identity (R13.1).
    ///
    /// Deliberately NOT `start_child_under_recorded_id`, and the differences
    /// are the whole point of the hatch:
    ///
    /// - **No `parent_workflow_id`.** A hatch is not a child: there is no
    ///   lifecycle tie, no supervision edge, and nothing awaits its terminal.
    ///   Recording a parent edge would make the store's run-chain treat the
    ///   hatched workflow as spawned work, which is exactly the tie R13.1
    ///   says a hatch does not create.
    /// - **No inherited search attributes.** A child inherits the parent's
    ///   visibility metadata because it IS the parent's work; a detached
    ///   top-level workflow is not, so it starts with the attributes the
    ///   caller passed and nothing else.
    /// - **Routed version, not the caller's pin.** A child is pinned to its
    ///   parent's exact package version so a handoff chain stays coherent
    ///   across a redeploy. A hatch is a top-level start whose lifetime is
    ///   independent of the hatching run, so it resolves the CURRENT routed
    ///   version exactly as an operator start would.
    /// - **`InputAdmission::Declared`, exactly as an operator start.** The
    ///   input is measured against the hatched type's declared input schema
    ///   before any append. This used to fall through to the struct default
    ///   (`Trusted`), which is the engine-internal re-entry mode meaning "the
    ///   input is already contract-shaped" — true of a continue-as-new carry
    ///   the engine itself produced, and false of a hatch payload, which comes
    ///   from workflow code and may have come from outside it. The verb form
    ///   of this operation (`Engine::hatch_workflow`) goes through
    ///   `start_workflow_with_id` and is admitted as `Declared`, so the NIF
    ///   path was quietly the laxer of the two doors into the same start.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError`] when the workflow type is not loaded, the input
    /// fails declared-schema admission, or the start path fails. A
    /// [`aion_store::StoreError::SequenceConflict`] — the dedupe race, where a
    /// concurrent hatch of the same identity won the first append — is NOT
    /// handled here; the caller resolves it to the winner's workflow.
    pub(super) async fn start_hatched_under_recorded_id(
        &self,
        namespace: &str,
        workflow_type: &str,
        hatch_workflow_id: WorkflowId,
        input: aion_core::Payload,
        package_version: aion_core::PackageVersion,
    ) -> Result<WorkflowHandle, EngineError> {
        let loaded_version = crate::loader::parse_package_version(workflow_type, &package_version)?;
        start_workflow_with_options(
            StartWorkflowContext {
                store: Arc::clone(&self.store),
                visibility_store: Arc::clone(&self.visibility_store),
                catalog: Arc::clone(&self.catalog),
                runtime: Arc::clone(&self.runtime),
                supervision: Arc::clone(&self.supervision),
                registry: Arc::clone(&self.registry),
                signal_handoff: Some(Arc::clone(&self.signal_handoff)),
                search_attribute_schema: Arc::clone(&self.search_attribute_schema),
                monitor_tokio_handle: self.tokio_handle.clone(),
            },
            workflow_type,
            input,
            StartWorkflowOptions {
                // The deterministic hatch identity, derived by the caller from
                // (namespace, type, key) and re-derivable on every replay.
                workflow_id: Some(hatch_workflow_id),
                namespace: Some(namespace.to_owned()),
                // The version the caller resolved and pre-checked, so the
                // start cannot silently land on a different one than the
                // refusal path measured.
                loaded_version: Some(loaded_version),
                input_admission: crate::lifecycle::start::InputAdmission::Declared,
                ..StartWorkflowOptions::default()
            },
        )
        .await
    }

    /// The CURRENT routed package version for `workflow_type`, or `None` when
    /// no version of it is loaded on this engine.
    ///
    /// The hatch's counterpart to [`Self::package_version_for_child`]: a child
    /// is pinned to its parent's exact version, a hatch resolves the routed one
    /// exactly as an operator start would.
    ///
    /// # Errors
    ///
    /// Propagates catalog lookup failures. A failure is NOT "not loaded": the
    /// caller must refuse rather than treat an unreadable catalog as an absent
    /// workflow type.
    pub(crate) fn routed_package_version(
        &self,
        workflow_type: &str,
    ) -> Result<Option<aion_core::PackageVersion>, EngineError> {
        self.catalog.routed_version(workflow_type)
    }
}

pub(crate) struct NifChildEngine {
    bridge: Arc<ChildNifBridge>,
    parent: WorkflowHandle,
}

impl NifChildEngine {
    #[must_use]
    pub(crate) fn new(bridge: Arc<ChildNifBridge>, parent: WorkflowHandle) -> Self {
        Self { bridge, parent }
    }
}

impl EngineHandle for NifChildEngine {
    fn resolve_workflow(
        &self,
        workflow_id: &WorkflowId,
    ) -> Result<WorkflowResidency, EngineSeamError> {
        let handle = self
            .bridge
            .registry
            .list()
            .map_err(|error| EngineSeamError::Delivery {
                reason: error.to_string(),
            })?
            .into_iter()
            .find(|handle| handle.workflow_id() == workflow_id);
        match handle {
            Some(handle) if handle.residency() == HandleResidency::Resident => Ok(
                WorkflowResidency::Resident(WorkflowProcessHandle::new(handle.pid())),
            ),
            Some(_) => Ok(WorkflowResidency::NonResident),
            None => Ok(WorkflowResidency::Unknown),
        }
    }

    fn deliver_workflow_message(
        &self,
        process: WorkflowProcessHandle,
        message: WorkflowMailboxMessage,
    ) -> Result<(), EngineSeamError> {
        match message {
            WorkflowMailboxMessage::SignalReceived { .. } => self
                .bridge
                .runtime
                .deliver_signal_received(process.pid())
                .map_err(|error| EngineSeamError::Delivery {
                    reason: error.to_string(),
                }),
            other => Err(EngineSeamError::Delivery {
                reason: format!("unsupported child NIF message: {other:?}"),
            }),
        }
    }

    fn spawn_child_workflow(
        &self,
        request: ChildWorkflowSpawnRequest,
    ) -> Result<ChildWorkflowSpawnResult, EngineSeamError> {
        let parent_workflow_id = self.parent.workflow_id().clone();
        let parent_namespace = self.parent.namespace().to_owned();
        let child = self
            .bridge
            .tokio_handle
            .block_on(self.bridge.start_child_under_recorded_id(
                &parent_workflow_id,
                &parent_namespace,
                request,
            ))
            .map_err(|error| EngineSeamError::ChildSpawn {
                reason: error.to_string(),
            })?;
        Ok(ChildWorkflowSpawnResult {
            child_workflow_id: child.workflow_id().clone(),
            child_process: WorkflowProcessHandle::new(child.pid()),
        })
    }

    fn terminate_linked_child_workflow(
        &self,
        _parent_workflow_id: &WorkflowId,
        child_process: WorkflowProcessHandle,
        _correlation: u64,
    ) -> Result<(), EngineSeamError> {
        self.bridge
            .runtime
            .cancel_pid(child_process.pid())
            .map_err(|error| EngineSeamError::ChildTermination {
                reason: error.to_string(),
            })
    }

    fn terminate_linked_activity(
        &self,
        _parent_workflow_id: &WorkflowId,
        activity_process: crate::Pid,
        _correlation: u64,
    ) -> Result<(), EngineSeamError> {
        self.bridge
            .runtime
            .cancel_pid(activity_process)
            .map_err(|error| EngineSeamError::ChildTermination {
                reason: error.to_string(),
            })
    }

    fn arm_timer(&self, entry: TimerWheelEntry) -> Result<(), EngineSeamError> {
        let _ = entry;
        Err(EngineSeamError::TimerWheel {
            reason: "child NIF engine cannot arm timers".to_owned(),
        })
    }

    fn disarm_timer(
        &self,
        process: WorkflowProcessHandle,
        timer_id: &aion_core::TimerId,
    ) -> Result<(), EngineSeamError> {
        let _ = (process, timer_id);
        Err(EngineSeamError::TimerWheel {
            reason: "child NIF engine cannot disarm timers".to_owned(),
        })
    }

    fn record_workflow_event(
        &self,
        workflow_id: &WorkflowId,
        event: Event,
    ) -> Result<crate::engine_seam::RecordOutcome, EngineSeamError> {
        if workflow_id != self.parent.workflow_id() {
            return Err(EngineSeamError::Recorder {
                reason: format!("cannot record child event for unrelated workflow {workflow_id}"),
            });
        }
        record_child_event(&self.bridge.tokio_handle, &self.parent, event)
            .map(|()| crate::engine_seam::RecordOutcome::Recorded)
    }

    fn record_redelivered_timer_fire(
        &self,
        workflow_id: &WorkflowId,
        timer_id: &aion_core::TimerId,
    ) -> Result<crate::engine_seam::RedeliveredFire, EngineSeamError> {
        let _ = (workflow_id, timer_id);
        Err(EngineSeamError::Recorder {
            reason: "the child engine seam does not answer timer redeliveries".to_owned(),
        })
    }
}

fn record_child_event(
    tokio_handle: &Handle,
    parent: &WorkflowHandle,
    event: Event,
) -> Result<(), EngineSeamError> {
    let recorder = parent.recorder();
    tokio_handle
        .block_on(async {
            let mut recorder = recorder.lock().await;
            match event {
                Event::ChildWorkflowStarted {
                    child_workflow_id,
                    workflow_type,
                    input,
                    package_version,
                    envelope,
                } => {
                    recorder
                        .record_child_workflow_started(
                            envelope.recorded_at,
                            child_workflow_id,
                            workflow_type,
                            input,
                            package_version,
                        )
                        .await
                }
                Event::ChildWorkflowCompleted {
                    child_workflow_id,
                    result,
                    envelope,
                } => {
                    recorder
                        .record_child_workflow_completed(
                            envelope.recorded_at,
                            child_workflow_id,
                            result,
                        )
                        .await
                }
                Event::ChildWorkflowFailed {
                    child_workflow_id,
                    error,
                    envelope,
                } => {
                    recorder
                        .record_child_workflow_failed(
                            envelope.recorded_at,
                            child_workflow_id,
                            error,
                        )
                        .await
                }
                other => Err(DurabilityError::HistoryShape {
                    reason: format!("child NIF cannot record non-child event: {other:?}"),
                }),
            }
        })
        .map_err(|error| EngineSeamError::Recorder {
            reason: error.to_string(),
        })
}