soulseek-rs-modded/soulseek-rs-lib/src/actor/peer_actor.rs

use crate::actor::{Actor, ActorHandle, ConnectionState};
use crate::client::ClientOperation;
use crate::dispatcher::MessageDispatcher;
use crate::message::peer::{
    FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueRequest,
    PlaceInQueueResponse, QueueUploadHandler, SharedDirectory,
    SharedFileListResponseHandler, TransferRequest, TransferResponse,
    UploadFailedHandler,
};
use crate::message::server::MessageFactory;
use crate::message::{Handlers, Message, MessageReader, MessageType};
use crate::peer::Peer;
use crate::types::{Download, SearchResult, Transfer};
use crate::utils::lock::RwLockExt;
use crate::{debug, error, trace, warn};

use std::io::{self, Error, Write};
use std::net::TcpStream;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};

#[derive(Debug, Clone)]
pub enum PeerMessage {
    SendMessage(Message),
    FileSearchResult(SearchResult),
    TransferRequest(Transfer),
    UploadFailed(String, String),
    TransferResponse {
        token: u32,
        allowed: bool,
        reason: Option<String>,
    },
    PlaceInQueueResponse {
        filename: String,
        place: u32,
    },
    SetUsername(String),
    QueueUpload(String),
    RequestTransfer(Download),
    /// A peer queued one of our shared files for download (they sent us code 43).
    IncomingQueueUpload(String),
    /// A peer asked where their queued file sits (they sent us code 51).
    IncomingPlaceInQueueRequest(String),
    /// A peer asked to browse our shared files (they sent us code 4).
    ShareListRequested,
    /// A peer we are browsing sent us their shared-file listing (code 5).
    ShareListReceived(Vec<SharedDirectory>),
    /// Offer the queued file to that peer: send an upload TransferRequest.
    ServeUpload {
        token: u32,
        filename: String,
        size: u64,
    },
    ProcessRead,
}

pub struct PeerActor {
    peer: Arc<RwLock<Peer>>,
    stream: Option<TcpStream>,
    connection_state: ConnectionState,
    reader: MessageReader,
    client_channel: Sender<ClientOperation>,
    self_handle: Option<ActorHandle<PeerMessage>>,
    dispatcher: Option<MessageDispatcher<PeerMessage>>,
    dispatcher_receiver: Option<Receiver<PeerMessage>>,
    queued_messages: Vec<PeerMessage>,
    own_username: String,
    /// True when we initiated this connection (no stream supplied at
    /// construction), so we must send a `PeerInit` once connected.
    outbound: bool,
    /// Set once the connection is up. A pre-established failure on an outbound
    /// connection means the peer is unreachable (likely firewalled).
    established: bool,
    /// Set once this actor has reported a terminal outcome (disconnect or
    /// connect-failure). Guards against a second notification — e.g. when a
    /// replaced actor is later stopped — which would otherwise alias and evict
    /// a newer actor registered under the same username.
    disconnect_reported: bool,
    /// Registry-assigned unique id, echoed in terminal notifications so the
    /// client evicts only this actor and never a newer namesake.
    id: u64,
    /// Transfer tokens for uploads we are serving to this peer. A TransferResponse
    /// for one of these is our upload being accepted, not a download offer.
    serving_tokens: std::collections::HashSet<u32>,
    /// Bytes queued for the peer that the socket would not take yet. The stream
    /// is non-blocking (so socket reads never starve the mailbox), so a message
    /// larger than the send buffer — a full share listing is easily megabytes —
    /// only partly fits; the rest drains on later ticks.
    pending_write: Vec<u8>,
    /// When bytes last moved on this connection, in either direction. A peer
    /// that goes quiet is eventually reaped: nothing else ever closes an idle
    /// control connection, and each one holds a descriptor and a thread for
    /// the life of the process.
    last_activity: Instant,
}

/// Cap on queued outbound bytes. A peer that stops reading must not be able to
/// grow this without bound; dropping that connection is the right answer.
const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024;

/// How long a connected peer may stay silent before its control connection is
/// closed. Transfers run on their own sockets, and a peer that wants us again
/// simply reconnects — every serious client reaps idle peers this way. Five
/// minutes comfortably outlives a search's chatter and queue updates.
const IDLE_DISCONNECT: Duration = Duration::from_mins(5);

impl PeerActor {
    #[must_use]
    pub fn new(
        peer: Peer,
        stream: Option<TcpStream>,
        reader: Option<MessageReader>,
        client_channel: Sender<ClientOperation>,
        own_username: String,
        id: u64,
    ) -> Self {
        let outbound = stream.is_none();
        let connection_state = if stream.is_some() {
            ConnectionState::Connected
        } else {
            ConnectionState::Disconnected
        };

        Self {
            peer: Arc::new(RwLock::new(peer)),
            stream,
            connection_state,
            reader: reader.unwrap_or_default(),
            client_channel,
            self_handle: None,
            dispatcher: None,
            dispatcher_receiver: None,
            queued_messages: Vec::new(),
            own_username,
            outbound,
            established: false,
            disconnect_reported: false,
            id,
            serving_tokens: std::collections::HashSet::new(),
            pending_write: Vec::new(),
            last_activity: Instant::now(),
        }
    }

    pub fn set_self_handle(&mut self, handle: ActorHandle<PeerMessage>) {
        self.self_handle = Some(handle);
    }

    fn peer_username(&self) -> String {
        match self.peer.read_safe() {
            Ok(p) => p.username.clone(),
            Err(e) => {
                error!("[peer_actor] peer lock poisoned: {}", e);
                "<unknown>".to_string()
            }
        }
    }

    fn peer_snapshot(&self) -> Option<Peer> {
        match self.peer.read_safe() {
            Ok(p) => Some(p.clone()),
            Err(e) => {
                error!("[peer_actor] peer lock poisoned: {}", e);
                None
            }
        }
    }

    fn initialize_dispatcher(&mut self) {
        let (dispatcher_sender, dispatcher_receiver) =
            std::sync::mpsc::channel::<PeerMessage>();

        self.dispatcher_receiver = Some(dispatcher_receiver);

        let mut handlers = Handlers::new();
        handlers.register_handler(FileSearchResponse);
        handlers.register_handler(TransferRequest);
        handlers.register_handler(TransferResponse);
        handlers.register_handler(GetShareFileList);
        handlers.register_handler(UploadFailedHandler);
        handlers.register_handler(PlaceInQueueRequest);
        handlers.register_handler(PlaceInQueueResponse);
        handlers.register_handler(QueueUploadHandler);
        handlers.register_handler(SharedFileListResponseHandler);
        handlers.register_handler(PeerInit);

        self.dispatcher = Some(MessageDispatcher::new(
            "peer".to_string(),
            dispatcher_sender,
            handlers,
        ));
    }

    fn process_dispatcher_messages(&mut self) {
        let messages: Vec<PeerMessage> = self
            .dispatcher_receiver
            .as_ref()
            .map_or_else(Vec::new, |receiver| receiver.try_iter().collect());

        for msg in &messages {
            self.handle_message(msg.clone());
        }
    }

    fn handle_message(&mut self, msg: PeerMessage) {
        if matches!(self.connection_state, ConnectionState::Connecting { .. }) {
            match &msg {
                PeerMessage::SetUsername(_) | PeerMessage::ProcessRead => {}
                _ => {
                    self.queued_messages.push(msg);
                    return;
                }
            }
        }

        match msg {
            PeerMessage::SendMessage(message) => {
                self.send_message(message);
            }
            PeerMessage::FileSearchResult(file_search) => {
                self.handle_file_search_result(file_search);
            }
            PeerMessage::TransferRequest(transfer) => {
                self.handle_transfer_request(transfer);
            }
            PeerMessage::TransferResponse {
                token,
                allowed,
                reason,
            } => {
                self.handle_transfer_response(token, allowed, reason);
            }
            PeerMessage::PlaceInQueueResponse { filename, place } => {
                self.handle_place_in_queue_response(filename, place);
            }
            PeerMessage::SetUsername(username) => {
                self.handle_set_username(username);
            }
            PeerMessage::QueueUpload(filename) => {
                let message =
                    MessageFactory::build_queue_upload_message(&filename);
                self.send_message(message);
            }
            PeerMessage::IncomingQueueUpload(filename) => {
                self.handle_incoming_queue_upload(filename);
            }
            PeerMessage::IncomingPlaceInQueueRequest(filename) => {
                let requester_key = self.peer_username();
                if let Err(e) = self.client_channel.send(
                    ClientOperation::PlaceInQueueRequested {
                        requester_key,
                        filename,
                    },
                ) {
                    error!(
                        "[peer_actor] forward IncomingPlaceInQueueRequest: {}",
                        e
                    );
                }
            }
            PeerMessage::ServeUpload {
                token,
                filename,
                size,
            } => {
                self.handle_serve_upload(token, filename, size);
            }
            PeerMessage::ShareListRequested => {
                self.handle_share_list_requested();
            }
            PeerMessage::ShareListReceived(directories) => {
                self.handle_share_list_received(directories);
            }
            PeerMessage::RequestTransfer(download) => {
                let message = MessageFactory::build_transfer_request_message(
                    &download.filename,
                    download.token,
                );
                self.send_message(message);
            }
            PeerMessage::ProcessRead => {
                self.process_read();
            }
            PeerMessage::UploadFailed(username, filename) => {
                self.handle_upload_failed(username, filename);
            }
        }
    }

    fn handle_file_search_result(&self, file_search: SearchResult) {
        if let Err(e) = self
            .client_channel
            .send(ClientOperation::SearchResult(file_search))
        {
            error!(
                "[peer:{}] failed to forward search result: {}",
                self.peer_username(),
                e
            );
        }
    }

    /// A peer offers us a file. Record the peer's transfer token with the
    /// client, and let *that* step send the TransferResponse.
    ///
    /// The response is what invites the peer's file connection, and that
    /// connection is matched by the peer's token — so replying here would race
    /// the client loop that records the token. Under load the client loop is
    /// the one that loses, the file connection finds no download, and the
    /// transfer is dropped with the download stuck on Queued forever.
    fn handle_transfer_request(&self, transfer: Transfer) {
        let username = self.peer_username();
        debug!("[peer:{}] TransferRequest for {}", username, transfer.token);

        if let Err(e) =
            self.client_channel
                .send(ClientOperation::UpdateDownloadTokens(
                    transfer,
                    username.clone(),
                ))
        {
            error!(
                "[peer:{}] failed to send UpdateDownloadTokens: {}",
                username, e
            );
        }
    }

    fn handle_transfer_response(
        &mut self,
        token: u32,
        allowed: bool,
        reason: Option<String>,
    ) {
        let username = self.peer_username();
        debug!(
            "[peer:{}] transfer response token: {} allowed: {}",
            username, token, allowed
        );

        // If this token is one of our uploads, the peer just accepted
        // our offer — start streaming. This leaves the download path
        // (every other token) byte-for-byte unchanged.
        if self.serving_tokens.remove(&token) {
            if allowed {
                let _ = self
                    .client_channel
                    .send(ClientOperation::StartUpload { token });
            }
            return;
        }

        if allowed {
            debug!(
                "[peer:{}] Transfer allowed, ready to connect with token {:}",
                username, token
            );
            let Some(peer_snapshot) = self.peer_snapshot() else {
                return;
            };
            if let Err(e) =
                self.client_channel.send(ClientOperation::DownloadFromPeer(
                    token,
                    peer_snapshot,
                    allowed,
                ))
            {
                error!(
                    "[peer:{}] failed to send DownloadFromPeer: {}",
                    username, e
                );
            }
        } else if let Some(reason_text) = reason {
            debug!(
                "[peer:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...",
                username, reason_text, token
            );
        }
    }

    fn handle_place_in_queue_response(&self, filename: String, place: u32) {
        let username = self.peer_username();
        debug!(
            "[peer:{}] Place in queue response - file: {}, place: {}",
            username, filename, place
        );
        if let Err(e) =
            self.client_channel
                .send(ClientOperation::PlaceInQueueUpdate {
                    username: username.clone(),
                    filename,
                    place,
                })
        {
            error!(
                "[peer:{}] failed to forward PlaceInQueueUpdate: {}",
                username, e
            );
        }
    }

    fn handle_set_username(&self, username: String) {
        trace!("[peer:{}] SetUsername: {}", self.peer_username(), username);
        match self.peer.write_safe() {
            Ok(mut p) => p.username = username,
            Err(e) => {
                error!("[peer_actor] SetUsername write: {}", e);
            }
        }
    }

    fn handle_incoming_queue_upload(&self, filename: String) {
        // A peer wants to download one of our shared files. Ask the
        // client (which owns the shares) to prepare the upload.
        let requester_key = self.peer_username();
        if let Err(e) = self.client_channel.send(ClientOperation::QueueUpload {
            requester_key,
            filename,
        }) {
            error!("[peer_actor] forward IncomingQueueUpload: {}", e);
        }
    }

    fn handle_serve_upload(&mut self, token: u32, filename: String, size: u64) {
        self.serving_tokens.insert(token);
        let message = MessageFactory::build_upload_transfer_request(
            &filename, token, size,
        );
        self.send_message(message);
    }

    fn handle_share_list_requested(&self) {
        let requester_key = self.peer_username();
        if let Err(e) = self
            .client_channel
            .send(ClientOperation::ShareListRequested { requester_key })
        {
            error!("[peer_actor] forward ShareListRequested: {}", e);
        }
    }

    fn handle_share_list_received(&self, directories: Vec<SharedDirectory>) {
        let username = self.peer_username();
        if let Err(e) =
            self.client_channel.send(ClientOperation::BrowseResult {
                username,
                directories,
            })
        {
            error!("[peer_actor] forward BrowseResult: {}", e);
        }
    }

    fn handle_upload_failed(&self, username: String, filename: String) {
        if let Err(e) = self
            .client_channel
            .send(ClientOperation::UploadFailed(username, filename))
        {
            error!("[peer_actor] failed to forward UploadFailed: {}", e);
        }
    }

    fn process_read(&mut self) {
        if self.reader.buffer_len() > 0 {
            self.extract_and_process_messages();
        }

        {
            let Some(stream) = self.stream.as_mut() else {
                return;
            };

            match self.reader.read_from_socket(stream) {
                Ok(()) => {
                    self.last_activity = Instant::now();
                }
                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
                Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
                    if let Ok(peer_lock) = self.peer.read_safe() {
                        debug!(
                            "Read operation timed out for peer actor {:}:{:}",
                            peer_lock.host, peer_lock.port
                        );
                    }
                }
                // A peer hanging up a control connection it had already
                // established is ordinary: other clients close idle peer
                // sockets while a transfer runs on its own connection. Report
                // the clean disconnect it is — an error here fails every
                // download queued with that peer, the one streaming included.
                // A connection that never established is the other story: the
                // peer accepted and vanished, which is what the server-brokered
                // fallback exists for, so that one keeps the error path.
                Err(ref e)
                    if e.kind() == io::ErrorKind::UnexpectedEof
                        && self.established =>
                {
                    debug!(
                        "[peer:{}] peer closed the connection",
                        self.peer_username()
                    );
                    self.disconnect();
                    return;
                }
                Err(e) => {
                    let username = self.peer_username();
                    error!(
                        "[peer:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.",
                        username,
                        e,
                        e.kind()
                    );
                    self.disconnect_with_error(e);
                    return;
                }
            }
        }
        self.extract_and_process_messages();
    }

    fn extract_and_process_messages(&mut self) {
        let username = self.peer_username();
        let mut extracted_count = 0;
        loop {
            match self.reader.extract_message() {
                Ok(Some(mut message)) => {
                    extracted_count += 1;
                    trace!(
                        "[peer:{}] ← Message #{}: {:?}",
                        username,
                        extracted_count,
                        message
                            .get_message_name(
                                MessageType::Peer,
                                u32::from(message.get_message_code())
                            )
                            .map_err(|e| e.to_string())
                    );
                    if let Some(ref dispatcher) = self.dispatcher {
                        dispatcher.dispatch(&mut message);
                    } else {
                        warn!("[peer:{}] No dispatcher available!", username);
                    }
                }
                Err(e) => {
                    warn!(
                        "[peer:{}] Error extracting message: {}. Disconnecting peer.",
                        username, e
                    );
                    self.disconnect_with_error(e);
                    return;
                }
                Ok(None) => {
                    break;
                }
            }
        }

        self.process_dispatcher_messages();
    }

    fn send_message(&mut self, message: Message) {
        let username = self.peer_username();
        if self.stream.is_none() {
            error!("Cannot send message: stream is None");
            return;
        }

        trace!(
            "[peer:{}] ➡ {:?}",
            username,
            message
                .get_message_name(
                    MessageType::Peer,
                    u32::from_le_bytes(
                        message.get_slice(0, 4).try_into().unwrap_or_default()
                    )
                )
                .map_err(|e| e.to_string())
        );

        // Queue then drain, so messages always leave in order even when an
        // earlier one is still partly buffered.
        self.pending_write.extend_from_slice(&message.get_buffer());
        self.flush_pending_write();
    }

    /// Push as much of `pending_write` into the socket as it will take. A
    /// non-blocking socket refuses the rest with `WouldBlock`; that is normal
    /// backpressure, not a failure, so the remainder stays queued for the next
    /// tick. Only a real I/O error disconnects.
    fn flush_pending_write(&mut self) {
        if self.pending_write.is_empty() {
            return;
        }

        let username = self.peer_username();
        let Some(stream) = self.stream.as_mut() else {
            return;
        };

        let mut written = 0;
        let outcome = loop {
            match stream.write(&self.pending_write[written..]) {
                Ok(0) => {
                    break Err(Error::new(
                        io::ErrorKind::WriteZero,
                        "peer stopped accepting data",
                    ));
                }
                Ok(n) => {
                    written += n;
                    if written == self.pending_write.len() {
                        break Ok(());
                    }
                }
                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                    break Ok(());
                }
                Err(e) => break Err(e),
            }
        };
        self.pending_write.drain(..written);
        if written > 0 {
            self.last_activity = Instant::now();
        }

        if let Err(e) = outcome {
            error!(
                "[peer:{}] Error writing message: {}. Disconnecting.",
                username, e
            );
            self.disconnect_with_error(e);
            return;
        }

        // Backpressure is normal, but what a peer has still not taken after we
        // pushed everything we could is a peer that is not reading at all.
        if self.pending_write.len() > MAX_PENDING_WRITE {
            error!(
                "[peer:{}] Outbound queue stuck above {} bytes. Disconnecting.",
                username, MAX_PENDING_WRITE
            );
            self.disconnect_with_error(Error::new(
                io::ErrorKind::WriteZero,
                "peer is not draining our outbound queue",
            ));
        }
    }

    fn disconnect_with_error(&mut self, error: Error) {
        let username = self.peer_username();
        // Include the cause: a failed outbound dial reports nothing else, so
        // without it an unreachable peer is indistinguishable from a clean
        // disconnect in the logs.
        debug!("[peer:{}] disconnect: {}", username, error);

        self.stream.take();
        self.connection_state = ConnectionState::Disconnected;

        if self.disconnect_reported {
            return;
        }
        self.disconnect_reported = true;

        // A direct outbound connection that never established means the peer is
        // unreachable (likely firewalled): signal a connect failure so the
        // client can fall back to server-brokered connect. Anything else is a
        // normal disconnect.
        let op = if self.outbound && !self.established {
            ClientOperation::PeerConnectFailed(self.id, username)
        } else {
            ClientOperation::PeerDisconnected(
                self.id,
                username,
                Some(error.into()),
            )
        };

        if let Err(e) = self.client_channel.send(op) {
            error!("Failed to send disconnect notification: {}", e);
        }
    }
    fn disconnect(&mut self) {
        let username = self.peer_username();
        debug!("[peer:{}] disconnect", username);

        self.stream.take();
        self.connection_state = ConnectionState::Disconnected;

        if self.disconnect_reported {
            return;
        }
        self.disconnect_reported = true;

        if let Err(e) = self
            .client_channel
            .send(ClientOperation::PeerDisconnected(self.id, username, None))
        {
            error!("Failed to send disconnect notification: {}", e);
        }
    }

    fn initiate_connection(&mut self) {
        let (username, host, port) = match self.peer.read_safe() {
            Ok(peer) => (peer.username.clone(), peer.host.clone(), peer.port),
            Err(e) => {
                error!("[peer_actor] initiate_connection peer lock: {}", e);
                return;
            }
        };

        let socket_addr =
            format!("{host}:{port}").parse::<std::net::SocketAddr>();

        let addr = match socket_addr {
            Ok(addr) => addr,
            Err(e) => {
                error!(
                    "[peer:{}] Invalid socket address {}:{} - {}",
                    username, host, port, e
                );
                self.disconnect_with_error(io::Error::new(
                    io::ErrorKind::InvalidInput,
                    e,
                ));
                return;
            }
        };

        // Use connect_timeout to prevent blocking the thread for too long
        let timeout = Duration::from_secs(5);
        let stream = match TcpStream::connect_timeout(&addr, timeout) {
            Ok(stream) => stream,
            Err(e) => {
                self.disconnect_with_error(e);
                return;
            }
        };

        if let Err(e) = stream.set_nonblocking(true) {
            error!("[peer:{}] Failed to set non-blocking: {}", username, e);
            self.disconnect_with_error(e);
            return;
        }
        stream.set_nodelay(true).ok();
        self.stream = Some(stream);
        self.connection_state = ConnectionState::Connecting {
            since: Instant::now(),
        };
    }

    fn check_connection_status(&mut self) {
        let ConnectionState::Connecting { since } = self.connection_state
        else {
            return;
        };

        let username = self.peer_username();

        if since.elapsed() > Duration::from_secs(20) {
            error!("[peer:{}] Connection timeout after 20 seconds", username);
            self.disconnect_with_error(io::Error::new(
                io::ErrorKind::TimedOut,
                "Connection timeout",
            ));
            return;
        }

        let Some(ref stream) = self.stream else {
            return;
        };

        match stream.peer_addr() {
            Ok(_) => {
                self.connection_state = ConnectionState::Connected;
                self.on_connection_established();
            }
            Err(ref e) if e.kind() == io::ErrorKind::NotConnected => {}
            Err(e) => {
                error!("[peer:{}] Connection failed: {}", username, e);
                self.disconnect_with_error(e);
            }
        }
    }

    fn on_connection_established(&mut self) {
        let (username, connection_type, token) = match self.peer.read_safe() {
            Ok(peer) => (
                peer.username.clone(),
                peer.connection_type.clone(),
                peer.token,
            ),
            Err(e) => {
                error!(
                    "[peer_actor] on_connection_established peer lock: {}",
                    e
                );
                return;
            }
        };

        self.established = true;

        let Some(ref mut stream) = self.stream else {
            return;
        };

        // Connections we initiated must announce themselves; inbound peers
        // already sent us theirs, so we stay silent for them. A direct dial
        // (resolved via GetPeerAddress, token None) sends a PeerInit; a dial we
        // make because the server asked us to (a brokered ConnectToPeer carries
        // the server's token) must instead PierceFirewall with that token so
        // the remote peer can correlate its own request. `get_buffer` prepends
        // the length prefix the peer's MessageReader expects.
        if self.outbound {
            let handshake = match token {
                Some(server_token) => {
                    MessageFactory::build_pierce_firewall_message(server_token)
                }
                None => MessageFactory::build_peer_init_message(
                    &self.own_username,
                    connection_type,
                    0,
                ),
            };
            if let Err(e) = stream.write_all(&handshake.get_buffer()) {
                error!(
                    "[peer:{}] Failed to send outbound handshake: {}",
                    username, e
                );
                self.disconnect_with_error(e);
                return;
            }
        }

        self.initialize_dispatcher();

        let queued = std::mem::take(&mut self.queued_messages);
        for msg in queued {
            self.handle_message(msg);
        }

        // Tell the client the control connection is live so any downloads
        // queued for this peer can now be requested over a handshaken stream.
        if self.outbound {
            let _ = self
                .client_channel
                .send(ClientOperation::PeerConnected(username));
        }

        if let Some(ref handle) = self.self_handle {
            handle.send(PeerMessage::ProcessRead).ok();
        }

        self.process_read();
    }
}

impl Actor for PeerActor {
    type Message = PeerMessage;

    fn handle(&mut self, msg: Self::Message) {
        self.handle_message(msg);
    }

    fn on_start(&mut self) {
        if self.stream.is_none() {
            self.initiate_connection();
        } else {
            self.connection_state = ConnectionState::Connected;
            self.on_connection_established();
        }
    }

    fn on_stop(&mut self) {
        let username = self.peer_username();
        trace!("[peer:{}] actor stopping", username);
        self.disconnect();
    }

    fn tick(&mut self) {
        match self.connection_state {
            ConnectionState::Connecting { .. } => {
                self.check_connection_status();
            }
            ConnectionState::Connected => {
                if self.stream.is_some() {
                    // a backlogged message drains one socket buffer
                    // per 100ms tick. Enough for a share listing; if a future
                    // message stream needs more throughput, wake the actor as
                    // soon as the socket is writable instead of on the tick.
                    self.flush_pending_write();
                    self.process_read();
                    self.flush_pending_write();

                    if self.stream.is_some()
                        && self.last_activity.elapsed() > IDLE_DISCONNECT
                    {
                        debug!(
                            "[peer:{}] idle for {}s, closing",
                            self.peer_username(),
                            IDLE_DISCONNECT.as_secs()
                        );
                        self.disconnect();
                    }
                }
            }
            ConnectionState::Disconnected => {}
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::peer::{ConnectionType, Peer};
    use std::net::{TcpListener, TcpStream};
    use std::sync::mpsc::Receiver;

    /// A connected inbound actor over a real loopback socket, plus the far
    /// end (kept alive so reads see silence, not EOF).
    fn connected_actor() -> (PeerActor, Receiver<ClientOperation>, TcpStream) {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let stream = TcpStream::connect(addr).unwrap();
        stream.set_nonblocking(true).unwrap();
        let (far_end, _) = listener.accept().unwrap();

        let (mut actor, rx) = make_actor(Some(stream));
        actor.on_start();
        (actor, rx, far_end)
    }

    fn make_actor(
        stream: Option<TcpStream>,
    ) -> (PeerActor, Receiver<ClientOperation>) {
        let (tx, rx) = std::sync::mpsc::channel();
        let peer = Peer::new(
            "bob".to_string(),
            ConnectionType::P,
            "127.0.0.1".to_string(),
            0,
            None,
            0,
            0,
            0,
        );
        let actor = PeerActor::new(peer, stream, None, tx, "me".to_string(), 7);
        (actor, rx)
    }

    #[test]
    fn a_timed_out_connect_parks_the_actor_in_disconnected() {
        let (mut actor, rx) = make_actor(None);
        actor.connection_state = ConnectionState::Connecting {
            since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(),
        };

        actor.tick();

        assert!(
            matches!(actor.connection_state, ConnectionState::Disconnected),
            "a timed-out connect must leave Connecting"
        );
        match rx.try_recv() {
            Ok(ClientOperation::PeerConnectFailed(7, username)) => {
                assert_eq!(username, "bob");
            }
            other => panic!("expected PeerConnectFailed, got {other:?}"),
        }
    }

    #[test]
    fn tick_reaps_a_peer_idle_past_the_deadline() {
        let (mut actor, rx, _far_end) = connected_actor();

        actor.tick();
        assert!(actor.stream.is_some(), "a fresh connection stays open");

        actor.last_activity = Instant::now()
            .checked_sub(IDLE_DISCONNECT + Duration::from_secs(1))
            .unwrap();
        actor.tick();

        assert!(actor.stream.is_none(), "an idle stream must be closed");
        match rx.try_recv() {
            Ok(ClientOperation::PeerDisconnected(7, username, None)) => {
                assert_eq!(username, "bob");
            }
            other => panic!("expected a clean PeerDisconnected, got {other:?}"),
        }
    }
}