//! The adapter that drives `tsclientlib` on a background task and //! exposes a typed channel-and-future API to the rest of Chanora. //! //! Threading model: //! //! * `ProtocolClient::connect` spawns a tokio task that owns the //! `tsclientlib::Connection` (which is not `Send`-safe to move //! across awaits in some shapes — keeping it inside a single task //! sidesteps the problem entirely). //! * The task exposes its life via a `oneshot` that fires when the //! initial state snapshot is ready. //! * Snapshot reads are served by sending a request over an //! `mpsc::channel`; the task replies on a `oneshot` per request. //! * Outbound voice packets are submitted via a separate mpsc; //! inbound voice packets are forwarded out via a broadcast channel //! so multiple sinks (recorder, audio mixer, …) can subscribe. //! * Disconnect is requested via a `oneshot`; the task drains //! `tsclientlib`'s outbound events and exits. use std::net::SocketAddr; use std::time::{Duration, Instant}; use base64::prelude::*; use chanora_resolver::ChanoraResolver; use futures::prelude::*; use std::collections::HashMap; use tokio::sync::{mpsc, oneshot}; use tracing::{info, warn}; use tsclientlib::data::{self, Channel, Client}; use tsclientlib::messages::s2c::{InClientDbInfoPart, InMessage}; use tsclientlib::prelude::*; use tsclientlib::{ ChannelId as TsChannelId, ClientId as TsClientId, Connection, ConnectionStats, DisconnectOptions, Identity, MessageHandle, OutCommandExt, StreamItem, Version, }; use tsproto_packets::packets::{Direction, Flags, InAudioBuf, OutCommand, OutPacket, PacketType}; use tsproto_types::ClientType; use crate::dto::{ ChannelId, ChannelInfo, ChatMessage, ClientId, ClientInfo, ClientProfile, MessageTarget, ProtocolDelta, ServerActivity, ServerSnapshot, }; use crate::poke_limiter::PokeLimiter; use crate::ProtocolError; const SPEAKING_ACTIVITY_WINDOW: Duration = Duration::from_millis(750); const INBOUND_VOICE_SEND_TIMEOUT: Duration = Duration::from_millis(40); const PROFILE_REFRESH_RESULT_TIMEOUT: Duration = Duration::from_secs(3); const OUTBOUND_VOICE_PACKETS_PER_TICK: usize = 8; const DISCONNECT_REPLY_TIMEOUT: Duration = Duration::from_secs(1); const DISCONNECT_EVENT_DRAIN_TIMEOUT: Duration = Duration::from_millis(500); type PendingMoves = HashMap< MessageHandle, ( u64, Option>>, std::time::Instant, ), >; struct EventChannels { voice_in: mpsc::Sender, chat: mpsc::Sender, activity: mpsc::Sender, delta: mpsc::Sender, } #[derive(Debug, PartialEq, Eq)] enum SendTimeoutError { Timeout(T), Closed(T), } async fn send_with_timeout( tx: &mpsc::Sender, value: T, timeout_duration: Duration, ) -> Result<(), SendTimeoutError> { match tokio::time::timeout(timeout_duration, tx.reserve()).await { Ok(Ok(permit)) => { permit.send(value); Ok(()) } Ok(Err(_)) => Err(SendTimeoutError::Closed(value)), Err(_) => Err(SendTimeoutError::Timeout(value)), } } fn drain_voice_packets_for_tick( voice_out_rx: &mut mpsc::Receiver, max_packets: usize, mut send: impl FnMut(T) -> Result<(), E>, ) -> usize { let mut drained = 0; for _ in 0..max_packets { let packet = match voice_out_rx.try_recv() { Ok(packet) => packet, Err(_) => break, }; let _ = send(packet); drained += 1; } drained } async fn bounded_drain_stream(stream: S, timeout_duration: Duration) where S: futures::Stream, { let _ = tokio::time::timeout(timeout_duration, stream.for_each(|_| future::ready(()))).await; } /// Pick the TeamSpeak `client_version`/platform/signature triple /// (sourced from `ReSpeak/tsdeclarations/Versions.csv`, baked into /// `tsproto-types` at vendor-time) that best matches the *runtime* /// platform Chanora is executing on. /// /// Per-platform selection (set by the project owner): /// /// * Windows / Linux / macOS desktops announce as TeamSpeak **5 /// beta51** — the latest TS5 desktop signature in the vendored /// `tsproto-types` enum. TS5 servers expect this shape; TS3 /// servers are happy to accept any well-formed signed client /// descriptor and have no codec-version coupling. /// * Android announces as **3.5.0__7** (the latest Android /// signature in the vendored enum). /// * iOS announces as **3.5.6** (latest iOS signature in the /// vendored enum). /// /// All five variants are guaranteed to exist in the generated /// `Version` enum at compile time; if upstream rotates the CSV the /// build will fail loudly here rather than silently fall back. fn pick_client_version() -> Version { Version::Windows_3_X_X__1 } /// Typed configuration for a connection attempt. #[derive(Debug, Clone)] pub struct ConnectConfig { /// Server address: `hostname[:port]` or TSDNS name. pub address: String, /// Nickname to use on the server. pub nickname: String, /// Optional server password. pub password: Option, /// Optional pre-existing identity (base64 string accepted by /// `tsclientlib::Identity::new_from_str`). If `None`, a fresh /// identity is generated and **not persisted** — production callers /// should provide one from secure identity storage. pub identity: Option, /// How long to wait for the initial state snapshot before /// returning `ProtocolError::Timeout`. pub ready_timeout: Duration, /// Optional already-resolved socket address from core's invisible /// prefetch cache. When present, the protocol layer skips address /// resolution but still opens a normal TS3 connection only after /// the user requested Connect. pub resolved_address: Option, } impl Default for ConnectConfig { fn default() -> Self { Self { address: String::new(), nickname: "Chanora".to_string(), password: None, identity: None, ready_timeout: Duration::from_secs(10), resolved_address: None, } } } enum Request { Snapshot(oneshot::Sender>), Disconnect(oneshot::Sender<()>), /// Move self to a channel. Optional channel password. MoveToChannel { channel_id: u64, password: Option, reply: oneshot::Sender>, }, /// Move self to a channel without waiting for the reply. MoveToChannelNoWait { channel_id: u64, password: Option, }, /// Update own client mute state (input and/or output). SetMuted { input: Option, output: Option, reply: oneshot::Sender>, }, /// Send a text message to a target. SendTextMessage { /// Message content. message: String, /// Target scope. target: MessageTarget, /// Reply channel for outcome. reply: oneshot::Sender>, }, /// Fetch richer profile/connection details for an online client. FetchClientProfile { client_id: u64, reply: oneshot::Sender>, }, } /// Why a [`ProtocolClient`] task ended. Distinguishes a user-driven /// disconnect (the supervisor must NOT retry) from a network-driven /// loss (the supervisor should consider retrying). #[derive(Debug, Clone)] pub enum DisconnectReason { /// The caller explicitly called [`ProtocolClient::disconnect`] /// or dropped the handle. UserRequested, /// The underlying tsclientlib event stream ended. StreamEnded, /// A protocol-layer error caused the task to abort. Error(String), } /// Async handle owning a live protocol connection. Drop = disconnect. pub struct ProtocolClient { tx: mpsc::Sender, /// Submit outbound voice packets here. Built by `chanora_audio` /// via [`Self::voice_out`]. voice_out_tx: mpsc::Sender, /// Inbound voice packets land here. Consumed by `chanora_audio`. /// Wrapped in a `Mutex>` so the consumer can take it /// exactly once. voice_in_rx: std::sync::Mutex>>, /// Fires exactly once when the connection task exits, with the /// reason. Used by the supervisor in `chanora_core` to drive /// auto-reconnect. Wrapped in a Mutex> so it can be /// taken once by the supervisor and never resurfaced. lost_rx: std::sync::Mutex>>, /// Inbound chat message stream from the connection task. The /// receiver is taken by the supervisor and forwarded to UI. chat_rx: std::sync::Mutex>>, /// Inbound server-activity stream from the connection task. activity_rx: std::sync::Mutex>>, /// Inbound state-delta stream from the connection task. delta_rx: std::sync::Mutex>>, } /// One inbound voice packet from a remote client. pub struct InboundVoice { /// The remote client this audio came from. pub from_client: u64, /// Raw packet bytes for `AudioHandler::handle_packet`. pub packet: InAudioBuf, } /// A cheap, clone-free probe handle for the watchdog. Owns its own /// clone of the connection task's request channel. #[derive(Clone)] pub struct SnapshotProbe { tx: mpsc::Sender, } impl SnapshotProbe { /// Issue a single snapshot RPC. Returns the same error shape as /// [`ProtocolClient::snapshot`]. Suitable for use under a /// `tokio::time::timeout`. pub async fn probe(&self) -> Result { let (tx, rx) = oneshot::channel(); self.tx .send(Request::Snapshot(tx)) .await .map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?; rx.await .map_err(|_| ProtocolError::Lost("snapshot reply dropped".to_string()))? } } impl ProtocolClient { /// Generate a fresh, persistable TS3 identity string. The /// returned value is the canonical `counter`V`base64key` form /// accepted by [`ConnectConfig::identity`] and by tsclientlib's /// `Identity::new_from_str`. Callers should persist it via /// `chanora_storage` so subsequent connects reuse the same /// identity and the server sees the same client UID. pub fn generate_identity() -> String { let id = Identity::create(); format!("{}V{}", id.counter(), id.key().to_ts()) } /// Dial the server and wait for the initial state snapshot. The /// returned client is ready for [`Self::snapshot`] and /// [`Self::disconnect`] calls. pub async fn connect(cfg: ConnectConfig) -> Result { if cfg.address.trim().is_empty() { return Err(ProtocolError::Invalid("address is empty".to_string())); } if cfg.nickname.trim().is_empty() { return Err(ProtocolError::Invalid("nickname is empty".to_string())); } let (tx, rx) = mpsc::channel::(8); let (voice_out_tx, voice_out_rx) = mpsc::channel::(64); let (voice_in_tx, voice_in_rx) = mpsc::channel::(64); let (chat_tx, chat_rx) = mpsc::channel::(64); let (activity_tx, activity_rx) = mpsc::channel::(128); let (delta_tx, delta_rx) = mpsc::channel::(256); let (ready_tx, ready_rx) = oneshot::channel::>(); let (lost_tx, lost_rx) = oneshot::channel::(); tokio::spawn(connection_task( cfg.clone(), rx, voice_out_rx, EventChannels { voice_in: voice_in_tx, chat: chat_tx, activity: activity_tx, delta: delta_tx, }, ready_tx, lost_tx, )); match tokio::time::timeout(cfg.ready_timeout, ready_rx).await { Ok(Ok(Ok(()))) => Ok(Self { tx, voice_out_tx, voice_in_rx: std::sync::Mutex::new(Some(voice_in_rx)), lost_rx: std::sync::Mutex::new(Some(lost_rx)), chat_rx: std::sync::Mutex::new(Some(chat_rx)), activity_rx: std::sync::Mutex::new(Some(activity_rx)), delta_rx: std::sync::Mutex::new(Some(delta_rx)), }), Ok(Ok(Err(e))) => Err(e), Ok(Err(_)) => Err(ProtocolError::Backend( "connection task exited before signalling ready".to_string(), )), Err(_) => Err(ProtocolError::Timeout), } } /// Read a typed snapshot of the current server state. pub async fn snapshot(&self) -> Result { let (tx, rx) = oneshot::channel(); self.tx .send(Request::Snapshot(tx)) .await .map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?; rx.await .map_err(|_| ProtocolError::Lost("snapshot reply dropped".to_string()))? } /// Fetch richer profile and live connection details for one online client. pub async fn client_profile(&self, client_id: u64) -> Result { let (tx, rx) = oneshot::channel(); self.tx .send(Request::FetchClientProfile { client_id, reply: tx, }) .await .map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?; rx.await .map_err(|_| ProtocolError::Lost("client_profile reply dropped".to_string()))? } /// Disconnect cleanly. Blocks until the task exits. pub async fn disconnect(self) { let (tx, rx) = oneshot::channel(); let request_path = async { if self.tx.send(Request::Disconnect(tx)).await.is_ok() { let _ = rx.await; } }; if tokio::time::timeout(DISCONNECT_REPLY_TIMEOUT, request_path) .await .is_err() { warn!( target: "chanora_protocol", timeout_ms = DISCONNECT_REPLY_TIMEOUT.as_millis() as u64, "disconnect request did not complete before timeout" ); } } /// Move our own client into a channel. `password` is optional /// for password-protected channels. pub async fn move_to_channel( &self, channel_id: u64, password: Option, ) -> Result<(), ProtocolError> { let (tx, rx) = oneshot::channel(); self.tx .send(Request::MoveToChannel { channel_id, password, reply: tx, }) .await .map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?; rx.await .map_err(|_| ProtocolError::Lost("move_to_channel reply dropped".to_string()))? } /// Queue a move command and return once it has been accepted by /// the protocol task. The server-side result is still observed /// asynchronously by the task for logging and reconciliation. pub async fn queue_move_to_channel( &self, channel_id: u64, password: Option, ) -> Result<(), ProtocolError> { self.tx .send(Request::MoveToChannelNoWait { channel_id, password, }) .await .map_err(|_| ProtocolError::Lost("connection task is gone".to_string())) } /// Update mute state on our own client. Pass `Some(_)` for the /// fields you want to change, `None` to leave a field as-is. pub async fn set_muted( &self, input: Option, output: Option, ) -> Result<(), ProtocolError> { let (tx, rx) = oneshot::channel(); self.tx .send(Request::SetMuted { input, output, reply: tx, }) .await .map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?; rx.await .map_err(|_| ProtocolError::Lost("set_muted reply dropped".to_string()))? } /// Sender for outbound voice packets. Clone freely. pub fn voice_out(&self) -> mpsc::Sender { self.voice_out_tx.clone() } /// Clone the request channel so a watchdog can issue probes /// without holding a `&self` reference across the await. The /// returned [`SnapshotProbe`] is `Send + 'static` and dispatches /// a single snapshot RPC against this protocol task. pub fn snapshot_probe(&self) -> SnapshotProbe { SnapshotProbe { tx: self.tx.clone(), } } /// Take the inbound-voice receiver. Returns `None` if it has /// already been taken; only one consumer is allowed. pub fn take_voice_in(&self) -> Option> { self.voice_in_rx.lock().ok().and_then(|mut g| g.take()) } /// Put a previously-taken voice_in receiver back so a /// follow-up `take_voice_in()` succeeds. Used by the core's /// `start_audio` to recover from a failed /// `AudioEngine::start_with_gate` — without this a single /// engine-construction failure would permanently poison the /// voice channel and force a reconnect to fix. pub fn put_voice_in(&self, rx: mpsc::Receiver) { if let Ok(mut g) = self.voice_in_rx.lock() { // If a consumer is already in possession we drop the // duplicate rather than overwriting; this branch // should not be reachable in practice because the only // caller (start_audio) takes-then-puts inside the same // critical section. if g.is_none() { *g = Some(rx); } } } /// Take the loss-notifier. Returns `None` if it has already been /// taken. The supervisor in `chanora_core` consumes this to /// drive auto-reconnect; nothing else should call it. pub fn take_loss_notifier(&self) -> Option> { self.lost_rx.lock().ok().and_then(|mut g| g.take()) } /// Take the inbound-chat receiver. Returns `None` if it has /// already been taken; only one consumer is allowed. pub fn take_chat_rx(&self) -> Option> { self.chat_rx.lock().ok().and_then(|mut g| g.take()) } /// Put a previously-taken chat_rx receiver back. pub fn put_chat_rx(&self, rx: mpsc::Receiver) { if let Ok(mut g) = self.chat_rx.lock() { if g.is_none() { *g = Some(rx); } } } /// Take the inbound server-activity receiver. Returns `None` if it has /// already been taken; only one consumer is allowed. pub fn take_activity_rx(&self) -> Option> { self.activity_rx.lock().ok().and_then(|mut g| g.take()) } /// Put a previously-taken activity receiver back. pub fn put_activity_rx(&self, rx: mpsc::Receiver) { if let Ok(mut g) = self.activity_rx.lock() { if g.is_none() { *g = Some(rx); } } } /// Takes ownership of the delta receiver channel. Returns `None` if already /// taken. Must be called exactly once during initialization to subscribe to /// incremental state changes. pub fn take_delta_rx(&self) -> Option> { self.delta_rx.lock().ok().and_then(|mut g| g.take()) } /// Send a text message to the specified target. pub async fn send_text_message( &self, message: String, target: MessageTarget, ) -> Result<(), ProtocolError> { let (tx, rx) = oneshot::channel(); self.tx .send(Request::SendTextMessage { message, target, reply: tx, }) .await .map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?; rx.await .map_err(|_| ProtocolError::Lost("send_text_message reply dropped".to_string()))? } } async fn connection_task( cfg: ConnectConfig, mut rx: mpsc::Receiver, mut voice_out_rx: mpsc::Receiver, channels: EventChannels, ready_tx: oneshot::Sender>, lost_tx: oneshot::Sender, ) { // Box the lost_tx so each exit branch can move it. let mut lost_tx = Some(lost_tx); // Macro: report the disconnect reason and return from the task. macro_rules! exit { ($reason:expr) => {{ if let Some(tx) = lost_tx.take() { let _ = tx.send($reason); } return; }}; } macro_rules! fail_ready { ($err:expr) => {{ let err = $err; let reason = DisconnectReason::Error(format!("{err}")); let _ = ready_tx.send(Err(err)); exit!(reason); }}; ($err:expr, $reason:expr) => {{ let _ = ready_tx.send(Err($err)); exit!($reason); }}; } let resolved = match server_socket_from_config(&cfg) { Some(addr) => { info!( target: "chanora_protocol", input = %cfg.address, resolved = %addr, "using prefetched server address" ); addr } None => { let addr = match resolve_server_socket(&cfg.address).await { Ok(addr) => addr, Err(err) => fail_ready!(err), }; info!( target: "chanora_protocol", input = %cfg.address, resolved = %addr, "server address resolved" ); addr } }; // Pass the resolved SocketAddr directly to tsclientlib so it // skips its own resolver entirely (tsclientlib accepts // SocketAddr via the From for ServerAddress impl). let client_version = pick_client_version(); info!( target: "chanora_protocol", client_version = %client_version, "selected TS3 client_version for this platform" ); let mut builder = Connection::build(resolved) .name(cfg.nickname.clone()) .version(client_version); let identity = match cfg.identity.as_deref() { Some(value) => match Identity::new_from_str(value) { Ok(identity) => identity, Err(err) => { let msg = err.to_string(); fail_ready!( ProtocolError::Identity(msg.clone()), DisconnectReason::Error(format!("identity: {msg}")) ); } }, None => Identity::create(), }; builder = builder.identity(identity); if let Some(pw) = &cfg.password { builder = builder.password(pw.clone()); } let mut con = match builder.connect() { Ok(c) => c, Err(e) => { let msg = format!("{e}"); fail_ready!( ProtocolError::Connect(msg.clone()), DisconnectReason::Error(format!("connect: {msg}")) ); } }; // Wait for the first BookEvents indicating the state snapshot is ready. let first = con .events() .try_filter(|e| future::ready(matches!(e, StreamItem::BookEvents(_)))) .next() .await; match first { Some(Ok(_)) => { info!(target: "chanora_protocol", "initial state snapshot received"); } Some(Err(e)) => { let msg = format!("{e}"); fail_ready!( ProtocolError::DisconnectedEarly(msg.clone()), DisconnectReason::Error(format!("disconnected early: {msg}")) ); } None => { let msg = "event stream ended before snapshot".to_string(); fail_ready!( ProtocolError::DisconnectedEarly(msg.clone()), DisconnectReason::Error(msg) ); } } // Subscribe to the full server tree so snapshot() returns more than just our channel. if let Ok(state) = con.get_state() { if let Err(e) = state.server.set_subscribed(true).send(&mut con) { warn!(target: "chanora_protocol", error = %e, "could not subscribe to server tree"); } } // Settle: pump events for ~2 s so the subscribed tree arrives // before the first snapshot. The upstream packet codec emits // out-of-order command-packet warnings here; they are harmless // and the final state still converges. let settle_until = std::time::Instant::now() + Duration::from_secs(2); while std::time::Instant::now() < settle_until { let ev = tokio::time::timeout(Duration::from_millis(100), con.events().next()).await; match ev { Ok(Some(Ok(_))) => continue, Ok(Some(Err(e))) => { warn!(target: "chanora_protocol", error = %e, "event error during settle"); } Ok(None) => { let msg = "stream closed during settle".to_string(); fail_ready!( ProtocolError::DisconnectedEarly(msg), DisconnectReason::StreamEnded ); } Err(_) => { /* no event available right now; keep waiting */ } } } let _ = ready_tx.send(Ok(())); // Pending `client_move` requests: each one is keyed by the // `MessageHandle` tsclientlib returns from `send_with_result`. // When the corresponding `StreamItem::MessageResult` arrives we // resolve the oneshot back to the caller. Entries also carry a // deadline so a server that never replies doesn't leak the // reply channel — at most 3 s of pending state per move. let mut pending_moves: PendingMoves = HashMap::new(); let mut voice_activity: HashMap = HashMap::new(); let mut poke_limiter = PokeLimiter::new(); // Main loop: pump events, service requests, forward voice. loop { // 1. Send a bounded batch of outbound voice packets first — they're // time-sensitive, but control requests must still make progress. drain_voice_packets_for_tick(&mut voice_out_rx, OUTBOUND_VOICE_PACKETS_PER_TICK, |pkt| { if let Err(e) = con.send_audio(pkt) { warn!(target: "chanora_protocol", error = %e, "send_audio failed"); } Ok::<(), ()>(()) }); // 2. Advance event stream by at most one event with a small timeout. let pump = async { let mut ev_stream = con.events(); tokio::time::timeout(Duration::from_millis(20), ev_stream.next()).await }; match pump.await { Ok(Some(Ok(item))) => match item { StreamItem::Audio(buf) => { handle_audio_stream_item(&channels.voice_in, &mut voice_activity, buf).await; } other => handle_non_audio_stream_item( &con, other, &channels.chat, &channels.activity, &channels.delta, &mut pending_moves, &mut poke_limiter, ), }, Ok(Some(Err(e))) => { warn!(target: "chanora_protocol", error = %e, "event error"); // Some errors are transient; treat persistent ones // as a loss after the next iteration. } Ok(None) => { warn!(target: "chanora_protocol", "event stream ended"); exit!(DisconnectReason::StreamEnded); } Err(_) => { /* no event in 20 ms */ } } // 2b. Sweep stale pending_moves whose deadline has passed. // The server should always reply within ~1 s; 3 s is a // generous ceiling. Expired entries fall back to Ok() so // the caller's snapshot-confirmation polling still has a // chance to detect success — better than a fake // ServerRejected for legacy servers that never reply to // move requests. if !pending_moves.is_empty() { let now = std::time::Instant::now(); let expired: Vec = pending_moves .iter() .filter_map(|(handle, (_, _, deadline))| { if now >= *deadline { Some(*handle) } else { None } }) .collect(); for handle in expired { if let Some((_target_channel, Some(reply), _)) = pending_moves.remove(&handle) { let _ = reply.send(Ok(())); } } } // 3. Service at most one control request (non-blocking). match rx.try_recv() { Ok(Request::Snapshot(reply)) => { let snap = build_snapshot(&con, &voice_activity); let _ = reply.send(snap); } Ok(Request::MoveToChannel { channel_id, password, reply, }) => match move_self_to(&mut con, channel_id, password.as_deref()) { Ok(handle) => { let deadline = std::time::Instant::now() + Duration::from_secs(3); pending_moves.insert(handle, (channel_id, Some(reply), deadline)); } Err(e) => { let _ = reply.send(Err(e)); } }, Ok(Request::MoveToChannelNoWait { channel_id, password, }) => match move_self_to(&mut con, channel_id, password.as_deref()) { Ok(handle) => { let deadline = std::time::Instant::now() + Duration::from_secs(3); pending_moves.insert(handle, (channel_id, None, deadline)); } Err(e) => { warn!( target: "chanora_protocol", error = %e, channel_id, "fire-and-forget client_move could not be queued" ); } }, Ok(Request::SetMuted { input, output, reply, }) => { let r = set_self_muted(&mut con, input, output); let _ = reply.send(r); } Ok(Request::SendTextMessage { message, target, reply, }) => { let r = send_text_message(&mut con, &message, target); let _ = reply.send(r); } Ok(Request::FetchClientProfile { client_id, reply }) => { let r = fetch_client_profile( &mut con, client_id, &channels, &mut pending_moves, &mut voice_activity, &mut poke_limiter, ) .await; let _ = reply.send(r); } Ok(Request::Disconnect(reply)) => { let _ = con.disconnect(DisconnectOptions::new()); bounded_drain_stream(con.events(), DISCONNECT_EVENT_DRAIN_TIMEOUT).await; let _ = reply.send(()); info!(target: "chanora_protocol", "clean disconnect"); exit!(DisconnectReason::UserRequested); } Err(mpsc::error::TryRecvError::Empty) => {} Err(mpsc::error::TryRecvError::Disconnected) => { let _ = con.disconnect(DisconnectOptions::new()); bounded_drain_stream(con.events(), DISCONNECT_EVENT_DRAIN_TIMEOUT).await; info!(target: "chanora_protocol", "handle dropped; implicit disconnect"); exit!(DisconnectReason::UserRequested); } } } } async fn handle_audio_stream_item( voice_in_tx: &mpsc::Sender, voice_activity: &mut HashMap, buf: InAudioBuf, ) { let from = packet_sender_id(&buf); if let Some(from) = from { voice_activity.insert(from, Instant::now()); let inbound = InboundVoice { from_client: from, packet: buf, }; match send_with_timeout(voice_in_tx, inbound, INBOUND_VOICE_SEND_TIMEOUT).await { Ok(()) => {} Err(SendTimeoutError::Timeout(_)) => { warn!( target: "chanora_protocol", from_client = from, timeout_ms = INBOUND_VOICE_SEND_TIMEOUT.as_millis() as u64, "inbound voice queue stayed full; dropping packet" ); } Err(SendTimeoutError::Closed(_)) => { warn!( target: "chanora_protocol", from_client = from, "inbound voice consumer closed; dropping packet" ); } } } } fn handle_non_audio_stream_item( con: &Connection, item: StreamItem, chat_tx: &mpsc::Sender, activity_tx: &mpsc::Sender, delta_tx: &mpsc::Sender, pending_moves: &mut PendingMoves, poke_limiter: &mut PokeLimiter, ) { match item { StreamItem::BookEvents(events) => { for ev in events { if let tsclientlib::events::Event::PropertyChanged { id: ts_bookkeeping::events::PropertyId::ClientChannel(client_id), .. } = &ev { let own_client = con.get_state().ok().map(|state| state.own_client); if let Ok(state) = con.get_state() { if let Some(client) = state.clients.get(client_id) { let _ = delta_tx.try_send(ProtocolDelta::ClientMoved { client_id: client_id.0 as u64, new_channel_id: client.channel.0, }); } } if own_client == Some(*client_id) { let current_channel = con.get_state().ok().and_then(|state| { state.clients.get(client_id).map(|client| client.channel.0) }); if let Some(current_channel) = current_channel { let matched: Vec = pending_moves .iter() .filter_map(|(handle, (target_channel, _, _))| { if *target_channel == current_channel { Some(*handle) } else { None } }) .collect(); for handle in matched { if let Some((_, reply, _)) = pending_moves.remove(&handle) { info!( target: "chanora_protocol", channel_id = current_channel, "client_move resolved by authoritative self channel change" ); if let Some(reply) = reply { let _ = reply.send(Ok(())); } } } } } } if let Some(activity) = format_server_activity(con, &ev) { let _ = activity_tx.try_send(ServerActivity { message: activity }); } forward_delta(con, &ev, delta_tx); if let tsclientlib::events::Event::Message { target, invoker, message, } = ev { let (mapped, poke_strength) = match target { tsclientlib::MessageTarget::Server => (MessageTarget::Server, None), tsclientlib::MessageTarget::Channel => (MessageTarget::Channel, None), tsclientlib::MessageTarget::Client(id) => { (MessageTarget::Client(id.0 as u64), None) } tsclientlib::MessageTarget::Poke(id) => { let own_client_id = con.get_state().ok().map(|state| state.own_client.0 as u64); let strength = poke_limiter.record(invoker.id.0 as u64, own_client_id); (MessageTarget::Poke(id.0 as u64), Some(strength)) } }; let _ = chat_tx.try_send(ChatMessage { sender_id: ClientId(invoker.id.0 as u64), sender_name: sanitize(&invoker.name), message: sanitize(&message), target: mapped, poke_strength, }); } } } StreamItem::MessageResult(handle, result) => { if let Some((_target_channel, reply, _deadline)) = pending_moves.remove(&handle) { let mapped = match result { Ok(()) => Ok(()), Err(cmd_err) => { let code = cmd_err.error as u32; let message = cmd_err.error.to_string(); info!( target: "chanora_protocol", code, message = %message, "server rejected client_move" ); Err(ProtocolError::ServerRejected { code, message }) } }; if let Some(reply) = reply { let _ = reply.send(mapped); } else if let Err(err) = mapped { info!( target: "chanora_protocol", error = %err, "client_move completed in background with error" ); } } } StreamItem::Audio(_) => unreachable!("audio handled separately"), _ => {} } } async fn resolve_server_socket(address: &str) -> Result { let resolver = ChanoraResolver::new().map_err(|err| ProtocolError::DnsFailed { host: address.to_string(), reason: format!("resolver initialization failed: {err}"), })?; let resolved_address = resolver .resolve_client_address(address) .await .map_err(|err| ProtocolError::DnsFailed { host: address.to_string(), reason: err.to_string(), })?; resolved_address .parse::() .map_err(|err| ProtocolError::DnsFailed { host: address.to_string(), reason: format!("resolver returned invalid socket address '{resolved_address}': {err}"), }) } fn server_socket_from_config(cfg: &ConnectConfig) -> Option { cfg.resolved_address } /// Move our own client into `channel_id` with an optional password. /// Looks up our `own_client` in the current state and dispatches the /// generated `client_move` command via `send_with_result`. The /// returned `MessageHandle` is correlated by the connection loop /// against the next `StreamItem::MessageResult` so we can surface /// typed `ServerRejected` errors (no permission, wrong password, /// channel full, etc.) per the TS3 error catalogue. fn move_self_to( con: &mut Connection, channel_id: u64, password: Option<&str>, ) -> Result { let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; let own_id = state.own_client; let own_client = state .clients .get(&own_id) .ok_or_else(|| ProtocolError::Backend("own_client not in state".to_string()))?; let target = TsChannelId(channel_id); let mut part = own_client.client_move(target); if let Some(pw) = password { part = part.set_password(pw); } let handle = part .send_with_result(con) .map_err(|e| ProtocolError::Backend(format!("client_move send: {e}")))?; info!(target: "chanora_protocol", channel_id, "client_move sent"); Ok(handle) } /// Send a `clientupdate` with the requested mute fields set. `None` /// fields are omitted so callers can toggle just one flag. fn set_self_muted( con: &mut Connection, input: Option, output: Option, ) -> Result<(), ProtocolError> { if input.is_none() && output.is_none() { return Ok(()); } let part = { let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; let mut p = state.client_update(); if let Some(v) = input { p = p.set_input_muted(v); } if let Some(v) = output { p = p.set_output_muted(v); } p }; part.send(con) .map_err(|e| ProtocolError::Backend(format!("client_update send: {e}")))?; info!(target: "chanora_protocol", ?input, ?output, "client_update sent"); Ok(()) } fn send_text_message( con: &mut Connection, message: &str, target: MessageTarget, ) -> Result<(), ProtocolError> { use tsproto_types::TextMessageTargetMode; match target { MessageTarget::Server => { send_text_to_mode(con, message, TextMessageTargetMode::Server, "server")?; } MessageTarget::Channel => { // Fix: previously channel messages were sent via // state.server.send_textmessage() which always uses // TextMessageTargetMode::Server. Now correctly uses // TextMessageTargetMode::Channel so the message is // scoped to the current channel, not server-wide. send_text_to_mode(con, message, TextMessageTargetMode::Channel, "channel")?; } MessageTarget::Client(client_id) => { let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; let client = find_client_by_id(state.clients.values(), client_id)?; client .send_textmessage(message) .send(con) .map_err(|e| ProtocolError::Backend(format!("send_textmessage(client): {e}")))?; } MessageTarget::Poke(client_id) => { let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; let client = find_client_by_id(state.clients.values(), client_id)?; client .poke(message) .send(con) .map_err(|e| ProtocolError::Backend(format!("poke: {e}")))?; } } info!(target: "chanora_protocol", len = message.len(), ?target, "text message sent"); Ok(()) } fn send_text_to_mode( con: &mut Connection, message: &str, target: tsproto_types::TextMessageTargetMode, label: &str, ) -> Result<(), ProtocolError> { use ts_bookkeeping::messages::c2s; c2s::OutSendTextMessageMessage::new(&mut std::iter::once(c2s::OutSendTextMessagePart { target, target_client_id: None, message: message.into(), })) .send(con) .map_err(|e| ProtocolError::Backend(format!("send_textmessage({label}): {e}"))) } async fn fetch_client_profile( con: &mut Connection, client_id: u64, channels: &EventChannels, pending_moves: &mut PendingMoves, voice_activity: &mut HashMap, poke_limiter: &mut PokeLimiter, ) -> Result { let target_id = TsClientId(client_id as u16); let ( database_id, uid_b64, has_optional, has_connection, is_own, needs_server_groups, needs_channel_groups, ) = { let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; let client = state .clients .get(&target_id) .ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?; ( client.database_id, client.uid.as_ref().map(|uid| uid_to_b64(uid.as_ref())), client.optional_data.is_some(), client.connection_data.is_some(), state.own_client == target_id, state.server_groups.is_empty(), state.channel_groups.is_empty(), ) }; let refresh_plan = client_profile_refresh_plan( is_own, has_optional, has_connection, needs_server_groups, needs_channel_groups, ); if refresh_plan.needs_server_groups { let _ = request_messages( con, build_command("servergrouplist", &[], &[]), channels, pending_moves, voice_activity, poke_limiter, ) .await; } if refresh_plan.needs_channel_groups { let _ = request_messages( con, build_command("channelgrouplist", &[], &[]), channels, pending_moves, voice_activity, poke_limiter, ) .await; } if refresh_plan.needs_client_variables { if let Err(e) = request_messages( con, build_command( "clientgetvariables", &[("clid", client_id.to_string())], &[], ), channels, pending_moves, voice_activity, poke_limiter, ) .await { warn!( target: "chanora_protocol", client_id, error = %e, "clientgetvariables failed" ); } } if refresh_plan.needs_connection_info { if let Err(e) = request_messages( con, build_command("getconnectioninfo", &[("clid", client_id.to_string())], &[]), channels, pending_moves, voice_activity, poke_limiter, ) .await { warn!( target: "chanora_protocol", client_id, error = %e, "getconnectioninfo failed (ping/packet_loss will be unavailable)" ); } } let db_info = if refresh_plan.needs_client_db_info { request_client_db_info( con, database_id, channels, pending_moves, voice_activity, poke_limiter, ) .await .ok() } else { None }; let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; let client = state .clients .get(&target_id) .ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?; let optional = client.optional_data.as_ref(); let connection = client.connection_data.as_ref(); let net_stats: Option<&ConnectionStats> = if is_own { con.get_network_stats().ok() } else { None }; let server_group_names: HashMap<_, _> = state .server_groups .iter() .map(|(id, group)| (*id, group.name.clone())) .collect(); let channel_group_names: HashMap<_, _> = state .channel_groups .iter() .map(|(id, group)| (*id, group.name.clone())) .collect(); let unique_id = db_info .as_ref() .map(|info| uid_to_b64(info.uid.as_ref())) .or(uid_b64) .unwrap_or_default(); let avatar_path = if client.avatar_hash.is_empty() || unique_id.is_empty() { String::new() } else { format!("/avatar_{}", uid_to_avatar_path(&unique_id)) }; Ok(ClientProfile { id: ClientId(client.id.0 as u64), channel: ChannelId(client.channel.0), name: db_info .as_ref() .map(|info| sanitize(&info.name)) .unwrap_or_else(|| sanitize(&client.name)), unique_id, database_id: Some(client.database_id.0), country_code: client.country_code.clone(), description: db_info .as_ref() .map(|info| sanitize(&info.description)) .unwrap_or_else(|| sanitize(&client.description)), version: optional .map(|info| info.version.clone()) .unwrap_or_default(), platform: optional .map(|info| info.platform.clone()) .unwrap_or_default(), created_unix_seconds: optional .map(|info| info.created.unix_timestamp()) .or_else(|| db_info.as_ref().map(|info| info.created.unix_timestamp())), last_connected_unix_seconds: optional .map(|info| info.last_connected.unix_timestamp()) .or_else(|| { db_info .as_ref() .map(|info| info.last_connected.unix_timestamp()) }), connections_total: optional .map(|info| u64::from(info.connections_total)) .or_else(|| { db_info .as_ref() .map(|info| u64::from(info.connections_total)) }), online_seconds: connection .and_then(|info| info.connected_time.map(|duration| duration.whole_seconds())), idle_milliseconds: connection.map(|info| duration_millis(info.idle_time)), ping_milliseconds: net_stats .and_then(|s| std_duration_millis(s.rtt)) .or_else(|| connection.and_then(|info| info.ping.map(duration_millis))), ping_deviation_milliseconds: net_stats .and_then(|s| std_duration_millis(s.rtt_dev)) .or_else(|| connection.and_then(|info| info.ping_deviation.map(duration_millis))), client_address: connection .and_then(|info| info.client_address.map(|address| address.to_string())) .unwrap_or_default(), server_groups: format_server_groups(client.server_groups.iter(), &server_group_names), channel_group: channel_group_names .get(&client.channel_group) .cloned() .unwrap_or_else(|| format!("Unknown ({})", client.channel_group.0)), avatar_path, bytes_downloaded_month: optional .map(|info| info.bytes_downloaded_month) .or_else(|| db_info.as_ref().map(|info| info.bytes_downloaded_month)), bytes_uploaded_month: optional .map(|info| info.bytes_uploaded_month) .or_else(|| db_info.as_ref().map(|info| info.bytes_uploaded_month)), bytes_downloaded_total: optional .map(|info| info.bytes_downloaded_total) .or_else(|| db_info.as_ref().map(|info| info.bytes_downloaded_total)), bytes_uploaded_total: optional .map(|info| info.bytes_uploaded_total) .or_else(|| db_info.as_ref().map(|info| info.bytes_uploaded_total)), packet_loss_client_to_server_total: net_stats .map(|s| s.get_packetloss()) .or_else(|| connection.map(|info| info.client_to_server_packetloss_total)), packet_loss_server_to_client_total: net_stats .map(|s| s.get_packetloss_s2c_total()) .or_else(|| connection.and_then(|info| info.server_to_client_packetloss_total)), }) } fn build_command(name: &str, args: &[(&str, String)], flags: &[&str]) -> OutCommand { let mut command = OutCommand::new(Direction::C2S, Flags::empty(), PacketType::Command, name); for flag in flags { command.write_arg(flag, &""); } for (key, value) in args { command.write_arg(key, value); } command } #[derive(Debug, Clone, Copy, PartialEq, Eq)] struct ClientProfileRefreshPlan { needs_server_groups: bool, needs_channel_groups: bool, needs_client_variables: bool, needs_connection_info: bool, needs_client_db_info: bool, } fn client_profile_refresh_plan( is_own: bool, has_optional: bool, has_connection: bool, needs_server_groups: bool, needs_channel_groups: bool, ) -> ClientProfileRefreshPlan { ClientProfileRefreshPlan { needs_server_groups, needs_channel_groups, needs_client_variables: !has_optional || !is_own, needs_connection_info: !has_connection || !is_own, needs_client_db_info: !has_optional || !is_own, } } async fn request_messages( con: &mut Connection, command: OutCommand, channels: &EventChannels, pending_moves: &mut PendingMoves, voice_activity: &mut HashMap, poke_limiter: &mut PokeLimiter, ) -> Result, ProtocolError> { let handle = command .send_with_result(con) .map_err(|e| ProtocolError::Backend(format!("send command: {e}")))?; let mut messages = Vec::new(); let deadline = Instant::now() + PROFILE_REFRESH_RESULT_TIMEOUT; loop { let remaining = deadline.saturating_duration_since(Instant::now()); if remaining.is_zero() { return Err(ProtocolError::Backend( "profile refresh command timed out".to_string(), )); } let item = { let mut stream = con.events(); tokio::time::timeout(remaining, stream.next()) .await .map_err(|_| { ProtocolError::Backend("profile refresh command timed out".to_string()) })? .ok_or_else(|| ProtocolError::Lost("event stream ended".to_string()))? .map_err(|e| ProtocolError::Lost(e.to_string()))? }; match item { StreamItem::MessageEvent(message) => messages.push(message), StreamItem::MessageResult(reply, status) if reply == handle => { status.map_err(|error| ProtocolError::ServerRejected { code: error.error as u32, message: error.error.to_string(), })?; return Ok(messages); } StreamItem::Audio(buf) => { handle_audio_stream_item(&channels.voice_in, voice_activity, buf).await; } other => handle_non_audio_stream_item( con, other, &channels.chat, &channels.activity, &channels.delta, pending_moves, poke_limiter, ), } } } async fn request_client_db_info( con: &mut Connection, dbid: tsclientlib::ClientDbId, channels: &EventChannels, pending_moves: &mut PendingMoves, voice_activity: &mut HashMap, poke_limiter: &mut PokeLimiter, ) -> Result { let messages = request_messages( con, build_command("clientdbinfo", &[("cldbid", dbid.0.to_string())], &[]), channels, pending_moves, voice_activity, poke_limiter, ) .await?; for message in messages { if let InMessage::ClientDbInfo(info) = message { if let Some(row) = info.iter().next() { return Ok(row.clone()); } } } Err(ProtocolError::Backend( "clientdbinfo returned no row".to_string(), )) } fn format_server_groups<'a>( groups: impl IntoIterator, names: &HashMap, ) -> Vec { let mut rendered = groups .into_iter() .map(|group| { names .get(group) .cloned() .unwrap_or_else(|| format!("Unknown ({})", group.0)) }) .collect::>(); rendered.sort(); rendered } fn duration_millis(duration: time::Duration) -> i64 { duration .whole_milliseconds() .clamp(i64::MIN as i128, i64::MAX as i128) as i64 } fn std_duration_millis(duration: std::time::Duration) -> Option { duration.as_millis().try_into().ok() } fn uid_to_b64(uid: &tsclientlib::Uid) -> String { BASE64_STANDARD.encode(&uid.0) } fn uid_to_avatar_path(uid_b64: &str) -> String { let decoded = BASE64_STANDARD.decode(uid_b64).unwrap_or_default(); let mut rendered = String::with_capacity(decoded.len() * 2); for byte in decoded { rendered.push((b'a' + (byte >> 4)) as char); rendered.push((b'a' + (byte & 0x0f)) as char); } rendered } fn find_client_by_id<'a>( clients: impl IntoIterator, client_id: u64, ) -> Result<&'a Client, ProtocolError> { clients .into_iter() .find(|client| client.id.0 as u64 == client_id) .ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found"))) } /// Extract the originating `client_id` from an inbound voice packet. fn packet_sender_id(buf: &InAudioBuf) -> Option { use tsproto_packets::packets::AudioData; match buf.data().data() { AudioData::S2C { from, .. } => Some(*from as u64), AudioData::S2CWhisper { from, .. } => Some(*from as u64), _ => None, } } /// Sort TeamSpeak channels by the linked-list ordering carried in /// each channel's `order` field (predecessor pointer), producing a /// root-first depth-first list suitable for direct UI rendering. /// /// Generic over `T` + an `extract` closure so unit tests can supply /// a lightweight fixture struct without constructing a live /// `tsclientlib::data::Channel`. fn sort_channels_tree<'a>(channels: &'a [&'a Channel]) -> Vec<&'a Channel> { sort_channels_tree_by(channels, |c| (c.id.0, c.parent.0, c.order.0)) } /// Inner implementation that operates on any slice of items via an /// extractor returning `(id, parent, order)` u64 triples. Pure; /// tested directly by the `#[cfg(test)]` block below without /// touching `tsclientlib`. fn sort_channels_tree_by<'a, T>( items: &'a [&'a T], extract: impl Fn(&T) -> (u64, u64, u64), ) -> Vec<&'a T> { // Bucket by parent and build (per parent) the predecessor->item // lookup so we can walk the chain in O(n). let mut by_parent: HashMap> = HashMap::new(); for &item in items { let (_id, parent, _order) = extract(item); by_parent.entry(parent).or_default().push(item); } let mut ordered_per_parent: HashMap> = HashMap::new(); for (parent, siblings) in by_parent.into_iter() { let mut successor: HashMap = HashMap::with_capacity(siblings.len()); for &c in &siblings { let (_id, _parent, order) = extract(c); // First-write wins: if the server emits two channels // with the same predecessor (corrupted state), keep // the first and fall through the leftover path for // the duplicates. successor.entry(order).or_insert(c); } let mut ordered: Vec<&'a T> = Vec::with_capacity(siblings.len()); let mut cursor: u64 = 0; let mut visited: std::collections::HashSet = std::collections::HashSet::new(); while let Some(next) = successor.get(&cursor).copied() { let (id, _parent, _order) = extract(next); if !visited.insert(id) { // Cycle guard. Should not happen on a well-formed // server snapshot but cheap to defend against. break; } ordered.push(next); cursor = id; } // Anything we did not reach (broken predecessor pointer or // duplicate predecessor) gets appended sorted by id so the // UI does not silently drop channels. let reached: std::collections::HashSet = ordered.iter().map(|c| extract(c).0).collect(); let mut leftover: Vec<&'a T> = siblings .into_iter() .filter(|c| !reached.contains(&extract(c).0)) .collect(); leftover.sort_by_key(|c| extract(c).0); ordered.extend(leftover); ordered_per_parent.insert(parent, ordered); } // Emit root list first then each subtree depth-first. let mut out: Vec<&'a T> = Vec::with_capacity(items.len()); emit_subtree(&ordered_per_parent, 0, &mut out, &extract); // Defensive: if a channel's `parent` does not appear anywhere // in the emitted tree (orphaned subtree) append it so it isn't // lost. We track emitted ids and dump anything else. let emitted: std::collections::HashSet = out.iter().map(|c| extract(c).0).collect(); let mut orphans: Vec<&'a T> = items .iter() .copied() .filter(|c| !emitted.contains(&extract(c).0)) .collect(); orphans.sort_by_key(|c| extract(c).0); out.extend(orphans); out } fn emit_subtree<'a, T>( by_parent: &HashMap>, root_id: u64, out: &mut Vec<&'a T>, extract: &impl Fn(&T) -> (u64, u64, u64), ) { let Some(children) = by_parent.get(&root_id) else { return; }; for &ch in children { out.push(ch); let (child_id, _parent, _order) = extract(ch); emit_subtree(by_parent, child_id, out, extract); } } fn build_snapshot( con: &Connection, voice_activity: &HashMap, ) -> Result { let state: &data::Connection = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; // TeamSpeak channel ordering: the `order` field on a channel is // NOT a numeric rank but the id of the channel that should // appear immediately before this one within the same parent. // `order == ChannelId(0)` marks the head of a parent's child // list. The previous implementation sorted by `order.0` // numerically, which produced a stable-but-arbitrary order // that did not match the TS3 client display order and was // reported by users as "channel sort in not correct". // // Correct algorithm: // 1. Bucket channels by parent. // 2. Within each bucket, walk the linked list starting from // the entry whose `order == ChannelId(0)` and following // each successive channel via its successor map until the // chain terminates. // 3. Emit channels root-first depth-first, so callers see a // pre-ordered tree without needing to re-sort. // // Defensive fallback: any siblings the linked-list walk // cannot reach (e.g. the server sent a cycle or a dangling // predecessor) are appended at the end of the bucket sorted // by id so the UI doesn't lose channels. let all_channels: Vec<&Channel> = state.channels.values().collect(); let channels: Vec<&Channel> = sort_channels_tree(&all_channels); let clients: Vec<&Client> = state.clients.values().collect(); let channels_dto: Vec = channels .iter() .map(|c| ChannelInfo { id: ChannelId(c.id.0), parent: ChannelId(c.parent.0), name: sanitize(&c.name), order: c.order.0 as i64, has_password: c.has_password.unwrap_or(false), needed_talk_power: c.needed_talk_power, }) .collect(); let clients_dto: Vec = clients .iter() .map(|c| ClientInfo { id: ClientId(c.id.0 as u64), channel: ChannelId(c.channel.0), name: sanitize(&c.name), input_muted: c.input_muted, output_muted: c.output_muted || c.output_only_muted, is_speaking: voice_activity .get(&(c.id.0 as u64)) .is_some_and(|last| last.elapsed() <= SPEAKING_ACTIVITY_WINDOW), is_server_query: is_server_query_client_type(&c.client_type), talk_power: c.talk_power, talk_power_granted: c.talk_power_granted, }) .collect(); Ok(ServerSnapshot { server_name: sanitize(&state.server.name), welcome_message: sanitize(&state.server.welcome_message), platform: sanitize(&state.server.platform), version: sanitize(&state.server.version), channels: channels_dto, clients: clients_dto, own_client_id: state.own_client.0 as u64, }) } fn is_server_query_client_type(client_type: &ClientType) -> bool { matches!(client_type, ClientType::Query { .. }) } /// Light sanitisation of strings before they cross the protocol /// boundary. The redaction policy proper lives in /// `chanora_diagnostics`; this filter only strips control characters /// that would break terminal output or Flutter rendering. fn sanitize(s: &str) -> String { s.chars() .filter(|c| !c.is_control() || *c == '\t' || *c == '\n') .collect() } fn quoted(s: &str) -> String { format!("\"{}\"", sanitize(s)) } fn activity_channel_name(con: &Connection, id: tsclientlib::ChannelId) -> Option { con.get_state() .ok() .and_then(|state| state.channels.get(&id)) .map(|channel| quoted(&channel.name)) } fn activity_client_name(con: &Connection, id: tsclientlib::ClientId) -> Option { con.get_state() .ok() .and_then(|state| state.clients.get(&id)) .filter(|client| !is_server_query_client_type(&client.client_type)) .map(|client| quoted(&client.name)) } fn activity_client(con: &Connection, id: tsclientlib::ClientId) -> Option { con.get_state() .ok() .and_then(|state| state.clients.get(&id)) .filter(|client| !is_server_query_client_type(&client.client_type)) .cloned() } fn activity_channel_group_name( con: &Connection, id: tsclientlib::ChannelGroupId, ) -> Option { con.get_state() .ok() .and_then(|state| state.channel_groups.get(&id)) .map(|group| quoted(&group.name)) } fn activity_server_group_name(con: &Connection, id: tsclientlib::ServerGroupId) -> Option { con.get_state() .ok() .and_then(|state| state.server_groups.get(&id)) .map(|group| quoted(&group.name)) } fn activity_invoker_name(invoker: Option<&tsclientlib::Invoker>) -> String { invoker .map(|invoker| quoted(&invoker.name)) .unwrap_or_else(|| "\"Server\"".to_string()) } fn format_server_activity(con: &Connection, ev: &tsclientlib::events::Event) -> Option { use ts_bookkeeping::events::{Event, PropertyId, PropertyValue}; use tsproto_types::Reason; match ev { Event::PropertyAdded { id: PropertyId::Client(client_id), extra, .. } => { extra.reason?; let client = activity_client(con, *client_id)?; let channel = activity_channel_name(con, client.channel)?; Some(format!( "{} connected to channel {}", quoted(&client.name), channel )) } Event::PropertyRemoved { id: PropertyId::Client(_), old, extra, .. } => { let PropertyValue::Client(client) = old else { return None; }; if is_server_query_client_type(&client.client_type) { return None; } match extra.reason { Some(Reason::Clientdisconnect) => { Some(format!("{} disconnected (Leaving)", quoted(&client.name))) } Some(Reason::ClientdisconnectServerShutdown) | Some(Reason::Serverstop) => Some( format!("{} disconnected (server shutdown)", quoted(&client.name)), ), _ => Some(format!( "{} dropped (connection lost)", quoted(&client.name) )), } } Event::PropertyChanged { id: PropertyId::ClientChannel(client_id), old, invoker, .. } => { let PropertyValue::ChannelId(from_channel_id) = old else { return None; }; let client = activity_client(con, *client_id)?; let from = activity_channel_name(con, *from_channel_id)?; let to = activity_channel_name(con, client.channel)?; if invoker.as_ref().map(|invoker| invoker.id) == Some(*client_id) || invoker.is_none() { Some(format!( "{} switched from channel {} to {}", quoted(&client.name), from, to )) } else { Some(format!( "{} was moved from channel {} to {} by {}", quoted(&client.name), from, to, activity_invoker_name(invoker.as_ref()) )) } } Event::PropertyChanged { id: PropertyId::ClientChannelGroup(client_id), invoker, .. } => { let client = activity_client(con, *client_id)?; let group = activity_channel_group_name(con, client.channel_group)?; Some(format!( "Channel group {} was assigned to {} by {}.", group, quoted(&client.name), activity_invoker_name(invoker.as_ref()) )) } Event::PropertyAdded { id: PropertyId::ClientServerGroup(client_id, group_id), invoker, .. } => { let client = activity_client_name(con, *client_id)?; let group = activity_server_group_name(con, *group_id)?; Some(format!( "Server group {} was assigned to {} by {}.", group, client, activity_invoker_name(invoker.as_ref()) )) } Event::PropertyRemoved { id: PropertyId::ClientServerGroup(client_id, group_id), invoker, .. } => { let client = activity_client_name(con, *client_id)?; let group = activity_server_group_name(con, *group_id)?; Some(format!( "Server group {} was removed from {} by {}.", group, client, activity_invoker_name(invoker.as_ref()) )) } _ => None, } } const _: () = { // Compile-time assertion that ChannelId(0) maps to what tsclientlib // also considers the root. let _ = TsChannelId(0); }; #[cfg(test)] mod tests { use super::{ bounded_drain_stream, client_profile_refresh_plan, drain_voice_packets_for_tick, is_server_query_client_type, send_with_timeout, server_socket_from_config, sort_channels_tree_by, std_duration_millis, ConnectConfig, ProtocolClient, Request, SendTimeoutError, DISCONNECT_REPLY_TIMEOUT, }; use futures::stream; use std::time::Duration; use tokio::sync::mpsc; use tsproto_types::ClientType; /// Lightweight fixture mirroring just the (id, parent, order) /// triple that the linked-list sort needs. Avoids constructing /// a real `tsclientlib::data::Channel` (which requires a live /// connection) in unit tests. #[derive(Debug, PartialEq)] struct FakeChannel { id: u64, parent: u64, order: u64, } fn extract(c: &FakeChannel) -> (u64, u64, u64) { (c.id, c.parent, c.order) } #[test] fn query_client_type_maps_to_server_query_flag() { assert!(!is_server_query_client_type(&ClientType::Normal)); assert!(is_server_query_client_type(&ClientType::Query { admin: false })); assert!(is_server_query_client_type(&ClientType::Query { admin: true })); } #[test] fn connect_config_can_carry_prefetched_socket_address() { let addr: std::net::SocketAddr = "127.0.0.1:9987".parse().unwrap(); let cfg = ConnectConfig { address: "example.com".to_string(), nickname: "Tester".to_string(), password: None, identity: None, ready_timeout: Duration::from_secs(1), resolved_address: Some(addr), }; assert_eq!(cfg.resolved_address, Some(addr)); } #[test] fn server_socket_from_config_prefers_prefetched_address() { let addr: std::net::SocketAddr = "127.0.0.1:9987".parse().unwrap(); let cfg = ConnectConfig { address: "example.com".to_string(), resolved_address: Some(addr), nickname: "Tester".to_string(), password: None, identity: None, ready_timeout: Duration::from_secs(1), }; assert_eq!(server_socket_from_config(&cfg), Some(addr)); } #[test] fn std_duration_millis_returns_none_when_duration_exceeds_i64() { let overflow = Duration::from_millis(i64::MAX as u64) + Duration::from_millis(1); assert_eq!(std_duration_millis(overflow), None); } #[test] fn non_self_profile_refresh_requests_variables_connection_and_db_info() { let plan = client_profile_refresh_plan(false, true, true, false, false); assert!(plan.needs_client_variables); assert!(plan.needs_connection_info); assert!(plan.needs_client_db_info); } #[test] fn own_profile_refresh_skips_complete_optional_and_connection_requests() { let plan = client_profile_refresh_plan(true, true, true, false, false); assert!(!plan.needs_server_groups); assert!(!plan.needs_channel_groups); assert!(!plan.needs_client_variables); assert!(!plan.needs_connection_info); assert!(!plan.needs_client_db_info); } #[test] fn own_profile_refresh_requests_missing_group_lists() { let plan = client_profile_refresh_plan(true, true, true, true, true); assert!(plan.needs_server_groups); assert!(plan.needs_channel_groups); } #[test] fn channel_sort_linked_list_under_one_parent() { // Server emits four root-level channels in arbitrary HashMap // iteration order. order=0 -> first; order=X means "comes // after the channel with id=X". Expected emitted order is // the linked-list walk: a -> b -> c -> d. let a = FakeChannel { id: 100, parent: 0, order: 0, }; let b = FakeChannel { id: 200, parent: 0, order: 100, }; let c = FakeChannel { id: 300, parent: 0, order: 200, }; let d = FakeChannel { id: 400, parent: 0, order: 300, }; // Deliberately shuffled inputs. let inputs: Vec<&FakeChannel> = vec![&c, &a, &d, &b]; let sorted = sort_channels_tree_by(&inputs, extract); let ids: Vec = sorted.iter().map(|c| c.id).collect(); assert_eq!(ids, vec![100, 200, 300, 400]); } #[test] fn channel_sort_disconnected_predecessor_falls_back_by_id() { // a is the head. b correctly chains. c claims predecessor // = 999 which does not exist among the siblings. c must // not be dropped — it falls back to the leftover bucket // appended sorted by id at the end. let a = FakeChannel { id: 100, parent: 0, order: 0, }; let b = FakeChannel { id: 200, parent: 0, order: 100, }; let c = FakeChannel { id: 300, parent: 0, order: 999, }; let inputs: Vec<&FakeChannel> = vec![&c, &a, &b]; let sorted = sort_channels_tree_by(&inputs, extract); let ids: Vec = sorted.iter().map(|c| c.id).collect(); assert_eq!(ids, vec![100, 200, 300]); } #[test] fn channel_sort_emits_subtree_depth_first() { // Tree: // root (id=0, implicit) // ├── a (id=10, order=0) // │ ├── a1 (id=11, parent=10, order=0) // │ └── a2 (id=12, parent=10, order=11) // └── b (id=20, order=10) // Expected emission: a, a1, a2, b let a = FakeChannel { id: 10, parent: 0, order: 0, }; let a1 = FakeChannel { id: 11, parent: 10, order: 0, }; let a2 = FakeChannel { id: 12, parent: 10, order: 11, }; let b = FakeChannel { id: 20, parent: 0, order: 10, }; let inputs: Vec<&FakeChannel> = vec![&b, &a2, &a, &a1]; let sorted = sort_channels_tree_by(&inputs, extract); let ids: Vec = sorted.iter().map(|c| c.id).collect(); assert_eq!(ids, vec![10, 11, 12, 20]); } #[test] fn channel_sort_does_not_loop_on_cycle() { // a says "comes after b"; b says "comes after a". The // walk must terminate (cycle guard) and both channels // must still appear in the output via the leftover path. let a = FakeChannel { id: 1, parent: 0, order: 2, }; let b = FakeChannel { id: 2, parent: 0, order: 1, }; let inputs: Vec<&FakeChannel> = vec![&a, &b]; let sorted = sort_channels_tree_by(&inputs, extract); // Both reachable in some deterministic order (id-sorted // in the leftover bucket since cursor=0 finds nothing). let ids: Vec = sorted.iter().map(|c| c.id).collect(); assert_eq!(ids.len(), 2); assert!(ids.contains(&1)); assert!(ids.contains(&2)); } #[tokio::test] async fn send_with_timeout_waits_for_capacity_instead_of_dropping() { let (tx, mut rx) = mpsc::channel(1); tx.send(1_u8).await.expect("seed first item"); let drain = tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(10)).await; assert_eq!(rx.recv().await, Some(1)); rx.recv().await }); send_with_timeout(&tx, 2_u8, Duration::from_millis(100)) .await .expect("second item should enqueue once capacity frees"); assert_eq!(drain.await.expect("drain task should succeed"), Some(2)); } #[tokio::test] async fn send_with_timeout_times_out_when_capacity_stays_full() { let (tx, _rx) = mpsc::channel(1); tx.send(1_u8).await.expect("seed first item"); let result = send_with_timeout(&tx, 2_u8, Duration::from_millis(10)).await; assert_eq!(result, Err(SendTimeoutError::Timeout(2))); } #[tokio::test] async fn disconnect_request_send_is_bounded_when_request_channel_is_full() { let (tx, _rx) = mpsc::channel(1); let (reply_tx, _reply_rx) = tokio::sync::oneshot::channel(); tx.send(Request::Snapshot(reply_tx)) .await .expect("seed first request"); let (disconnect_tx, _disconnect_rx) = tokio::sync::oneshot::channel(); let result = send_with_timeout( &tx, Request::Disconnect(disconnect_tx), Duration::from_millis(10), ) .await; assert!(matches!(result, Err(SendTimeoutError::Timeout(_)))); } #[tokio::test] async fn protocol_client_disconnect_returns_when_request_channel_is_full() { let (tx, _rx) = mpsc::channel(1); let (snapshot_tx, _snapshot_rx) = tokio::sync::oneshot::channel(); tx.send(Request::Snapshot(snapshot_tx)) .await .expect("seed first request"); let (voice_out_tx, _voice_out_rx) = mpsc::channel(1); let (_voice_in_tx, voice_in_rx) = mpsc::channel(1); let (_lost_tx, lost_rx) = tokio::sync::oneshot::channel(); let (_chat_tx, chat_rx) = mpsc::channel(1); let (_activity_tx, activity_rx) = mpsc::channel(1); let (_delta_tx, delta_rx) = mpsc::channel(1); let client = ProtocolClient { tx, voice_out_tx, voice_in_rx: std::sync::Mutex::new(Some(voice_in_rx)), lost_rx: std::sync::Mutex::new(Some(lost_rx)), chat_rx: std::sync::Mutex::new(Some(chat_rx)), activity_rx: std::sync::Mutex::new(Some(activity_rx)), delta_rx: std::sync::Mutex::new(Some(delta_rx)), }; tokio::time::timeout( DISCONNECT_REPLY_TIMEOUT + Duration::from_millis(100), client.disconnect(), ) .await .expect("disconnect should not wait indefinitely for request channel capacity"); } #[tokio::test] async fn voice_drain_stops_at_per_tick_budget() { let (tx, mut rx) = mpsc::channel(8); for value in 0_u8..5 { tx.send(value).await.expect("seed voice packet"); } let mut sent = Vec::new(); let drained = drain_voice_packets_for_tick(&mut rx, 2, |value| { sent.push(value); Ok::<(), ()>(()) }); assert_eq!(drained, 2); assert_eq!(sent, vec![0, 1]); assert_eq!(rx.len(), 3); } #[tokio::test] async fn disconnect_stream_drain_returns_after_timeout() { let start = tokio::time::Instant::now(); bounded_drain_stream(stream::pending::<()>(), Duration::from_millis(10)).await; assert!(start.elapsed() < Duration::from_millis(100)); } } fn forward_delta( con: &Connection, ev: &tsclientlib::events::Event, delta_tx: &mpsc::Sender, ) { use ts_bookkeeping::events::{Event, PropertyId, PropertyValue}; match ev { Event::PropertyAdded { id: PropertyId::Client(client_id), .. } => { if let Ok(state) = con.get_state() { if let Some(client) = state.clients.get(client_id) { let _ = delta_tx.try_send(ProtocolDelta::ClientJoined { client_id: client_id.0 as u64, channel_id: client.channel.0, name: client.name.clone(), input_muted: client.input_muted, output_muted: client.output_muted || client.output_only_muted, is_server_query: is_server_query_client_type(&client.client_type), talk_power: client.talk_power, talk_power_granted: client.talk_power_granted, }); } } } Event::PropertyRemoved { id: PropertyId::Client(_), old: PropertyValue::Client(client), .. } => { let _ = delta_tx.try_send(ProtocolDelta::ClientLeft { client_id: client.id.0 as u64, name: client.name.clone(), }); } Event::PropertyChanged { id: PropertyId::ClientChannel(_), .. } => {} Event::PropertyChanged { id: PropertyId::Client(client_id), .. } => { if let Ok(state) = con.get_state() { if let Some(client) = state.clients.get(client_id) { let _ = delta_tx.try_send(ProtocolDelta::ClientUpdated { client_id: client_id.0 as u64, input_muted: client.input_muted, output_muted: client.output_muted || client.output_only_muted, is_server_query: is_server_query_client_type(&client.client_type), talk_power: client.talk_power, talk_power_granted: client.talk_power_granted, }); } } } Event::PropertyAdded { id: PropertyId::Channel(channel_id), .. } => { if let Ok(state) = con.get_state() { if let Some(channel) = state.channels.get(channel_id) { let _ = delta_tx.try_send(ProtocolDelta::ChannelAdded { id: channel_id.0, parent: channel.parent.0, name: channel.name.clone(), order: channel.order.0 as i64, has_password: channel.has_password.unwrap_or(false), needed_talk_power: channel.needed_talk_power, }); } } } Event::PropertyRemoved { id: PropertyId::Channel(_), old: PropertyValue::Channel(channel), .. } => { let _ = delta_tx.try_send(ProtocolDelta::ChannelRemoved { id: channel.id.0 }); } Event::PropertyChanged { id: PropertyId::Channel(channel_id), .. } => { if let Ok(state) = con.get_state() { if let Some(channel) = state.channels.get(channel_id) { let _ = delta_tx.try_send(ProtocolDelta::ChannelUpdated { id: channel_id.0, name: channel.name.clone(), has_password: channel.has_password.unwrap_or(false), needed_talk_power: channel.needed_talk_power, }); } } } _ => {} } }