//! 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 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::prelude::*; use tsclientlib::{ ChannelId as TsChannelId, Connection, DisconnectOptions, Identity, MessageHandle, OutCommandExt, StreamItem, Version, }; use tsproto_packets::packets::{InAudioBuf, OutPacket}; use tsproto_types::ClientType; use crate::dto::{ ChannelId, ChannelInfo, ChatMessage, ClientId, ClientInfo, MessageTarget, ServerActivity, ServerSnapshot, }; use crate::ProtocolError; const SPEAKING_ACTIVITY_WINDOW: Duration = Duration::from_millis(750); /// 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 { #[cfg(target_os = "windows")] { Version::Windows_5_0_0_beta51 } #[cfg(target_os = "macos")] { Version::macOS_5_0_0_beta51 } #[cfg(target_os = "ios")] { Version::iOS_3_5_6 } #[cfg(target_os = "android")] { Version::Android_3_5_0__7 } #[cfg(target_os = "linux")] { Version::Linux_5_0_0_beta51 } #[cfg(not(any( target_os = "windows", target_os = "macos", target_os = "ios", target_os = "android", target_os = "linux" )))] { // Last-ditch fallback for unanticipated targets (BSDs, // Solaris-likes). Linux signature is the closest analogue. Version::Linux_5_0_0_beta51 } } /// 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, } 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), } } } 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>, }, } /// 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>>, } /// 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 (ready_tx, ready_rx) = oneshot::channel::>(); let (lost_tx, lost_rx) = oneshot::channel::(); tokio::spawn(connection_task( cfg.clone(), rx, voice_out_rx, voice_in_tx, chat_tx, activity_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)), }), 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()))? } /// Disconnect cleanly. Blocks until the task exits. pub async fn disconnect(self) { let (tx, rx) = oneshot::channel(); if self.tx.send(Request::Disconnect(tx)).await.is_ok() { let _ = rx.await; } } /// 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); } } } /// 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, voice_in_tx: mpsc::Sender, chat_tx: mpsc::Sender, activity_tx: mpsc::Sender, 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 resolve_server_socket(&cfg.address).await { Ok(addr) => addr, Err(err) => fail_ready!(err), }; info!( target: "chanora_protocol", input = %cfg.address, resolved = %resolved, "server address resolved" ); // 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: HashMap< MessageHandle, ( u64, Option>>, std::time::Instant, ), > = HashMap::new(); let mut pending_text_messages: HashMap< MessageHandle, ( MessageTarget, oneshot::Sender>, std::time::Instant, ), > = HashMap::new(); let mut voice_activity: HashMap = HashMap::new(); // Main loop: pump events, service requests, forward voice. loop { // 1. Drain any outbound voice packets first — they're time-sensitive. while let Ok(pkt) = voice_out_rx.try_recv() { if let Err(e) = con.send_audio(pkt) { warn!(target: "chanora_protocol", error = %e, "send_audio failed"); } } // 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) => { let from = packet_sender_id(&buf); if let Some(from) = from { voice_activity.insert(from, Instant::now()); if voice_in_tx .try_send(InboundVoice { from_client: from, packet: buf, }) .is_err() { // Subscriber is too slow or absent; drop. } } } 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 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 }); } if let tsclientlib::events::Event::Message { target, invoker, message, } = ev { let mapped = match target { tsclientlib::MessageTarget::Server => MessageTarget::Server, tsclientlib::MessageTarget::Channel => MessageTarget::Channel, tsclientlib::MessageTarget::Client(id) => { MessageTarget::Client(id.0 as u64) } tsclientlib::MessageTarget::Poke(id) => { MessageTarget::Poke(id.0 as u64) } }; let _ = chat_tx.try_send(ChatMessage { sender_id: ClientId(invoker.id.0 as u64), sender_name: sanitize(&invoker.name), message: sanitize(&message), target: mapped, }); } } } StreamItem::MessageResult(handle, result) => { if let Some((_target_channel, reply, _deadline)) = pending_moves.remove(&handle) { let mapped = map_command_result(result, "client_move"); 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" ); } } else if let Some((_target, reply, _deadline)) = pending_text_messages.remove(&handle) { let mapped = map_command_result(result, "text_message"); let _ = reply.send(mapped); } } _ => { /* book / message / other events: ignore */ } } } 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, reply, _)) = pending_moves.remove(&handle) { if let Some(reply) = reply { let _ = reply.send(Ok(())); } } } } if !pending_text_messages.is_empty() { let now = std::time::Instant::now(); let expired: Vec = pending_text_messages .iter() .filter_map(|(handle, (_, _, deadline))| { if now >= *deadline { Some(*handle) } else { None } }) .collect(); for handle in expired { if let Some((_target, reply, _)) = pending_text_messages.remove(&handle) { let _ = reply.send(Err(ProtocolError::Timeout)); } } } // 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, }) => match send_text_message(&mut con, &message, target) { Ok(handle) => { let deadline = std::time::Instant::now() + Duration::from_secs(3); pending_text_messages.insert(handle, (target, reply, deadline)); } Err(e) => { let _ = reply.send(Err(e)); } }, Ok(Request::Disconnect(reply)) => { let _ = con.disconnect(DisconnectOptions::new()); con.events().for_each(|_| future::ready(())).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()); con.events().for_each(|_| future::ready(())).await; info!(target: "chanora_protocol", "handle dropped; implicit disconnect"); exit!(DisconnectReason::UserRequested); } } } } 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}"), }) } /// 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 { 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)?; let handle = client .send_textmessage(message) .send_with_result(con) .map_err(|e| ProtocolError::Backend(format!("send_textmessage(client): {e}")))?; info!(target: "chanora_protocol", len = message.len(), ?target, "text message queued"); Ok(handle) } 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)?; let handle = client .poke(message) .send_with_result(con) .map_err(|e| ProtocolError::Backend(format!("poke: {e}")))?; info!(target: "chanora_protocol", len = message.len(), ?target, "text message queued"); Ok(handle) } } } fn send_text_to_mode( con: &mut Connection, message: &str, target: tsproto_types::TextMessageTargetMode, label: &str, ) -> Result { use ts_bookkeeping::messages::c2s; c2s::OutSendTextMessageMessage::new(&mut std::iter::once(c2s::OutSendTextMessagePart { target, target_client_id: None, message: message.into(), })) .send_with_result(con) .map_err(|e| ProtocolError::Backend(format!("send_textmessage({label}): {e}"))) } fn map_command_result( result: Result<(), tsclientlib::CommandError>, action: &str, ) -> Result<(), ProtocolError> { 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", action, code, message = %message, "server rejected command" ); Err(ProtocolError::ServerRejected { code, message }) } } } 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, .. } => { if extra.reason.is_none() { return None; } 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::{is_server_query_client_type, sort_channels_tree_by}; 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 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)); } }