chanora/crates/chanora_protocol/src/adapter.rs

//! 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::time::Duration;

use futures::prelude::*;
use tokio::sync::{mpsc, oneshot};
use tracing::{info, warn};

use tsclientlib::data::{self, Channel, Client};
use tsclientlib::{
    ChannelId as TsChannelId, Connection, DisconnectOptions, Identity, OutCommandExt, StreamItem,
};
use tsproto_packets::packets::{InAudioBuf, OutPacket};

use crate::dto::{ChannelId, ChannelInfo, ClientId, ClientInfo, ServerSnapshot};
use crate::ProtocolError;

/// 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<String>,
    /// 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 must wire this to `chanora_storage::SecretStorageRepository`.
    pub identity: Option<String>,
    /// 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<Result<ServerSnapshot, ProtocolError>>),
    Disconnect(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<Request>,
    /// Submit outbound voice packets here. Built by `chanora_audio`
    /// via [`Self::voice_out`].
    voice_out_tx: mpsc::Sender<OutPacket>,
    /// Inbound voice packets land here. Consumed by `chanora_audio`.
    /// Wrapped in a `Mutex<Option<_>>` so the consumer can take it
    /// exactly once.
    voice_in_rx: std::sync::Mutex<Option<mpsc::Receiver<InboundVoice>>>,
    /// 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<Option<_>> so it can be
    /// taken once by the supervisor and never resurfaced.
    lost_rx: std::sync::Mutex<Option<oneshot::Receiver<DisconnectReason>>>,
}

/// 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<Request>,
}

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<ServerSnapshot, ProtocolError> {
        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 {
    /// 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<Self, ProtocolError> {
        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::<Request>(8);
        let (voice_out_tx, voice_out_rx) = mpsc::channel::<OutPacket>(64);
        let (voice_in_tx, voice_in_rx) = mpsc::channel::<InboundVoice>(64);
        let (ready_tx, ready_rx) = oneshot::channel::<Result<(), ProtocolError>>();
        let (lost_tx, lost_rx) = oneshot::channel::<DisconnectReason>();

        tokio::spawn(connection_task(
            cfg.clone(),
            rx,
            voice_out_rx,
            voice_in_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)),
            }),
            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<ServerSnapshot, ProtocolError> {
        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;
        }
    }

    /// Sender for outbound voice packets. Clone freely.
    pub fn voice_out(&self) -> mpsc::Sender<OutPacket> {
        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<mpsc::Receiver<InboundVoice>> {
        self.voice_in_rx.lock().ok().and_then(|mut g| g.take())
    }

    /// 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<oneshot::Receiver<DisconnectReason>> {
        self.lost_rx.lock().ok().and_then(|mut g| g.take())
    }
}

async fn connection_task(
    cfg: ConnectConfig,
    mut rx: mpsc::Receiver<Request>,
    mut voice_out_rx: mpsc::Receiver<OutPacket>,
    voice_in_tx: mpsc::Sender<InboundVoice>,
    ready_tx: oneshot::Sender<Result<(), ProtocolError>>,
    lost_tx: oneshot::Sender<DisconnectReason>,
) {
    // 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;
        }};
    }

    // Resolve the hostname OURSELVES using the platform resolver.
    // tsclientlib's built-in hickory-resolver reads /etc/resolv.conf,
    // which does not exist on Android or iOS — by side-stepping it
    // here we get hostname connects working on every platform.
    let addrs = match crate::resolver::resolve(&cfg.address).await {
        Ok(a) => a,
        Err(e) => {
            let msg = format!("{e}");
            let _ = ready_tx.send(Err(e));
            exit!(DisconnectReason::Error(msg));
        }
    };
    // Pick the first address (IPv4 preferred by the resolver's
    // ordering). Future retry logic could fall back to subsequent
    // addresses; one is enough for the Beta connect flow.
    let resolved = addrs[0];
    info!(
        target: "chanora_protocol",
        input = %cfg.address,
        resolved = %resolved,
        "dns resolved"
    );

    // Pass the resolved SocketAddr directly to tsclientlib so it
    // skips its own resolver entirely (tsclientlib accepts
    // SocketAddr via the From<SocketAddr> for ServerAddress impl).
    let mut builder = Connection::build(resolved).name(cfg.nickname.clone());

    let identity = match cfg.identity.as_deref() {
        Some(s) => match Identity::new_from_str(s) {
            Ok(id) => id,
            Err(e) => {
                let msg = format!("{e}");
                let _ = ready_tx.send(Err(ProtocolError::Identity(msg.clone())));
                exit!(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}");
            let _ = ready_tx.send(Err(ProtocolError::Connect(msg.clone())));
            exit!(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}");
            let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
            exit!(DisconnectReason::Error(format!("disconnected early: {msg}")));
        }
        None => {
            let msg = "event stream ended before snapshot".to_string();
            let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
            exit!(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();
                let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
                exit!(DisconnectReason::StreamEnded);
            }
            Err(_) => { /* no event available right now; keep waiting */ }
        }
    }

    let _ = ready_tx.send(Ok(()));

    // 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))) => {
                if let StreamItem::Audio(buf) = item {
                    // Extract `from` client id then forward.
                    let from = packet_sender_id(&buf);
                    if let Some(from) = from {
                        if voice_in_tx
                            .try_send(InboundVoice {
                                from_client: from,
                                packet: buf,
                            })
                            .is_err()
                        {
                            // Subscriber is too slow or absent; drop.
                        }
                    }
                }
            }
            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 */ }
        }

        // 3. Service at most one control request (non-blocking).
        match rx.try_recv() {
            Ok(Request::Snapshot(reply)) => {
                let snap = build_snapshot(&con);
                let _ = reply.send(snap);
            }
            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);
            }
        }
    }
}

/// Extract the originating `client_id` from an inbound voice packet.
fn packet_sender_id(buf: &InAudioBuf) -> Option<u64> {
    use tsproto_packets::packets::AudioData;
    match buf.data().data() {
        AudioData::S2C { from, .. } => Some(*from as u64),
        AudioData::S2CWhisper { from, .. } => Some(*from as u64),
        _ => None,
    }
}

fn build_snapshot(con: &Connection) -> Result<ServerSnapshot, ProtocolError> {
    let state: &data::Connection = con
        .get_state()
        .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;

    let mut channels: Vec<&Channel> = state.channels.values().collect();
    channels.sort_by_key(|c| c.order.0);
    let clients: Vec<&Client> = state.clients.values().collect();

    let channels_dto: Vec<ChannelInfo> = 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,
        })
        .collect();

    let clients_dto: Vec<ClientInfo> = clients
        .iter()
        .map(|c| ClientInfo {
            id: ClientId(c.id.0 as u64),
            channel: ChannelId(c.channel.0),
            name: sanitize(&c.name),
        })
        .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,
    })
}

/// 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()
}

#[allow(dead_code)]
const _ROOT_MATCHES_UPSTREAM: () = {
    // Compile-time assertion that ChannelId(0) maps to what tsclientlib
    // also considers the root.
    let _ = TsChannelId(0);
};