cordis-core 0.0.4

use std::sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
};

use futures::{FutureExt, future::BoxFuture};
use tokio::sync::{mpsc, oneshot, watch};

use crate::{
    Context, Error, Plugin, PluginContext, PluginHandle, PluginStatus, Result,
    plugin::{ActivationId, FailurePhase, PluginCommand, PluginId},
    runtime::{DependencySnapshot, Runtime},
    scope::ScopeInner,
};

type ApplyFn = Arc<dyn Fn(PluginContext) -> BoxFuture<'static, Result<()>> + Send + Sync>;

struct Activation {
    scope: Arc<ScopeInner>,
    snapshot: DependencySnapshot,
}

#[derive(Default)]
pub(crate) struct ControlPlane {
    registrations: std::sync::Mutex<Vec<RegistrationRecord>>,
    transition: tokio::sync::Mutex<()>,
}

#[derive(Clone)]
struct RegistrationRecord {
    id: PluginId,
    name: &'static str,
    dependencies: Vec<crate::Dependency>,
    provides: Vec<crate::ServiceDeclaration>,
    commands: mpsc::Sender<PluginCommand>,
    status: watch::Receiver<PluginStatus>,
}

/// Root runtime and owner of every persistent plugin registration.
pub struct App {
    runtime: Arc<Runtime>,
    root: Arc<ScopeInner>,
    control: Arc<ControlPlane>,
    started: AtomicBool,
    shutdown: AtomicBool,
}

impl App {
    pub fn new() -> Self {
        let runtime = Runtime::new();
        let root = ScopeInner::new(runtime.next_id(), "application");
        Self {
            runtime,
            root,
            control: Arc::new(ControlPlane::default()),
            started: AtomicBool::new(false),
            shutdown: AtomicBool::new(false),
        }
    }

    pub fn context(&self) -> Context {
        Context::root(self.runtime.clone())
    }

    pub fn is_started(&self) -> bool {
        self.started.load(Ordering::Acquire)
    }

    /// Registers a plugin and performs its first reconciliation.
    ///
    /// Missing required services produce a suspended handle, not an error.
    /// The same registration is automatically reactivated when declared
    /// service generations appear, disappear, or change.
    pub async fn install<P: Plugin>(&self, plugin: P, config: P::Config) -> Result<PluginHandle> {
        let name = plugin.name();
        let dependencies = plugin.dependencies();
        let provides = plugin.provides();
        let plugin = Arc::new(plugin);
        let config = Arc::new(config);
        let apply: ApplyFn = Arc::new(move |ctx| {
            let plugin = plugin.clone();
            let config = config.clone();
            Box::pin(async move { plugin.apply(ctx, config).await })
        });
        self.install_apply(name, dependencies, provides, apply)
            .await
    }

    pub async fn install_erased(
        &self,
        plugin: Arc<dyn crate::ErasedPlugin>,
        config: crate::ErasedConfig,
    ) -> Result<PluginHandle> {
        let name = plugin.name();
        let dependencies = plugin.dependencies();
        let provides = plugin.provides();
        let apply: ApplyFn = Arc::new(move |ctx| plugin.apply(ctx, config.clone()));
        self.install_apply(name, dependencies, provides, apply)
            .await
    }

    async fn install_apply(
        &self,
        name: &'static str,
        dependencies: Vec<crate::Dependency>,
        provides: Vec<crate::ServiceDeclaration>,
        apply: ApplyFn,
    ) -> Result<PluginHandle> {
        if self.shutdown.load(Ordering::Acquire) {
            return Err(Error::ApplicationShutdown);
        }
        self.ensure_acyclic(name, &dependencies, &provides)?;
        let id = PluginId(self.runtime.next_id());
        let initial = PluginStatus::Suspended {
            missing: dependencies
                .iter()
                .filter(|dependency| dependency.required)
                .map(|dependency| dependency.name)
                .collect::<Vec<_>>()
                .into(),
        };
        let (status_tx, status_rx) = watch::channel(initial.clone());
        let diagnostics = Arc::new(std::sync::Mutex::new(vec![crate::PluginDiagnostic {
            at: std::time::SystemTime::now(),
            status: initial,
        }]));
        let diagnostic_log = diagnostics.clone();
        let mut diagnostic_status = status_rx.clone();
        tokio::spawn(async move {
            while diagnostic_status.changed().await.is_ok() {
                diagnostic_log
                    .lock()
                    .expect("diagnostics lock poisoned")
                    .push(crate::PluginDiagnostic {
                        at: std::time::SystemTime::now(),
                        status: diagnostic_status.borrow().clone(),
                    });
            }
        });
        let (command_tx, command_rx) = mpsc::channel(16);
        let (initialized_tx, initialized_rx) = oneshot::channel();
        let services = self.runtime.subscribe_services();
        let runtime = self.runtime.clone();
        let control = self.control.clone();

        tokio::spawn(run_plugin(
            runtime,
            control,
            id,
            name,
            dependencies.clone(),
            provides.clone(),
            apply,
            command_rx,
            services,
            status_tx,
            initialized_tx,
        ));

        let shutdown_tx = command_tx.clone();
        if let Err(error) = self.root.push(Box::new(move || {
            Box::pin(async move {
                let (reply_tx, reply_rx) = oneshot::channel();
                if shutdown_tx
                    .send(PluginCommand::Dispose(reply_tx))
                    .await
                    .is_err()
                {
                    return Ok(());
                }
                reply_rx.await.unwrap_or(Ok(()))
            })
        })) {
            let (reply_tx, _) = oneshot::channel();
            let _ = command_tx.send(PluginCommand::Dispose(reply_tx)).await;
            return Err(error);
        }

        initialized_rx.await.map_err(|_| Error::PluginDisposed)??;
        self.control
            .registrations
            .lock()
            .expect("registration lock poisoned")
            .push(RegistrationRecord {
                id,
                name,
                dependencies,
                provides,
                commands: command_tx.clone(),
                status: status_rx.clone(),
            });
        Ok(PluginHandle::new(
            id,
            name,
            command_tx,
            status_rx,
            diagnostics,
            self.control.clone(),
        ))
    }

    fn ensure_acyclic(
        &self,
        name: &'static str,
        dependencies: &[crate::Dependency],
        provides: &[crate::ServiceDeclaration],
    ) -> Result<()> {
        let records = self
            .control
            .registrations
            .lock()
            .expect("registration lock poisoned");
        let mut declarations: Vec<_> = records
            .iter()
            .filter(|record| !matches!(*record.status.borrow(), PluginStatus::Disposed))
            .map(|record| {
                (
                    record.name,
                    record.dependencies.clone(),
                    record.provides.clone(),
                )
            })
            .collect();
        declarations.push((name, dependencies.to_vec(), provides.to_vec()));
        if topological_order(&declarations).len() != declarations.len() {
            let names = declarations
                .iter()
                .map(|(name, _, _)| *name)
                .collect::<Vec<_>>()
                .join(" -> ");
            return Err(Error::DependencyCycle(names));
        }
        Ok(())
    }

    pub async fn start(&self) -> Result<()> {
        if self.shutdown.load(Ordering::Acquire) {
            return Err(Error::ApplicationShutdown);
        }
        if !self.started.swap(true, Ordering::AcqRel) {
            self.context().emit(crate::Ready).await?;
        }
        Ok(())
    }

    /// Disposes consumers before their declared providers, independent of
    /// installation order, then clears application-owned resources.
    pub async fn shutdown(&self) -> Result<()> {
        if self.shutdown.swap(true, Ordering::AcqRel) {
            return Ok(());
        }

        let _transition = self.control.transition.lock().await;
        let records = self
            .control
            .registrations
            .lock()
            .expect("registration lock poisoned")
            .clone();
        let declarations = records
            .iter()
            .map(|record| {
                (
                    record.name,
                    record.dependencies.clone(),
                    record.provides.clone(),
                )
            })
            .collect::<Vec<_>>();
        let mut first_error = None;
        for index in topological_order(&declarations).into_iter().rev() {
            let record = &records[index];
            if matches!(*record.status.borrow(), PluginStatus::Disposed) {
                continue;
            }
            let (reply_tx, reply_rx) = oneshot::channel();
            if record
                .commands
                .send(PluginCommand::Dispose(reply_tx))
                .await
                .is_ok()
            {
                if let Ok(Err(error)) = reply_rx.await {
                    first_error.get_or_insert(error);
                }
            }
        }
        if let Err(error) = self.root.dispose().await {
            first_error.get_or_insert(error);
        }
        first_error.map_or(Ok(()), Err)
    }
}

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

#[derive(Clone, Copy)]
enum ControlAction {
    Reload,
    Quiesce,
    Resume,
    Dispose,
}

async fn send_control(sender: &mpsc::Sender<PluginCommand>, action: ControlAction) -> Result<()> {
    let (reply_tx, reply_rx) = oneshot::channel();
    let command = match action {
        ControlAction::Reload => PluginCommand::Reload(reply_tx),
        ControlAction::Quiesce => PluginCommand::Quiesce(reply_tx),
        ControlAction::Resume => PluginCommand::Resume(reply_tx),
        ControlAction::Dispose => PluginCommand::Dispose(reply_tx),
    };
    sender
        .send(command)
        .await
        .map_err(|_| Error::PluginDisposed)?;
    reply_rx.await.map_err(|_| Error::PluginDisposed)?
}

impl ControlPlane {
    pub(crate) async fn reload(&self, id: PluginId) -> Result<()> {
        self.transition(id, false).await
    }

    pub(crate) async fn dispose(&self, id: PluginId) -> Result<()> {
        self.transition(id, true).await
    }

    async fn transition(&self, id: PluginId, disposing: bool) -> Result<()> {
        let _guard = self.transition.lock().await;
        let records = self
            .registrations
            .lock()
            .expect("registration lock poisoned")
            .clone();
        let target = records
            .iter()
            .position(|record| record.id == id)
            .ok_or(Error::PluginDisposed)?;
        let declarations = records
            .iter()
            .map(|record| {
                (
                    record.name,
                    record.dependencies.clone(),
                    record.provides.clone(),
                )
            })
            .collect::<Vec<_>>();
        let order = topological_order(&declarations);
        let dependents = dependent_indices(&declarations, target);

        for index in order.iter().rev().copied() {
            if dependents.contains(&index)
                && !matches!(*records[index].status.borrow(), PluginStatus::Disposed)
            {
                send_control(&records[index].commands, ControlAction::Quiesce).await?;
            }
        }

        let target_action = if disposing {
            ControlAction::Dispose
        } else {
            ControlAction::Reload
        };
        let target_result = send_control(&records[target].commands, target_action).await;

        for index in order {
            if dependents.contains(&index)
                && !matches!(*records[index].status.borrow(), PluginStatus::Disposed)
            {
                let _ = send_control(&records[index].commands, ControlAction::Resume).await;
            }
        }
        target_result
    }
}

fn graph_plan(
    control: &ControlPlane,
    id: PluginId,
) -> Result<(
    Vec<RegistrationRecord>,
    Vec<usize>,
    std::collections::HashSet<usize>,
)> {
    let records = control
        .registrations
        .lock()
        .expect("registration lock poisoned")
        .clone();
    let target = records
        .iter()
        .position(|record| record.id == id)
        .ok_or(Error::PluginDisposed)?;
    let declarations = records
        .iter()
        .map(|record| {
            (
                record.name,
                record.dependencies.clone(),
                record.provides.clone(),
            )
        })
        .collect::<Vec<_>>();
    let order = topological_order(&declarations);
    let dependents = dependent_indices(&declarations, target);
    Ok((records, order, dependents))
}

async fn quiesce_plan(
    records: &[RegistrationRecord],
    order: &[usize],
    dependents: &std::collections::HashSet<usize>,
) -> Result<()> {
    for index in order.iter().rev().copied() {
        if dependents.contains(&index)
            && !matches!(*records[index].status.borrow(), PluginStatus::Disposed)
        {
            send_control(&records[index].commands, ControlAction::Quiesce).await?;
        }
    }
    Ok(())
}

async fn resume_plan(
    records: &[RegistrationRecord],
    order: &[usize],
    dependents: &std::collections::HashSet<usize>,
) {
    for index in order.iter().copied() {
        if dependents.contains(&index)
            && !matches!(*records[index].status.borrow(), PluginStatus::Disposed)
        {
            let _ = send_control(&records[index].commands, ControlAction::Resume).await;
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_plugin(
    runtime: Arc<Runtime>,
    control: Arc<ControlPlane>,
    id: PluginId,
    name: &'static str,
    dependencies: Vec<crate::Dependency>,
    provides: Vec<crate::ServiceDeclaration>,
    apply: ApplyFn,
    mut commands: mpsc::Receiver<PluginCommand>,
    mut services: watch::Receiver<u64>,
    status: watch::Sender<PluginStatus>,
    initialized: oneshot::Sender<Result<()>>,
) {
    let mut active = None;
    let mut failed_generations = None;
    let mut quiesced = false;
    let initial = reconcile(
        &runtime,
        id,
        name,
        &dependencies,
        &apply,
        &status,
        &mut active,
        &mut failed_generations,
        false,
    )
    .await;
    let _ = initialized.send(Ok(()));
    if initial.is_err() {
        // The status channel contains the actionable failure.
    }

    loop {
        tokio::select! {
            biased;
            command = commands.recv() => {
                let Some(command) = command else { break };
                match command {
                    PluginCommand::Reload(reply) | PluginCommand::Retry(reply) => {
                        let result = reconcile(
                            &runtime, id, name, &dependencies, &apply, &status,
                            &mut active, &mut failed_generations, true,
                        ).await;
                        let _ = reply.send(result);
                    }
                    PluginCommand::Quiesce(reply) => {
                        quiesced = true;
                        let result = dispose_activation(&runtime, id, &status, &mut active).await;
                        if result.is_ok() {
                            status.send_replace(PluginStatus::Suspended { missing: Arc::new([]) });
                        }
                        let _ = reply.send(result);
                    }
                    PluginCommand::Resume(reply) => {
                        quiesced = false;
                        let result = reconcile(
                            &runtime, id, name, &dependencies, &apply, &status,
                            &mut active, &mut failed_generations, false,
                        ).await;
                        let _ = reply.send(result);
                    }
                    PluginCommand::Dispose(reply) => {
                        let result = dispose_activation(&runtime, id, &status, &mut active).await;
                        status.send_replace(PluginStatus::Disposed);
                        let _ = reply.send(result);
                        break;
                    }
                }
            }
            changed = services.changed() => {
                if changed.is_err() { break; }
                tokio::time::sleep(std::time::Duration::from_millis(1)).await;
                while services.has_changed().unwrap_or(false) {
                    services.borrow_and_update();
                }
                if quiesced { continue; }
                let snapshot = runtime.dependency_snapshot(&dependencies);
                let changes_active_provider = active.as_ref().is_some_and(|activation| {
                    activation.snapshot.generations != snapshot.generations
                        || !snapshot.missing.is_empty()
                });
                if changes_active_provider && !provides.is_empty() {
                    let _transition = control.transition.lock().await;
                    if let Ok((records, order, dependents)) = graph_plan(&control, id) {
                        let _ = quiesce_plan(&records, &order, &dependents).await;
                        let _ = reconcile(
                            &runtime, id, name, &dependencies, &apply, &status,
                            &mut active, &mut failed_generations, false,
                        ).await;
                        resume_plan(&records, &order, &dependents).await;
                    }
                } else {
                    let _ = reconcile(
                        &runtime, id, name, &dependencies, &apply, &status,
                        &mut active, &mut failed_generations, false,
                    ).await;
                }
            }
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn reconcile(
    runtime: &Arc<Runtime>,
    plugin_id: PluginId,
    name: &'static str,
    dependencies: &[crate::Dependency],
    apply: &ApplyFn,
    status: &watch::Sender<PluginStatus>,
    active: &mut Option<Activation>,
    failed_generations: &mut Option<Vec<(std::any::TypeId, Option<u64>)>>,
    force: bool,
) -> Result<()> {
    let mut snapshot = runtime.dependency_snapshot(dependencies);

    if !snapshot.missing.is_empty() {
        dispose_activation(runtime, plugin_id, status, active).await?;
        *failed_generations = None;
        status.send_replace(PluginStatus::Suspended {
            missing: snapshot.missing,
        });
        return Ok(());
    }

    let unchanged = active
        .as_ref()
        .is_some_and(|activation| activation.snapshot.generations == snapshot.generations);
    if unchanged && !force {
        return Ok(());
    }
    if !force
        && failed_generations
            .as_ref()
            .is_some_and(|failed| *failed == snapshot.generations)
    {
        return Ok(());
    }

    dispose_activation(runtime, plugin_id, status, active).await?;

    // A dependency may change while apply is running. Discard stale staging
    // activations and retry against the latest snapshot, with a small bound to
    // avoid pathological self-triggered loops.
    for _ in 0..8 {
        status.send_replace(PluginStatus::Starting {
            revision: snapshot.revision,
        });
        let activation_id = ActivationId(runtime.next_id());
        let scope = ScopeInner::new(activation_id.0, name);
        let context = Context::for_scope(runtime.clone(), activation_id.0);
        let plugin_context = PluginContext::new(context, scope.clone());

        let applied = std::panic::AssertUnwindSafe(apply(plugin_context))
            .catch_unwind()
            .await;
        let applied = match applied {
            Ok(result) => result,
            Err(payload) => Err(Error::panic(payload)),
        };
        if let Err(error) = applied {
            let _ = scope.dispose().await;
            *failed_generations = Some(snapshot.generations.clone());
            status.send_replace(PluginStatus::Failed {
                phase: FailurePhase::Apply,
                message: error.to_string().into(),
                revision: snapshot.revision,
            });
            return Err(error);
        }

        let after_apply = runtime.dependency_snapshot(dependencies);
        if after_apply.generations != snapshot.generations || !after_apply.missing.is_empty() {
            scope.dispose().await?;
            if !after_apply.missing.is_empty() {
                status.send_replace(PluginStatus::Suspended {
                    missing: after_apply.missing,
                });
                return Ok(());
            }
            snapshot = after_apply;
            continue;
        }

        if let Err(error) = scope.commit().await {
            let _ = scope.dispose().await;
            *failed_generations = Some(snapshot.generations.clone());
            status.send_replace(PluginStatus::Failed {
                phase: FailurePhase::Apply,
                message: error.to_string().into(),
                revision: snapshot.revision,
            });
            return Err(error);
        }
        runtime.commit_owner(activation_id.0);
        *failed_generations = None;
        status.send_replace(PluginStatus::Active {
            activation: activation_id,
            revision: snapshot.revision,
        });
        *active = Some(Activation { scope, snapshot });
        if let Err(error) = runtime
            .emit_serial(crate::Fork {
                plugin: plugin_id,
                activation: activation_id,
            })
            .await
        {
            let _ = dispose_activation(runtime, plugin_id, status, active).await;
            status.send_replace(PluginStatus::Failed {
                phase: FailurePhase::Apply,
                message: error.to_string().into(),
                revision: runtime.dependency_snapshot(dependencies).revision,
            });
            return Err(error);
        }
        return Ok(());
    }

    let error = Error::cleanup("dependency snapshot did not stabilize");
    status.send_replace(PluginStatus::Failed {
        phase: FailurePhase::Apply,
        message: error.to_string().into(),
        revision: snapshot.revision,
    });
    Err(error)
}

async fn dispose_activation(
    runtime: &Arc<Runtime>,
    plugin_id: PluginId,
    status: &watch::Sender<PluginStatus>,
    active: &mut Option<Activation>,
) -> Result<()> {
    let Some(activation) = active.take() else {
        return Ok(());
    };
    let activation_id = ActivationId(activation.scope.id);
    status.send_replace(PluginStatus::Stopping {
        activation: activation_id,
    });
    let event_error = runtime
        .emit_serial(crate::Dispose {
            plugin: plugin_id,
            activation: activation_id,
        })
        .await
        .err();
    let cleanup_error = activation.scope.dispose().await.err();
    if let Some(error) = event_error.or(cleanup_error) {
        status.send_replace(PluginStatus::Failed {
            phase: FailurePhase::Dispose,
            message: error.to_string().into(),
            revision: activation.snapshot.revision,
        });
        return Err(error);
    }
    Ok(())
}

fn topological_order(
    declarations: &[(
        &'static str,
        Vec<crate::Dependency>,
        Vec<crate::ServiceDeclaration>,
    )],
) -> Vec<usize> {
    let count = declarations.len();
    let mut outgoing = vec![Vec::new(); count];
    let mut indegree = vec![0usize; count];

    for (provider_index, (_, _, provided)) in declarations.iter().enumerate() {
        for (consumer_index, (_, dependencies, _)) in declarations.iter().enumerate() {
            let linked = provided.iter().any(|service| {
                dependencies
                    .iter()
                    .any(|dependency| dependency.key == service.key)
            });
            if linked {
                outgoing[provider_index].push(consumer_index);
                indegree[consumer_index] += 1;
            }
        }
    }

    let mut ready = std::collections::VecDeque::new();
    for (index, degree) in indegree.iter().enumerate() {
        if *degree == 0 {
            ready.push_back(index);
        }
    }
    let mut order = Vec::with_capacity(count);
    while let Some(index) = ready.pop_front() {
        order.push(index);
        for consumer in &outgoing[index] {
            indegree[*consumer] -= 1;
            if indegree[*consumer] == 0 {
                ready.push_back(*consumer);
            }
        }
    }
    order
}

fn dependent_indices(
    declarations: &[(
        &'static str,
        Vec<crate::Dependency>,
        Vec<crate::ServiceDeclaration>,
    )],
    provider: usize,
) -> std::collections::HashSet<usize> {
    let mut result = std::collections::HashSet::new();
    let mut pending = vec![provider];
    while let Some(current) = pending.pop() {
        let provided = &declarations[current].2;
        for (index, (_, dependencies, _)) in declarations.iter().enumerate() {
            if index == provider || result.contains(&index) {
                continue;
            }
            if provided.iter().any(|service| {
                dependencies
                    .iter()
                    .any(|dependency| dependency.key == service.key)
            }) {
                result.insert(index);
                pending.push(index);
            }
        }
    }
    result
}