tauri-plugin-deeprespond-im 1.0.0-alpha

//! Linux：直接经 D-Bus 调用 freedesktop 通知服务（org.freedesktop.Notifications）。
//! replaces_id 替换同一 id 的旧通知；按钮和点击经 ActionInvoked 信号交回（"default" 为点击通知本身）；
//! NotificationClosed 后忘掉这条通知。应用退出后通知上的按钮失效，所以没有“启动点击”。

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use futures_util::StreamExt;
use tauri::{AppHandle, Runtime};
use zbus::message::Type as MessageType;
use zbus::zvariant::Value;
use zbus::{Connection, MatchRule, MessageStream};

use super::{Clear, Payload, Request, Sink};
use crate::error::{Error, Result};

const DEST: &str = "org.freedesktop.Notifications";
const PATH: &str = "/org/freedesktop/Notifications";
const IFACE: &str = "org.freedesktop.Notifications";
/// 点击通知本身的动作键（规范约定）。
const DEFAULT_ACTION: &str = "default";

pub(super) struct Backend {
    app_name: String,
    desktop_entry: String,
    sink: Sink,
    /// 测试时指定的连接；为空时用共用的会话总线。
    bus: Option<Connection>,
    conn: tokio::sync::OnceCell<Option<Connection>>,
    caps: tokio::sync::OnceCell<Vec<String>>,
    shown: Arc<Mutex<Shown>>,
    warned: AtomicBool,
}

/// 已显示的通知：插件的 id 与通知服务分配的编号互查。
#[derive(Default)]
struct Shown {
    by_id: HashMap<String, (u32, Payload)>,
    by_server: HashMap<u32, String>,
}

impl Shown {
    fn insert(&mut self, server_id: u32, payload: Payload) {
        if let Some((old, _)) = self.by_id.get(&payload.id) {
            if *old != server_id {
                self.by_server.remove(old);
            }
        }
        self.by_server.insert(server_id, payload.id.clone());
        self.by_id.insert(payload.id.clone(), (server_id, payload));
    }

    fn remove_server(&mut self, server_id: u32) {
        if let Some(id) = self.by_server.remove(&server_id) {
            if self.by_id.get(&id).is_some_and(|(s, _)| *s == server_id) {
                self.by_id.remove(&id);
            }
        }
    }

    /// 取出要关闭的通知的编号（同时从表中删除）。
    fn take(&mut self, target: &Clear) -> Vec<u32> {
        let ids: Vec<String> = match target {
            Clear::Id(id) => self.by_id.contains_key(id).then(|| id.clone()).into_iter().collect(),
            Clear::Conversation(cid) => self.by_id.iter().filter(|(_, (_, p))| p.conversation_id.as_deref() == Some(cid.as_str())).map(|(id, _)| id.clone()).collect(),
            Clear::All => self.by_id.keys().cloned().collect(),
        };
        let mut out = Vec::new();
        for id in ids {
            if let Some((server_id, _)) = self.by_id.remove(&id) {
                self.by_server.remove(&server_id);
                out.push(server_id);
            }
        }
        out
    }
}

/// Notify 的参数（与 D-Bus 无关，便于测试）。
#[derive(Debug, PartialEq)]
pub(crate) struct NotifyArgs {
    pub app_name: String,
    pub replaces_id: u32,
    pub app_icon: String,
    pub summary: String,
    pub body: String,
    pub actions: Vec<String>,
    pub hints: Vec<(&'static str, Hint)>,
    pub expire_timeout: i32,
}

#[derive(Debug, PartialEq, Clone)]
pub(crate) enum Hint {
    Str(String),
    Bool(bool),
    Byte(u8),
}

/// 按请求和通知服务的能力组装 Notify 的参数。
pub(crate) fn notify_args(app_name: &str, desktop_entry: &str, replaces_id: u32, req: &Request, caps: &[String]) -> NotifyArgs {
    let has = |cap: &str| caps.iter().any(|c| c == cap);
    let call = req.kind == "call";
    let mut actions = Vec::new();
    if has("actions") {
        actions.push(DEFAULT_ACTION.to_string());
        actions.push("打开".to_string());
        for action in req.effective_actions() {
            actions.push(action.id);
            actions.push(action.title);
        }
    }
    let mut hints = vec![("desktop-entry", Hint::Str(desktop_entry.to_string())), ("urgency", Hint::Byte(if call { 2 } else { 1 }))];
    if req.kind == "message" {
        hints.push(("category", Hint::Str("im.received".into())));
    }
    if req.sound {
        hints.push(("sound-name", Hint::Str("message-new-instant".into())));
    } else {
        hints.push(("suppress-sound", Hint::Bool(true)));
    }
    // 支持正文标记的服务把 < & 当作标记：转义，按纯文本显示
    let body = if has("body-markup") { escape_markup(&req.body) } else { req.body.clone() };
    NotifyArgs {
        app_name: app_name.to_string(),
        replaces_id,
        app_icon: desktop_entry.to_string(),
        summary: req.title.clone(),
        body,
        actions,
        hints,
        // 来电一直显示到被处理（接听、拒绝、取消时由 SDK 清除），其他按服务的默认时长
        expire_timeout: if call { 0 } else { -1 },
    }
}

pub(crate) fn escape_markup(text: &str) -> String {
    text.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;")
}

/// ActionInvoked 的动作键 → 点击的按钮（点击通知本身为空）。
pub(crate) fn action_of(key: &str) -> Option<String> {
    (key != DEFAULT_ACTION).then(|| key.to_string())
}

impl Backend {
    pub fn new<R: Runtime>(app: &AppHandle<R>, _identifier: &str, sink: Sink) -> Self {
        Self::with_bus(app.package_info().name.clone(), crate::system::desktop_entry(app), sink, None)
    }

    pub(crate) fn with_bus(app_name: String, desktop_entry: String, sink: Sink, bus: Option<Connection>) -> Self {
        Self {
            app_name,
            desktop_entry,
            sink,
            bus,
            conn: tokio::sync::OnceCell::new(),
            caps: tokio::sync::OnceCell::new(),
            shown: Arc::new(Mutex::new(Shown::default())),
            warned: AtomicBool::new(false),
        }
    }

    /// 第一次使用时连接并订阅信号（先订阅再发第一条通知，不会漏掉点击）。
    async fn connection(&self) -> Option<Connection> {
        self.conn
            .get_or_init(|| async {
                let conn = match &self.bus {
                    Some(conn) => conn.clone(),
                    None => crate::system::session_bus().await?,
                };
                match subscribe(&conn).await {
                    Ok(stream) => {
                        let (shown, sink) = (self.shown.clone(), self.sink.clone());
                        tauri::async_runtime::spawn(listen(stream, shown, sink));
                    }
                    Err(err) => log::warn!("订阅通知服务的信号失败，点击通知不会交回：{err}"),
                }
                Some(conn)
            })
            .await
            .clone()
    }

    async fn caps(&self, conn: &Connection) -> Vec<String> {
        self.caps
            .get_or_init(|| async {
                let reply = conn.call_method(Some(DEST), PATH, Some(IFACE), "GetCapabilities", &()).await;
                reply.and_then(|m| m.body().deserialize::<Vec<String>>()).unwrap_or_default()
            })
            .await
            .clone()
    }

    pub async fn show(&self, req: &Request, payload: &Payload) -> Result<()> {
        let Some(conn) = self.connection().await else {
            if !self.warned.swap(true, Ordering::Relaxed) {
                log::warn!("没有 D-Bus 会话总线，系统通知不显示");
            }
            return Ok(());
        };
        let caps = self.caps(&conn).await;
        let replaces = self.shown.lock().unwrap().by_id.get(&req.id).map(|(s, _)| *s).unwrap_or(0);
        let args = notify_args(&self.app_name, &self.desktop_entry, replaces, req, &caps);
        let hints: HashMap<&str, Value<'_>> = args
            .hints
            .iter()
            .map(|(k, v)| {
                let value = match v {
                    Hint::Str(s) => Value::from(s.as_str()),
                    Hint::Bool(b) => Value::from(*b),
                    Hint::Byte(b) => Value::from(*b),
                };
                (*k, value)
            })
            .collect();
        let actions: Vec<&str> = args.actions.iter().map(String::as_str).collect();
        let body = (args.app_name.as_str(), args.replaces_id, args.app_icon.as_str(), args.summary.as_str(), args.body.as_str(), actions, hints, args.expire_timeout);
        let reply = conn.call_method(Some(DEST), PATH, Some(IFACE), "Notify", &body).await.map_err(|e| Error::internal(format!("通知服务：{e}")))?;
        let server_id: u32 = reply.body().deserialize().map_err(|e| Error::internal(format!("通知服务的回复：{e}")))?;
        self.shown.lock().unwrap().insert(server_id, payload.clone());
        Ok(())
    }

    pub async fn clear(&self, target: &Clear) -> Result<()> {
        let ids = self.shown.lock().unwrap().take(target);
        if ids.is_empty() {
            return Ok(());
        }
        let Some(conn) = self.connection().await else { return Ok(()) };
        for id in ids {
            if let Err(err) = conn.call_method(Some(DEST), PATH, Some(IFACE), "CloseNotification", &(id,)).await {
                log::debug!("关闭通知 {id} 失败：{err}");
            }
        }
        Ok(())
    }

    pub async fn permission(&self) -> &'static str {
        "granted"
    }

    pub async fn request_permission(&self) -> &'static str {
        "granted"
    }
}

async fn subscribe(conn: &Connection) -> zbus::Result<MessageStream> {
    let rule = MatchRule::builder().msg_type(MessageType::Signal).sender(DEST)?.path(PATH)?.interface(IFACE)?.build();
    MessageStream::for_match_rule(rule, conn, None).await
}

/// ActionInvoked、NotificationClosed：只处理本应用显示的通知（信号可能广播给所有客户端）。
async fn listen(mut stream: MessageStream, shown: Arc<Mutex<Shown>>, sink: Sink) {
    while let Some(msg) = stream.next().await {
        let Ok(msg) = msg else { continue };
        let header = msg.header();
        let Some(member) = header.member() else { continue };
        match member.as_str() {
            "ActionInvoked" => {
                let Ok((server_id, key)) = msg.body().deserialize::<(u32, String)>() else { continue };
                let payload = {
                    let shown = shown.lock().unwrap();
                    shown.by_server.get(&server_id).and_then(|id| shown.by_id.get(id)).map(|(_, p)| p.clone())
                };
                if let Some(payload) = payload {
                    sink(payload.with_action(action_of(&key)), false);
                }
            }
            "NotificationClosed" => {
                if let Ok((server_id, _reason)) = msg.body().deserialize::<(u32, u32)>() {
                    shown.lock().unwrap().remove_server(server_id);
                }
            }
            _ => {}
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::notify::Action;
    use std::io::BufRead;
    use std::time::Duration;
    use zbus::zvariant::OwnedValue;

    fn request() -> Request {
        Request { id: "dr:conv:c1".into(), title: "张三".into(), body: "a < b & c".into(), kind: "message".into(), conversation_id: Some("c1".into()), seq: Some(7), sound: true, ..Default::default() }
    }

    #[test]
    fn args_follow_capabilities() {
        let caps = vec!["actions".to_string(), "body".to_string(), "body-markup".to_string()];
        let args = notify_args("示例", "com.example.im", 0, &request(), &caps);
        assert_eq!(args.actions, vec!["default", "打开"]);
        assert_eq!(args.body, "a &lt; b &amp; c");
        assert_eq!(args.expire_timeout, -1);
        assert!(args.hints.contains(&("sound-name", Hint::Str("message-new-instant".into()))));
        assert!(args.hints.contains(&("desktop-entry", Hint::Str("com.example.im".into()))));
        assert!(args.hints.contains(&("category", Hint::Str("im.received".into()))));

        // 来电：默认按钮、不过期、紧急；关掉声音
        let mut call = request();
        call.kind = "call".into();
        call.sound = false;
        let args = notify_args("示例", "com.example.im", 9, &call, &caps);
        assert_eq!(args.replaces_id, 9);
        assert_eq!(args.actions, vec!["default", "打开", "accept", "接听", "decline", "拒绝"]);
        assert_eq!(args.expire_timeout, 0);
        assert!(args.hints.contains(&("urgency", Hint::Byte(2))));
        assert!(args.hints.contains(&("suppress-sound", Hint::Bool(true))));

        // 服务不支持按钮、不支持标记：不带按钮，正文原样
        let mut custom = request();
        custom.actions = Some(vec![Action { id: "x".into(), title: "X".into() }]);
        let args = notify_args("示例", "e", 0, &custom, &["body".to_string()]);
        assert!(args.actions.is_empty());
        assert_eq!(args.body, "a < b & c");
    }

    #[test]
    fn action_keys() {
        assert_eq!(action_of("default"), None);
        assert_eq!(action_of("decline"), Some("decline".to_string()));
    }

    #[test]
    fn shown_table() {
        let mut shown = Shown::default();
        let mut p = request().payload();
        shown.insert(1, p.clone());
        // 同一 id 的替换可能换了编号
        shown.insert(2, p.clone());
        assert!(!shown.by_server.contains_key(&1));
        p.id = "other".into();
        p.conversation_id = Some("c2".into());
        shown.insert(3, p);
        assert_eq!(shown.take(&Clear::Conversation("c1".into())), vec![2]);
        assert_eq!(shown.take(&Clear::Id("dr:conv:c1".into())), Vec::<u32>::new());
        shown.remove_server(3);
        assert!(shown.by_id.is_empty());
    }

    /// 收到的 Notify：标题、replaces_id、按钮、提示。
    type NotifyCall = (String, u32, Vec<String>, HashMap<String, OwnedValue>);

    /// 测试用的通知服务：记下收到的调用，编号从 100 开始。
    #[derive(Default, Clone)]
    struct FakeServer {
        calls: Arc<Mutex<Vec<NotifyCall>>>,
        closed: Arc<Mutex<Vec<u32>>>,
        next: Arc<Mutex<u32>>,
    }

    #[zbus::interface(name = "org.freedesktop.Notifications")]
    impl FakeServer {
        #[allow(clippy::too_many_arguments)]
        async fn notify(&self, _app_name: String, replaces_id: u32, _app_icon: String, summary: String, _body: String, actions: Vec<String>, hints: HashMap<String, OwnedValue>, _expire_timeout: i32) -> u32 {
            self.calls.lock().unwrap().push((summary, replaces_id, actions, hints));
            if replaces_id != 0 {
                return replaces_id;
            }
            let mut next = self.next.lock().unwrap();
            *next += 1;
            100 + *next
        }

        async fn close_notification(&self, id: u32) {
            self.closed.lock().unwrap().push(id);
        }

        async fn get_capabilities(&self) -> Vec<String> {
            vec!["actions".into(), "body".into()]
        }
    }

    /// 起一个独立的会话总线（dbus-daemon），不依赖桌面环境；没有 dbus-daemon 的环境跳过。
    fn start_bus() -> Option<(std::process::Child, String)> {
        let mut child = std::process::Command::new("dbus-daemon").args(["--session", "--nofork", "--print-address=1"]).stdout(std::process::Stdio::piped()).spawn().ok()?;
        let mut line = String::new();
        std::io::BufReader::new(child.stdout.take()?).read_line(&mut line).ok()?;
        Some((child, line.trim().to_string()))
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn end_to_end_with_fake_server() {
        let Some((mut daemon, address)) = start_bus() else {
            eprintln!("没有 dbus-daemon，跳过");
            return;
        };
        let server = FakeServer::default();
        let server_conn = zbus::connection::Builder::address(address.as_str()).unwrap().name(DEST).unwrap().serve_at(PATH, server.clone()).unwrap().build().await.unwrap();
        let client = zbus::connection::Builder::address(address.as_str()).unwrap().build().await.unwrap();
        let clicks: Arc<Mutex<Vec<Payload>>> = Arc::default();
        let sink_clicks = clicks.clone();
        let sink: Sink = Arc::new(move |p, launched| {
            assert!(!launched);
            sink_clicks.lock().unwrap().push(p);
        });
        let backend = Backend::with_bus("示例".into(), "com.example.im".into(), sink, Some(client));

        let mut call = request();
        call.kind = "call".into();
        let mut data = serde_json::Map::new();
        data.insert("call_id".into(), serde_json::json!("k1"));
        call.data = Some(data);
        backend.show(&call, &call.payload()).await.unwrap();
        // 同一 id 再显示一次：带上 replaces_id
        backend.show(&call, &call.payload()).await.unwrap();
        {
            let calls = server.calls.lock().unwrap();
            assert_eq!(calls.len(), 2);
            assert_eq!(calls[0].1, 0);
            assert_eq!(calls[1].1, 101);
            assert_eq!(calls[0].2, vec!["default", "打开", "accept", "接听", "decline", "拒绝"]);
            assert_eq!(calls[0].3.get("urgency").map(|v| v.downcast_ref::<u8>().unwrap()), Some(2));
        }

        // 服务发出点击和按钮的信号
        server_conn.emit_signal(None::<&str>, PATH, IFACE, "ActionInvoked", &(101u32, "default")).await.unwrap();
        server_conn.emit_signal(None::<&str>, PATH, IFACE, "ActionInvoked", &(101u32, "decline")).await.unwrap();
        // 别的应用的通知：忽略
        server_conn.emit_signal(None::<&str>, PATH, IFACE, "ActionInvoked", &(55u32, "default")).await.unwrap();
        for _ in 0..50 {
            if clicks.lock().unwrap().len() >= 2 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
        {
            let clicks = clicks.lock().unwrap();
            assert_eq!(clicks.len(), 2, "{clicks:?}");
            assert_eq!(clicks[0].action, None);
            assert_eq!(clicks[0].id, "dr:conv:c1");
            assert_eq!(clicks[1].action.as_deref(), Some("decline"));
            assert_eq!(clicks[1].data.as_ref().and_then(|d| d.get("call_id")).and_then(|v| v.as_str()), Some("k1"));
        }

        // 按会话清除：关闭这条通知
        backend.clear(&Clear::Conversation("c1".into())).await.unwrap();
        assert_eq!(*server.closed.lock().unwrap(), vec![101]);
        // 关闭后服务发出 NotificationClosed：再显示时不再替换
        server_conn.emit_signal(None::<&str>, PATH, IFACE, "NotificationClosed", &(101u32, 3u32)).await.unwrap();
        backend.show(&request(), &request().payload()).await.unwrap();
        assert_eq!(server.calls.lock().unwrap()[2].1, 0);

        let _ = daemon.kill();
        let _ = daemon.wait();
    }
}