tauri-plugin-deeprespond-im 1.0.0-alpha

//! 锁与频道（《4.实现与测试》第 3.3 节）：代替 WebView 的 Web Locks 和 BroadcastChannel，在全部窗口之间生效。
//! - 锁：按名称排他，可以 ifAvailable、steal；记在“页面实例 + 请求 ID”上，页面关闭、崩溃、重新加载时全部释放。
//! - 频道：转发给持有同名频道的其他页面，不回发给自己；同一发送者的消息按顺序到达（在一个锁内依次交给各页面的 Channel）。

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Mutex;

use serde_json::{json, Value};
use tokio::sync::oneshot;

use crate::error::{Error, Result};
use crate::hub::{Hub, Instance};

/// 单条频道消息的上限（附件不经过频道，经路径交给 Rust）。
pub const MAX_MESSAGE_BYTES: usize = 1024 * 1024;

/// 锁的持有者或等待者：页面实例与请求 ID。
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct Owner {
    label: String,
    instance: u64,
    request: String,
}

struct Waiter {
    owner: Owner,
    tx: oneshot::Sender<Result<bool>>,
}

#[derive(Default)]
struct LockState {
    holder: Option<Owner>,
    queue: VecDeque<Waiter>,
}

#[derive(Default)]
struct Inner {
    locks: HashMap<String, LockState>,
    /// 频道名 → 加入了的页面（label, 实例）
    channels: HashMap<String, HashSet<(String, u64)>>,
    /// 页面不在了时替它发到频道的消息（Web SDK 的 bye：窗口被关闭、崩溃时页面自己来不及发）
    wills: HashMap<(String, u64, String), Value>,
}

#[derive(Default)]
pub struct Bus {
    inner: Mutex<Inner>,
    /// 频道消息的发送顺序：转发在这个锁内进行，同一发送者的消息不会交错。
    post_order: Mutex<()>,
}

pub enum LockGrant {
    Granted,
    NotAvailable,
    Wait(oneshot::Receiver<Result<bool>>),
}

impl Bus {
    /// 请求锁。steal 时抢走当前的持有者（通知它 lock_stolen），不影响排队的请求；
    /// if_available 时取不到立即返回 NotAvailable，不排队。
    pub fn request(&self, hub: &Hub, name: &str, instance: &Instance, request: &str, steal: bool, if_available: bool) -> LockGrant {
        let owner = Owner { label: instance.label.clone(), instance: instance.id, request: request.to_string() };
        let mut inner = self.inner.lock().unwrap();
        let state = inner.locks.entry(name.to_string()).or_default();
        if steal {
            if let Some(previous) = state.holder.replace(owner) {
                hub_notify_stolen(hub, &previous, name);
            }
            return LockGrant::Granted;
        }
        if state.holder.is_none() && state.queue.is_empty() {
            state.holder = Some(owner);
            return LockGrant::Granted;
        }
        if if_available {
            return LockGrant::NotAvailable;
        }
        let (tx, rx) = oneshot::channel();
        state.queue.push_back(Waiter { owner, tx });
        LockGrant::Wait(rx)
    }

    /// 释放一个请求：持有的交给下一个等待者，排队中的取消（等待者得到 aborted）。
    pub fn release(&self, label: &str, instance: u64, request: &str) {
        let mut inner = self.inner.lock().unwrap();
        let mut empty = Vec::new();
        for (name, state) in inner.locks.iter_mut() {
            if state.holder.as_ref().is_some_and(|h| h.label == label && h.instance == instance && h.request == request) {
                state.holder = None;
                grant_next(state);
            } else if let Some(pos) = state.queue.iter().position(|w| w.owner.label == label && w.owner.instance == instance && w.owner.request == request) {
                if let Some(waiter) = state.queue.remove(pos) {
                    let _ = waiter.tx.send(Err(Error::aborted()));
                }
                if state.holder.is_none() {
                    grant_next(state);
                }
            }
            if state.holder.is_none() && state.queue.is_empty() {
                empty.push(name.clone());
            }
        }
        for name in empty {
            inner.locks.remove(&name);
        }
    }

    /// 登记页面不在了时替它发到频道 name 的消息。
    pub fn set_will(&self, name: &str, instance: &Instance, msg: Value) {
        let mut inner = self.inner.lock().unwrap();
        inner.wills.insert((instance.label.clone(), instance.id, name.to_string()), msg);
    }

    /// 页面实例不在了：释放它持有的全部锁、取消它的等待，退出它加入的频道，并替它发出登记的消息。
    pub fn release_instance(&self, hub: &Hub, label: &str, instance: u64) {
        let wills: Vec<(String, Value)> = {
            let mut inner = self.inner.lock().unwrap();
            let keys: Vec<_> = inner.wills.keys().filter(|(l, i, _)| l == label && *i == instance).cloned().collect();
            keys.into_iter().filter_map(|k| inner.wills.remove(&k).map(|v| (k.2, v))).collect()
        };
        self.release_locks_and_channels(label, instance);
        for (name, msg) in wills {
            let event = json!({ "t": "bus", "name": name, "msg": msg });
            let targets: Vec<(String, u64)> = self.inner.lock().unwrap().channels.get(&name).map(|m| m.iter().cloned().collect()).unwrap_or_default();
            for (l, id) in targets {
                if let Some(target) = hub.get(&l) {
                    if target.id == id {
                        target.send(event.clone());
                    }
                }
            }
        }
    }

    fn release_locks_and_channels(&self, label: &str, instance: u64) {
        let mut inner = self.inner.lock().unwrap();
        for state in inner.locks.values_mut() {
            state.queue.retain(|w| !(w.owner.label == label && w.owner.instance == instance));
            if state.holder.as_ref().is_some_and(|h| h.label == label && h.instance == instance) {
                state.holder = None;
                grant_next(state);
            }
        }
        inner.locks.retain(|_, s| s.holder.is_some() || !s.queue.is_empty());
        for members in inner.channels.values_mut() {
            members.retain(|(l, i)| !(l == label && *i == instance));
        }
        inner.channels.retain(|_, m| !m.is_empty());
    }

    pub fn join(&self, name: &str, instance: &Instance) {
        let mut inner = self.inner.lock().unwrap();
        inner.channels.entry(name.to_string()).or_default().insert((instance.label.clone(), instance.id));
    }

    pub fn leave(&self, name: &str, instance: &Instance) {
        let mut inner = self.inner.lock().unwrap();
        if let Some(members) = inner.channels.get_mut(name) {
            members.remove(&(instance.label.clone(), instance.id));
            if members.is_empty() {
                inner.channels.remove(name);
            }
        }
    }

    /// 转发频道消息给同名频道的其他页面。消息超过 1 MB 的拒绝。
    pub fn post(&self, hub: &Hub, name: &str, from: &Instance, msg: &Value, size: usize) -> Result<()> {
        if size > MAX_MESSAGE_BYTES {
            return Err(Error::validation("too_large", format!("频道消息 {size} 字节，超过 1 MB")));
        }
        let _order = self.post_order.lock().unwrap();
        let targets: Vec<(String, u64)> = {
            let inner = self.inner.lock().unwrap();
            match inner.channels.get(name) {
                Some(members) => members.iter().filter(|(l, i)| !(l == &from.label && *i == from.id)).cloned().collect(),
                None => Vec::new(),
            }
        };
        let event = json!({ "t": "bus", "name": name, "msg": msg });
        for (label, id) in targets {
            if let Some(instance) = hub.get(&label) {
                if instance.id == id {
                    instance.send(event.clone());
                }
            }
        }
        Ok(())
    }

    /// 测试：某个锁当前的持有者。
    #[cfg(test)]
    pub fn holder(&self, name: &str) -> Option<(String, String)> {
        let inner = self.inner.lock().unwrap();
        inner.locks.get(name).and_then(|s| s.holder.as_ref().map(|h| (h.label.clone(), h.request.clone())))
    }
}

fn grant_next(state: &mut LockState) {
    while let Some(waiter) = state.queue.pop_front() {
        let owner = waiter.owner.clone();
        if waiter.tx.send(Ok(true)).is_ok() {
            state.holder = Some(owner);
            return;
        }
        // 等待的命令已经结束（页面关闭）：跳过
    }
}

fn hub_notify_stolen(hub: &Hub, previous: &Owner, name: &str) {
    if let Some(instance) = hub.get(&previous.label) {
        if instance.id == previous.instance {
            instance.send(json!({ "t": "lock_stolen", "name": name, "id": previous.request }));
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn instance(label: &str, id: u64) -> Instance {
        Instance::for_test(label, id)
    }

    #[tokio::test]
    async fn queue_and_release_in_order() {
        let hub = Hub::default();
        let bus = Bus::default();
        let a = instance("a", 1);
        let b = instance("b", 2);
        let c = instance("c", 3);
        assert!(matches!(bus.request(&hub, "L", &a, "r1", false, false), LockGrant::Granted));
        let LockGrant::Wait(wb) = bus.request(&hub, "L", &b, "r2", false, false) else { panic!("b should wait") };
        let LockGrant::Wait(wc) = bus.request(&hub, "L", &c, "r3", false, false) else { panic!("c should wait") };
        assert!(matches!(bus.request(&hub, "L", &c, "r4", false, true), LockGrant::NotAvailable));
        bus.release("a", 1, "r1");
        assert!(wb.await.unwrap().unwrap());
        assert_eq!(bus.holder("L"), Some(("b".into(), "r2".into())));
        bus.release_instance(&hub, "b", 2);
        assert!(wc.await.unwrap().unwrap());
        assert_eq!(bus.holder("L"), Some(("c".into(), "r3".into())));
    }

    #[tokio::test]
    async fn steal_keeps_queue_and_cancel_aborts_waiter() {
        let hub = Hub::default();
        let bus = Bus::default();
        let a = instance("a", 1);
        let b = instance("b", 2);
        let c = instance("c", 3);
        assert!(matches!(bus.request(&hub, "L", &a, "r1", false, false), LockGrant::Granted));
        let LockGrant::Wait(wc) = bus.request(&hub, "L", &c, "r3", false, false) else { panic!() };
        assert!(matches!(bus.request(&hub, "L", &b, "r2", true, false), LockGrant::Granted));
        assert_eq!(bus.holder("L"), Some(("b".into(), "r2".into())));
        // a 的释放不影响 b（已被抢走）
        bus.release("a", 1, "r1");
        assert_eq!(bus.holder("L"), Some(("b".into(), "r2".into())));
        bus.release("c", 3, "r3");
        assert_eq!(wc.await.unwrap().unwrap_err().code, "aborted");
        bus.release("b", 2, "r2");
        assert_eq!(bus.holder("L"), None);
    }

    #[test]
    fn channel_keeps_order_and_skips_sender() {
        let hub = Hub::default();
        let bus = Bus::default();
        let got = std::sync::Arc::new(Mutex::new(Vec::new()));
        let a = hub.insert_for_test(Instance::for_test("a", 1));
        let b = hub.insert_for_test(Instance::with_sink("b", 2, got.clone()));
        let own = std::sync::Arc::new(Mutex::new(Vec::new()));
        let c = hub.insert_for_test(Instance::with_sink("c", 3, own.clone()));
        bus.join("ch", &a);
        bus.join("ch", &b);
        bus.join("other", &c);
        for i in 0..50 {
            bus.post(&hub, "ch", &a, &serde_json::json!(i), 2).unwrap();
        }
        let seen: Vec<i64> = got.lock().unwrap().iter().map(|e| e["msg"].as_i64().unwrap()).collect();
        assert_eq!(seen, (0..50).collect::<Vec<_>>());
        assert!(own.lock().unwrap().is_empty());
        // 页面不在了：退出它的频道，替它发出登记的消息
        bus.set_will("ch", &a, serde_json::json!("bye-a"));
        bus.release_instance(&hub, "b", 2);
        bus.post(&hub, "ch", &a, &serde_json::json!(99), 2).unwrap();
        assert_eq!(got.lock().unwrap().len(), 50);
        let d = hub.insert_for_test(Instance::with_sink("d", 4, own.clone()));
        bus.join("ch", &d);
        bus.release_instance(&hub, "a", 1);
        assert_eq!(own.lock().unwrap().last().unwrap()["msg"], serde_json::json!("bye-a"));
    }

    #[test]
    fn oversized_message_rejected() {
        let hub = Hub::default();
        let bus = Bus::default();
        let a = instance("a", 1);
        bus.join("ch", &a);
        let err = bus.post(&hub, "ch", &a, &Value::Null, MAX_MESSAGE_BYTES + 1).unwrap_err();
        assert_eq!(err.reason.as_deref(), Some("too_large"));
    }
}