ek-station 0.4.13

//! cfg_load —— 环节4:配置回灌接通宿主装配面(boot_with_cfg / cfgload)。
//!
//! 钉面:
//! - t1 载入直灌 + 时机:预写 `<ns>.ini` → boot_with_cfg → apply_config 早于
//!   on_start(工厂探针直接记帐;boot 建的是厂例,seen 句柄共享);
//! - t2 冷启动零动:未写文件 = Fresh = apply_config 不被调;
//! - t3 格式错即报:current 坏 + 无 `.bak` → boot_with_cfg = Err(Cfg)
//!   (BD-U4 语义:单一硬报错点);
//! - t4 boot()(不带 Store)零盘面:零 Load 零回灌(存量调用面零移植);
//! - t5 ns 撞名 → NsDuplicate;
//! - t6 apply_config 拒收 → ApplyRejected;
//! - t7 svctl.start 复用同一单点(cfg_store 带 Store 时,起「可配置工厂」
//!   实例同样回灌;档案预写 → 起探针 → 记帐到)。

use ek_station::{StationConfig, StationError, boot, boot_with_cfg, default_kinds};
use std::borrow::Cow;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use ek_core::section::{ApplyReport, Configurable, Field, SectionMap};
use ek_core::{MainEngine, Plugin, PluginContext};
use ek_event::Event;
use ek_station::wire::{SVC_START, StartReq, StartRes};
use std::sync::OnceLock;

// ---- 手写可配置探针 ----

type Ledger = Arc<Mutex<Vec<String>>>;

struct ProbeCfg {
    name: &'static str,
    ns: &'static str,
    reject: bool,
    ledger: Ledger,
}

impl Configurable for ProbeCfg {
    fn config_ns(&self) -> &str {
        self.ns
    }
    fn apply_config(&self, map: &SectionMap) -> Result<ApplyReport, String> {
        if self.reject {
            return Err("拒绝该段(测试哨)".into());
        }
        let mut ledger = self.ledger.lock().unwrap();
        for (k, f) in map.iter() {
            ledger.push(format!("apply:{k}={}", f.plain_value().unwrap_or_default()));
        }
        Ok(ApplyReport::all_adopted())
    }
    fn dump_config(&self) -> SectionMap {
        SectionMap::new(1)
    }
}

impl Plugin for ProbeCfg {
    fn name(&self) -> &str {
        self.name
    }
    fn routes(&self) -> Vec<Cow<'static, str>> {
        Vec::new()
    }
    fn on_mail(&self, _: &Event) {}
    fn on_start(&self) {
        self.ledger.lock().unwrap().push("STARTED".into());
    }
    // ek-core 0.4.0 的 opt-in 口诀:有配置面 = 显式覆写(缺省 None)。
    fn as_configurable(&self) -> Option<&dyn Configurable> {
        Some(self)
    }
}

fn kind_of(
    name: &'static str,
    ns: &'static str,
    ledger: Ledger,
    reject: bool,
) -> ek_station::PlugKind {
    ek_station::PlugKind {
        name,
        doc: "探针(可配置,测试专用)",
        build: Arc::new(move || {
            Arc::new(ProbeCfg {
                name,
                ns,
                reject,
                ledger: Arc::clone(&ledger),
            }) as Arc<dyn Plugin>
        }),
        build_named: None,
        name_validator: None,
    }
}

fn tmp_root(tag: &str) -> PathBuf {
    let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("target")
        .join("tmp")
        .join("cfg-load")
        .join(format!("{tag}-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&root);
    std::fs::create_dir_all(&root).unwrap();
    root
}

fn led() -> Ledger {
    Arc::new(Mutex::new(Vec::new()))
}

#[test]
fn t1_load_before_start() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let root = tmp_root("t1");
    let store = ek_cfg::Store::with_root(root);
    store
        .save("tprobe", &SectionMap::new(7).set("k1", Field::plain("v1")))
        .unwrap();

    let ledger = led();
    let kinds = default_kinds();
    let mut kinds = kinds;
    kinds.push(kind_of("tprobe", "tprobe", Arc::clone(&ledger), false));
    let engine = Arc::new(MainEngine::new(None));
    boot_with_cfg(
        &engine,
        &kinds,
        &StationConfig {
            plugins: vec!["cfg".to_string(), "tprobe".into()],
        },
        Some(&store),
    )
    .expect("boot_with_cfg");
    engine.start().unwrap();

    let got = ledger.lock().unwrap();
    assert!(got.contains(&"apply:k1=v1".to_string()), "未回灌:{got:?}");
    // 时机钉:apply_config 必须早于 STARTED。
    assert_eq!(got.first().map(String::as_str), Some("apply:k1=v1"));
    assert!(got.contains(&"STARTED".to_string()));
    engine.close_all(Duration::from_secs(2));
}

#[test]
fn t2_fresh_is_noop() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let root = tmp_root("t2");
    let store = ek_cfg::Store::with_root(root);
    let ledger = led();
    let mut kinds = default_kinds();
    kinds.push(kind_of("tprobe", "tprobe", Arc::clone(&ledger), false));
    let engine = Arc::new(MainEngine::new(None));
    boot_with_cfg(
        &engine,
        &kinds,
        &StationConfig {
            plugins: vec!["cfg".to_string(), "tprobe".into()],
        },
        Some(&store),
    )
    .unwrap();
    engine.start().unwrap();
    let got = ledger.lock().unwrap();
    assert!(
        !got.iter().any(|l| l.starts_with("apply:")),
        "Fresh 不应回灌:{got:?}"
    );
    assert!(got.contains(&"STARTED".to_string()));
    engine.close_all(Duration::from_secs(2));
}

#[test]
fn t3_corrupt_legacy_archive_is_hard_error() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    // 步骤 5 单模:段只住在库里,坏配置的**显形面**搬到迁移入口 ——
    // 旧档(迁移前 `*.ini`)解析硬错 = migrate-db 拒收(不半迁不吞)。
    // station 侧不再有「坏 .ini → Cfg 错」这条(文件形已删)。
    let root = tmp_root("t3");
    std::fs::create_dir_all(&root).unwrap();
    std::fs::write(root.join("tprobe.ini"), b"junk_no_rev\nk = v\n").unwrap();
    let err = ek_cfg::migrate_file_to_db(&root).expect_err("坏旧档必显形");
    assert!(err.to_string().contains("tprobe"), "显形点名坏段: {err}");
    // 坏档保持原样(未改名 = 拒收不留半迁)
    assert!(root.join("tprobe.ini").exists(), "拒收不得挪档");
}

#[test]
fn t4_boot_no_store_is_pure_assembly() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let ledger = led();
    let mut kinds = default_kinds();
    kinds.push(kind_of("tprobe", "tprobe", Arc::clone(&ledger), false));
    let engine = Arc::new(MainEngine::new(None));
    boot(
        &engine,
        &kinds,
        &StationConfig {
            plugins: vec!["cfg".to_string(), "tprobe".into()],
        },
    )
    .unwrap();
    engine.start().unwrap();
    let got = ledger.lock().unwrap();
    // 不带 Store:零 Load 零回灌(存量测试/调用面零移植语义)。
    assert!(!got.iter().any(|l| l.starts_with("apply:")), "{got:?}");
    assert!(got.contains(&"STARTED".to_string()));
    engine.close_all(Duration::from_secs(2));
}

#[test]
fn t5_ns_duplicate_rejected() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let root = tmp_root("t5");
    let store = ek_cfg::Store::with_root(root);
    let mut kinds = default_kinds();
    kinds.push(kind_of("tprobe1", "tprobe", led(), false));
    kinds.push(kind_of("tprobe2", "tprobe", led(), false));
    let engine = Arc::new(MainEngine::new(None));
    let Err(err) = boot_with_cfg(
        &engine,
        &kinds,
        &StationConfig {
            plugins: vec!["cfg".to_string(), "tprobe1".into(), "tprobe2".into()],
        },
        Some(&store),
    ) else {
        panic!("撞名必须 Err");
    };
    assert!(matches!(err, StationError::NsDuplicate(_)), "{err:?}");
    engine.close_all(Duration::from_secs(2));
}

#[test]
fn t6_reject_surfaces() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let root = tmp_root("t6");
    let store = ek_cfg::Store::with_root(root);
    store
        .save("tprobe", &SectionMap::new(1).set("k", Field::plain("v")))
        .unwrap();
    let ledger = led();
    let mut kinds = default_kinds();
    kinds.push(kind_of("tprobe", "tprobe", Arc::clone(&ledger), true));
    let engine = Arc::new(MainEngine::new(None));
    let Err(err) = boot_with_cfg(
        &engine,
        &kinds,
        &StationConfig {
            plugins: vec!["cfg".to_string(), "tprobe".into()],
        },
        Some(&store),
    ) else {
        panic!("拒收必须 Err");
    };
    assert!(matches!(err, StationError::ApplyRejected { .. }), "{err:?}");
    assert!(err.to_string().contains("拒绝该段"), "错因透出:{err}");
    engine.close_all(Duration::from_secs(2));
}

#[test]
fn t8_real_oms_through_loader() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    // 环节5(e2e):真 `ek_oms::OmsPlugin`(自带 Configurable,opt-in)过同一
    // 载入单点:预写 oms 段(rate_limit=12)→ apply_persisted → 内核限额生效。
    let root = tmp_root("t8");
    let store = ek_cfg::Store::with_root(root);
    store
        .save(
            "oms",
            &SectionMap::new(11).set("rate_limit", Field::plain("12")),
        )
        .unwrap();

    let oms = std::sync::Arc::new(ek_oms::OmsPlugin::new());
    ek_station::cfgload::apply_persisted(oms.as_ref() as &dyn Plugin, Some(&store))
        .expect("真 oms 插件回灌");
    // 生效视窗:dump 读回锁内值;rev 单调 = 11 + 1。
    let d = ek_core::section::Configurable::dump_config(oms.as_ref() as &dyn Configurable);
    assert_eq!(d.get_plain("rate_limit"), Some("12"));
    assert_eq!(d.rev, 12);
}

#[test]
fn t7_svctl_start_uses_same_loader() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    // svctl.start 复用 cfgload 单点:带 cfg_store 的 Supervisor,start 起探针工厂,
    // 档案预写 → 起例 → 记帐到(start_by 里 apply_persisted 在 add 之前)。
    let root = tmp_root("t7");
    let store = ek_cfg::Store::with_root(root);
    store
        .save("tprobe", &SectionMap::new(9).set("k2", Field::plain("v2")))
        .unwrap();

    use ek_station::Supervisor;

    struct StartProbe {
        ctx: OnceLock<PluginContext>,
        out: Arc<Mutex<String>>,
    }
    impl Plugin for StartProbe {
        fn name(&self) -> &str {
            "startprobe"
        }
        fn routes(&self) -> Vec<Cow<'static, str>> {
            vec![Cow::Borrowed("eGo.")]
        }
        fn on_init(&self, ctx: &PluginContext) {
            let _ = self.ctx.set(ctx.clone());
        }
        fn on_mail(&self, _event: &ek_event::Event) {
            let Some(ctx) = self.ctx.get() else { return };
            let deadline = std::time::Instant::now() + Duration::from_secs(3);
            loop {
                let r = ctx.ask_service::<StartReq, StartRes>(
                    SVC_START,
                    StartReq {
                        factory: "tprobe".into(),
                        name: String::new(),
                    },
                    Duration::from_secs(2),
                );
                let ok = match &r {
                    Ok(StartRes::Started { .. }) => "Started".to_string(),
                    Ok(StartRes::Rejected { reason }) => format!("Rejected:{reason}"),
                    Err(e) => format!("Err:{e:?}"),
                };
                *self.out.lock().unwrap() = ok;
                if *self.out.lock().unwrap() != "Err:UnknownService"
                    || std::time::Instant::now() > deadline
                {
                    break;
                }
                std::thread::sleep(Duration::from_millis(50));
            }
        }
    }

    let ledger = led();
    let mut kinds = default_kinds();
    kinds.push(kind_of("tprobe", "tprobe", Arc::clone(&ledger), false));

    let out = Arc::new(Mutex::new(String::new()));
    let engine = Arc::new(MainEngine::new(None));
    engine
        .add_plugin(Arc::new(Supervisor::new(
            Arc::clone(&engine),
            kinds.clone(),
            Some(store.clone()),
        )))
        .unwrap();
    engine
        .add_plugin(Arc::new(StartProbe {
            ctx: OnceLock::new(),
            out: Arc::clone(&out),
        }))
        .unwrap();
    engine.start().unwrap();
    std::thread::sleep(Duration::from_millis(200));
    engine
        .event_engine()
        .put(ek_event::Event::with_data("eGo.", std::sync::Arc::new(())))
        .unwrap();
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    while std::time::Instant::now() < deadline && out.lock().unwrap().is_empty() {
        std::thread::sleep(Duration::from_millis(30));
    }
    let say = out.lock().unwrap().clone();
    assert!(
        say.starts_with("Started") || say.starts_with("Rejected"),
        "svctl.start 未回:{say}"
    );
    if let Ok(ledger_got) = ledger.try_lock() {
        assert!(
            ledger_got.contains(&"apply:k2=v2".to_string()),
            "svctl.start 路未回灌:{ledger_got:?}"
        );
    } else {
        std::thread::sleep(Duration::from_millis(200));
        assert!(
            ledger.lock().unwrap().contains(&"apply:k2=v2".to_string()),
            "svctl.start 路未回灌"
        );
    }
    engine.close_all(Duration::from_secs(3));
}