ek-core 0.1.0

//! `MainEngine` —— 纯插件宿主:单表注册、(typed)获取、(deps)拓扑 start、
//! 逆序五步 close、单插件 remove(允许+警告)、运行统计查询。零业务方法。
//!
//! close 五步(每插件,逆依赖序依次执行):
//! ①卸 forwarder(unregister 全部 HandlerId)
//! ②on_stop 钩子(业务清理)
//! ③关邮箱(宿主侧 Sender 置 None;全部 send 端消费尽 → 信道断开)
//! ④worker 排空(crossbeam recv 先取尽缓冲再报 Disconnected)
//! ⑤join 带超时:超时则告警 + 剥离句柄(线程自然排空收尾,Rust 不杀线程)

use crate::context::PluginContext;
use crate::error::PluginError;
use crate::forwarder::{make_forwarder, Forwarder};
use crate::log_plugin::{LogData, EVENT_LOG, LOG_LEVEL_WARN};
use crate::plugin::{CLOSE_JOIN_TIMEOUT, Plugin, valid_plugin_name};
use crate::registry::{EntryStats, PluginEntry, PluginStateCell, PluginStateCode, Registry};
use crate::worker::spawn_worker;
use ek_event::{EventEngine, HandlerId};
use std::any::Any;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant};

/// 插件生命周期状态(worker 状态机,AtomicU8 CAS 迁移)。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PluginState {
    Ready,
    Running,
    Stopping,
    Exited,
    Quarantined,
}

impl PluginState {
    pub(crate) fn from_code(c: u8) -> Self {
        match c {
            0 => Self::Ready,
            1 => Self::Running,
            2 => Self::Stopping,
            3 => Self::Exited,
            _ => Self::Quarantined,
        }
    }
    fn from_code_enum(code: PluginStateCode) -> Self {
        Self::from_code(code as u8)
    }
}

/// 插件信息快照(get_plugin)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PluginInfo {
    pub name: String,
    pub state: PluginState,
    pub deps: Vec<String>,
}

/// 邮箱运行统计(mail_stats)。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct MailStats {
    /// forwarder 溢出丢弃累计。
    pub spilled: u64,
    /// handler panic 累计。
    pub panics: u64,
}

fn log_event(source: &str, msg: String) -> ek_event::Event {
    ek_event::Event::with_data(
        EVENT_LOG,
        Arc::new(LogData {
            level: LOG_LEVEL_WARN,
            source: source.to_string(),
            msg,
        }),
    )
}

pub struct MainEngine {
    event: Arc<EventEngine>,
    registry: Arc<Registry>,
    /// start() 装配顺序纪录(close 反向使用);None = 未 start。
    started_order: Mutex<Option<Vec<String>>>,
    closed: RwLock<bool>,
}

impl MainEngine {
    /// 注入式创建:`Some(bus)` 复用外部引擎;`None` 自建(start() 时拉起)。
    pub fn new(event: Option<Arc<EventEngine>>) -> Self {
        Self {
            event: event.unwrap_or_else(|| Arc::new(EventEngine::new())),
            registry: Arc::new(Registry::new()),
            started_order: Mutex::new(None),
            closed: RwLock::new(false),
        }
    }

    /// 宿主级引擎句柄(测试/装配工具可用;插件不可见——ctx 不暴露)。
    pub fn event_engine(&self) -> &Arc<EventEngine> {
        &self.event
    }

    fn closed_check(&self) -> Result<(), PluginError> {
        if *self.closed.read().unwrap() {
            Err(PluginError::Stopped)
        } else {
            Ok(())
        }
    }

    fn host_log(&self, msg: String) {
        let _ = self.event.put(log_event("ek-core", msg));
    }

    /// 统一格式注册(名/D2、重名、同名自依赖、deps 存在性校验)。
    pub fn add_plugin(&self, plugin: Arc<impl Plugin + 'static>) -> Result<(), PluginError> {
        self.closed_check()?;
        if self.started_order.lock().unwrap().is_some() {
            return Err(PluginError::HostStarted);
        }
        let name = plugin.name().to_string();
        if !valid_plugin_name(&name) {
            return Err(PluginError::IllegalName(name));
        }
        let deps: Vec<String> = plugin.deps().iter().map(|d| d.to_string()).collect();
        if deps.iter().any(|d| d == &name) {
            return Err(PluginError::DepCycle(vec![name]));
        }
        let mailbox_cap = plugin.mailbox_cap();
        let any: Arc<dyn Any + Send + Sync> = Arc::clone(&plugin) as Arc<dyn Any + Send + Sync>;
        let dyn_plugin: Arc<dyn Plugin> = plugin;

        let mut w = self.registry.state_of.write().unwrap();
        if w.entries.contains_key(&name) {
            return Err(PluginError::Duplicate(name));
        }
        for d in &deps {
            if !w.entries.contains_key(d) {
                return Err(PluginError::UnknownDep(d.clone()));
            }
        }
        let entry = Arc::new(PluginEntry {
            name: name.clone(),
            plugin: dyn_plugin,
            any,
            state: PluginStateCell::new(PluginStateCode::Ready),
            mailbox_cap,
            deps,
            mailbox: Mutex::new(None),
            forwarder_ids: Mutex::new(Vec::new()),
            worker: Mutex::new(None),
            stats: Arc::new(EntryStats::default()),
        });
        w.order.push(entry.name.clone());
        w.entries.insert(entry.name.clone(), entry);
        Ok(())
    }

    /// 状态查询。
    pub fn get_state(&self, name: &str) -> Option<PluginState> {
        self.registry.get(name).map(|e| e.state.get())
    }

    /// 信息快照。
    pub fn get_plugin(&self, name: &str) -> Option<PluginInfo> {
        self.registry.get(name).map(|e| PluginInfo {
            name: e.name.clone(),
            state: e.state.get(),
            deps: e.deps.clone(),
        })
    }

    /// 类型安全获取(get_typed):同一分配的 Arc<dyn Any> downcast。
    pub fn get_typed<T: Plugin + 'static>(&self, name: &str) -> Option<Arc<T>> {
        let entry = self.registry.get(name)?;
        Arc::downcast::<T>(Arc::clone(&entry.any)).ok()
    }

    /// 注册名列表(注册序)。
    pub fn plugin_names(&self) -> Vec<String> {
        self.registry.names()
    }

    /// 运行统计(溢出丢弃/panic 计数)。未注册 → None。
    pub fn mail_stats(&self, name: &str) -> Option<MailStats> {
        self.registry.get(name).map(|e| MailStats {
            spilled: e.stats.spilled.load(Ordering::Acquire),
            panics: e.stats.panics.load(Ordering::Acquire),
        })
    }

    /// 启动全部插件:总线 start → deps 拓扑序逐个装配
    /// (on_init → 建邮箱 → spawn worker → state=Running → 注册 forwarder → on_start)。
    /// 依赖环 → DepCycle;重复 start 幂等 Ok;已 close → Stopped。
    pub fn start(&self) -> Result<(), PluginError> {
        let mut started = self.started_order.lock().unwrap();
        if started.is_some() {
            return Ok(());
        }
        self.closed_check()?;
        self.event.start();
        let order = self.registry.topo_order()?;
        let mut launched = Vec::with_capacity(order.len());
        for entry in order {
            let ctx = Arc::new(PluginContext::new(
                Arc::clone(&self.event),
                Arc::clone(&self.registry),
                entry.name.clone(),
            ));
            entry.plugin.on_init(&ctx);
            let (tx, rx) = crossbeam_channel::bounded::<ek_event::Event>(entry.mailbox_cap);
            *entry.mailbox.lock().unwrap() = Some(tx.clone());
            let handler = spawn_worker(
                Arc::clone(&entry.plugin),
                rx,
                entry.state.clone(),
                Arc::clone(&entry.stats),
                Arc::clone(&ctx),
                Arc::clone(&self.event),
            );
            *entry.worker.lock().unwrap() = Some(handler);
            // 先 Running 再装 forwarder:避免 ask 预检在装配窗口把目标误判 Ready
            let _ = entry.state.cas(PluginStateCode::Ready, PluginStateCode::Running);
            let mut ids: Vec<HandlerId> = Vec::new();
            for ty in entry.plugin.routes() {
                let fwd = Forwarder::new(
                    tx.clone(),
                    Arc::clone(&entry.stats),
                    Arc::clone(&self.event),
                    entry.name.clone(),
                );
                let id = self.event.register(ty.into_owned(), make_forwarder(fwd));
                ids.push(id);
            }
            *entry.forwarder_ids.lock().unwrap() = ids;
            entry.plugin.on_start();
            launched.push(entry.name.clone());
        }
        *started = Some(launched);
        Ok(())
    }

    /// D4 五步闭停单个插件(不摘表)。
    ///
    /// on_stop 语义(执行期复审 F1 修正):凡曾装配过的插件(Running/Stopping/Exited/
    /// Quarantined)close 时都调 on_stop —— 熔断或自然退出的插件也需要业务清理;
    /// 仅 Ready(从未 start)不调。
    fn close_one(&self, entry: &Arc<PluginEntry>, deadline: Instant) {
        let pre_state = entry.state.get();
        let call_stop = pre_state != PluginState::Ready;
        // 挂起迁移:Running → Stopping(容错其他状态)
        let _ = entry.state.cas(PluginStateCode::Running, PluginStateCode::Stopping);
        // ① 卸 forwarder
        let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
        for id in ids {
            self.event.unregister(id);
        }
        // ② on_stop
        if call_stop {
            entry.plugin.on_stop();
        }
        // ③ 关邮箱(宿主侧 Sender drop → 全部 send 端消费尽)
        *entry.mailbox.lock().unwrap() = None;
        // ④⑤ 排空等待 + join(带超时,超时告警 + 剥离)
        while !entry.stats.done.load(Ordering::Acquire) && Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(2));
        }
        if let Some(handle) = entry.worker.lock().unwrap().take() {
            if entry.stats.done.load(Ordering::Acquire) {
                let _ = handle.join();
            } else {
                self.host_log(format!(
                    "close timeout: worker of {:?} still busy; detach and keep draining",
                    entry.name
                ));
            }
        }
        // 兜底:Stopping → Exited(worker 已退出但未来得及 CAS 的场合)
        let _ = entry.state.cas(PluginStateCode::Stopping, PluginStateCode::Exited);
    }

    /// 全局闭停:逆 start 序逐插件五步;然后清插件表。总线留给调用方(close_all 停总线)。
    pub fn close(&self, timeout: Duration) {
        let order = self.started_order.lock().unwrap().take();
        *self.closed.write().unwrap() = true;
        let deadline = Instant::now() + timeout.max(Duration::from_millis(1));
        if let Some(order) = order {
            for name in order.iter().rev() {
                if let Some(entry) = self.registry.get(name) {
                    self.close_one(&entry, deadline);
                }
            }
        }
        self.registry.clear();
    }

    /// close + 停总线(终态)。
    pub fn close_all(&self, timeout: Duration) {
        self.close(timeout);
        self.event.stop();
    }

    /// 移除单个插件:Running 者执行五步(UNKNOWN 表项报错);仍被其他插件 deps 引用
    /// 时,eLog 警告(定稿 policy:允许移除 + 警告;依赖方后续 ask 走预检/断连失败)。
    pub fn remove_plugin(&self, name: &str) -> Result<(), PluginError> {
        self.closed_check()?;
        let entry = self
            .registry
            .get(name)
            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
        let dependents = self.registry.any_depends_on(name);
        if !dependents.is_empty() {
            self.host_log(format!(
                "remove_plugin {:?} still depended on by {:?}",
                name, dependents
            ));
        }
        let deadline = Instant::now() + CLOSE_JOIN_TIMEOUT;
        if entry.state.get() == PluginState::from_code_enum(PluginStateCode::Running) {
            self.close_one(&entry, deadline);
        } else {
            // Ready / Exited / Quarantined:摘 forwarder(可能残留于熔断前)+ 关邮箱
            let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
            for id in ids {
                self.event.unregister(id);
            }
            *entry.mailbox.lock().unwrap() = None;
        }
        self.registry
            .remove(name)
            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
        Ok(())
    }
}