ek-core 0.5.2

//! discovery(0.2.0):服务自报与发现、动态生命周期、目录改版、类型闸门、版本迁移。
//!
//! DV1 发现命中(能力索引 + 现场明细 + services_of;Ready 即可查)
//! DV2 未命中 → 插件自治降级(eLog WARN)
//! DV3 ePlugUp 迟绑定(动态 add 上线 → 订阅者自主重配)
//! DV4 publish_services 目录改版(eSrvChg → 刷新;旧 id 无人提供 → UnknownService)
//! DV5 多提供方 failover + 版本被动迁移(ePlugDown → 自动重配新版本)
//! DV6 ask_service 三闸门(UnknownService / InvalidPayload 错型 / 契约匹配成功)
//! DV7 动态生命周期:start 后 add → 即刻 Running;close 反序覆盖动态插件;close 后 Stopped

mod common;

use common::{load, push_record, snapshot, wait_until};
use ek_core::{
    AskError, MainEngine, PlugNotice, PluginContext, PluginState, ServiceDesc, envelope::Envelope,
};
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";
const SVC_ALIAS: &str = "order.peg";

fn svc_desc() -> ServiceDesc {
    ServiceDesc::describe::<Ping, Pong>(SVC, "Echo: Ping 反射为 Pong")
}

// ======================== 公用:插件侧 ctx 槽位 ========================

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

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

/// on_init 钩子:把 ctx 抓进槽位,mail 闭包按需取用。
fn hook_ctx(slot: CtxSlot) -> impl Fn(&PluginContext) + Send + Sync + 'static {
    move |ctx: &PluginContext| {
        *slot.lock().unwrap() = Some(ctx.clone());
    }
}

// ======================== DV1 ========================

// DV1 发现命中:on_init 里 find_service / services_of 直读目录(add 即入索引,Ready 可查)
#[test]
fn dv1_discovery_hit() {
    let eng = MainEngine::new(None);
    eng.add_plugin(
        common::test_plugin("oms")
            .init_hook(|ctx: &PluginContext| {
                let refs = ctx.find_service(SVC);
                assert_eq!(refs.len(), 1);
                assert_eq!(refs[0].name, "broker");
                assert_eq!(refs[0].version, "1.0.0");
                assert_eq!(refs[0].service.id, SVC);
                assert_eq!(
                    refs[0].service.req_type,
                    std::any::type_name::<Ping>(),
                    "闸门底座:type_name 现场一致"
                );
                assert!(refs[0].service.doc.contains("Pong"));
                // 现场目录(完整细节)与索引一致
                let list = ctx.services_of("broker").expect("broker declared services");
                assert_eq!(list.len(), 1);
                assert_eq!(list[0].id, SVC);
                // 未申报的服务无人提供
                assert!(ctx.find_service("absent.thing").is_empty());
                match ctx.ask_service::<Ping, Pong>(
                    "absent.thing",
                    Ping { n: 1 },
                    Duration::from_millis(100),
                ) {
                    Err(AskError::UnknownService { service }) => {
                        assert_eq!(service, "absent.thing")
                    }
                    other => panic!("expected UnknownService, got {other:?}"),
                }
            })
            .build(),
    )
    .unwrap();
    eng.add_plugin(
        common::test_plugin("broker")
            .version("1.0.0")
            .services(vec![svc_desc()])
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    eng.close_all(Duration::from_secs(1));
}

// ======================== DV2 ========================

// DV2 未命中 → 插件自治降级(降级旗标是插件自己的状态;宿主零日志,不提供任何日志出口)
#[test]
fn dv2_discovery_miss_then_degrade() {
    let eng = MainEngine::new(None);
    let degraded = common::counter();
    let d2 = Arc::clone(&degraded);
    eng.add_plugin(
        common::test_plugin("oms")
            .init_hook(move |ctx: &PluginContext| {
                if ctx.find_service(SVC).is_empty() {
                    // 自持降级旗标;诊断文本走插件自身通道(0.3.0 纪律:宿主不涉)
                    common::bump(&d2);
                }
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    assert!(wait_until(Duration::from_secs(2), || load(&degraded) == 1));
    eng.close_all(Duration::from_secs(1));
}

// ======================== DV3 ========================

// DV3 ePlugUp 迟绑定:A 先行无提供方 → B 动态 add 上线 → A 收 ePlugUp → 自主重配
#[test]
fn dv3_late_binding_via_plug_up() {
    let eng = MainEngine::new(None);
    let paired = common::records::<String>();
    let p2 = Arc::clone(&paired);
    eng.add_plugin(
        common::test_plugin("a")
            .routes_str(&["ePlugUp"])
            .mail(move |e: &Event| {
                if e.event_type == "ePlugUp" {
                    if let Some(n) = e.data.downcast_ref::<PlugNotice>() {
                        if n.name == "broker" {
                            push_record(&p2, format!("paired:{}/{}", n.name, n.version));
                        }
                    }
                }
            })
            .init_hook(|ctx: &PluginContext| {
                assert!(
                    ctx.find_service(SVC).is_empty(),
                    "no provider yet at a.on_init"
                );
                // ctx.plugins():单一披露面,此时尚且只见自己
                assert!(ctx.plugins().iter().any(|p| p.name == "a"));
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();

    // 运行期动态 add:登记 → launch_one(同步装配,即刻 Running)→ ePlugUp 双发
    eng.add_plugin(
        common::test_plugin("broker")
            .version("2.0.0")
            .services(vec![svc_desc()])
            .build(),
    )
    .unwrap();
    assert_eq!(eng.get_state("broker"), Some(PluginState::Running));
    assert!(wait_until(Duration::from_secs(2), || snapshot(&paired)

        == vec!["paired:broker/2.0.0".to_string()]));
    eng.close_all(Duration::from_secs(1));
}

// ======================== DV4 ========================

/// 可变目录提供方(测试件):目录随配置演化 —— 先换自己的 services() 返回,
/// 再 publish_services 申报;宿主现场取到的始终是插件当前目录。
/// (教训已固化为纪律:申报前必须先换目录,否则 live 细节与索引脱节。)
struct DynProvider {
    catalog: Mutex<Vec<ServiceDesc>>,
    ctx: Mutex<Option<PluginContext>>,
}

impl ek_core::Plugin for DynProvider {
    fn name(&self) -> &str {
        "broker"
    }
    fn version(&self) -> std::borrow::Cow<'static, str> {
        "1.0.0".into()
    }
    fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
        vec!["eRepub".into()]
    }
    fn services(&self) -> Vec<ServiceDesc> {
        self.catalog.lock().unwrap().clone()
    }
    fn on_init(&self, ctx: &PluginContext) {
        *self.ctx.lock().unwrap() = Some(ctx.clone());
    }
    fn on_mail(&self, event: &Event) {
        if event.event_type != "eRepub" {
            return;
        }
        // 改版:先换目录,再申报(宿主重织索引 + eSrvChg 广播)
        {
            let mut cat = self.catalog.lock().unwrap();
            cat.clear();
            cat.push(ServiceDesc::describe::<Ping, Pong>(
                SVC_ALIAS,
                "改版后只提供 peg",
            ));
        }
        let Some(ctx) = self.ctx.lock().unwrap().as_ref().cloned() else {
            return;
        };
        ctx.publish_services(self.services());
    }
}

// DV4 publish_services 目录改版 → eSrvChg → 消费者刷新;旧 id 立刻无人提供
#[test]
fn dv4_catalog_revision() {
    let eng = MainEngine::new(None);
    let seen = common::records::<String>();
    let s2 = Arc::clone(&seen);

    // broker:可变目录件
    eng.add_plugin(Arc::new(DynProvider {
        catalog: Mutex::new(vec![svc_desc()]),
        ctx: Mutex::new(None),
    }))
    .unwrap();
    // a:订阅 eSrvChg → 重查刷新认知 + 验证旧 id now UnknownService
    let a_ctx = ctx_slot();
    let actx = Arc::clone(&a_ctx);
    eng.add_plugin(
        common::test_plugin("a")
            .routes_str(&["eSrvChg"])
            .init_hook(hook_ctx(a_ctx))
            .mail(move |e: &Event| {
                if e.event_type == "eSrvChg" {
                    if let Some(n) = e.data.downcast_ref::<PlugNotice>() {
                        if n.name != "broker" {
                            return;
                        }
                        let ctx = actx.lock().unwrap().as_ref().expect("a ctx").clone();
                        let place = ctx.find_service(SVC).len();
                        let peg = ctx.find_service(SVC_ALIAS).len();
                        let ask = ctx.ask_service::<Ping, Pong>(
                            SVC,
                            Ping { n: 1 },
                            Duration::from_millis(300),
                        );
                        push_record(
                            &s2,
                            format!("rev:ver={}/place={place}/peg={peg}", n.version),
                        );
                        match ask {
                            Err(AskError::UnknownService { service }) => {
                                assert_eq!(service, SVC);
                                push_record(&s2, "old-id→UnknownService".to_string());
                            }
                            other => {
                                panic!("expected UnknownService after revision, got {other:?}")
                            }
                        }
                    }
                }
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();

    // 触发改版
    eng.event_engine()
        .put(Event::with_data("eRepub", Arc::new(0u8)))
        .unwrap();
    assert!(
        wait_until(Duration::from_secs(2), || snapshot(&seen).len() == 2),
        "seen={:?}",
        snapshot(&seen)
    );
    let got = snapshot(&seen);
    assert!(
        got[0].contains("rev:ver=1.0.0/place=0/peg=1"),
        "got {got:?}"
    );
    assert_eq!(got[1], "old-id→UnknownService");
    eng.close_all(Duration::from_secs(1));
}

// ======================== DV5 ========================

// DV5 failover + 版本被动迁移:ePlugDown(计划内移除)→ A 重查并接上下一家(带版本)
#[test]
fn dv5_failover_version_migration() {
    let eng = MainEngine::new(None);
    let mig = common::records::<String>();
    let a_ctx = ctx_slot();
    let paired = Arc::new(Mutex::new(String::new()));
    let p_init = Arc::clone(&paired);
    let p_mail = Arc::clone(&paired);
    let m_init = Arc::clone(&mig);
    let m_mail = Arc::clone(&mig);
    let a_init = Arc::clone(&a_ctx);
    let mail_actx = Arc::clone(&a_ctx);

    eng.add_plugin(
        common::test_plugin("a")
            .routes_str(&["ePlugDown"])
            .init_hook(move |ctx: &PluginContext| {
                // 配对注册序第一个提供方(注册序:broker_old 先入索引)
                if let Some(first) = ctx.find_service(SVC).first() {
                    *p_init.lock().unwrap() = first.name.clone();
                    push_record(&m_init, format!("pair:{}/{}", first.name, first.version));
                }
                *a_init.lock().unwrap() = Some(ctx.clone());
            })
            .mail(move |e: &Event| {
                if e.event_type == "ePlugDown" {
                    let Some(n) = e.data.downcast_ref::<PlugNotice>() else {
                        return;
                    };
                    let Some(ctx) = mail_actx.lock().unwrap().as_ref().cloned() else {
                        return;
                    };
                    let mut cur = p_mail.lock().unwrap();
                    if *cur != n.name {
                        return; // 下线的不是我配对的那家,无视
                    }
                    // 被动跟随:重查 → 接上第一个 Running 的下一个提供方
                    if let Some(next) = ctx
                        .find_service(SVC)
                        .into_iter()
                        .find(|p| p.state == PluginState::Running)
                    {
                        push_record(&m_mail, format!("move:{}/{}", next.name, next.version));
                        *cur = next.name.clone();
                    } else {
                        push_record(&m_mail, "move:none".to_string());
                        cur.clear();
                    }
                }
            })
            .build(),
    )
    .unwrap();
    eng.add_plugin(
        common::test_plugin("broker_old")
            .version("1.0.0")
            .services(vec![svc_desc()])
            .build(),
    )
    .unwrap();
    eng.add_plugin(
        common::test_plugin("broker_new")
            .version("1.1.0")
            .services(vec![svc_desc()])
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    assert!(wait_until(Duration::from_secs(2), || snapshot(&mig)

        == vec!["pair:broker_old/1.0.0".to_string()]));

    // 计划内移除旧提供方 → ePlugDown → A 被动迁移到 broker_new(1.1.0)
    eng.remove_plugin("broker_old").unwrap();
    assert!(wait_until(Duration::from_secs(2), || snapshot(&mig)

        == vec![
            "pair:broker_old/1.0.0".to_string(),
            "move:broker_new/1.1.0".to_string()
        ]));
    // 索引已修剪:broker_old 不再是幽灵 providing 单位
    let eng_ctx = eng.get_typed::<common::TestPlugin>("a");
    assert!(eng_ctx.is_some());
    eng.close_all(Duration::from_secs(1));
}

// ======================== DV6 ========================

// DV6 ask_service 三闸门:UnknownService / InvalidPayload(错型) / 契约匹配成功
#[test]
fn dv6_ask_service_gates() {
    let eng = MainEngine::new(None);
    let seen = common::records::<String>();
    let s2 = Arc::clone(&seen);
    let a_ctx = ctx_slot();
    let actx = Arc::clone(&a_ctx);
    let kick_no = common::counter();
    let k2 = Arc::clone(&kick_no);

    // provider:只提供 alias;契约处理 Ping → Pong
    eng.add_plugin(
        common::test_plugin("broker")
            .routes_str(&["eReq.broker"])
            .services(vec![ServiceDesc::describe::<Ping, Pong>(
                SVC_ALIAS,
                "peg 提供",
            )])
            .mail(|e: &Event| {
                Envelope::handle::<Ping, Pong>(e, |req| Pong { n: req.n });
            })
            .build(),
    )
    .unwrap();
    eng.add_plugin(
        common::test_plugin("a")
            .routes_str(&["eKick"])
            .init_hook(hook_ctx(a_ctx))
            .mail(move |e: &Event| {
                if e.event_type != "eKick" {
                    return;
                }
                let n = load(&k2);
                if n == 0 {
                    // 闸门①:无人提供该服务(宿主索引空)
                    let ctx = actx.lock().unwrap().as_ref().expect("a ctx").clone();
                    let r = ctx.ask_service::<Ping, Pong>(
                        "absent.svc",
                        Ping { n: 1 },
                        Duration::from_millis(200),
                    );
                    match r {
                        Err(AskError::UnknownService { service }) => {
                            assert_eq!(service, "absent.svc");
                            push_record(&s2, "g1:UnknownService".to_string());
                        }
                        other => panic!("expected UnknownService, got {other:?}"),
                    }
                } else if n == 1 {
                    // 闸门②:服务在,但 Req 型别与声明不符 → 立即 InvalidPayload(不发包)
                    let ctx = actx.lock().unwrap().as_ref().expect("a ctx").clone();
                    let r = ctx.ask_service::<String, Pong>(
                        SVC_ALIAS,
                        "wrong-type-payload".to_string(),
                        Duration::from_millis(200),
                    );
                    match r {
                        Err(AskError::InvalidPayload { target }) => {
                            assert_eq!(target, "broker");
                            push_record(&s2, "g2:InvalidPayload".to_string());
                        }
                        other => panic!("expected InvalidPayload, got {other:?}"),
                    }
                } else {
                    // 正路:契约匹配 → 成功
                    let ctx = actx.lock().unwrap().as_ref().expect("a ctx").clone();
                    let r = ctx.ask_service::<Ping, Pong>(
                        SVC_ALIAS,
                        Ping { n: 7 },
                        Duration::from_millis(500),
                    );
                    match r {
                        Ok(p) => {
                            assert_eq!(p, Pong { n: 7 });
                            push_record(&s2, format!("g3:ok={}", p.n));
                        }
                        other => panic!("expected Ok(Pong), got {other:?}"),
                    }
                }
                common::bump(&k2);
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    for _ in 0..3u64 {
        eng.event_engine()
            .put(Event::with_data("eKick", Arc::new(0u8)))
            .unwrap();
    }
    assert!(wait_until(Duration::from_secs(3), || snapshot(&seen).len() == 3));
    let got = snapshot(&seen);
    assert!(got.iter().any(|s| s == "g1:UnknownService"), "{got:?}");
    assert!(got.iter().any(|s| s == "g2:InvalidPayload"), "{got:?}");
    assert!(got.iter().any(|s| s == "g3:ok=7"), "{got:?}");
    eng.close_all(Duration::from_secs(1));
}

// ======================== DV7 ========================

// DV7 动态生命周期:start 后 add → 即刻 Running;close 反序覆盖动态插件;close 后 Stopped
#[test]
fn dv7_dynamic_add_and_close_coverage() {
    let eng = MainEngine::new(None);
    let seq = common::records::<String>();
    let s_start = Arc::clone(&seq);
    let s_stop = Arc::clone(&seq);
    eng.add_plugin(
        common::test_plugin("first")
            .start_hook(move || push_record(&s_start, "start:first".to_string()))
            .stop_hook(move || push_record(&s_stop, "stop:first".to_string()))
            .build(),
    )
    .unwrap();
    eng.start().unwrap();

    // 动态 add:同步装配,即刻 Running
    let l_start = Arc::clone(&seq);
    let l_stop = Arc::clone(&seq);
    eng.add_plugin(
        common::test_plugin("late")
            .services(vec![ServiceDesc::describe::<Ping, Pong>(SVC, "动态提供方")])
            .start_hook(move || push_record(&l_start, "start:late".to_string()))
            .stop_hook(move || push_record(&l_stop, "stop:late".to_string()))
            .build(),
    )
    .unwrap();
    assert_eq!(eng.get_state("late"), Some(PluginState::Running));
    assert!(snapshot(&seq).contains(&"start:late".to_string()));

    // close 反序覆盖动态插件:late 后注册 → 先关
    eng.close(Duration::from_secs(2));
    let got = snapshot(&seq);
    let pos_first_stop = got.iter().position(|s| s == "stop:first").unwrap();
    let pos_late_stop = got.iter().position(|s| s == "stop:late").unwrap();
    assert!(pos_late_stop < pos_first_stop, "late closes first: {got:?}");

    // close 后一切变更被拒(Stopped)
    let eng2 = MainEngine::new(None);
    eng2.start().unwrap();
    eng2.close_all(Duration::from_secs(1));
    match eng2.add_plugin(common::test_plugin("x3").build()) {
        Err(ek_core::PluginError::Stopped) => {}
        other => panic!("expected Stopped, got {other:?}"),
    }
}