ek-core 0.2.0

//! dynamic_lifecycle(0.2.0 补强):运行期 add/remove 的编排纪律。
//!
//! DL1 迟到插件的 routes 即时通电(晚装车道不加漏)
//! DL2 同名重注册 = 换血(旧实例冻结/新实例上位/ePlugUp 计数=2)
//! DL3 运行期重名 → Duplicate(且原件毫发无损)
//! DL4 remove 不存在者 → UnknownName
//! DL5 start 前 remove(Ready)→ Ok 且不调 on_stop
//! DL6 忙提供方被移除:在途 ask 仍交付(worker 排空语义),之后 TargetGone
//! DL7 两个动态插件 + close 反序覆盖(last 注册最先关)

mod common;

use common::{load, push_record, snapshot, wait_until};
use ek_core::{envelope::Envelope, MainEngine, PlugNotice, PluginState};
use ek_event::Event;
use std::sync::{Arc, Mutex};
use std::time::Duration;

#[derive(Debug)]
pub struct Ping {
    pub n: u64,
}
#[derive(Debug, PartialEq)]
pub struct Pong {
    pub n: u64,
}

const SVC: &str = "order.place";

type CtxSlot = Arc<Mutex<Option<ek_core::PluginContext>>>;

fn ctx_slot() -> CtxSlot {
    Arc::new(Mutex::new(None))
}

// DL1 迟到插件的 routes 即时通电
#[test]
fn dl1_late_routes_get_wired() {
    let eng = MainEngine::new(None);
    let c1 = common::counter();
    let c1b = Arc::clone(&c1);
    eng.add_plugin(
        common::test_plugin("c1")
            .routes_str(&["eTick."])
            .cap(128)
            .mail(move |_e: &Event| {
                common::bump(&c1b);
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    // 动态 add(注意:add_plugin 是同步装配 —— 之后它就完整在线)
    let c2 = common::counter();
    let c2b = Arc::clone(&c2);
    eng.add_plugin(
        common::test_plugin("c2")
            .routes_str(&["eTick."])
            .cap(128)
            .mail(move |_e: &Event| {
                common::bump(&c2b);
            })
            .build(),
    )
    .unwrap();
    for i in 0..100u64 {
        eng.event_engine().put(Event::with_data("eTick.", Arc::new(i))).unwrap();
    }
    assert!(wait_until(Duration::from_secs(3), || {
        load(&c1) == 100 && load(&c2) == 100
    }));
    eng.close_all(Duration::from_secs(1));
}

// DL2 同名重注册 = 换血:旧实例冻结、新实例上位、ePlugUp 共两发
#[test]
fn dl2_same_name_readd_replaces() {
    let eng = MainEngine::new(None);
    let ups = common::counter();
    let u2 = Arc::clone(&ups);
    eng.add_plugin(
        common::test_plugin("watch")
            .routes_str(&["ePlugUp"])
            .mail(move |e: &Event| {
                if e.data.downcast_ref::<PlugNotice>().map(|n| n.name == "worker").unwrap_or(false) {
                    common::bump(&u2);
                }
            })
            .build(),
    )
    .unwrap();
    let stop_seen = common::records::<String>();
    let s_start = Arc::clone(&stop_seen);
    let s_stop = Arc::clone(&stop_seen);
    let ticks_a = common::counter();
    let ta = Arc::clone(&ticks_a);
    eng.add_plugin(
        common::test_plugin("worker")
            .routes_str(&["eTick."])
            .cap(128)
            .mail(move |_e: &Event| {
                common::bump(&ta);
            })
            .start_hook(move || push_record(&s_start, "start#1".to_string()))
            .stop_hook(move || push_record(&s_stop, "stop#1".to_string()))
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    eng.event_engine().put(Event::new("eTick.")).unwrap();
    assert!(wait_until(Duration::from_secs(2), || load(&ticks_a) == 1));

    // 换血:remove → add(新独立计数器)
    eng.remove_plugin("worker").unwrap();
    let ticks_b = common::counter();
    let tb = Arc::clone(&ticks_b);
    eng.add_plugin(
        common::test_plugin("worker")
            .routes_str(&["eTick."])
            .cap(128)
            .mail(move |_e: &Event| {
                common::bump(&tb);
            })
            .build(),
    )
    .unwrap();
    eng.event_engine().put(Event::new("eTick.")).unwrap();
    eng.event_engine().put(Event::new("eTick.")).unwrap();
    assert!(wait_until(Duration::from_secs(3), || load(&ticks_b) == 2));
    // 旧实例冻结(不再挤入任何邮件)、ePlugUp 共 2 发、旧件 on_stop 已调
    assert_eq!(load(&ticks_a), 1);
    assert_eq!(load(&ups), 2);
    assert!(snapshot(&stop_seen).contains(&"stop#1".to_string()));
    assert_eq!(eng.get_state("worker"), Some(PluginState::Running));
    eng.close_all(Duration::from_secs(1));
}

// DL3 运行期重名 → Duplicate;原件继续工作
#[test]
fn dl3_duplicate_while_running() {
    let eng = MainEngine::new(None);
    let c1 = common::counter();
    let c1b = Arc::clone(&c1);
    eng.add_plugin(
        common::test_plugin("worker")
            .routes_str(&["eTick."])
            .mail(move |_e: &Event| {
                common::bump(&c1b);
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    // 重名拒绝
    match eng.add_plugin(common::test_plugin("worker").build()) {
        Err(ek_core::PluginError::Duplicate(n)) => assert_eq!(n, "worker"),
        other => panic!("expected Duplicate, got {other:?}"),
    }
    // 原件毫发无损
    for i in 0..5u64 {
        eng.event_engine().put(Event::with_data("eTick.", Arc::new(i))).unwrap();
    }
    assert!(wait_until(Duration::from_secs(2), || load(&c1) == 5));
    assert_eq!(eng.get_state("worker"), Some(PluginState::Running));
    eng.close_all(Duration::from_secs(1));
}

// DL4 remove 不存在者 → UnknownName
#[test]
fn dl4_remove_unknown() {
    let eng = MainEngine::new(None);
    match eng.remove_plugin("no_such") {
        Err(ek_core::PluginError::UnknownName(n)) => assert_eq!(n, "no_such"),
        other => panic!("expected UnknownName, got {other:?}"),
    }
    // close 前提下也拒(Stopped 优先)
    eng.start().unwrap();
    eng.close_all(Duration::from_secs(1));
    match eng.remove_plugin("x") {
        Err(ek_core::PluginError::Stopped) => {}
        other => panic!("expected Stopped, got {other:?}"),
    }
}

// DL5 start 前 remove(Ready)→ Ok 且不调 on_stop
#[test]
fn dl5_remove_ready_no_stop_hook() {
    let eng = MainEngine::new(None);
    let stopped = common::counter();
    let s_stop = Arc::clone(&stopped);
    eng.add_plugin(
        common::test_plugin("ghostborn")
            .stop_hook(move || {
                common::bump(&s_stop);
            })
            .build(),
    )
    .unwrap();
    assert_eq!(eng.get_state("ghostborn"), Some(PluginState::Ready));
    eng.remove_plugin("ghostborn").unwrap();
    assert_eq!(eng.get_state("ghostborn"), None);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(load(&stopped), 0, "Ready 不调 on_stop（D4 判据）");
    eng.close_all(Duration::from_secs(1));
}

// DL6 忙提供方被移除:在途 ask 仍交付(⑤步排空语义);移除后索引已剪除 → ask_service = UnknownService
#[test]
fn dl6_busy_remove_inflight_ask_delivered() {
    let eng = MainEngine::new(None);
    let started_flag = common::counter();
    let s_flag = Arc::clone(&started_flag);
    eng.add_plugin(
        common::test_plugin("broker")
            .routes_str(&["eReq.broker"])
            .services(vec![ek_core::ServiceDesc::describe::<Ping, Pong>(
                SVC,
                " slow echo: 600ms per req",
            )])
            .mail(move |e: &Event| {
                // 进入处理期即打点
                if load(&s_flag) == 0 {
                    common::bump(&s_flag);
                }
                Envelope::handle::<Ping, Pong>(e, |q| {
                    std::thread::sleep(Duration::from_millis(600));
                    Pong { n: q.n }
                });
            })
            .build(),
    )
    .unwrap();
    let a_ctx = ctx_slot();
    let a_init = Arc::clone(&a_ctx);
    let a_mail = Arc::clone(&a_ctx);
    let a_seen = common::records::<String>();
    let a2 = Arc::clone(&a_seen);
    eng.add_plugin(
        common::test_plugin("a")
            .routes_str(&["eKick"])
            .init_hook(move |ctx| {
                *a_init.lock().unwrap() = Some(ctx.clone());
            })
            .mail(move |e: &Event| {
                if e.event_type != "eKick" {
                    return;
                }
                let ctx = a_mail.lock().unwrap().as_ref().expect("ctx").clone();
                let r = ctx.ask_service::<Ping, Pong>(SVC, Ping { n: 7 }, Duration::from_secs(3));
                push_record(
                    &a2,
                    match r {
                        Ok(p) => format!("ok:{}", p.n),
                        // 移除后:索引已被修剪(F3)→ 无人提供 = UnknownService(快速,不烧超时)
                        Err(ek_core::AskError::UnknownService { service }) => {
                            assert_eq!(service, SVC);
                            "unknown".to_string()
                        }
                        other => panic!("unexpected {other:?}"),
                    },
                );
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();

    // 第一问:确认已在处理中(600ms 窗口内),然后移除 broker(五步在第④步会等 worker 收工)
    eng.event_engine().put(Event::with_data("eKick", Arc::new(0u8))).unwrap();
    assert!(wait_until(Duration::from_millis(500), || load(&started_flag) == 1));
    eng.remove_plugin("broker").unwrap();
    // 在途问仍获得回复(排空交付)
    assert!(wait_until(Duration::from_secs(3), || snapshot(&a_seen) == vec!["ok:7".to_string()]));
    // 第二问:目标已移除且索引被剪 → 快速 UnknownService
    eng.event_engine().put(Event::with_data("eKick", Arc::new(1u8))).unwrap();
    assert!(
        wait_until(Duration::from_secs(3), || snapshot(&a_seen)
            == vec!["ok:7".to_string(), "unknown".to_string()]),
        "{:?}",
        snapshot(&a_seen)
    );
    eng.close_all(Duration::from_secs(1));
}

// DL7 两个动态插件 + close 反序覆盖(last 注册最先关)
#[test]
fn dl7_two_late_and_reverse_close() {
    let eng = MainEngine::new(None);
    let seq = common::records::<String>();
    let f_start = Arc::clone(&seq);
    let f_stop = Arc::clone(&seq);
    eng.add_plugin(
        common::test_plugin("first")
            .start_hook(move || push_record(&f_start, "start:first".into()))
            .stop_hook(move || push_record(&f_stop, "stop:first".into()))
            .build(),
    )
    .unwrap();
    eng.start().unwrap();

    let s1 = Arc::clone(&seq);
    let s1b = Arc::clone(&seq);
    eng.add_plugin(
        common::test_plugin("late1")
            .start_hook(move || push_record(&s1, "start:late1".into()))
            .stop_hook(move || push_record(&s1b, "stop:late1".into()))
            .build(),
    )
    .unwrap();
    let s2a = Arc::clone(&seq);
    let s2b = Arc::clone(&seq);
    eng.add_plugin(
        common::test_plugin("late2")
            .start_hook(move || push_record(&s2a, "start:late2".into()))
            .stop_hook(move || push_record(&s2b, "stop:late2".into()))
            .build(),
    )
    .unwrap();
    assert_eq!(eng.get_state("late1"), Some(PluginState::Running));
    assert_eq!(eng.get_state("late2"), Some(PluginState::Running));

    eng.close(Duration::from_secs(2));
    let got = snapshot(&seq);
    assert!(got.contains(&"start:first".into()) && got.contains(&"start:late1".into()) && got.contains(&"start:late2".into()));
    assert!(got.contains(&"stop:late2".into()) && got.contains(&"stop:late1".into()) && got.contains(&"stop:first".into()));
    // 反序:late2 → late1 → first
    let p = |name: &str| got.iter().position(|s| s == name).unwrap();
    assert!(p("stop:late2") < p("stop:late1") && p("stop:late1") < p("stop:first"), "{got:?}");
}