From fd22c7d8d75c15f8551481826bddb18c8358cc68 Mon Sep 17 00:00:00 2001 From: Ivan FB Date: Sat, 18 Jul 2026 17:31:28 +0200 Subject: [PATCH] feat: add the channel-chat example MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A standalone room-chat TUI on the reliable channels API, with no MLS or libchat stack in the way — the channels API on its own. One process per participant (--nick/--room), --peer to wire two local instances directly or logos.dev discovery otherwise. The author's nick travels inside the payload because a received event's sender_id reports the receiving channel's own participant id, not the originator's. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ai43sF2rymPFMG9iki9fwL --- bin/channel-chat/Cargo.toml | 22 ++ bin/channel-chat/README.md | 66 ++++++ bin/channel-chat/src/main.rs | 379 +++++++++++++++++++++++++++++++++++ 3 files changed, 467 insertions(+) create mode 100644 bin/channel-chat/Cargo.toml create mode 100644 bin/channel-chat/README.md create mode 100644 bin/channel-chat/src/main.rs diff --git a/bin/channel-chat/Cargo.toml b/bin/channel-chat/Cargo.toml new file mode 100644 index 0000000..3e4d5cf --- /dev/null +++ b/bin/channel-chat/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "channel-chat" +version = "0.1.0" +edition = "2024" + +[[bin]] +name = "channel-chat" +path = "src/main.rs" + +[dependencies] +# Workspace dependencies (sorted) +base64 = { workspace = true } +waku-bindings = { workspace = true } + +# External dependencies (sorted) +anyhow = "1.0" +clap = { version = "4", features = ["derive"] } +crossterm = "0.29" +ratatui = "0.29" +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +tokio = { version = "1", features = ["full"] } diff --git a/bin/channel-chat/README.md b/bin/channel-chat/README.md new file mode 100644 index 0000000..4f1b1b5 --- /dev/null +++ b/bin/channel-chat/README.md @@ -0,0 +1,66 @@ +# channel-chat + +A room chat built directly on logos-delivery's **reliable channels** API, via +[`waku-bindings`](../../vendor/logos-delivery-rust-bindings). It is deliberately +small: no MLS, no libchat stack, no encryption of the chat itself — just the +channels API, so the API is the only thing on screen. + +For the full encrypted client see [`bin/chat-cli`](../chat-cli), which uses +these same channels underneath as its transport. + +## Running + +One process per participant, each with its own nick: + +```sh +cargo run -p channel-chat -- --nick Alice --room lobby # terminal 1 +cargo run -p channel-chat -- --nick Bob --room lobby # terminal 2 +``` + +Type and press Enter to send; Esc quits. The node takes a few seconds to start +and find peers on `logos.dev`, so the first message may need a retry. Each +instance prints its own multiaddr at startup; to wire two local instances +together without waiting on discovery, pass one to the other: + +```sh +cargo run -p channel-chat -- --nick Bob --room lobby --peer /ip4/127.0.0.1/tcp/60001/p2p/16Uiu2... +``` + +| Flag | Default | Meaning | +|---|---|---| +| `--nick` | *(required)* | Display name, and the channel's SDS participant id | +| `--room` | `lobby` | Room to join — everyone in a room shares one channel | +| `--port` | `0` (OS picks) | TCP port for the node | +| `--peer` | *(none)* | Multiaddr to dial directly, instead of waiting on discovery | + +## What it does + +1. Starts a node on the `logos.dev` preset. The preset is what supplies the + entry nodes *and* autosharding — a channel's shard is derived from its + content topic, so a node without autosharding cannot carry one. +2. Registers the three channel listeners — `received`, `sent`, `error` — + **before** starting the node. +3. Opens channel `{room}` on content topic `/logos-chat-example/1/{room}/proto`. +4. **Subscribes to that content topic.** Creating a channel does not subscribe; + ingress arrives through the messaging layer, so without this nothing lands. +5. Sends with `channel_send`, and closes the channel on quit. + +## Notes + +- **Run each participant under a different `--nick`.** The nick is both the + participant id and (via `./data-channel-chat-{nick}`) the storage root. The + persistency layer is a process-wide singleton keyed on that root, and SDS rows + are keyed by channel id, so two participants sharing a root would each load + the other's causal history and silently drop the other's messages as replays. + This is also why participants are separate processes rather than tasks. +- Sends are acknowledged asynchronously, so the outcome shows up in the status + bar (`sent` / `SEND FAILED`) rather than at the call site. That reliability — + acks, retransmission, causal ordering — is what channels add over a raw relay + publish. +- A node does not receive its own channel messages, so your own lines are echoed + locally. +- **The author's nick travels inside the payload, not in `sender_id`.** A + received event's `sender_id` is filled in from the *receiving* channel's own + participant id (`reliable_channel.nim`'s `reportReceived` emits + `self.senderId`), so reading the author off it would attribute every incoming + message to whoever is reading it. Worth knowing before reaching for that field. diff --git a/bin/channel-chat/src/main.rs b/bin/channel-chat/src/main.rs new file mode 100644 index 0000000..28baa43 --- /dev/null +++ b/bin/channel-chat/src/main.rs @@ -0,0 +1,379 @@ +//! A room chat over logos-delivery's reliable channels. +//! +//! Every participant runs one process, joins a channel named after the room and +//! types into it. There is no encryption of the chat itself and no libchat +//! stack here — the point is the channels API on its own. +//! +//! ```sh +//! cargo run -p channel-chat -- --nick Alice --room lobby +//! cargo run -p channel-chat -- --nick Bob --room lobby +//! ``` +//! +//! ## What channels buy +//! +//! Sending is `channel_send` rather than a relay publish, so the node runs SDS +//! underneath: acknowledgement, retransmission of unacknowledged messages, and +//! causal ordering. The cost is that delivery is no longer fire-and-forget, so +//! this example subscribes to the outcome events (`sent` / `error`) and shows +//! them in the status bar — without that, a failed send is silent. + +use std::io; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use clap::Parser; +use crossterm::{ + event::{self, Event, KeyCode, KeyEventKind}, + execute, + terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode}, +}; +use ratatui::{ + Frame, Terminal, + backend::{Backend, CrosstermBackend}, + layout::{Constraint, Direction, Layout}, + style::{Color, Modifier, Style}, + text::{Line, Span}, + widgets::{Block, Borders, List, ListItem, Paragraph}, +}; +use serde::{Deserialize, Serialize}; +use waku_bindings::LogosDeliveryCtx; + +/// What actually travels on the channel. +/// +/// The nick rides inside the payload rather than being read off the event's +/// `sender_id`: logos-delivery fills that field in from the *receiving* +/// channel's own participant id (`reliable_channel.nim`'s `reportReceived` +/// emits `self.senderId`), so every inbound message would otherwise be +/// attributed to whoever is reading it. +#[derive(Serialize, Deserialize)] +struct ChatLine { + nick: String, + text: String, +} + +/// The network to join. It also settles sharding: a channel's shard is derived +/// from its content topic, so the node needs autosharding, and this preset +/// supplies it (cluster 2, 8 shards) along with the entry nodes that let two +/// participants find each other. +const NETWORK_PRESET: &str = "logos.dev"; +const NODE_TIMEOUT: Duration = Duration::from_secs(30); +const DIAL_TIMEOUT_MS: u32 = 10_000; +const NOOP_ENCRYPTION: &str = "noop"; +/// Poll cadence for the input loop; also caps how long quitting takes. +const INPUT_POLL: Duration = Duration::from_millis(200); + +#[derive(Parser, Debug)] +#[command( + name = "channel-chat", + about = "Room chat over logos-delivery reliable channels" +)] +struct Cli { + /// Your display name. Doubles as the channel's SDS participant id. + #[arg(long, short)] + nick: String, + + /// The room to join. Everyone in a room shares one channel. + #[arg(long, short, default_value = "lobby")] + room: String, + + /// TCP port for the node. `0` lets the OS pick a free one. + #[arg(long, default_value_t = 0)] + port: usize, + + /// Multiaddr of a peer to dial directly. The preset's entry nodes normally + /// bring participants together on their own; this is for wiring two local + /// instances up without waiting on discovery (each prints its address at + /// startup). + #[arg(long)] + peer: Option, +} + +fn content_topic_for(room: &str) -> String { + format!("/logos-chat-example/1/{room}/proto") +} + +/// Lines shown in the chat pane, shared with the node's event callbacks. +type Log = Arc>>; + +struct App { + input: String, + nick: String, + room: String, + channel_id: String, + content_topic: String, + messages: Log, + /// The last channel outcome (`sent` / `error`) the node reported. + status: Arc>, + peer: Option, + node: LogosDeliveryCtx, +} + +impl App { + async fn new(cli: &Cli) -> Result { + // Per-nick, and this is load-bearing: the persistency layer is a + // process-wide singleton keyed on this path, and SDS rows are keyed by + // channel id. Two participants sharing a path would each load the + // other's causal history and silently drop their messages as replays. + let storage_path = format!("./data-channel-chat-{}", cli.nick); + + // Hand-rolled rather than the typed `WakuNodeConfig`, which has no + // `preset`/`mode` field — and the preset is what brings the entry nodes + // and autosharding a channel needs. `discv5UdpPort` tracks the TCP port + // so a second instance on one machine collides on neither. + let config = serde_json::json!({ + "logLevel": "FATAL", // anything louder corrupts the TUI + "mode": "Core", + "preset": NETWORK_PRESET, + "tcpPort": cli.port, + "discv5UdpPort": cli.port, + "local-storage-path": storage_path, + }) + .to_string(); + + let node = LogosDeliveryCtx::new_async(config, NODE_TIMEOUT) + .await + .map_err(anyhow::Error::msg) + .context("failed to create the node")?; + + Ok(Self { + input: String::new(), + nick: cli.nick.clone(), + room: cli.room.clone(), + channel_id: cli.room.clone(), + content_topic: content_topic_for(&cli.room), + messages: Arc::new(Mutex::new(Vec::new())), + status: Arc::new(Mutex::new("starting...".into())), + peer: cli.peer.clone(), + node, + }) + } + + /// Register the listeners, start the node, then open and subscribe the room + /// channel. + async fn start(&self) -> Result<()> { + let messages = self.messages.clone(); + // Registered before the node starts, so nothing can arrive unobserved. + self.node + .add_on_channel_message_received_listener(move |event| { + let line = BASE64 + .decode(&event.payload) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()); + let rendered = match line { + Some(line) => format!("[{}]: {}", line.nick, line.text), + None => "".to_string(), + }; + messages.lock().unwrap().push(rendered); + }); + + // The two events that make a reliable send observable. + let status = self.status.clone(); + self.node + .add_on_channel_message_sent_listener(move |event| { + *status.lock().unwrap() = format!("sent (request {})", event.request_id); + }); + + let status = self.status.clone(); + self.node + .add_on_channel_message_error_listener(move |event| { + *status.lock().unwrap() = format!("SEND FAILED: {}", event.error); + }); + + self.node + .start_node_async() + .await + .map_err(anyhow::Error::msg) + .context("failed to start the node")?; + + // Printed before the TUI takes the screen, so another instance can be + // pointed at this one with --peer. + if let Ok(addresses) = self.node.waku_listen_addresses_async().await { + println!("Listening on: {addresses}"); + } + + if let Some(peer) = &self.peer { + self.node + .waku_connect_async(peer.clone(), DIAL_TIMEOUT_MS) + .await + .map_err(anyhow::Error::msg) + .context("failed to dial the peer")?; + } + + self.node + .channel_create_async( + self.channel_id.clone(), + self.content_topic.clone(), + // The participant id. Distinct per peer, or SDS treats two + // peers as one and drops the second's messages as replays. + self.nick.clone(), + NOOP_ENCRYPTION.to_string(), + ) + .await + .map_err(anyhow::Error::msg) + .context("failed to create the channel")?; + + // Creating a channel does not subscribe to its content topic: ingress + // arrives through the messaging layer, so without this nothing lands. + self.node + .subscribe_async(self.content_topic.clone()) + .await + .map_err(anyhow::Error::msg) + .context("failed to subscribe to the channel topic")?; + + *self.status.lock().unwrap() = format!("joined #{}", self.room); + Ok(()) + } + + async fn send(&self, text: String) { + let line = ChatLine { + nick: self.nick.clone(), + text: text.clone(), + }; + let encoded = match serde_json::to_vec(&line) { + Ok(bytes) => bytes, + Err(e) => { + *self.status.lock().unwrap() = format!("SEND FAILED: {e}"); + return; + } + }; + + let req = waku_bindings::ChannelSendRequest { + payload: BASE64.encode(&encoded), + ephemeral: false, + }; + + match self + .node + .channel_send_async(self.channel_id.clone(), req) + .await + { + // Echoed locally: a node does not receive its own channel messages. + Ok(_) => self + .messages + .lock() + .unwrap() + .push(format!("[{}]: {}", self.nick, text)), + Err(e) => *self.status.lock().unwrap() = format!("SEND FAILED: {e}"), + } + } + + async fn stop(self) { + let _ = self.node.channel_close_async(self.channel_id.clone()).await; + let _ = self.node.stop_node_async().await; + // The node itself is torn down by LogosDeliveryCtx's Drop impl. + } +} + +async fn run(terminal: &mut Terminal, app: &mut App) -> Result<()> { + loop { + terminal.draw(|f| ui(f, &*app))?; + + // Non-blocking so the draw above keeps picking up messages that arrive + // on the node's callback threads while no key is pressed. + if !event::poll(INPUT_POLL)? { + continue; + } + + let Event::Key(key) = event::read()? else { + continue; + }; + // Windows reports press and release; without this every key doubles. + if key.kind != KeyEventKind::Press { + continue; + } + + match key.code { + KeyCode::Esc => return Ok(()), + KeyCode::Enter => { + let text = app.input.trim().to_string(); + if !text.is_empty() { + app.send(text).await; + app.input.clear(); + } + } + KeyCode::Char(c) => app.input.push(c), + KeyCode::Backspace => { + app.input.pop(); + } + _ => {} + } + } +} + +fn ui(f: &mut Frame, app: &App) { + let chunks = Layout::default() + .direction(Direction::Vertical) + .margin(1) + .constraints([ + Constraint::Length(1), // status + Constraint::Min(1), // messages + Constraint::Length(3), // input + ]) + .split(f.area()); + + let status = app.status.lock().unwrap().clone(); + let failed = status.starts_with("SEND FAILED"); + let header = Paragraph::new(Line::from(vec![ + Span::styled( + format!(" #{} as {} ", app.room, app.nick), + Style::default().add_modifier(Modifier::BOLD), + ), + Span::raw("— Esc to quit — "), + Span::styled( + status, + Style::default().fg(if failed { Color::Red } else { Color::DarkGray }), + ), + ])); + f.render_widget(header, chunks[0]); + + // Newest last, and only what fits: the pane does not scroll. + let height = chunks[1].height.saturating_sub(2) as usize; + let messages = app.messages.lock().unwrap(); + let items: Vec = messages + .iter() + .rev() + .take(height) + .rev() + .map(|m| ListItem::new(Line::from(m.as_str()))) + .collect(); + f.render_widget( + List::new(items).block(Block::default().borders(Borders::ALL).title("Chat")), + chunks[1], + ); + + f.render_widget( + Paragraph::new(app.input.as_str()) + .block(Block::default().borders(Borders::ALL).title("Message")), + chunks[2], + ); + f.set_cursor_position(( + chunks[2].x + app.input.chars().count() as u16 + 1, + chunks[2].y + 1, + )); +} + +#[tokio::main] +async fn main() -> Result<()> { + let cli = Cli::parse(); + + println!("Starting logos-delivery node, this takes a few seconds..."); + let mut app = App::new(&cli).await?; + app.start().await?; + + enable_raw_mode()?; + let mut stdout = io::stdout(); + execute!(stdout, EnterAlternateScreen)?; + let mut terminal = Terminal::new(CrosstermBackend::new(stdout))?; + + let res = run(&mut terminal, &mut app).await; + + disable_raw_mode()?; + execute!(terminal.backend_mut(), LeaveAlternateScreen)?; + terminal.show_cursor()?; + + app.stop().await; + res +}