Skip to main content

root/
lib.rs

1mod arrow;
2mod directory;
3mod mount;
4mod node;
5pub use arrow::query::{Query, SearchDescriptionResult};
6//以后要把内存中的 handler task Sender Receiver 和 数据分开处理
7pub use mount::{Mount, Root};
8
9use std::cell::RefCell;
10use std::sync::{Arc, LazyLock};
11
12use anyhow::{Result, anyhow};
13use dynamic::{Dynamic, MsgPack, MsgUnpack, Type};
14use rand::RngExt;
15
16use tokio::sync::mpsc;
17
18pub type Msg<T> = (T, Option<mpsc::Sender<T>>);
19pub type MsgSender<T> = mpsc::Sender<Msg<T>>;
20pub type MsgReceiver<T> = mpsc::Receiver<Msg<T>>;
21
22pub fn tx_rx<T: Send>() -> (MsgSender<T>, MsgReceiver<T>) {
23    let (tx, rx) = mpsc::channel(1024);
24    (tx, rx)
25}
26
27pub fn block_on_async<F, T>(f: F) -> T
28where
29    F: FnOnce() -> std::pin::Pin<Box<dyn Future<Output = T> + Send>> + 'static + Send,
30    T: Send + 'static,
31{
32    if tokio::runtime::Handle::try_current().is_ok() {
33        let (tx, rx) = tokio::sync::oneshot::channel();
34        tokio::task::spawn(async move {
35            // spawn 任务若 panic,tx 在 send 前被 drop,rx.await 会收到 Err,
36            // 由下面的 match 转成带上下文的 panic,而非裸 unwrap 丢失信息。
37            let result = f().await;
38            let _ = tx.send(result);
39        });
40        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async {
41            match rx.await {
42                Ok(value) => value,
43                // rx 收到 Err 说明 tx 被 drop(spawn 任务 panic 或提前退出):带上下文 panic。
44                Err(_) => panic!("block_on_async: spawn 任务在发送结果前异常退出"),
45            }
46        }))
47    } else {
48        let rt = tokio::runtime::Runtime::new().unwrap();
49        rt.block_on(f())
50    }
51}
52
53use smol_str::SmolStr;
54pub fn start_task<F>(info: Dynamic, f: F) -> Dynamic
55where
56    F: FnOnce() -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>> + 'static + Send,
57{
58    let id = uuid::Uuid::new_v4().to_string();
59    let task = SmolStr::new(format!("local/tasks/{}", id));
60    let r = if tokio::runtime::Handle::try_current().is_ok() {
61        Object::Task(tokio::task::spawn(async move { f().await }), info)
62    } else {
63        Object::ThreadTask(
64            std::thread::spawn(move || {
65                let rt = tokio::runtime::Runtime::new().unwrap();
66                rt.block_on(f())
67            }),
68            info,
69        )
70    };
71    let _ = add(&task, r);
72    log::info!("start task {:?}", task);
73    id.into()
74}
75
76#[macro_export]
77macro_rules! sync_await {
78    ($future:expr) => {
79        $crate::block_on_async(move || Box::pin($future))
80    };
81}
82
83pub fn send<T: Send + 'static>(tx: &MsgSender<T>, msg: T) -> Result<()> {
84    tx.try_send((msg, None)).map_err(|e| anyhow!("发送失败: {}", e))
85}
86
87pub fn call<T: Send + 'static>(tx: &MsgSender<T>, msg: T) -> Result<T> {
88    let (reply_tx, reply_rx) = mpsc::channel::<T>(1024);
89    tx.try_send((msg, Some(reply_tx))).map_err(|e| anyhow!("发送失败: {}", e))?;
90    // 原先用 try_recv() 立即返回,但处理方在另一 task 里异步消费,根本没机会在
91    // try_send 和 try_recv 之间执行,导致几乎必然返回"接收回复失败"。
92    // 改为阻塞等待 reply:把 reply_rx move 进 async 块,复用 block_on_async 的
93    // block_in_place 机制等待回复。
94    sync_await!(async move {
95        let mut rx = reply_rx;
96        rx.recv().await
97    })
98    .ok_or_else(|| anyhow!("接收回复失败: channel 关闭"))
99}
100
101use tokio::task::JoinHandle;
102
103use crate::node::Node;
104
105/// ROOT 挂载的 native 处理函数。
106///
107/// 用 `Arc<dyn Fn>` 而非 `fn` 指针,这样既能挂无状态的顶层 `fn`,
108/// 也能挂带捕获环境的闭包(`Arc::new(move |msg| { ... 用到外部状态 ... })`)。
109/// newtype + 手动 Debug 让 `Object` 仍可派生 Debug(`Arc<dyn Fn>` 本身不实现 Debug)。
110#[derive(Clone)]
111pub struct NativeHandler(Arc<dyn Fn(Dynamic) -> Dynamic + Send + Sync>);
112
113impl NativeHandler {
114    /// 用顶层 `fn` 构造(零开销,等价于原先的 `fn` 指针)。
115    pub fn from_fn(f: fn(Dynamic) -> Dynamic) -> Self {
116        Self(Arc::new(f))
117    }
118
119    /// 用任意 `Fn` 闭包构造(可带捕获,需 `Send + Sync`)。
120    pub fn from_closure<F>(f: F) -> Self
121    where
122        F: Fn(Dynamic) -> Dynamic + Send + Sync + 'static,
123    {
124        Self(Arc::new(f))
125    }
126
127    /// 调用处理函数。
128    pub fn call(&self, msg: Dynamic) -> Dynamic {
129        (self.0)(msg)
130    }
131}
132
133impl std::fmt::Debug for NativeHandler {
134    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
135        f.debug_struct("NativeHandler").finish_non_exhaustive()
136    }
137}
138
139#[derive(Debug)]
140pub enum Object {
141    Value(Dynamic),                                           //基本的值
142    Native(NativeHandler),                                    //函数处理对象(支持 fn 与闭包)
143    Func(i64, Type),                                          //裸指针
144    Tx(MsgSender<Dynamic>, Dynamic),                          //包括 Tx 信息
145    Task(JoinHandle<Result<()>>, Dynamic),                    //异步任务
146    ThreadTask(std::thread::JoinHandle<Result<()>>, Dynamic), //同步任务
147}
148
149impl Into<Object> for Dynamic {
150    fn into(self) -> Object {
151        Object::Value(self)
152    }
153}
154
155impl Object {
156    pub fn value(&self) -> Dynamic {
157        match self {
158            Self::Value(v) => v.deep_clone(),
159            Self::Task(_, info) => info.deep_clone(),
160            Self::ThreadTask(_, info) => info.deep_clone(),
161            Self::Tx(_, info) => info.deep_clone(),
162            _ => Dynamic::Null,
163        }
164    }
165}
166
167impl Default for Object {
168    fn default() -> Self {
169        Self::Value(Dynamic::Null)
170    }
171}
172
173impl MsgPack for Object {
174    fn encode(&self, buf: &mut Vec<u8>) {
175        match self {
176            Self::Value(v) => v.encode(buf),
177            _ => {}
178        }
179    }
180}
181
182impl MsgUnpack for Object {
183    fn decode(buf: &[u8]) -> Result<(Self, usize)> {
184        let (v, len) = Dynamic::decode(buf)?;
185        Ok((Self::Value(v), len))
186    }
187}
188
189static ROOT: LazyLock<Root<Object>> = LazyLock::new(|| {
190    let root = Root::<Object>::new();
191    if let Ok((mount, name)) = root.get_mount("local/fight") {
192        mount.add(name, Object::Native(NativeHandler::from_fn(fight)));
193    }
194    root
195});
196
197thread_local! {
198    static SEND_STACK: RefCell<Vec<String>> = RefCell::new(Vec::new());
199}
200
201fn builtin_native(name: &str) -> Option<NativeHandler> {
202    match name {
203        "fight" | "native.fight" | "native::fight" => Some(NativeHandler::from_fn(fight)),
204        _ => None,
205    }
206}
207
208fn get_value(input: &Dynamic, keys: &[&str]) -> Option<Dynamic> {
209    keys.iter().find_map(|key| input.get_dynamic(key))
210}
211
212fn get_string(input: &Dynamic, keys: &[&str]) -> Option<String> {
213    get_value(input, keys).map(|value| value.as_str().to_string())
214}
215
216fn get_i64(input: &Dynamic, keys: &[&str]) -> Option<i64> {
217    get_value(input, keys).and_then(|value| value.as_int())
218}
219
220fn get_f64(input: &Dynamic, keys: &[&str]) -> Option<f64> {
221    get_value(input, keys).and_then(|value| value.as_float())
222}
223
224fn get_range(input: &Dynamic, range_keys: &[&str], min_keys: &[&str], max_keys: &[&str], default_min: i64, default_max: i64) -> (i64, i64) {
225    if let Some(range) = get_value(input, range_keys) {
226        let min = range.get_idx(0).and_then(|v| v.as_int()).unwrap_or(default_min);
227        let max = range.get_idx(1).and_then(|v| v.as_int()).unwrap_or(default_max);
228        return normalize_range(min, max, default_min, default_max);
229    }
230
231    let min = get_i64(input, min_keys).unwrap_or(default_min);
232    let max = get_i64(input, max_keys).unwrap_or(default_max);
233    normalize_range(min, max, default_min, default_max)
234}
235
236fn normalize_range(min: i64, max: i64, default_min: i64, default_max: i64) -> (i64, i64) {
237    let min = min.max(0);
238    let max = max.max(0);
239    if min == 0 && max == 0 {
240        (default_min, default_max)
241    } else if min <= max {
242        (min, max)
243    } else {
244        (max, min)
245    }
246}
247
248fn role_name(role: &Dynamic) -> String {
249    get_string(role, &["name", "名称"]).unwrap_or_else(|| "未命名".to_string())
250}
251
252fn role_hp(role: &Dynamic) -> i64 {
253    get_i64(role, &["hp", "health", "生命"]).unwrap_or(0).max(0)
254}
255
256fn role_max_hp(role: &Dynamic) -> i64 {
257    get_i64(role, &["max_hp", "max_health", "最大生命"]).unwrap_or_else(|| role_hp(role)).max(1)
258}
259
260fn role_speed(role: &Dynamic) -> i64 {
261    get_i64(role, &["speed", "initiative", "速度"]).unwrap_or(0)
262}
263
264fn role_attack_range(role: &Dynamic) -> (i64, i64) {
265    get_range(role, &["attack", "攻击"], &["attack_min", "min_attack", "攻击下限"], &["attack_max", "max_attack", "攻击上限"], 8, 15)
266}
267
268fn role_defense_range(role: &Dynamic) -> (i64, i64) {
269    get_range(role, &["defense", "防御"], &["defense_min", "min_defense", "防御下限"], &["defense_max", "max_defense", "防御上限"], 1, 4)
270}
271
272fn role_block_chance(role: &Dynamic) -> f64 {
273    get_f64(role, &["block_chance", "block", "格挡几率"]).unwrap_or(0.15).clamp(0.0, 0.95)
274}
275
276fn role_block_ratio(role: &Dynamic) -> f64 {
277    get_f64(role, &["block_ratio", "格挡减伤"]).unwrap_or(0.5).clamp(0.0, 1.0)
278}
279
280fn role_crit_chance(role: &Dynamic) -> f64 {
281    get_f64(role, &["crit_chance", "crit", "暴击几率"]).unwrap_or(0.1).clamp(0.0, 1.0)
282}
283
284fn role_crit_ratio(role: &Dynamic) -> f64 {
285    get_f64(role, &["crit_ratio", "暴击倍率"]).unwrap_or(1.5).max(1.0)
286}
287
288fn set_i64(role: &Dynamic, key: &str, value: i64) {
289    role.set_dynamic(key.into(), value);
290}
291
292fn set_bool(role: &Dynamic, key: &str, value: bool) {
293    role.set_dynamic(key.into(), value);
294}
295
296fn normalize_role(input: &Dynamic, side: &'static str, default_name: &str) -> Dynamic {
297    let role = if input.is_map() { input.deep_clone() } else { Dynamic::map(Default::default()) };
298    let name = get_string(&role, &["name", "名称"]).unwrap_or_else(|| default_name.to_string());
299    let hp = get_i64(&role, &["hp", "health", "生命"]).unwrap_or(100).max(1);
300    let (attack_min, attack_max) = role_attack_range(&role);
301    let (defense_min, defense_max) = role_defense_range(&role);
302
303    role.set_dynamic("name".into(), name);
304    role.set_dynamic("side".into(), side);
305    set_i64(&role, "hp", hp);
306    set_i64(&role, "max_hp", role_max_hp(&role).max(hp));
307    set_i64(&role, "attack_min", attack_min);
308    set_i64(&role, "attack_max", attack_max);
309    set_i64(&role, "defense_min", defense_min);
310    set_i64(&role, "defense_max", defense_max);
311    role.set_dynamic("block_chance".into(), role_block_chance(&role));
312    role.set_dynamic("block_ratio".into(), role_block_ratio(&role));
313    role.set_dynamic("crit_chance".into(), role_crit_chance(&role));
314    role.set_dynamic("crit_ratio".into(), role_crit_ratio(&role));
315    set_i64(&role, "speed", role_speed(&role));
316    set_bool(&role, "alive", hp > 0);
317    role
318}
319
320fn make_fight_record(entries: [(&str, Dynamic); 10]) -> Dynamic {
321    let mut map = std::collections::BTreeMap::<SmolStr, Dynamic>::new();
322    for (key, value) in entries {
323        map.insert(key.into(), value);
324    }
325    Dynamic::map(map)
326}
327
328fn fight(msg: Dynamic) -> Dynamic {
329    let left_input = msg.get_dynamic("left").or_else(|| msg.get_dynamic("a")).unwrap_or(Dynamic::Null);
330    let right_input = msg.get_dynamic("right").or_else(|| msg.get_dynamic("b")).unwrap_or(Dynamic::Null);
331    if !left_input.is_map() || !right_input.is_map() {
332        let mut error = std::collections::BTreeMap::new();
333        error.insert("error".into(), "fight expects {left: {...}, right: {...}}".into());
334        return Dynamic::map(error);
335    }
336
337    let left = normalize_role(&left_input, "left", "左侧");
338    let right = normalize_role(&right_input, "right", "右侧");
339    let max_rounds = get_i64(&msg, &["max_rounds", "最大回合"]).unwrap_or(50).clamp(1, 500) as usize;
340
341    let mut rng = rand::rng();
342    let mut process = Vec::new();
343    let mut records = Vec::new();
344    let mut left_turn = if role_speed(&left) == role_speed(&right) { rng.random_bool(0.5) } else { role_speed(&left) > role_speed(&right) };
345
346    let mut round_count = 0usize;
347    while role_hp(&left) > 0 && role_hp(&right) > 0 && round_count < max_rounds {
348        round_count += 1;
349        let (attacker, defender) = if left_turn { (&left, &right) } else { (&right, &left) };
350
351        let (attack_min, attack_max) = role_attack_range(attacker);
352        let (defense_min, defense_max) = role_defense_range(defender);
353        let attack_roll = rng.random_range(attack_min..=attack_max);
354        let defense_roll = rng.random_range(defense_min..=defense_max);
355        let blocked = rng.random_bool(role_block_chance(defender));
356        let crit = rng.random_bool(role_crit_chance(attacker));
357
358        let mut damage = (attack_roll - defense_roll).max(1);
359        if crit {
360            damage = ((damage as f64) * role_crit_ratio(attacker)).round() as i64;
361        }
362        if blocked {
363            damage = ((damage as f64) * (1.0 - role_block_ratio(defender))).round() as i64;
364        }
365        damage = damage.max(if blocked { 0 } else { 1 });
366
367        let defender_hp = (role_hp(defender) - damage).max(0);
368        let defender_max_hp = role_max_hp(defender);
369        set_i64(defender, "hp", defender_hp);
370        set_bool(defender, "alive", defender_hp > 0);
371
372        let line = format!(
373            "第{}回合 {} 攻击 {},攻击 {},防御 {},{}{}造成 {} 点伤害,{} 剩余 {} / {}",
374            round_count,
375            role_name(attacker),
376            role_name(defender),
377            attack_roll,
378            defense_roll,
379            if crit { "触发暴击," } else { "" },
380            if blocked { "被格挡后," } else { "" },
381            damage,
382            role_name(defender),
383            defender_hp,
384            defender_max_hp,
385        );
386        process.push(line.clone().into());
387        records.push(make_fight_record([
388            ("round", (round_count as i64).into()),
389            ("attacker", role_name(attacker).into()),
390            ("attacker_side", get_string(attacker, &["side"]).unwrap_or_default().into()),
391            ("defender", role_name(defender).into()),
392            ("defender_side", get_string(defender, &["side"]).unwrap_or_default().into()),
393            ("attack_roll", attack_roll.into()),
394            ("defense_roll", defense_roll.into()),
395            ("blocked", blocked.into()),
396            ("critical", crit.into()),
397            ("damage", damage.into()),
398        ]));
399        if let Some(last) = records.last() {
400            last.insert("defender_hp", defender_hp);
401            last.insert("text", line);
402        }
403
404        left_turn = !left_turn;
405    }
406
407    let left_hp = role_hp(&left);
408    let right_hp = role_hp(&right);
409    let left_name = role_name(&left);
410    let right_name = role_name(&right);
411    let winner = if left_hp == right_hp {
412        "draw".to_string()
413    } else if left_hp > right_hp {
414        left_name.clone()
415    } else {
416        right_name.clone()
417    };
418    let loser = if winner == "draw" {
419        String::new()
420    } else if winner == left_name {
421        right_name
422    } else {
423        left_name
424    };
425
426    let mut result = std::collections::BTreeMap::new();
427    result.insert("winner".into(), winner.into());
428    result.insert("loser".into(), loser.into());
429    result.insert("draw".into(), (left_hp == right_hp).into());
430    result.insert("round_count".into(), (round_count as i64).into());
431    result.insert("process".into(), Dynamic::list(process));
432    result.insert("records".into(), Dynamic::list(records));
433    result.insert("left".into(), left);
434    result.insert("right".into(), right);
435    Dynamic::map(result)
436}
437
438pub fn mount_memory(name: &str) -> bool {
439    ROOT.mount_memory(name)
440}
441
442pub fn mount_redis(name: &str, url: &str) -> Result<bool> {
443    ROOT.mount_redis(name, url)
444}
445
446pub fn mount_fjall(name: &str, data_dir: &str) -> Result<bool> {
447    ROOT.mount_fjall(name, data_dir)
448}
449
450pub fn get_mount<'a>(name: &'a str) -> Result<(Mount<Object>, &'a str)> {
451    ROOT.get_mount(name)
452}
453
454pub fn mount_dir(name: &str, host_dir: &str) -> Result<bool> {
455    ROOT.mount_dir(name, host_dir)
456}
457
458pub fn add(name: &str, obj: Object) -> Result<bool> {
459    let (m, name) = get_mount(name)?;
460    // Dir 后端的 add 走文件 I/O(impl Mount<Object>::dir_add)。
461    if matches!(m, Mount::Dir { .. }) {
462        return Ok(m.dir_add(name, obj));
463    }
464    let mut obj = obj;
465    let expire = take_object_expire(&mut obj);
466    let added = m.add(name, obj);
467    if added {
468        apply_redis_expire(&m, name, expire);
469    }
470    Ok(added)
471}
472
473/// `add_native` 第二个参数的 trait 重载。
474///
475/// 既支持按内建名挂载(`add_native("local/h", "fight")`,查 `builtin_native`),
476/// 也支持直接挂任意 `Fn` 闭包(`add_native("local/h", |msg| { ... })`,可带捕获)。
477/// 顶层 `fn` 也会走闭包分支(无捕获闭包)。
478pub trait IntoNativeHandler {
479    fn into_handler(self) -> Option<NativeHandler>;
480}
481
482/// `&str`:按内建名查找(如 "fight")。未知名返回 None,`add_native` 返回 false。
483impl IntoNativeHandler for &str {
484    fn into_handler(self) -> Option<NativeHandler> {
485        builtin_native(self)
486    }
487}
488
489/// 任意 `Fn(Dynamic) -> Dynamic + Send + Sync` 闭包(含无捕获的顶层 `fn`)。
490/// 允许 `root::add_native("local/h", |msg| { ... })` 直接挂闭包。
491impl<F> IntoNativeHandler for F
492where
493    F: Fn(Dynamic) -> Dynamic + Send + Sync + 'static,
494{
495    fn into_handler(self) -> Option<NativeHandler> {
496        Some(NativeHandler::from_closure(self))
497    }
498}
499
500/// 挂载一个 native 处理函数到 ROOT 树。
501///
502/// 第二个参数既可以是内建名(`&str`,如 "fight"),也可以是任意
503/// `Fn(Dynamic) -> Dynamic + Send + Sync` 闭包(可带捕获环境)。
504///
505/// 注意:示例为说明用途,需在已初始化 ROOT 树的上下文中调用。
506///
507/// ```no_run
508/// # use dynamic::Dynamic;
509/// # use root;
510/// // 按名挂内建
511/// let _ = root::add_native("local/fight", "fight");
512/// // 挂顶层 fn
513/// fn h(msg: Dynamic) -> Dynamic { Dynamic::Null }
514/// let _ = root::add_native("local/h", h);
515/// // 挂带捕获的闭包
516/// let state = 42i64;
517/// let closure = move |msg: Dynamic| -> Dynamic { Dynamic::from(state) };
518/// let _ = root::add_native("local/h2", closure);
519/// ```
520pub fn add_native<H: IntoNativeHandler>(name: &str, handler: H) -> Result<bool> {
521    let Some(handler) = handler.into_handler() else {
522        return Ok(false);
523    };
524    let (m, name) = get_mount(name)?;
525    Ok(m.add(name, Object::Native(handler)))
526}
527
528pub fn add_value<T: Into<Dynamic>>(name: &str, val: T) -> Result<bool> {
529    let (m, name) = get_mount(name)?;
530    let mut value = val.into();
531    let expire = take_dynamic_expire(&mut value);
532    let added = m.add(name, Object::Value(value));
533    if added {
534        apply_redis_expire(&m, name, expire);
535    }
536    Ok(added)
537}
538
539pub fn get(name: &str) -> Result<Dynamic> {
540    let (m, name) = get_mount(name)?;
541    // Dir 后端:文件 → decode;目录 → Null(不报错);不存在 → Err。
542    if matches!(m, Mount::Dir { .. }) {
543        return m.dir_get(name);
544    }
545    m.get(name, |obj| obj.value())
546}
547
548pub fn dir(name: &str) -> Result<Dynamic> {
549    let (m, name) = get_mount(name)?;
550    m.dir(name).map(Into::into)
551}
552
553pub fn keys(name: &str) -> Result<Dynamic> {
554    let (m, name) = get_mount(name)?;
555    m.keys(name).map(Into::into)
556}
557
558pub fn contains(name: &str) -> bool {
559    if let Ok((m, name)) = get_mount(name) {
560        if matches!(m, Mount::Dir { .. }) {
561            return m.dir_contains(name);
562        }
563        return m.contains(name);
564    }
565    false
566}
567
568pub fn remove(name: &str) -> Result<Dynamic> {
569    let (m, name) = get_mount(name)?;
570    if matches!(m, Mount::Dir { .. }) {
571        return m.dir_remove(name);
572    }
573    match m.remove(name) {
574        Ok(Object::Value(v)) => Ok(v),
575        _ => Err(anyhow!("没有删除对象")),
576    }
577}
578
579pub fn add_list(name: &str) -> Result<()> {
580    let (m, name) = get_mount(name)?;
581    m.add_list(name);
582    Ok(())
583}
584
585pub fn push(name: &str, value: Dynamic) -> Result<usize> {
586    let (m, name) = get_mount(name)?;
587    let mut value = value;
588    let expire = take_dynamic_expire(&mut value);
589    let len = m.push(name, Object::Value(value))?;
590    apply_redis_expire(&m, name, expire);
591    Ok(len)
592}
593
594/// 按 idx 从列表移除元素(符合 sparse slot 设计:不 pack,索引稳定)。
595/// 用于 WS 连接等需要稳定 idx 的场景,断开时回收列表槽位避免无限增长。
596pub fn remove_idx(name: &str, idx: usize) -> Result<Dynamic> {
597    let (m, name) = get_mount(name)?;
598    match m.remove_idx(name, idx)? {
599        Object::Value(v) => Ok(v),
600        _ => Ok(Dynamic::Null),
601    }
602}
603
604pub fn add_map(name: &str) -> Result<()> {
605    let (m, name) = get_mount(name)?;
606    m.add_map(name);
607    Ok(())
608}
609
610pub fn insert(name: &str, key: &str, value: Dynamic) -> Result<()> {
611    let (m, name) = get_mount(name)?;
612    let mut value = value;
613    let expire = take_dynamic_expire(&mut value);
614    let _ = m.insert(name, key, Object::Value(value));
615    apply_redis_expire(&m, name, expire);
616    Ok(())
617}
618
619fn take_object_expire(obj: &mut Object) -> Option<i64> {
620    if let Object::Value(value) = obj { take_dynamic_expire(value) } else { None }
621}
622
623fn take_dynamic_expire(value: &mut Dynamic) -> Option<i64> {
624    value.remove_dynamic("@expire").and_then(|expire| expire.as_int()).filter(|expire| *expire > 0)
625}
626
627fn apply_redis_expire(mount: &Mount<Object>, name: &str, expire: Option<i64>) {
628    let Some(expire) = expire else {
629        return;
630    };
631    // expire 是附加属性,失败不应让 add/insert 报错(主数据已写入),
632    // 但必须记录日志,否则过期未设置会静默导致 key 永驻。
633    if let Mount::Redis { client, rl: _ } = mount
634        && let Ok(mut conn) = client.get_connection()
635    {
636        if let Err(e) = conn.expire::<&str, ()>(name, expire) {
637            log::warn!("redis expire {} {}s 失败: {e:#}", name, expire);
638        }
639    }
640}
641
642pub fn get_key(name: &str, key: &str) -> Result<Dynamic> {
643    let (m, name) = get_mount(name)?;
644    m.get_key(name, key, |obj| obj.value())
645}
646
647/// 原子并发更新一个节点的值。返回更新后的值。
648/// 节点必须已存在,否则返回 Err。
649/// 原子并发更新一个 Object 节点的值。返回更新后的值。
650///
651/// **警告**:闭包 `f` 在 Memory 后端持有 scc bucket 级写锁(自旋锁,不可重入)期间执行。
652/// 闭包内**不要再调用任何 ROOT 操作**(如 `root::get`/`root::update`/`root::send_msg`),
653/// 否则若目标 name 与当前 name 落入同一 hash bucket,会自旋死锁。
654/// 闭包应只对传入的 `Dynamic` 做本地计算与修改。
655pub fn update<F>(name: &str, mut f: F) -> Result<Dynamic>
656where
657    F: FnMut(Dynamic) -> Dynamic + Send + 'static,
658{
659    let (m, name) = get_mount(name)?;
660    // Dir 后端的 update:读文件 → 应用 f → 写回。不保证原子,
661    // 但与 Memory/Redis/Fjall 后端的 update 整体语义一致(都是单节点 RMW)。
662    if matches!(m, Mount::Dir { .. }) {
663        let mut f_owned = f;
664        return m.dir_update(name, move |current| f_owned(current));
665    }
666    m.get_mut(name, move |obj| match obj {
667        Object::Value(v) => {
668            let current = std::mem::take(v);
669            let next = f(current);
670            let result = next.deep_clone();
671            *v = next;
672            result
673        }
674        _ => Dynamic::Null,
675    })
676}
677
678/// 原子并发更新一个 map 节点内某个 key 的值。返回更新后的值。
679///
680/// **警告**:同 [`update`],闭包在 Memory 后端持有写锁期间执行,
681/// 闭包内不要回调 ROOT 操作,否则可能自旋死锁。
682pub fn update_key<F>(name: &str, key: &str, mut f: F) -> Result<Dynamic>
683where
684    F: FnMut(Dynamic) -> Dynamic + Send + 'static,
685{
686    let (m, name) = get_mount(name)?;
687    m.get_key_mut(name, key, move |obj| match obj {
688        Object::Value(v) => {
689            let current = std::mem::take(v);
690            let next = f(current);
691            let result = next.deep_clone();
692            *v = next;
693            result
694        }
695        _ => Dynamic::Null,
696    })
697}
698
699use redis::Commands;
700pub fn get_list(name: &str) -> Result<Vec<Dynamic>> {
701    let (m, name) = get_mount(name)?;
702    match m {
703        Mount::Memory(m) => m
704            .read_sync(name, |_, v| match v {
705                Node::List(l) => l.iter().map(|(_, item)| item.value()).collect(),
706                _ => Vec::new(),
707            })
708            .ok_or(anyhow!("未发现 {}", name)),
709        Mount::Redis { client, rl: _ } => {
710            let mut conn = client.get_connection()?;
711            let items: Vec<Vec<u8>> = conn.lrange(name, 0, -1)?;
712            let items: Vec<Dynamic> = items.into_iter().map(|buf| Dynamic::decode(buf.as_slice()).map(|(v, _)| v).unwrap_or(Dynamic::Null)).collect();
713            Ok(items)
714        }
715        Mount::Fjall { .. } => {
716            let len = m.len(name)?;
717            let mut items = Vec::new();
718            for idx in 0..len {
719                if let Ok(value) = m.get_idx(name, idx, |obj| obj.value()) {
720                    items.push(value);
721                }
722            }
723            Ok(items)
724        }
725        // Dir 后端没有 List 概念,但保留接口语义:返回目录下一级 entry 列表
726        //(全是 string name,跟 dir() 一致;调用方拿 Dynamic 列表).
727        Mount::Dir { base } => {
728            let path = crate::mount::safe_path(&base, &name)
729                .ok_or_else(|| anyhow!("path 非法: {}", name))?;
730            if !path.exists() {
731                return Err(anyhow!("{} 不存在", name));
732            }
733            if !path.is_dir() {
734                return Err(anyhow!("{} 不是目录", name));
735            }
736            let mut items = Vec::new();
737            for entry in std::fs::read_dir(&path).map_err(|e| anyhow!("读取 {} 失败: {}", path.display(), e))? {
738                let entry = entry.map_err(|e| anyhow!("读取目录项失败: {}", e))?;
739                if let Some(s) = entry.file_name().to_str() {
740                    items.push(Dynamic::from(s.to_string()));
741                }
742            }
743            Ok(items)
744        }
745    }
746}
747
748#[derive(Debug)]
749enum MyFn {
750    Null,
751    Sender(MsgSender<Dynamic>),
752    Native(NativeHandler),
753    Script(i64, Type),
754}
755
756impl MyFn {
757    fn call(&self, msg: Dynamic) -> Result<Dynamic> {
758        match self {
759            MyFn::Sender(tx) => call(tx, msg),
760            MyFn::Native(f) => Ok(f.call(msg)),
761            MyFn::Script(ptr, ty) => dynamic::call_fn(*ptr, ty.clone(), Box::new(msg)).map(|r| r.as_ref().clone()),
762            MyFn::Null => Ok(Dynamic::Null),
763        }
764    }
765}
766
767impl From<&Object> for MyFn {
768    fn from(obj: &Object) -> Self {
769        match obj {
770            Object::Tx(tx, _) => MyFn::Sender(tx.clone()),
771            Object::Task(_, _) => {
772                // Task 节点不可作为函数调用;读它不应有 abort 副作用(否则 send_msg
773                // 读取 Task 句柄会杀掉正在运行的后台任务)。
774                MyFn::Null
775            }
776            Object::Func(ptr, ty) => MyFn::Script(*ptr, ty.clone()),
777            Object::Native(f) => MyFn::Native(f.clone()),
778            _ => MyFn::Null,
779        }
780    }
781}
782
783/// SEND_STACK 的 RAII guard:无论 send_msg/send_idx_msg 的调用是正常返回还是 panic,
784/// Drop 都会弹出栈帧,避免 thread_local 栈残留导致后续误判"递归路径检测"。
785struct SendStackGuard;
786impl Drop for SendStackGuard {
787    fn drop(&mut self) {
788        SEND_STACK.with(|stack| {
789            let _ = stack.borrow_mut().pop();
790        });
791    }
792}
793
794pub fn send_msg(name: &str, msg: Dynamic) -> Result<Dynamic> {
795    let entered = SEND_STACK.with(|stack| {
796        let mut stack = stack.borrow_mut();
797        if stack.len() >= 64 {
798            return Err(anyhow!("root::send call depth exceeded at {}", name));
799        }
800        if stack.iter().any(|item| item == name) {
801            return Err(anyhow!("root::send recursive path detected: {}", name));
802        }
803        stack.push(name.to_string());
804        Ok(())
805    });
806    entered?;
807    // guard 在此作用域结束(含 panic unwind)时 Drop,保证 SEND_STACK 不残留。
808    let _guard = SendStackGuard;
809    let (m, name) = get_mount(name)?;
810    let f: MyFn = m.get(name, |obj| obj.into())?;
811    f.call(msg)
812}
813
814pub fn send_idx_msg(name: &str, idx: usize, msg: Dynamic) -> Result<Dynamic> {
815    let stack_name = format!("{}[{}]", name, idx);
816    let entered = SEND_STACK.with(|stack| {
817        let mut stack = stack.borrow_mut();
818        if stack.len() >= 64 {
819            return Err(anyhow!("root::send_idx call depth exceeded at {}", stack_name));
820        }
821        if stack.iter().any(|item| item == &stack_name) {
822            return Err(anyhow!("root::send_idx recursive path detected: {}", stack_name));
823        }
824        stack.push(stack_name);
825        Ok(())
826    });
827    entered?;
828    let _guard = SendStackGuard;
829    let (m, name) = get_mount(name)?;
830    let f: MyFn = m.get_idx(name, idx, |obj| obj.into())?;
831    f.call(msg)
832}
833
834#[cfg(test)]
835mod tests {
836    use super::*;
837
838    fn fight_request() -> Dynamic {
839        let left = Dynamic::map(Default::default());
840        left.insert("name", "战士");
841        left.insert("hp", 30);
842        left.insert("attack_min", 8);
843        left.insert("attack_max", 12);
844        left.insert("defense_min", 1);
845        left.insert("defense_max", 3);
846        left.insert("block_chance", 0.2);
847
848        let right = Dynamic::map(Default::default());
849        right.insert("name", "盗贼");
850        right.insert("hp", 24);
851        right.insert("attack_min", 6);
852        right.insert("attack_max", 10);
853        right.insert("defense_min", 0);
854        right.insert("defense_max", 2);
855        right.insert("block_chance", 0.1);
856
857        let msg = Dynamic::map(Default::default());
858        msg.insert("left", left);
859        msg.insert("right", right);
860        msg
861    }
862
863    #[test]
864    fn add_native_registers_builtin_fight_handler() {
865        assert!(add_native("local/test/fight_handler", "native.fight").unwrap());
866        let result = send_msg("local/test/fight_handler", fight_request()).unwrap();
867        assert!(result.is_map());
868        assert!(result.get_dynamic("winner").is_some());
869        assert!(result.get_dynamic("process").is_some_and(|v| v.is_list()));
870    }
871
872    #[test]
873    fn add_native_accepts_closure_with_captured_state() {
874        // 挂带捕获环境的闭包:原先 Object::Native 是 fn 指针,不支持闭包;
875        // 改成 NativeHandler(Arc<dyn Fn>) 后应能挂任意 Fn 闭包。
876        let multiplier: i64 = 10;
877        let closure = move |msg: Dynamic| -> Dynamic {
878            let n = msg.as_int().unwrap_or(0);
879            Dynamic::from(n * multiplier)
880        };
881        assert!(add_native("local/test/closure_handler", closure).unwrap());
882        let result = send_msg("local/test/closure_handler", Dynamic::from(5i64)).unwrap();
883        assert_eq!(result.as_int(), Some(50));
884    }
885
886    #[test]
887    fn add_native_accepts_bare_fn() {
888        // 顶层 fn 也应能直接挂(走 Fn 闭包分支)
889        fn echo(msg: Dynamic) -> Dynamic {
890            msg
891        }
892        assert!(add_native("local/test/echo", echo).unwrap());
893        let result = send_msg("local/test/echo", Dynamic::from(42i64)).unwrap();
894        assert_eq!(result.as_int(), Some(42));
895    }
896
897    #[test]
898    fn add_native_unknown_builtin_name_returns_false() {
899        // 未知名按 &str 分支返回 None,add_native 返回 false 而非报错
900        assert!(!add_native("local/test/unknown", "no_such_builtin").unwrap());
901    }
902
903    #[test]
904    fn local_fight_handler_is_available_by_default() {
905        assert!(contains("local/fight"));
906        let result = send_msg("local/fight", fight_request()).unwrap();
907        let rounds = result.get_dynamic("round_count").and_then(|v| v.as_int()).unwrap_or_default();
908        assert!(rounds > 0);
909    }
910
911    #[test]
912    fn take_dynamic_expire_strips_positive_expire_metadata() {
913        let mut value = dynamic::map!("@expire"=> 30, "name"=> "zust");
914
915        assert_eq!(take_dynamic_expire(&mut value), Some(30));
916        assert!(!value.contains("@expire"));
917        assert_eq!(value.get_dynamic("name").unwrap().as_str(), "zust");
918    }
919
920    #[test]
921    fn take_dynamic_expire_ignores_non_positive_expire_metadata() {
922        let mut value = dynamic::map!("@expire"=> 0, "name"=> "zust");
923
924        assert_eq!(take_dynamic_expire(&mut value), None);
925        assert!(!value.contains("@expire"));
926    }
927
928    #[test]
929    fn update_increments_value_atomically() {
930        let path = format!("local/test/update_inc/{}", uuid::Uuid::new_v4());
931        add_value(&path, 0_i64).unwrap();
932
933        let next = update(&path, |v| v + Dynamic::I64(1)).unwrap();
934        assert_eq!(next.as_int(), Some(1));
935        assert_eq!(get(&path).unwrap().as_int(), Some(1));
936
937        let next = update(&path, |v| v + Dynamic::I64(41)).unwrap();
938        assert_eq!(next.as_int(), Some(42));
939        assert_eq!(get(&path).unwrap().as_int(), Some(42));
940    }
941
942    #[test]
943    fn update_concurrent_threads_no_lost_update() {
944        let path = format!("local/test/update_concurrent/{}", uuid::Uuid::new_v4());
945        add_value(&path, 0_i64).unwrap();
946
947        const THREADS: i64 = 8;
948        const ITERS: i64 = 1000;
949
950        let mut handles = Vec::new();
951        for _ in 0..THREADS {
952            let p = path.clone();
953            handles.push(std::thread::spawn(move || {
954                for _ in 0..ITERS {
955                    update(&p, |v| v + Dynamic::I64(1)).unwrap();
956                }
957            }));
958        }
959        for h in handles {
960            h.join().unwrap();
961        }
962
963        assert_eq!(get(&path).unwrap().as_int(), Some(THREADS * ITERS));
964    }
965
966    #[test]
967    fn update_key_increments_map_value() {
968        let path = format!("local/test/update_key/{}", uuid::Uuid::new_v4());
969        add_map(&path).unwrap();
970        insert(&path, "n", 0_i64.into()).unwrap();
971
972        const THREADS: i64 = 8;
973        const ITERS: i64 = 500;
974
975        let mut handles = Vec::new();
976        for _ in 0..THREADS {
977            let p = path.clone();
978            handles.push(std::thread::spawn(move || {
979                for _ in 0..ITERS {
980                    update_key(&p, "n", |v| v + Dynamic::I64(1)).unwrap();
981                }
982            }));
983        }
984        for h in handles {
985            h.join().unwrap();
986        }
987
988        assert_eq!(get_key(&path, "n").unwrap().as_int(), Some(THREADS * ITERS));
989    }
990
991    #[test]
992    fn keys_returns_memory_map_keys() {
993        let path = format!("local/test/keys/{}", uuid::Uuid::new_v4());
994        add_map(&path).unwrap();
995        insert(&path, "0", "zero".into()).unwrap();
996        insert(&path, "1", "one".into()).unwrap();
997
998        let key_list = keys(&path).unwrap();
999        let mut keys = (0..key_list.len()).map(|idx| key_list.get_idx(idx).unwrap().as_str().to_string()).collect::<Vec<_>>();
1000        keys.sort();
1001        assert_eq!(keys, vec!["0".to_string(), "1".to_string()]);
1002    }
1003
1004    #[test]
1005    fn dir_returns_immediate_child_names() {
1006        let path = format!("local/test/public_dir/{}", uuid::Uuid::new_v4());
1007        add_value(&format!("{path}/a"), 1_i64).unwrap();
1008        add_value(&format!("{path}/sub/item"), 2_i64).unwrap();
1009
1010        let entries = dir(&path).unwrap();
1011        let mut entries = (0..entries.len()).map(|idx| entries.get_idx(idx).unwrap().as_str().to_string()).collect::<Vec<_>>();
1012        entries.sort();
1013        assert_eq!(entries, vec!["a".to_string(), "sub".to_string()]);
1014    }
1015
1016    #[test]
1017    fn mount_fjall_uses_explicit_mount_name() {
1018        let data_dir = std::env::temp_dir().join(format!("zust-root-fjall-named-{}", uuid::Uuid::new_v4()));
1019        let data_dir_str = data_dir.to_str().unwrap();
1020        let mount_name = format!("fjall_{}", uuid::Uuid::new_v4().simple());
1021
1022        assert!(mount_fjall(&mount_name, data_dir_str).unwrap());
1023        add_value(&format!("{mount_name}/item"), 9_i64).unwrap();
1024        assert_eq!(get(&format!("{mount_name}/item")).unwrap().as_int(), Some(9));
1025        assert!(!contains("fjall/item"));
1026
1027        let _ = std::fs::remove_dir_all(data_dir);
1028    }
1029
1030    #[test]
1031    fn update_returns_err_when_missing() {
1032        let path = format!("local/test/update_missing/{}", uuid::Uuid::new_v4());
1033        let result = update(&path, |v| v + Dynamic::I64(1));
1034        assert!(result.is_err());
1035    }
1036
1037    #[test]
1038    fn root_get_returns_independent_dynamic_values() {
1039        let path = format!("local/test/root_get_clone/{}", uuid::Uuid::new_v4());
1040        let nested = dynamic::map!("score"=> 1);
1041        let value = dynamic::map!("nested"=> nested.clone(), "items"=> Dynamic::list(vec![nested]));
1042
1043        add_value(&path, value).unwrap();
1044
1045        let first = get(&path).unwrap();
1046        first.get_dynamic("nested").unwrap().insert("score", 2);
1047        first.get_dynamic("items").unwrap().get_idx(0).unwrap().insert("score", 3);
1048
1049        let second = get(&path).unwrap();
1050        assert_eq!(second.get_dynamic("nested").unwrap().get_dynamic("score").and_then(|v| v.as_int()), Some(1));
1051        assert_eq!(second.get_dynamic("items").unwrap().get_idx(0).unwrap().get_dynamic("score").and_then(|v| v.as_int()), Some(1));
1052    }
1053}