softchat/

managed_transport.rs

//! Optional SDK-owned native relay transport.
//!
//! The low-level [`crate::AccountTransportAction`] boundary remains public for
//! hosts that need to own their socket. This module is the batteries-included
//! alternative: one dedicated worker owns WSS, certificate verification,
//! optional Noise framing, retries, and the existing Rust-owned account
//! transport state machine.

use std::collections::VecDeque;
use std::fmt;
use std::future::Future;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{SyncSender, sync_channel};
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant as StdInstant, SystemTime, UNIX_EPOCH};

use futures_util::{SinkExt, StreamExt};
use rustls::ClientConfig;
#[cfg(target_os = "android")]
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
#[cfg(not(target_os = "android"))]
use rustls_platform_verifier::ConfigVerifierExt;
use tokio::net::TcpStream;
use tokio::runtime::Builder as RuntimeBuilder;
use tokio::sync::{Notify, mpsc as async_mpsc};
use tokio::time::{Instant, sleep_until, timeout};
use tokio_rustls::TlsConnector;
use tokio_rustls::client::TlsStream;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::error::Error as WebSocketError;
use tokio_tungstenite::tungstenite::http::{HeaderValue, header::SEC_WEBSOCKET_PROTOCOL};
use tokio_tungstenite::tungstenite::protocol::{Message, WebSocketConfig};
use tokio_tungstenite::{WebSocketStream, client_async_with_config};
use url::Url;

use crate::runtime_facade::AccountRuntimeHandle;
use crate::{
    AccountTransportAction, AccountTransportActionKind, AccountTransportBatch,
    AccountTransportResult, AccountTransportResultKind, AccountTypingIndicator,
    MAX_NOISE_FRAME_CIPHERTEXT, MAX_RELAY_FRAME_BYTES, NoiseClientHandshake, NoiseTransport,
    RelayConnectionState, SoftchatError, SyncEngineSnapshot,
};

const REVISION_WAKE_INTERVAL: Duration = Duration::from_millis(250);
const CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
const SEND_TIMEOUT: Duration = Duration::from_secs(15);
const CLOSE_TIMEOUT: Duration = Duration::from_secs(2);
const NOISE_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
const START_TIMEOUT: Duration = Duration::from_secs(30);
const COMMAND_TIMEOUT: Duration = Duration::from_secs(30);
const CONTROL_CHANNEL_CAPACITY: usize = 16;
const MAX_WAIT_MILLIS: u64 = 300_000;
const READ_BUFFER_BYTES: usize = 64 * 1024;
const MAX_WRITE_BUFFER_BYTES: usize = MAX_RELAY_FRAME_BYTES * 2;
const NOISE_WEBSOCKET_SUBPROTOCOL: &str = "nostr-noise-nk.v1.25519-chachapoly-sha256";

static NEXT_RUN_SEQUENCE: AtomicU64 = AtomicU64::new(1);

type NativeWebSocket = WebSocketStream<TlsStream<TcpStream>>;

/// Stable redacted reason for the most recent managed-transport failure.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum ManagedTransportFailure {
    /// The host reported that no usable network is available.
    NetworkUnavailable,
    /// DNS, TCP, timeout, or remote disconnect failure.
    Connection,
    /// Certificate verification or TLS negotiation failed.
    Tls,
    /// The HTTP upgrade or WebSocket stream failed its protocol contract.
    WebSocket,
    /// The configured Noise NK handshake or authenticated framing failed.
    Noise,
    /// A relay frame violated the bounded Softchat transport profile.
    Protocol,
    /// The Rust-owned account state machine or persistence transition failed.
    Account,
}

/// Complete observable state of one SDK-owned account transport.
#[derive(Clone, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ManagedTransportSnapshot {
    /// Monotonic process-local update sequence.
    pub sequence: u64,
    /// Durable account revision observed by the transport.
    pub revision: i64,
    /// Rust-owned relay connection state.
    pub connection_state: RelayConnectionState,
    /// Whether synchronization and durable delivery currently have no work.
    pub idle: bool,
    /// Current authenticated, non-durable typing indicators.
    pub typing_indicators: Vec<AccountTypingIndicator>,
    /// Current Rust-owned synchronization state.
    pub synchronization: Option<SyncEngineSnapshot>,
    /// Most recent redacted failure, cleared after the session becomes ready.
    pub last_failure: Option<ManagedTransportFailure>,
    /// Latest host reachability hint.
    pub network_available: bool,
    /// Whether the native worker has terminated.
    pub stopped: bool,
}

impl ManagedTransportSnapshot {
    fn initial(revision: i64) -> Self {
        Self {
            sequence: 0,
            revision,
            connection_state: RelayConnectionState::Disconnected,
            idle: false,
            typing_indicators: Vec::new(),
            synchronization: None,
            last_failure: None,
            network_available: true,
            stopped: false,
        }
    }

    fn same_content(&self, other: &Self) -> bool {
        self.revision == other.revision
            && self.connection_state == other.connection_state
            && self.idle == other.idle
            && self.typing_indicators == other.typing_indicators
            && self.synchronization == other.synchronization
            && self.last_failure == other.last_failure
            && self.network_available == other.network_available
            && self.stopped == other.stopped
    }
}

#[derive(Debug)]
struct SnapshotState {
    value: Mutex<ManagedTransportSnapshot>,
    changed: Condvar,
}

impl SnapshotState {
    fn new(revision: i64) -> Self {
        Self {
            value: Mutex::new(ManagedTransportSnapshot::initial(revision)),
            changed: Condvar::new(),
        }
    }

    fn current(&self) -> ManagedTransportSnapshot {
        lock(&self.value).clone()
    }

    fn publish(&self, mut next: ManagedTransportSnapshot) {
        let mut current = lock(&self.value);
        if current.same_content(&next) {
            return;
        }
        next.sequence = current.sequence.saturating_add(1);
        *current = next;
        self.changed.notify_all();
    }

    fn wait_after(&self, sequence: u64, wait: Duration) -> ManagedTransportSnapshot {
        let deadline = StdInstant::now().checked_add(wait);
        let mut current = lock(&self.value);
        while current.sequence <= sequence && !current.stopped {
            let Some(deadline) = deadline else {
                break;
            };
            let remaining = deadline.saturating_duration_since(StdInstant::now());
            if remaining.is_zero() {
                break;
            }
            match self.changed.wait_timeout(current, remaining) {
                Ok((guard, result)) => {
                    current = guard;
                    if result.timed_out() {
                        break;
                    }
                }
                Err(poisoned) => {
                    let (guard, _) = poisoned.into_inner();
                    current = guard;
                    break;
                }
            }
        }
        current.clone()
    }
}

enum ManagerCommand {
    ForceResync(SyncSender<Result<(), SoftchatError>>),
}

#[derive(Debug)]
struct ManagerSignals {
    wake_pending: AtomicBool,
    network_available: AtomicBool,
    network_changed: AtomicBool,
    notify: Notify,
}

impl ManagerSignals {
    fn new() -> Self {
        Self {
            wake_pending: AtomicBool::new(false),
            network_available: AtomicBool::new(true),
            network_changed: AtomicBool::new(false),
            notify: Notify::new(),
        }
    }

    fn request_wake(&self) {
        if !self.wake_pending.swap(true, Ordering::AcqRel) {
            self.notify.notify_one();
        }
    }

    fn set_network_available(&self, available: bool) {
        if self.network_available.swap(available, Ordering::AcqRel) != available {
            self.network_changed.store(true, Ordering::Release);
            self.notify.notify_one();
        }
    }
}

/// SDK-owned WSS/Noise executor for one [`crate::SoftchatAccount`].
///
/// Each instance owns one named OS thread with a current-thread Tokio runtime.
/// It drives the same correlated account transport actions exposed by the
/// low-level API, so protocol, delivery, retry, authentication, and sync truth
/// remain in one Rust state machine.
#[cfg_attr(feature = "native-bindings", derive(uniffi::Object))]
pub struct ManagedAccountTransport {
    account: Arc<AccountRuntimeHandle>,
    commands: async_mpsc::Sender<ManagerCommand>,
    signals: Arc<ManagerSignals>,
    snapshots: Arc<SnapshotState>,
    shutdown_requested: Arc<AtomicBool>,
    shutdown_notify: Arc<Notify>,
    worker: Mutex<Option<JoinHandle<()>>>,
}

impl fmt::Debug for ManagedAccountTransport {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ManagedAccountTransport")
            .field("account_id", &self.account.account_id())
            .field("snapshot", &self.snapshots.current())
            .finish_non_exhaustive()
    }
}

#[cfg_attr(feature = "native-bindings", uniffi::export)]
impl ManagedAccountTransport {
    /// Start the only active managed or low-level transport for this account.
    ///
    /// # Errors
    ///
    /// Returns a stable account, relay, persistence, or worker-start error.
    #[cfg_attr(feature = "native-bindings", uniffi::constructor)]
    pub fn new(account: Arc<AccountRuntimeHandle>) -> Result<Self, SoftchatError> {
        let revision = account.database_info()?.revision;
        let snapshots = Arc::new(SnapshotState::new(revision));
        let shutdown_requested = Arc::new(AtomicBool::new(false));
        let shutdown_notify = Arc::new(Notify::new());
        let signals = Arc::new(ManagerSignals::new());
        let (commands, command_receiver) = async_mpsc::channel(CONTROL_CHANNEL_CAPACITY);
        let (started_sender, started_receiver) = sync_channel(1);
        let run_sequence = NEXT_RUN_SEQUENCE.fetch_add(1, Ordering::Relaxed);
        let run_id = format!("managed-{}-{run_sequence}", std::process::id());

        let thread_account = Arc::clone(&account);
        let thread_snapshots = Arc::clone(&snapshots);
        let thread_shutdown = Arc::clone(&shutdown_requested);
        let thread_notify = Arc::clone(&shutdown_notify);
        let thread_signals = Arc::clone(&signals);
        let worker = thread::Builder::new()
            .name(format!("softchat-transport-{run_sequence}"))
            .spawn(move || {
                let runtime = RuntimeBuilder::new_current_thread()
                    .enable_io()
                    .enable_time()
                    .build()
                    .map_err(|_| SoftchatError::InternalFailure);
                let runtime = match runtime {
                    Ok(runtime) => runtime,
                    Err(error) => {
                        let _ = started_sender.send(Err(error));
                        return;
                    }
                };
                let initial = thread_account.start_account_transport(run_id);
                let initial = match initial {
                    Ok(initial) => initial,
                    Err(error) => {
                        let _ = started_sender.send(Err(error));
                        return;
                    }
                };
                let manager = TransportManager::new(
                    thread_account,
                    command_receiver,
                    thread_signals,
                    thread_snapshots,
                    thread_shutdown,
                    thread_notify,
                    initial,
                );
                let _ = started_sender.send(Ok(()));
                runtime.block_on(manager.run());
            })
            .map_err(|_| SoftchatError::InternalFailure)?;

        match started_receiver.recv_timeout(START_TIMEOUT) {
            Ok(Ok(())) => Ok(Self {
                account,
                commands,
                signals,
                snapshots,
                shutdown_requested,
                shutdown_notify,
                worker: Mutex::new(Some(worker)),
            }),
            Ok(Err(error)) => {
                let _ = worker.join();
                Err(error)
            }
            Err(_) => {
                shutdown_requested.store(true, Ordering::Release);
                shutdown_notify.notify_one();
                let _ = worker.join();
                Err(SoftchatError::InternalFailure)
            }
        }
    }

    /// Return the latest non-blocking state snapshot.
    #[must_use]
    pub fn snapshot(&self) -> ManagedTransportSnapshot {
        self.snapshots.current()
    }

    /// Block for a newer snapshot or the bounded timeout.
    ///
    /// This is intended for authored language facades to adapt to `Flow` or an
    /// equivalent async stream on a background dispatcher.
    #[must_use]
    pub fn wait_for_update(
        &self,
        after_sequence: u64,
        timeout_millis: u64,
    ) -> ManagedTransportSnapshot {
        self.snapshots.wait_after(
            after_sequence,
            Duration::from_millis(timeout_millis.min(MAX_WAIT_MILLIS)),
        )
    }

    /// Coalesce an immediate durable-revision wake hint.
    ///
    /// The worker also checks revision-backed work every 250 ms, so direct
    /// Rust consumers do not need to implement a notifier for correctness.
    pub fn wake(&self) {
        self.signals.request_wake();
    }

    /// Update the host reachability hint.
    pub fn set_network_available(&self, available: bool) {
        self.signals.set_network_available(available);
    }

    /// Forget synchronization checkpoints and reconnect from retained truth.
    ///
    /// # Errors
    ///
    /// Returns a stable account/synchronization error or a redacted worker
    /// failure when the command cannot be completed within 30 seconds.
    pub fn force_resync(&self) -> Result<(), SoftchatError> {
        let (sender, receiver) = sync_channel(1);
        self.commands
            .try_send(ManagerCommand::ForceResync(sender))
            .map_err(|error| match error {
                async_mpsc::error::TrySendError::Full(_) => SoftchatError::InternalFailure,
                async_mpsc::error::TrySendError::Closed(_) => SoftchatError::AccountClosed,
            })?;
        receiver
            .recv_timeout(COMMAND_TIMEOUT)
            .map_err(|_| SoftchatError::InternalFailure)?
    }

    /// Wait until work committed before this call is delivered and sync is idle.
    ///
    /// Returns `false` on timeout, disconnection, or shutdown.
    ///
    /// # Errors
    ///
    /// Returns a stable account error when the account is already closed.
    pub fn flush(&self, timeout_millis: u64) -> Result<bool, SoftchatError> {
        let target_revision = self.account.database_info()?.revision;
        self.wake();
        let timeout_duration = Duration::from_millis(timeout_millis.min(MAX_WAIT_MILLIS));
        let deadline = StdInstant::now()
            .checked_add(timeout_duration)
            .ok_or(SoftchatError::InternalFailure)?;
        let mut snapshot = self.snapshots.current();
        loop {
            if snapshot.stopped {
                return Ok(false);
            }
            if snapshot.connection_state == RelayConnectionState::Ready
                && snapshot.idle
                && snapshot.revision >= target_revision
            {
                return Ok(true);
            }
            let remaining = deadline.saturating_duration_since(StdInstant::now());
            if remaining.is_zero() {
                return Ok(false);
            }
            snapshot = self.snapshots.wait_after(snapshot.sequence, remaining);
        }
    }

    /// Stop the worker, close the socket, release leases, and join its thread.
    ///
    /// Repeated calls are harmless.
    pub fn shutdown(&self) {
        self.request_shutdown();
    }
}

impl ManagedAccountTransport {
    fn request_shutdown(&self) {
        if !self.shutdown_requested.swap(true, Ordering::AcqRel) {
            // `notify_one` retains a permit when shutdown races the start of a
            // TCP/TLS/WebSocket await. That keeps joining deterministic even
            // when no async waiter was registered at the exact call instant.
            self.shutdown_notify.notify_one();
        }
        if let Some(worker) = lock(&self.worker).take()
            && worker.join().is_err()
        {
            let mut snapshot = self.snapshots.current();
            snapshot.last_failure = Some(ManagedTransportFailure::Account);
            snapshot.stopped = true;
            self.snapshots.publish(snapshot);
        }
    }
}

impl Drop for ManagedAccountTransport {
    fn drop(&mut self) {
        self.request_shutdown();
    }
}

struct PendingConnect {
    action: AccountTransportAction,
    ready_at: Instant,
}

struct PendingReceive {
    action: AccountTransportAction,
    deadline: Instant,
}

struct RelaySocket {
    stream: NativeWebSocket,
    noise: Option<Arc<NoiseTransport>>,
}

impl RelaySocket {
    fn discard(self) {
        if let Some(noise) = self.noise {
            noise.shutdown();
        }
    }

    async fn close(mut self) {
        if let Some(noise) = self.noise.take() {
            noise.shutdown();
        }
        let _ = timeout(CLOSE_TIMEOUT, self.stream.close(None)).await;
    }

    async fn send_text(&mut self, text: String) -> Result<(), ManagedTransportFailure> {
        if text.len() > MAX_RELAY_FRAME_BYTES {
            return Err(ManagedTransportFailure::Protocol);
        }
        if let Some(noise) = &self.noise {
            let frames = noise
                .encrypt(text.into_bytes())
                .map_err(|_| ManagedTransportFailure::Noise)?;
            for frame in frames {
                if frame.len() > MAX_NOISE_FRAME_CIPHERTEXT {
                    return Err(ManagedTransportFailure::Noise);
                }
                self.stream
                    .send(Message::Binary(frame.into()))
                    .await
                    .map_err(map_write_error)?;
            }
        } else {
            self.stream
                .send(Message::Text(text.into()))
                .await
                .map_err(map_write_error)?;
        }
        self.stream.flush().await.map_err(map_write_error)
    }

    async fn receive(&mut self) -> WireRead {
        let next = self.stream.next().await;
        match next {
            None => WireRead::Disconnected,
            Some(Err(error)) => WireRead::Failure(map_read_error(&error)),
            Some(Ok(Message::Text(text))) => {
                if self.noise.is_some() || text.len() > MAX_RELAY_FRAME_BYTES {
                    WireRead::Failure(ManagedTransportFailure::Protocol)
                } else {
                    WireRead::Text(text.to_string())
                }
            }
            Some(Ok(Message::Binary(frame))) => {
                let Some(noise) = &self.noise else {
                    return WireRead::Failure(ManagedTransportFailure::Protocol);
                };
                if frame.len() > MAX_NOISE_FRAME_CIPHERTEXT {
                    return WireRead::Failure(ManagedTransportFailure::Noise);
                }
                match noise.decrypt(frame.to_vec()) {
                    Ok(None) => WireRead::Continue,
                    Ok(Some(plaintext)) => {
                        if plaintext.len() > MAX_RELAY_FRAME_BYTES {
                            return WireRead::Failure(ManagedTransportFailure::Protocol);
                        }
                        match String::from_utf8(plaintext) {
                            Ok(text) => WireRead::Text(text),
                            Err(_) => WireRead::Failure(ManagedTransportFailure::Noise),
                        }
                    }
                    Err(_) => WireRead::Failure(ManagedTransportFailure::Noise),
                }
            }
            Some(Ok(Message::Ping(_))) => match self.stream.flush().await {
                Ok(()) => WireRead::Continue,
                Err(error) => WireRead::Failure(map_read_error(&error)),
            },
            Some(Ok(Message::Pong(_))) => WireRead::Continue,
            Some(Ok(Message::Close(_))) => WireRead::Disconnected,
            Some(Ok(Message::Frame(_))) => WireRead::Failure(ManagedTransportFailure::Protocol),
        }
    }
}

enum WireRead {
    Text(String),
    Continue,
    Disconnected,
    Failure(ManagedTransportFailure),
}

struct TransportManager {
    account: Arc<AccountRuntimeHandle>,
    commands: async_mpsc::Receiver<ManagerCommand>,
    signals: Arc<ManagerSignals>,
    snapshots: Arc<SnapshotState>,
    shutdown_requested: Arc<AtomicBool>,
    shutdown_notify: Arc<Notify>,
    socket: Option<RelaySocket>,
    pending_connect: Option<PendingConnect>,
    pending_receive: Option<PendingReceive>,
    sends: VecDeque<AccountTransportAction>,
    closes: VecDeque<AccountTransportAction>,
    generation: u64,
    batch: AccountTransportBatch,
    last_failure: Option<ManagedTransportFailure>,
    network_available: bool,
    next_revision_wake: Instant,
}

impl TransportManager {
    fn new(
        account: Arc<AccountRuntimeHandle>,
        commands: async_mpsc::Receiver<ManagerCommand>,
        signals: Arc<ManagerSignals>,
        snapshots: Arc<SnapshotState>,
        shutdown_requested: Arc<AtomicBool>,
        shutdown_notify: Arc<Notify>,
        initial: AccountTransportBatch,
    ) -> Self {
        let mut manager = Self {
            account,
            commands,
            signals,
            snapshots,
            shutdown_requested,
            shutdown_notify,
            socket: None,
            pending_connect: None,
            pending_receive: None,
            sends: VecDeque::new(),
            closes: VecDeque::new(),
            generation: 0,
            batch: initial.clone(),
            last_failure: None,
            network_available: true,
            next_revision_wake: Instant::now() + REVISION_WAKE_INTERVAL,
        };
        manager.install_batch(initial);
        manager
    }

    async fn run(mut self) {
        let result = self.run_loop().await;
        if result.is_err() && self.last_failure.is_none() {
            self.last_failure = Some(ManagedTransportFailure::Account);
        }
        if let Some(socket) = self.socket.take() {
            socket.close().await;
        }
        if let Ok(batch) = self.account.cancel_account_transport() {
            self.install_batch(batch);
        }
        let mut snapshot = self.snapshot();
        snapshot.stopped = true;
        self.snapshots.publish(snapshot);
    }

    async fn run_loop(&mut self) -> Result<(), SoftchatError> {
        loop {
            if self.shutdown_requested.load(Ordering::Acquire) {
                return Ok(());
            }
            while let Ok(command) = self.commands.try_recv() {
                self.handle_command(command);
            }
            self.handle_pending_signals()?;

            if let Some(action) = self.sends.pop_front() {
                if !self.execute_send(action).await? {
                    return Ok(());
                }
                continue;
            }
            if let Some(action) = self.closes.pop_front() {
                if let Some(socket) = self.socket.take() {
                    socket.close().await;
                }
                self.apply_result(
                    &action,
                    AccountTransportResultKind::Disconnected,
                    String::new(),
                    None,
                )?;
                continue;
            }
            if self
                .pending_connect
                .as_ref()
                .is_some_and(|pending| pending.ready_at <= Instant::now())
            {
                let pending = self
                    .pending_connect
                    .take()
                    .ok_or(SoftchatError::InternalFailure)?;
                self.execute_connect(pending.action).await?;
                continue;
            }

            let event = self.next_event().await;
            match event {
                LoopEvent::Shutdown => return Ok(()),
                LoopEvent::Command(Some(command)) => {
                    self.handle_command(command);
                }
                LoopEvent::Command(None) => return Ok(()),
                LoopEvent::Signal => self.handle_pending_signals()?,
                LoopEvent::RevisionWake => self.wake_transport()?,
                LoopEvent::ConnectReady => {}
                LoopEvent::ReceiveTimedOut => {
                    let pending = self
                        .pending_receive
                        .take()
                        .ok_or(SoftchatError::InternalFailure)?;
                    self.apply_result(
                        &pending.action,
                        AccountTransportResultKind::TimedOut,
                        String::new(),
                        None,
                    )?;
                }
                LoopEvent::Wire(read) => self.handle_wire_read(read)?,
            }
        }
    }

    async fn next_event(&mut self) -> LoopEvent {
        let revision_wake = self.next_revision_wake;
        if let Some(pending) = &self.pending_connect {
            let connect_ready = pending.ready_at;
            return tokio::select! {
                _ = self.shutdown_notify.notified() => LoopEvent::Shutdown,
                command = self.commands.recv() => LoopEvent::Command(command),
                _ = self.signals.notify.notified() => LoopEvent::Signal,
                _ = sleep_until(connect_ready) => LoopEvent::ConnectReady,
                _ = sleep_until(revision_wake) => LoopEvent::RevisionWake,
            };
        }

        let Some(pending_receive) = &self.pending_receive else {
            return tokio::select! {
                _ = self.shutdown_notify.notified() => LoopEvent::Shutdown,
                command = self.commands.recv() => LoopEvent::Command(command),
                _ = self.signals.notify.notified() => LoopEvent::Signal,
                _ = sleep_until(revision_wake) => LoopEvent::RevisionWake,
            };
        };
        let receive_deadline = pending_receive.deadline;
        let Some(socket) = self.socket.as_mut() else {
            return LoopEvent::Wire(WireRead::Disconnected);
        };
        tokio::select! {
            _ = self.shutdown_notify.notified() => LoopEvent::Shutdown,
            command = self.commands.recv() => LoopEvent::Command(command),
            _ = self.signals.notify.notified() => LoopEvent::Signal,
            _ = sleep_until(receive_deadline) => LoopEvent::ReceiveTimedOut,
            _ = sleep_until(revision_wake) => LoopEvent::RevisionWake,
            read = socket.receive() => LoopEvent::Wire(read),
        }
    }

    fn handle_command(&mut self, command: ManagerCommand) {
        match command {
            ManagerCommand::ForceResync(response) => {
                let result = self.account.reset_account_sync(now_seconds()).map(|batch| {
                    self.reset_io();
                    self.install_batch(batch);
                });
                let _ = response.send(result);
            }
        }
    }

    fn handle_pending_signals(&mut self) -> Result<(), SoftchatError> {
        let network_changed = self.signals.network_changed.swap(false, Ordering::AcqRel);
        if network_changed {
            let available = self.signals.network_available.load(Ordering::Acquire);
            if self.network_available != available {
                self.network_available = available;
                self.publish_snapshot();
            }
            if !available {
                self.signals.wake_pending.store(false, Ordering::Release);
                return self.fail_for_network_unavailable();
            }
        }
        let wake_pending = self.signals.wake_pending.swap(false, Ordering::AcqRel);
        if network_changed || wake_pending {
            self.wake_transport()?;
        }
        Ok(())
    }

    fn fail_for_network_unavailable(&mut self) -> Result<(), SoftchatError> {
        if let Some(socket) = self.socket.take() {
            socket.discard();
        }
        let action = self
            .pending_connect
            .take()
            .map(|pending| pending.action)
            .or_else(|| self.sends.pop_front())
            .or_else(|| self.pending_receive.take().map(|pending| pending.action));
        if let Some(action) = action {
            self.apply_result(
                &action,
                AccountTransportResultKind::NetworkUnavailable,
                String::new(),
                Some(ManagedTransportFailure::NetworkUnavailable),
            )?;
        }
        Ok(())
    }

    fn wake_transport(&mut self) -> Result<(), SoftchatError> {
        self.next_revision_wake = Instant::now() + REVISION_WAKE_INTERVAL;
        let batch = self.account.wake_account_transport(now_seconds())?;
        self.install_batch(batch);
        Ok(())
    }

    async fn execute_connect(
        &mut self,
        action: AccountTransportAction,
    ) -> Result<(), SoftchatError> {
        if !self.network_available {
            return self.apply_result(
                &action,
                AccountTransportResultKind::NetworkUnavailable,
                String::new(),
                Some(ManagedTransportFailure::NetworkUnavailable),
            );
        }
        if let Some(socket) = self.socket.take() {
            socket.discard();
        }
        let connect = open_socket(&action);
        let result = tokio::select! {
            _ = self.shutdown_notify.notified() => return Ok(()),
            result = timeout(CONNECT_TIMEOUT, connect) => result,
        };
        match result {
            Ok(Ok(socket)) => {
                self.socket = Some(socket);
                self.apply_result(
                    &action,
                    AccountTransportResultKind::Connected,
                    String::new(),
                    None,
                )
            }
            Ok(Err(failure)) => self.apply_result(
                &action,
                failure.result_kind(),
                String::new(),
                Some(failure.snapshot_kind()),
            ),
            Err(_) => self.apply_result(
                &action,
                AccountTransportResultKind::Disconnected,
                String::new(),
                Some(ManagedTransportFailure::Connection),
            ),
        }
    }

    async fn execute_send(
        &mut self,
        action: AccountTransportAction,
    ) -> Result<bool, SoftchatError> {
        if !self.network_available {
            self.apply_result(
                &action,
                AccountTransportResultKind::NetworkUnavailable,
                String::new(),
                Some(ManagedTransportFailure::NetworkUnavailable),
            )?;
            return Ok(true);
        }
        let Some(socket) = self.socket.as_mut() else {
            self.apply_result(
                &action,
                AccountTransportResultKind::Disconnected,
                String::new(),
                Some(ManagedTransportFailure::Connection),
            )?;
            return Ok(true);
        };
        let send = socket.send_text(action.text_frame.clone());
        let result = await_send_or_control(
            send,
            &mut self.commands,
            &self.signals,
            &self.shutdown_notify,
        )
        .await;
        match result {
            SendWait::Shutdown => return Ok(false),
            SendWait::NetworkUnavailable => {
                if let Some(socket) = self.socket.take() {
                    socket.discard();
                }
                self.sends.push_front(action);
                self.handle_pending_signals()?;
            }
            SendWait::Command(command) => {
                if let Some(socket) = self.socket.take() {
                    socket.discard();
                }
                self.sends.push_front(action);
                return match command {
                    Some(command) => {
                        self.handle_command(command);
                        Ok(true)
                    }
                    None => Ok(false),
                };
            }
            SendWait::Written => self.apply_result(
                &action,
                AccountTransportResultKind::FrameWritten,
                String::new(),
                None,
            )?,
            SendWait::Failed(failure) => {
                if let Some(socket) = self.socket.take() {
                    socket.discard();
                }
                self.apply_result(
                    &action,
                    failure_result_kind(failure),
                    String::new(),
                    Some(failure),
                )?;
            }
            SendWait::TimedOut => {
                if let Some(socket) = self.socket.take() {
                    socket.discard();
                }
                self.apply_result(
                    &action,
                    AccountTransportResultKind::Disconnected,
                    String::new(),
                    Some(ManagedTransportFailure::Connection),
                )?;
            }
        }
        Ok(true)
    }

    fn handle_wire_read(&mut self, read: WireRead) -> Result<(), SoftchatError> {
        match read {
            WireRead::Continue => Ok(()),
            WireRead::Text(text) => {
                let pending = self
                    .pending_receive
                    .take()
                    .ok_or(SoftchatError::InternalFailure)?;
                self.apply_result(
                    &pending.action,
                    AccountTransportResultKind::FrameReceived,
                    text,
                    None,
                )
            }
            WireRead::Disconnected => {
                if let Some(socket) = self.socket.take() {
                    socket.discard();
                }
                let pending = self
                    .pending_receive
                    .take()
                    .ok_or(SoftchatError::InternalFailure)?;
                self.apply_result(
                    &pending.action,
                    AccountTransportResultKind::Disconnected,
                    String::new(),
                    Some(ManagedTransportFailure::Connection),
                )
            }
            WireRead::Failure(failure) => {
                if let Some(socket) = self.socket.take() {
                    socket.discard();
                }
                let pending = self
                    .pending_receive
                    .take()
                    .ok_or(SoftchatError::InternalFailure)?;
                self.apply_result(
                    &pending.action,
                    failure_result_kind(failure),
                    String::new(),
                    Some(failure),
                )
            }
        }
    }

    fn apply_result(
        &mut self,
        action: &AccountTransportAction,
        kind: AccountTransportResultKind,
        text_frame: String,
        failure: Option<ManagedTransportFailure>,
    ) -> Result<(), SoftchatError> {
        if let Some(failure) = failure {
            self.last_failure = Some(failure);
            self.account
                .logs
                .emit(crate::account_diagnostics::managed_failure(failure));
        }
        let batch = self.account.apply_account_transport_result(
            AccountTransportResult {
                run_id: action.run_id.clone(),
                action_id: action.action_id.clone(),
                generation: action.generation,
                kind,
                text_frame,
            },
            now_seconds(),
        )?;
        self.install_batch(batch);
        Ok(())
    }

    fn install_batch(&mut self, batch: AccountTransportBatch) {
        self.batch = batch.clone();
        if batch.connection_state == RelayConnectionState::Ready {
            self.last_failure = None;
        }
        for action in batch.actions {
            if action.generation < self.generation {
                continue;
            }
            if action.generation > self.generation {
                self.generation = action.generation;
                self.reset_io();
            }
            match action.kind {
                AccountTransportActionKind::Connect => {
                    self.pending_connect = Some(PendingConnect {
                        ready_at: Instant::now() + Duration::from_millis(action.delay_ms),
                        action,
                    });
                }
                AccountTransportActionKind::SendText => self.sends.push_back(action),
                AccountTransportActionKind::ReceiveText => {
                    if self.pending_receive.is_none() {
                        self.pending_receive = Some(PendingReceive {
                            deadline: Instant::now() + Duration::from_millis(action.delay_ms),
                            action,
                        });
                    }
                }
                AccountTransportActionKind::Close => self.closes.push_back(action),
            }
        }
        self.publish_snapshot();
    }

    fn reset_io(&mut self) {
        if let Some(socket) = self.socket.take() {
            socket.discard();
        }
        self.pending_connect = None;
        self.pending_receive = None;
        self.sends.clear();
        self.closes.clear();
    }

    fn snapshot(&self) -> ManagedTransportSnapshot {
        ManagedTransportSnapshot {
            sequence: 0,
            revision: self.batch.revision,
            connection_state: self.batch.connection_state,
            idle: self.batch.idle,
            typing_indicators: self.batch.typing_indicators.clone(),
            synchronization: self.batch.sync.clone(),
            last_failure: self.last_failure,
            network_available: self.network_available,
            stopped: false,
        }
    }

    fn publish_snapshot(&self) {
        self.snapshots.publish(self.snapshot());
    }
}

enum LoopEvent {
    Shutdown,
    Command(Option<ManagerCommand>),
    Signal,
    RevisionWake,
    ConnectReady,
    ReceiveTimedOut,
    Wire(WireRead),
}

enum SendWait {
    Shutdown,
    Command(Option<ManagerCommand>),
    NetworkUnavailable,
    Written,
    Failed(ManagedTransportFailure),
    TimedOut,
}

async fn await_send_or_control<F>(
    send: F,
    commands: &mut async_mpsc::Receiver<ManagerCommand>,
    signals: &ManagerSignals,
    shutdown_notify: &Notify,
) -> SendWait
where
    F: Future<Output = Result<(), ManagedTransportFailure>>,
{
    let send = timeout(SEND_TIMEOUT, send);
    tokio::pin!(send);
    loop {
        tokio::select! {
            _ = shutdown_notify.notified() => return SendWait::Shutdown,
            command = commands.recv() => return SendWait::Command(command),
            _ = signals.notify.notified() => {
                if signals.network_changed.load(Ordering::Acquire)
                    && !signals.network_available.load(Ordering::Acquire)
                {
                    return SendWait::NetworkUnavailable;
                }
            }
            result = &mut send => return match result {
                Ok(Ok(())) => SendWait::Written,
                Ok(Err(failure)) => SendWait::Failed(failure),
                Err(_) => SendWait::TimedOut,
            },
        }
    }
}

#[derive(Clone, Copy, Debug)]
enum ConnectFailure {
    Connection,
    Tls,
    WebSocket,
    Noise,
    Protocol,
    Authentication,
    RateLimited,
}

impl ConnectFailure {
    const fn snapshot_kind(self) -> ManagedTransportFailure {
        match self {
            Self::Connection => ManagedTransportFailure::Connection,
            Self::Tls | Self::Authentication => ManagedTransportFailure::Tls,
            Self::WebSocket | Self::RateLimited => ManagedTransportFailure::WebSocket,
            Self::Noise => ManagedTransportFailure::Noise,
            Self::Protocol => ManagedTransportFailure::Protocol,
        }
    }

    const fn result_kind(self) -> AccountTransportResultKind {
        match self {
            Self::Authentication | Self::Tls => AccountTransportResultKind::AuthenticationFailed,
            Self::RateLimited => AccountTransportResultKind::RateLimited,
            Self::Noise | Self::Protocol | Self::WebSocket => {
                AccountTransportResultKind::ProtocolFailed
            }
            Self::Connection => AccountTransportResultKind::Disconnected,
        }
    }
}

async fn open_socket(action: &AccountTransportAction) -> Result<RelaySocket, ConnectFailure> {
    let url = Url::parse(&action.network_url).map_err(|_| ConnectFailure::Protocol)?;
    if url.scheme() != "wss" || url.username() != "" || url.password().is_some() {
        return Err(ConnectFailure::Protocol);
    }
    let host = url.host_str().ok_or(ConnectFailure::Protocol)?.to_owned();
    let port = url
        .port_or_known_default()
        .ok_or(ConnectFailure::Protocol)?;
    let request = websocket_request(action)?;
    let tcp = TcpStream::connect((host.as_str(), port))
        .await
        .map_err(|_| ConnectFailure::Connection)?;
    let server_name = ServerName::try_from(host).map_err(|_| ConnectFailure::Protocol)?;
    let tls_config = managed_tls_config()?;
    let tls = TlsConnector::from(Arc::new(tls_config))
        .connect(server_name, tcp)
        .await
        .map_err(|_| ConnectFailure::Tls)?;
    let config = WebSocketConfig::default()
        .read_buffer_size(READ_BUFFER_BYTES)
        .write_buffer_size(0)
        .max_write_buffer_size(MAX_WRITE_BUFFER_BYTES)
        .max_message_size(Some(MAX_RELAY_FRAME_BYTES))
        .max_frame_size(Some(MAX_RELAY_FRAME_BYTES));
    let (mut stream, response) = client_async_with_config(request, tls, Some(config))
        .await
        .map_err(|error| map_handshake_error(&error))?;

    if !action.noise_remote_static_key.is_empty()
        && response
            .headers()
            .get(SEC_WEBSOCKET_PROTOCOL)
            .and_then(|value| value.to_str().ok())
            != Some(NOISE_WEBSOCKET_SUBPROTOCOL)
    {
        return Err(ConnectFailure::Noise);
    }

    let noise = if action.noise_remote_static_key.is_empty() {
        None
    } else {
        let handshake = NoiseClientHandshake::new(action.noise_remote_static_key.clone(), true)
            .map_err(|_| ConnectFailure::Noise)?;
        stream
            .send(Message::Binary(handshake.message_one().into()))
            .await
            .map_err(|_| ConnectFailure::Noise)?;
        let message_two = timeout(
            NOISE_HANDSHAKE_TIMEOUT,
            receive_noise_message_two(&mut stream),
        )
        .await
        .map_err(|_| ConnectFailure::Noise)??;
        Some(
            handshake
                .complete(message_two)
                .map_err(|_| ConnectFailure::Noise)?,
        )
    };
    Ok(RelaySocket { stream, noise })
}

#[cfg(not(target_os = "android"))]
fn managed_tls_config() -> Result<ClientConfig, ConnectFailure> {
    ClientConfig::with_platform_verifier().map_err(|_| ConnectFailure::Tls)
}

#[cfg(target_os = "android")]
fn managed_tls_config() -> Result<ClientConfig, ConnectFailure> {
    // Android's rustls-platform-verifier currently misclassifies valid
    // CRL-only Let's Encrypt chains as revoked. Use the audited Mozilla trust
    // anchors through rustls/webpki until that platform bug is fixed. This is
    // the same fully Rust-owned verification model used by common native Rust
    // clients; it does not accept user-installed Android roots.
    let roots = RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
    Ok(ClientConfig::builder()
        .with_root_certificates(roots)
        .with_no_client_auth())
}

fn websocket_request(
    action: &AccountTransportAction,
) -> Result<tokio_tungstenite::tungstenite::http::Request<()>, ConnectFailure> {
    let mut request = action
        .network_url
        .as_str()
        .into_client_request()
        .map_err(|_| ConnectFailure::Protocol)?;
    if !action.noise_remote_static_key.is_empty() {
        request.headers_mut().insert(
            SEC_WEBSOCKET_PROTOCOL,
            HeaderValue::from_static(NOISE_WEBSOCKET_SUBPROTOCOL),
        );
    }
    Ok(request)
}

async fn receive_noise_message_two(
    stream: &mut NativeWebSocket,
) -> Result<Vec<u8>, ConnectFailure> {
    loop {
        match stream.next().await {
            Some(Ok(Message::Binary(message))) => {
                if message.len() > MAX_NOISE_FRAME_CIPHERTEXT {
                    return Err(ConnectFailure::Noise);
                }
                return Ok(message.to_vec());
            }
            Some(Ok(Message::Ping(_))) => {
                stream.flush().await.map_err(|_| ConnectFailure::Noise)?;
            }
            Some(Ok(Message::Pong(_))) => {}
            Some(Ok(Message::Close(_))) | None => return Err(ConnectFailure::Connection),
            Some(Ok(Message::Text(_) | Message::Frame(_))) => {
                return Err(ConnectFailure::Noise);
            }
            Some(Err(error)) => return Err(map_handshake_error(&error)),
        }
    }
}

fn map_handshake_error(error: &WebSocketError) -> ConnectFailure {
    match error {
        WebSocketError::Http(response) if response.status().as_u16() == 429 => {
            ConnectFailure::RateLimited
        }
        WebSocketError::Http(response) if matches!(response.status().as_u16(), 401 | 403) => {
            ConnectFailure::Authentication
        }
        WebSocketError::Io(_)
        | WebSocketError::ConnectionClosed
        | WebSocketError::AlreadyClosed => ConnectFailure::Connection,
        WebSocketError::Tls(_) => ConnectFailure::Tls,
        WebSocketError::Http(_)
        | WebSocketError::HttpFormat(_)
        | WebSocketError::Protocol(_)
        | WebSocketError::Capacity(_)
        | WebSocketError::WriteBufferFull(_)
        | WebSocketError::Utf8(_)
        | WebSocketError::AttackAttempt
        | WebSocketError::Url(_) => ConnectFailure::WebSocket,
    }
}

fn map_write_error(error: WebSocketError) -> ManagedTransportFailure {
    match error {
        WebSocketError::Io(_)
        | WebSocketError::ConnectionClosed
        | WebSocketError::AlreadyClosed => ManagedTransportFailure::Connection,
        WebSocketError::Tls(_) => ManagedTransportFailure::Tls,
        _ => ManagedTransportFailure::WebSocket,
    }
}

fn map_read_error(error: &WebSocketError) -> ManagedTransportFailure {
    match error {
        WebSocketError::Io(_)
        | WebSocketError::ConnectionClosed
        | WebSocketError::AlreadyClosed => ManagedTransportFailure::Connection,
        WebSocketError::Tls(_) => ManagedTransportFailure::Tls,
        _ => ManagedTransportFailure::WebSocket,
    }
}

const fn failure_result_kind(failure: ManagedTransportFailure) -> AccountTransportResultKind {
    match failure {
        ManagedTransportFailure::NetworkUnavailable => {
            AccountTransportResultKind::NetworkUnavailable
        }
        ManagedTransportFailure::Connection => AccountTransportResultKind::Disconnected,
        ManagedTransportFailure::Tls => AccountTransportResultKind::AuthenticationFailed,
        ManagedTransportFailure::WebSocket
        | ManagedTransportFailure::Noise
        | ManagedTransportFailure::Protocol
        | ManagedTransportFailure::Account => AccountTransportResultKind::ProtocolFailed,
    }
}

fn now_seconds() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .ok()
        .and_then(|duration| i64::try_from(duration.as_secs()).ok())
        .unwrap_or(0)
}

fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    match mutex.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn temporary_directory(label: &str) -> Result<std::path::PathBuf, Box<dyn std::error::Error>> {
        let nonce = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
        let path =
            std::env::temp_dir().join(format!("softchat-{label}-{}-{nonce}", std::process::id()));
        std::fs::create_dir_all(&path)?;
        Ok(path)
    }

    #[test]
    fn snapshot_wait_is_bounded_and_sequence_based() {
        let state = SnapshotState::new(7);
        let initial = state.current();
        let timed_out = state.wait_after(initial.sequence, Duration::from_millis(1));
        assert_eq!(timed_out, initial);

        let mut updated = initial.clone();
        updated.network_available = false;
        state.publish(updated);
        let observed = state.wait_after(initial.sequence, Duration::ZERO);
        assert!(observed.sequence > initial.sequence);
        assert!(!observed.network_available);
    }

    #[test]
    fn managed_feature_rejects_insecure_or_credentialed_urls_before_io() {
        for value in [
            "ws://relay.example",
            "https://relay.example",
            "wss://[email protected]",
            "not a url",
        ] {
            let parsed = Url::parse(value);
            let valid = parsed.as_ref().is_ok_and(|url| {
                url.scheme() == "wss" && url.username().is_empty() && url.password().is_none()
            });
            assert!(!valid);
        }
    }

    #[test]
    fn noise_websocket_request_advertises_the_deployed_subprotocol() -> Result<(), ConnectFailure> {
        let mut action = AccountTransportAction {
            run_id: "test-run".to_owned(),
            action_id: "test-action".to_owned(),
            generation: 1,
            kind: AccountTransportActionKind::Connect,
            configured_url: "wss+noise://relay.example".to_owned(),
            network_url: "wss://relay.example/".to_owned(),
            noise_remote_static_key: vec![7_u8; 32],
            text_frame: String::new(),
            delay_ms: 0,
        };
        let noise = websocket_request(&action)?;
        assert_eq!(
            noise
                .headers()
                .get(SEC_WEBSOCKET_PROTOCOL)
                .and_then(|value| value.to_str().ok()),
            Some(NOISE_WEBSOCKET_SUBPROTOCOL)
        );

        action.noise_remote_static_key.clear();
        let plain = websocket_request(&action)?;
        assert!(plain.headers().get(SEC_WEBSOCKET_PROTOCOL).is_none());
        Ok(())
    }

    #[test]
    fn force_resync_control_preempts_a_stalled_send() -> Result<(), Box<dyn std::error::Error>> {
        let (sender, mut commands) = async_mpsc::channel(CONTROL_CHANNEL_CAPACITY);
        let (response_sender, _response_receiver) = sync_channel(1);
        sender.try_send(ManagerCommand::ForceResync(response_sender))?;
        let signals = ManagerSignals::new();
        let shutdown = Notify::new();
        let runtime = RuntimeBuilder::new_current_thread().enable_time().build()?;

        let selected = runtime.block_on(await_send_or_control(
            std::future::pending::<Result<(), ManagedTransportFailure>>(),
            &mut commands,
            &signals,
            &shutdown,
        ));

        assert!(matches!(
            selected,
            SendWait::Command(Some(ManagerCommand::ForceResync(_)))
        ));
        Ok(())
    }

    #[test]
    fn wake_and_reachability_hints_are_coalesced_to_latest_state() {
        let signals = ManagerSignals::new();
        for _ in 0..10_000 {
            signals.request_wake();
        }
        assert!(signals.wake_pending.swap(false, Ordering::AcqRel));
        assert!(!signals.wake_pending.swap(false, Ordering::AcqRel));

        for available in [false, true].into_iter().cycle().take(10_000) {
            signals.set_network_available(available);
        }
        assert!(signals.network_changed.swap(false, Ordering::AcqRel));
        assert!(signals.network_available.load(Ordering::Acquire));
        assert!(!signals.network_changed.swap(false, Ordering::AcqRel));
    }

    #[test]
    fn control_queue_is_bounded() {
        let (sender, _commands) = async_mpsc::channel(CONTROL_CHANNEL_CAPACITY);
        for _ in 0..CONTROL_CHANNEL_CAPACITY {
            let (response_sender, _response_receiver) = sync_channel(1);
            assert!(
                sender
                    .try_send(ManagerCommand::ForceResync(response_sender))
                    .is_ok()
            );
        }
        let (response_sender, _response_receiver) = sync_channel(1);
        assert!(matches!(
            sender.try_send(ManagerCommand::ForceResync(response_sender)),
            Err(async_mpsc::error::TrySendError::Full(_))
        ));
    }

    #[test]
    fn wake_does_not_preempt_a_valid_send() -> Result<(), Box<dyn std::error::Error>> {
        let (_sender, mut commands) = async_mpsc::channel(CONTROL_CHANNEL_CAPACITY);
        let signals = ManagerSignals::new();
        signals.request_wake();
        let shutdown = Notify::new();
        let runtime = RuntimeBuilder::new_current_thread().enable_time().build()?;

        let selected = runtime.block_on(await_send_or_control(
            async { Ok(()) },
            &mut commands,
            &signals,
            &shutdown,
        ));

        assert!(matches!(selected, SendWait::Written));
        assert!(signals.wake_pending.load(Ordering::Acquire));
        Ok(())
    }

    #[test]
    fn network_down_preempts_a_stalled_send() -> Result<(), Box<dyn std::error::Error>> {
        let (_sender, mut commands) = async_mpsc::channel(CONTROL_CHANNEL_CAPACITY);
        let signals = ManagerSignals::new();
        signals.set_network_available(false);
        let shutdown = Notify::new();
        let runtime = RuntimeBuilder::new_current_thread().enable_time().build()?;

        let selected = runtime.block_on(await_send_or_control(
            std::future::pending::<Result<(), ManagedTransportFailure>>(),
            &mut commands,
            &signals,
            &shutdown,
        ));

        assert!(matches!(selected, SendWait::NetworkUnavailable));
        Ok(())
    }

    #[test]
    fn failed_force_resync_preserves_existing_io_state() -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("managed-resync-order")?;
        let account = Arc::new(AccountRuntimeHandle::open(
            vec![1_u8; 32],
            directory.display().to_string(),
        )?);
        let initial = account.start_account_transport("resync-order".to_owned())?;
        let (_command_sender, command_receiver) = async_mpsc::channel(CONTROL_CHANNEL_CAPACITY);
        let signals = Arc::new(ManagerSignals::new());
        let snapshots = Arc::new(SnapshotState::new(initial.revision));
        let mut manager = TransportManager::new(
            Arc::clone(&account),
            command_receiver,
            signals,
            snapshots,
            Arc::new(AtomicBool::new(false)),
            Arc::new(Notify::new()),
            initial,
        );
        assert!(manager.pending_connect.is_some());
        account.close_account();
        let (response_sender, response_receiver) = sync_channel(1);

        manager.handle_command(ManagerCommand::ForceResync(response_sender));
        assert_eq!(response_receiver.recv()?, Err(SoftchatError::AccountClosed),);
        assert!(manager.pending_connect.is_some());

        drop(manager);
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn shutdown_interrupts_a_stalled_tls_handshake_and_joins_the_worker()
    -> Result<(), Box<dyn std::error::Error>> {
        let listener = std::net::TcpListener::bind(("127.0.0.1", 0))?;
        listener.set_nonblocking(true)?;
        let relay_url = format!("wss://127.0.0.1:{}/", listener.local_addr()?.port());
        let directory = temporary_directory("managed-shutdown")?;
        let account = Arc::new(AccountRuntimeHandle::open(
            vec![1_u8; 32],
            directory.display().to_string(),
        )?);
        account.add_account_relay(relay_url.clone(), 1_700_000_000)?;
        account.set_account_active_relay(relay_url, 1_700_000_001)?;

        let transport = ManagedAccountTransport::new(Arc::clone(&account))?;
        let accept_deadline = StdInstant::now() + Duration::from_secs(5);
        let accepted = loop {
            match listener.accept() {
                Ok((stream, _)) => break stream,
                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                    if StdInstant::now() >= accept_deadline {
                        return Err("managed transport did not open the test socket".into());
                    }
                    thread::sleep(Duration::from_millis(10));
                }
                Err(error) => return Err(error.into()),
            }
        };

        let shutdown_started = StdInstant::now();
        transport.shutdown();
        assert!(shutdown_started.elapsed() < Duration::from_secs(5));
        assert!(transport.snapshot().stopped);

        drop(accepted);
        drop(transport);
        account.close_account();
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }
}