ek-core 0.5.5

//! 请求应答信封与一次性信道(envelope + Oneshot)。
//!
//! 设计要点(P-H 执行期修订,审核记录于 TESTPLAN):
//! - `Envelope` **非泛型**:`payload: Arc<dyn Any + Send + Sync>` + erasure 回包
//!   `reply: OneshotSender<Box<dyn Any + Send + Sync>>`。
//!   原因:若信封双层泛型 `<Req,Res>`,一旦收发双方契约类型不同,收方整封 downcast 失败
//!   拿不到回包 Sender,"错型必回 InvalidPayload" 无法实现(A 的 Res 只在 A 侧存在)。
//!   抹平后的契约:B 拆 payload(错型→可回 Err(InvalidPayload)),A 收回包后自己 downcast
//!   判定(错型同样落 InvalidPayload)。任何路径都不存在类型混淆(全部经 Any TypeId 校验)。
//! - 一次性信道 = std::sync::mpsc(D5):零依赖;断连=TargetGone、超时=Timeout。

use ek_event::Event;
use std::any::Any;
use std::fmt;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{RecvTimeoutError, Sender as StdSender};
use std::time::Duration;

/// 寻址请求 type:`eReq.<target>`(精确 key,与 ek-event 精确匹配语义一致)。
pub fn e_req_type(target: &str) -> String {
    format!("eReq.{target}")
}

/// 定向通知 type:`eTo.<target>`。
pub fn e_to_type(target: &str) -> String {
    format!("eTo.{target}")
}

/// 协议级错误(B 明确回)。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReplyErr {
    /// 请求载荷类型与本插件契约不符。
    InvalidPayload,
}

impl fmt::Display for ReplyErr {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidPayload => write!(f, "invalid payload"),
        }
    }
}

/// 寻址信封(非泛型,payload 抹平)。
pub struct Envelope {
    /// 请求载荷(由发起方 Box 于 Arc 中;接收方 downcast 到自身契约类型)。
    pub payload: Arc<dyn Any + Send + Sync>,
    /// 义务回包:成功 `Box::new(Ok(Res))`,契约错 `Box::new(Err(ReplyErr::InvalidPayload))`。
    /// *handler panic 时 reply 随信封 drop → 信道关闭 → 发起方立即 TargetGone*。
    pub reply: OneshotSender<Box<dyn Any + Send + Sync>>,
}

impl Envelope {
    /// 发起方构造helper:由 `ctx.ask` 使用。
    pub fn new(
        payload: Arc<dyn Any + Send + Sync>,
    ) -> (Envelope, OneshotReceiver<Box<dyn Any + Send + Sync>>) {
        let (tx, rx) = oneshot();
        (Envelope { payload, reply: tx }, rx)
    }

    /// 接收方标准处理(B 侧唯一入口helper):
    /// - payload downcast 成功 → `handler(&req)`,回 `Ok(Res)`;
    /// - payload downcast 失败 → 回 `Err(ReplyErr::InvalidPayload)`;
    /// - event 载荷不是 Envelope → 静默忽略(错入 type 的兜底);
    /// - handler panic → 本函数帧先 unwind,reply 未 send → drop 断连(天然错误信令)。
    pub fn handle<Req, Res>(event: &Event, handler: impl FnOnce(&Req) -> Res)
    where
        Req: Any + Send + Sync,
        Res: Any + Send + Sync,
    {
        let Some(env) = event.data.downcast_ref::<Envelope>() else {
            return;
        };
        match Arc::clone(&env.payload).downcast::<Req>() {
            Ok(req) => {
                let res = handler(&*req);
                let _ = env.reply.send(Box::new(Ok::<Res, ReplyErr>(res)));
            }
            Err(_) => {
                let _ = env
                    .reply
                    .send(Box::new(Err::<Res, ReplyErr>(ReplyErr::InvalidPayload)));
            }
        }
    }
}

/// 一次性发送端(共享引用可发;AtomicBool 守卫保证协议上只有第一次 send 生效)。
///
/// 设计上的让步(实现期修订):worker 拿到的是 `&Event`,无法把 Envelope.reply move 出来,
/// 因此 send 取 `&self` + 内部 `sent` 标志 —— 第二次及以后调用直接 `Err(v)` 返回,协议上仍是一发。
pub struct OneshotSender<T> {
    tx: StdSender<T>,
    sent: AtomicBool,
}

impl<T> OneshotSender<T> {
    pub(crate) fn new_inner(tx: StdSender<T>) -> Self {
        Self {
            tx,
            sent: AtomicBool::new(false),
        }
    }
    /// 第一次调用生效;接收端已弃收则 `Err(v)` 原样归还;重复调用直接 `Err(v)`。
    pub fn send(&self, v: T) -> Result<(), T> {
        if self.sent.swap(true, Ordering::SeqCst) {
            return Err(v);
        }
        self.tx.send(v).map_err(|e| e.0)
    }
}

/// 接收端错误。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OneshotErr {
    /// 等满时限仍未回包。
    Timeout,
    /// 信道关闭(接收前发送端被 drop / 目标异常退出)。
    Closed,
}

/// 一次性接收端(recv 消费自身)。
pub struct OneshotReceiver<T>(std::sync::mpsc::Receiver<T>);

impl<T> OneshotReceiver<T> {
    pub(crate) fn new_inner(rx: std::sync::mpsc::Receiver<T>) -> Self {
        Self(rx)
    }
    /// 无限期等待;`Err(OneshotErr::Closed)` = 发送端先消失。
    pub fn recv(self) -> Result<T, OneshotErr> {
        self.0.recv().map_err(|_| OneshotErr::Closed)
    }
    /// 带时限等待。
    pub fn recv_timeout(self, d: Duration) -> Result<T, OneshotErr> {
        match self.0.recv_timeout(d) {
            Ok(v) => Ok(v),
            Err(RecvTimeoutError::Timeout) => Err(OneshotErr::Timeout),
            Err(RecvTimeoutError::Disconnected) => Err(OneshotErr::Closed),
        }
    }
}

/// 构造一对 one-shot 信道。
pub fn oneshot<T>() -> (OneshotSender<T>, OneshotReceiver<T>) {
    let (tx, rx) = std::sync::mpsc::channel();
    (OneshotSender::new_inner(tx), OneshotReceiver::new_inner(rx))
}