cordis_core/

plugin.rs

use std::{
    future::Future,
    sync::{Arc, Mutex},
    time::SystemTime,
};

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

use crate::{Dependency, Error, PluginContext, Result};

/// Strongly typed plugin API. Implementations can use native `async fn` on
/// Rust 1.85+; no proc-macro is required.
pub trait Plugin: Send + Sync + 'static {
    type Config: Send + Sync + 'static;

    fn name(&self) -> &'static str;

    fn dependencies(&self) -> Vec<Dependency> {
        Vec::new()
    }

    fn provides(&self) -> Vec<crate::ServiceDeclaration> {
        Vec::new()
    }

    fn apply(
        &self,
        ctx: PluginContext,
        config: Arc<Self::Config>,
    ) -> impl Future<Output = Result<()>> + Send;
}

pub type ErasedConfig = Arc<dyn std::any::Any + Send + Sync>;

/// Object-safe boundary for configuration-driven registries. Serialization is
/// deliberately left to the framework using this crate.
pub trait ErasedPlugin: Send + Sync + 'static {
    fn name(&self) -> &'static str;

    fn dependencies(&self) -> Vec<Dependency> {
        Vec::new()
    }

    fn provides(&self) -> Vec<crate::ServiceDeclaration> {
        Vec::new()
    }

    fn apply(&self, ctx: PluginContext, config: ErasedConfig) -> BoxFuture<'static, Result<()>>;
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct PluginId(pub(crate) u64);

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct ActivationId(pub(crate) u64);

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FailurePhase {
    Apply,
    Dispose,
}

#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum PluginStatus {
    Suspended {
        missing: Arc<[&'static str]>,
    },
    Starting {
        revision: u64,
    },
    Active {
        activation: ActivationId,
        revision: u64,
    },
    Stopping {
        activation: ActivationId,
    },
    Failed {
        phase: FailurePhase,
        message: Arc<str>,
        revision: u64,
    },
    Disposed,
}

#[derive(Clone, Debug)]
pub struct PluginDiagnostic {
    pub at: SystemTime,
    pub status: PluginStatus,
}

pub(crate) enum PluginCommand {
    Reload(oneshot::Sender<Result<()>>),
    Retry(oneshot::Sender<Result<()>>),
    Quiesce(oneshot::Sender<Result<()>>),
    Resume(oneshot::Sender<Result<()>>),
    Dispose(oneshot::Sender<Result<()>>),
}

/// Persistent plugin registration. Its activation may be repeatedly created
/// and disposed as dependency generations change.
#[must_use = "keep the handle to inspect status or call dispose().await"]
#[derive(Clone)]
pub struct PluginHandle {
    id: PluginId,
    name: &'static str,
    commands: mpsc::Sender<PluginCommand>,
    status: watch::Receiver<PluginStatus>,
    diagnostics: Arc<Mutex<Vec<PluginDiagnostic>>>,
    control: Arc<crate::app::ControlPlane>,
    dispose_lock: Arc<tokio::sync::Mutex<()>>,
    dispose_result: Arc<Mutex<Option<Option<Arc<str>>>>>,
}

impl PluginHandle {
    pub(crate) fn new(
        id: PluginId,
        name: &'static str,
        commands: mpsc::Sender<PluginCommand>,
        status: watch::Receiver<PluginStatus>,
        diagnostics: Arc<Mutex<Vec<PluginDiagnostic>>>,
        control: Arc<crate::app::ControlPlane>,
    ) -> Self {
        Self {
            id,
            name,
            commands,
            status,
            diagnostics,
            control,
            dispose_lock: Arc::new(tokio::sync::Mutex::new(())),
            dispose_result: Arc::new(Mutex::new(None)),
        }
    }

    pub fn id(&self) -> PluginId {
        self.id
    }

    pub fn name(&self) -> &'static str {
        self.name
    }

    pub fn status(&self) -> PluginStatus {
        self.status.borrow().clone()
    }

    pub fn subscribe(&self) -> watch::Receiver<PluginStatus> {
        self.status.clone()
    }

    pub fn diagnostics(&self) -> Vec<PluginDiagnostic> {
        self.diagnostics
            .lock()
            .expect("diagnostics lock poisoned")
            .clone()
    }

    pub async fn wait_active(&self) -> Result<ActivationId> {
        let mut status = self.status.clone();
        loop {
            let current = status.borrow().clone();
            match current {
                PluginStatus::Active { activation, .. } => return Ok(activation),
                PluginStatus::Failed { message, .. } => {
                    return Err(Error::PluginFailed(message.to_string()));
                }
                PluginStatus::Disposed => return Err(Error::PluginDisposed),
                _ => {}
            }
            status.changed().await.map_err(|_| Error::PluginDisposed)?;
        }
    }

    pub async fn reload(&self) -> Result<()> {
        self.control.reload(self.id).await
    }

    pub async fn retry(&self) -> Result<()> {
        self.request(PluginCommand::Retry).await
    }

    async fn request(
        &self,
        make: impl FnOnce(oneshot::Sender<Result<()>>) -> PluginCommand,
    ) -> Result<()> {
        let (tx, rx) = oneshot::channel();
        self.commands
            .send(make(tx))
            .await
            .map_err(|_| Error::PluginDisposed)?;
        rx.await.map_err(|_| Error::PluginDisposed)?
    }

    pub async fn dispose(&self) -> Result<()> {
        let _guard = self.dispose_lock.lock().await;
        if let Some(result) = self
            .dispose_result
            .lock()
            .expect("dispose result lock poisoned")
            .clone()
        {
            return result.map_or(Ok(()), |message| Err(Error::cleanup(message)));
        }

        if matches!(self.status(), PluginStatus::Disposed) {
            *self
                .dispose_result
                .lock()
                .expect("dispose result lock poisoned") = Some(None);
            return Ok(());
        }

        let result = self.control.dispose(self.id).await;
        *self
            .dispose_result
            .lock()
            .expect("dispose result lock poisoned") = Some(
            result
                .as_ref()
                .err()
                .map(|error| Arc::<str>::from(error.to_string())),
        );
        result
    }
}

/// Backwards-compatible name for the persistent plugin registration.
pub type PluginScope = PluginHandle;