qualia_client_core/
chat_mesh_service.rs

1//! Running chat-over-mesh service — drives [`ChatMeshBridge`] over a live [`MeshService`].
2//!
3//! This is the runtime that makes "chat over the mesh" flow. It owns a [`MeshService`] and a
4//! [`ChatMeshBridge`], and on its own thread it
5//! * publishes chat envelopes to peers (reliable, on the CHAT port),
6//! * drains inbound CHAT-port datagrams, feeds them through the reliable channel, sends ACKs back,
7//!   and **either** applies each newly-delivered [`RelayEnvelope`] to the session store **or**
8//!   forwards it to the caller (chosen at spawn), and
9//! * retransmits unacknowledged frames on a timer.
10//!
11//! It keeps a cloneable [`MeshControl`] so the desktop can add peers, drive handshakes and read
12//! status on the *same* mesh the chat loop is running — no second set of tunnels. The lower layers
13//! ([`ChatMeshBridge`], [`crate::mesh_channel`], `mesh_datagram`) are pure and independently tested;
14//! this module is the thin I/O loop binding them to the live tunnels.
15//!
16//! Native-only (the mesh is native-only); WASM chat uses the relay path.
17#![cfg(not(target_arch = "wasm32"))]
18
19use std::net::SocketAddr;
20use std::path::PathBuf;
21use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender, TryRecvError};
22use std::thread::JoinHandle;
23use std::time::{Duration, Instant};
24
25use qualia_core_db::p2p::mesh_datagram::{decode_datagram, encode_datagram, ports};
26use qualia_core_db::p2p::mesh_service::{MeshControl, MeshService};
27
28use crate::chat_mesh::ChatMeshBridge;
29use crate::chat_relay::RelayEnvelope;
30
31/// How often the loop wakes to service retransmits / drain inbound when idle.
32const LOOP_SLEEP: Duration = Duration::from_millis(10);
33
34/// A chat message received from a peer over the mesh: the sending peer's id and the envelope.
35pub type IncomingChat = (String, RelayEnvelope);
36
37/// Where the loop delivers newly-received envelopes.
38enum Sink {
39    /// Forward to the caller's channel (generic / tests).
40    Channel(Sender<IncomingChat>),
41    /// Apply directly to the session store under this root (the desktop path).
42    Apply(PathBuf),
43}
44
45enum Cmd {
46    Publish {
47        peers: Vec<String>,
48        env: Box<RelayEnvelope>,
49    },
50    Shutdown,
51}
52
53/// A handle to the running chat-over-mesh service.
54pub struct ChatMeshService {
55    control: MeshControl,
56    cmd_tx: Sender<Cmd>,
57    /// Present only in channel mode (`spawn`); `None` in apply mode (`spawn_applying`).
58    inbound_rx: Option<Receiver<IncomingChat>>,
59    handle: Option<JoinHandle<()>>,
60}
61
62impl ChatMeshService {
63    /// Spawn the chat loop over `mesh`, **forwarding** received envelopes to [`try_recv`] /
64    /// [`recv_timeout`]. Used by tests and callers that want to handle delivery themselves.
65    ///
66    /// [`try_recv`]: ChatMeshService::try_recv
67    /// [`recv_timeout`]: ChatMeshService::recv_timeout
68    pub fn spawn(mesh: MeshService) -> ChatMeshService {
69        let (inbound_tx, inbound_rx) = channel::<IncomingChat>();
70        Self::spawn_with_sink(mesh, Sink::Channel(inbound_tx), Some(inbound_rx))
71    }
72
73    /// Spawn the chat loop over `mesh`, **applying** received envelopes directly to the session store
74    /// under `storage_root` via [`crate::chat_relay::apply_incoming_envelope`] (dedup + agent-message
75    /// validation + UI notify). This is the desktop path: publish with [`publish`], and inbound chat
76    /// lands in the local sessions automatically.
77    ///
78    /// [`publish`]: ChatMeshService::publish
79    pub fn spawn_applying(mesh: MeshService, storage_root: PathBuf) -> ChatMeshService {
80        Self::spawn_with_sink(mesh, Sink::Apply(storage_root), None)
81    }
82
83    fn spawn_with_sink(
84        mesh: MeshService,
85        sink: Sink,
86        inbound_rx: Option<Receiver<IncomingChat>>,
87    ) -> ChatMeshService {
88        let control = mesh.control();
89        let (cmd_tx, cmd_rx) = channel::<Cmd>();
90        let handle = std::thread::Builder::new()
91            .name("chat-mesh".into())
92            .spawn(move || run(mesh, &cmd_rx, sink))
93            .expect("spawn chat-mesh thread");
94        ChatMeshService {
95            control,
96            cmd_tx,
97            inbound_rx,
98            handle: Some(handle),
99        }
100    }
101
102    // ── Mesh control pass-throughs (same mesh the chat loop drives) ──
103
104    /// Add a peer to the underlying mesh; returns the local bound address.
105    pub fn add_peer(
106        &self,
107        peer_id: &str,
108        peer_pubkey_hex: &str,
109        endpoint: Option<SocketAddr>,
110    ) -> Result<SocketAddr, String> {
111        self.control.add_peer(peer_id, peer_pubkey_hex, endpoint)
112    }
113
114    /// Point a peer's tunnel at `addr`.
115    pub fn set_peer_endpoint(&self, peer_id: &str, addr: SocketAddr) -> Result<(), String> {
116        self.control.set_peer_endpoint(peer_id, addr)
117    }
118
119    /// Initiate the handshake with a peer.
120    pub fn initiate_handshake(&self, peer_id: &str) -> Result<(), String> {
121        self.control.initiate_handshake(peer_id)
122    }
123
124    /// Peers currently in the mesh.
125    pub fn peers(&self) -> Result<Vec<String>, String> {
126        self.control.peers()
127    }
128
129    /// Whether a live session exists with a peer.
130    pub fn has_session(&self, peer_id: &str) -> Result<bool, String> {
131        self.control.has_session(peer_id)
132    }
133
134    /// Poll until a session with `peer_id` is established or `timeout` elapses.
135    pub fn wait_for_session(&self, peer_id: &str, timeout: Duration) -> bool {
136        self.control.wait_for_session(peer_id, timeout)
137    }
138
139    // ── Chat ──
140
141    /// Reliably send `env` to each peer in `peers` over the mesh.
142    pub fn publish(&self, peers: Vec<String>, env: RelayEnvelope) -> Result<(), String> {
143        self.cmd_tx
144            .send(Cmd::Publish {
145                peers,
146                env: Box::new(env),
147            })
148            .map_err(|_| "chat-mesh thread is gone".to_string())
149    }
150
151    /// Non-blocking receive of the next inbound chat message (channel mode only; `None` in apply mode).
152    pub fn try_recv(&self) -> Option<IncomingChat> {
153        self.inbound_rx.as_ref().and_then(|rx| rx.try_recv().ok())
154    }
155
156    /// Block up to `timeout` for the next inbound chat message (channel mode only).
157    pub fn recv_timeout(&self, timeout: Duration) -> Option<IncomingChat> {
158        let rx = self.inbound_rx.as_ref()?;
159        match rx.recv_timeout(timeout) {
160            Ok(v) => Some(v),
161            Err(RecvTimeoutError::Timeout) | Err(RecvTimeoutError::Disconnected) => None,
162        }
163    }
164
165    /// Stop the chat loop and join the thread (also runs on drop).
166    pub fn shutdown(&mut self) {
167        let _ = self.cmd_tx.send(Cmd::Shutdown);
168        if let Some(h) = self.handle.take() {
169            let _ = h.join();
170        }
171    }
172}
173
174impl Drop for ChatMeshService {
175    fn drop(&mut self) {
176        self.shutdown();
177    }
178}
179
180/// Frame a reliable-channel frame as a CHAT-port datagram and send it to `peer` over the mesh.
181fn send_chat_frame(mesh: &MeshService, peer: &str, frame: &[u8]) {
182    let dgram = encode_datagram(ports::CHAT, ports::CHAT, frame);
183    let _ = mesh.send(peer, dgram);
184}
185
186fn deliver(sink: &Sink, peer_id: &str, env: RelayEnvelope) -> bool {
187    match sink {
188        Sink::Channel(tx) => tx.send((peer_id.to_string(), env)).is_ok(),
189        Sink::Apply(root) => {
190            // Best-effort: a message for a session we don't have (or our own echo) is simply dropped.
191            let _ = crate::chat_relay::apply_incoming_envelope(root, &env.session_id, &env);
192            true
193        }
194    }
195}
196
197fn run(mesh: MeshService, cmd_rx: &Receiver<Cmd>, sink: Sink) {
198    let mut bridge = ChatMeshBridge::default();
199    let start = Instant::now();
200
201    loop {
202        let now = start.elapsed().as_millis() as u64;
203
204        // 1. Outbound: publish commands.
205        loop {
206            match cmd_rx.try_recv() {
207                Ok(Cmd::Publish { peers, env }) => {
208                    for out in bridge.broadcast(&peers, &env, now) {
209                        send_chat_frame(&mesh, &out.peer_did, &out.frame);
210                    }
211                }
212                Ok(Cmd::Shutdown) | Err(TryRecvError::Disconnected) => return,
213                Err(TryRecvError::Empty) => break,
214            }
215        }
216
217        // 2. Inbound: drain the mesh's decrypted datagrams.
218        while let Some(pkt) = mesh.try_recv() {
219            let Some(d) = decode_datagram(&pkt.inner) else {
220                continue;
221            };
222            if d.dst_port != ports::CHAT {
223                continue; // not ours (presence/QDP/etc.)
224            }
225            let inb = bridge.on_inbound(&pkt.peer_id, &d.payload, now);
226            for ack in inb.acks {
227                send_chat_frame(&mesh, &ack.peer_did, &ack.frame);
228            }
229            if let Some(env) = inb.delivered {
230                if !deliver(&sink, &pkt.peer_id, env) {
231                    return; // channel receiver gone
232                }
233            }
234        }
235
236        // 3. Retransmit anything unacknowledged whose RTO elapsed.
237        for out in bridge.on_tick(now) {
238            send_chat_frame(&mesh, &out.peer_did, &out.frame);
239        }
240
241        std::thread::sleep(LOOP_SLEEP);
242    }
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248    use qualia_core_db::p2p::social_webnet::SocialWebNet;
249    use qualia_core_db::p2p::wireguard_userspace::generate_keypair;
250
251    fn envelope(content: &str) -> RelayEnvelope {
252        RelayEnvelope {
253            session_id: "room-1".into(),
254            lamport: 5,
255            role: "user".into(),
256            content: content.into(),
257            author_did: "did:wf:alice".into(),
258            author_name: Some("Alice".into()),
259            reply_to_fragment: None,
260            timestamp: 1_700_000_000,
261            signature_hex: "sig".into(),
262            sub_agent_of: None,
263            agent_did: None,
264            model_id: None,
265            agent_backend: None,
266            outcome_sharing: None,
267        }
268    }
269
270    /// End-to-end over real loopback sockets: establish two mesh nodes, wrap each in a
271    /// `ChatMeshService`, and prove a chat envelope published on Alice's node is delivered to Bob's
272    /// inbound channel — the full "chat over the mesh" path through the running runtime.
273    #[test]
274    fn chat_envelope_flows_node_to_node_over_the_mesh() {
275        let a_keys = generate_keypair();
276        let b_keys = generate_keypair();
277        let (a_pub, b_pub) = (a_keys.public_hex(), b_keys.public_hex());
278
279        let to = Some(Duration::from_millis(50));
280        let ip = "127.0.0.1".parse().unwrap();
281        let a_mesh = MeshService::spawn(SocialWebNet::new(a_keys, ip, to));
282        let b_mesh = MeshService::spawn(SocialWebNet::new(b_keys, ip, to));
283
284        let a_local = a_mesh.add_peer("did:wf:bob", &b_pub, None).unwrap();
285        let b_local = b_mesh.add_peer("did:wf:alice", &a_pub, None).unwrap();
286        a_mesh.set_peer_endpoint("did:wf:bob", b_local).unwrap();
287        b_mesh.set_peer_endpoint("did:wf:alice", a_local).unwrap();
288        a_mesh.initiate_handshake("did:wf:bob").unwrap();
289        assert!(a_mesh.wait_for_session("did:wf:bob", Duration::from_secs(3)));
290        assert!(b_mesh.wait_for_session("did:wf:alice", Duration::from_secs(3)));
291
292        let alice = ChatMeshService::spawn(a_mesh);
293        let bob = ChatMeshService::spawn(b_mesh);
294
295        let env = envelope("hello Bob, over the SocialWebNet");
296        alice
297            .publish(vec!["did:wf:bob".to_string()], env.clone())
298            .unwrap();
299
300        let (from, got) = bob
301            .recv_timeout(Duration::from_secs(3))
302            .expect("Bob received a chat envelope");
303        assert_eq!(from, "did:wf:alice");
304        assert_eq!(got.content, "hello Bob, over the SocialWebNet");
305        assert_eq!(got.session_id, "room-1");
306    }
307}
qualia_client_core/chat_mesh_service.rs

qualia_client_core/
chat_mesh_service.rs
