peerspeak/src/presence_net.rs

//! Presence control transport (W7 P4) — the I/O edge for the friends-only idle
//! listener. The *policy* (who we answer, what we report) lives in pure form in
//! [`crate::presence`]; this module is just the wire: bind/probe/serve a
//! ping→pong over a dedicated ALPN, adapted from pixelpass's proven control plane
//! (same iroh version).
//!
//! Shape: **request/response, one exchange per connection.** A prober opens a
//! bi-stream, writes a [`ControlMsg::Ping`], finishes its send side, and reads
//! the peer's [`ControlMsg::Pong`] back. The server accepts, reads the ping, asks
//! its injected `handler` (which encapsulates [`crate::presence::should_answer`]
//! plus building the pong from current room state) what to reply, writes it, and
//! closes. A non-friend / invisible peer gets `None`, so the stream is closed
//! with no reply and the listener never reveals anything to a stranger.
//!
//! **Deferred to a later slice (the hard integration):** spawning `serve` on a
//! persistent endpoint that lives OUTSIDE the per-join room session, and the
//! outbound ping scheduler. The endpoint fork was DECIDED 2026-06-15 by a
//! throwaway spike: a **single persistent endpoint** (option b), NOT a second
//! always-on endpoint sharing our id (option a). The spike proved two endpoints
//! sharing one `SecretKey` can't coexist — inbound connections all land on one
//! instance and the other's ALPN fails the QUIC handshake ("error 120"). So this
//! `serve` will run as the FRIENDS_ALPN handler on the app's one persistent
//! `Router`; it still works against whatever `Endpoint` it's handed.

use crate::presence::ControlMsg;
use anyhow::{Context, Result, bail};
use iroh::endpoint::{Incoming, VarInt};
use iroh::{Endpoint, EndpointAddr, EndpointId};
use std::sync::Arc;
use std::time::Duration;

/// ALPN for the friends presence/control plane. Separate from the audio/gossip
/// ALPNs so a control dial never lands on a bare room endpoint and vice versa.
pub const FRIENDS_ALPN: &[u8] = crate::protocol::FRIENDS_ALPN;

/// Upper bound on a single control message — generous for a Pong carrying a
/// member ticket (~300 chars), but rejects a peer trying to make us buffer a
/// huge blob.
const MAX_MSG: usize = 64 * 1024;

/// Bound on each network phase so a half-dead peer or relay can't park us.
const IO_TIMEOUT: Duration = Duration::from_secs(10);

fn encode(msg: &ControlMsg) -> Result<Vec<u8>> {
    serde_json::to_vec(msg).context("failed to encode control message")
}

fn decode(bytes: &[u8]) -> Result<ControlMsg> {
    serde_json::from_slice(bytes).context("failed to decode control message")
}

/// Probe `peer` for presence: send a `Ping`, return their authenticated id and
/// `Pong`. An error means no usable reply (offline / unreachable / refused /
/// malformed) — the caller treats that as "appears offline". `peer` is usually a
/// bare [`EndpointId`] (friends store the stable id); a full [`EndpointAddr`] is
/// also accepted (and used by hermetic tests).
pub async fn probe(endpoint: &Endpoint, peer: impl Into<EndpointAddr>) -> Result<(EndpointId, ControlMsg)> {
    let conn = tokio::time::timeout(IO_TIMEOUT, endpoint.connect(peer, FRIENDS_ALPN))
        .await
        .context("timed out connecting to peer")?
        .context("failed to connect to peer")?;
    let from = conn.remote_id();

    let io = async {
        let (mut send, mut recv) = conn.open_bi().await.context("failed to open control stream")?;
        send.write_all(&encode(&ControlMsg::Ping)?)
            .await
            .context("failed to write ping")?;
        send.finish().context("failed to finish ping stream")?;
        let bytes = recv
            .read_to_end(MAX_MSG)
            .await
            .context("peer closed the stream without replying")?;
        decode(&bytes)
    };

    let result = tokio::time::timeout(IO_TIMEOUT, io)
        .await
        .context("timed out awaiting pong")?;
    conn.close(VarInt::from_u32(0), b"done");
    result.map(|msg| (from, msg))
}

/// A reply policy: given the *authenticated* remote id, decide whether and how to
/// answer a ping. `None` = don't answer (not a friend, or we're invisible). This
/// is where [`crate::presence::should_answer`] + the Pong contents are wired in
/// by the caller; keeping it injected means this transport carries no policy.
pub type Handler = Arc<dyn Fn(EndpointId) -> Option<ControlMsg> + Send + Sync>;

/// Run the presence accept loop on `endpoint`, replying to each ping per
/// `handler`. Returns when the endpoint stops accepting (i.e. it was closed).
pub async fn serve(endpoint: Endpoint, handler: Handler) {
    while let Some(incoming) = endpoint.accept().await {
        let handler = handler.clone();
        tokio::spawn(async move {
            if let Err(e) = handle(incoming, handler).await {
                crate::log_msg(&format!("presence: inbound connection failed: {e:#}"));
            }
        });
    }
}

async fn handle(incoming: Incoming, handler: Handler) -> Result<()> {
    let conn = incoming.await.context("inbound connection failed")?;
    exchange(&conn, &handler).await
}

/// One ping→pong exchange on an already-accepted connection: read the ping,
/// ask the policy, reply (or reveal nothing), close. Shared by the standalone
/// [`serve`] loop and the [`FriendsProtocol`] router handler.
async fn exchange(conn: &iroh::endpoint::Connection, handler: &Handler) -> Result<()> {
    // The authenticated remote id — NOT anything the peer puts in the payload.
    let from = conn.remote_id();
    let Some(reply) = handler(from) else {
        conn.close(VarInt::from_u32(0), b"not authorized");
        return Ok(());
    };

    let io = async {
        let (mut send, mut recv) = conn.accept_bi().await.context("failed to accept stream")?;
        let bytes = recv.read_to_end(MAX_MSG).await.context("failed to read ping")?;
        match decode(&bytes)? {
            ControlMsg::Ping => {}
            other => bail!("expected a ping, got {other:?}"),
        }
        send.write_all(&encode(&reply)?)
            .await
            .context("failed to write pong")?;
        send.finish().context("failed to finish reply stream")?;
        Ok::<_, anyhow::Error>(())
    };

    tokio::time::timeout(IO_TIMEOUT, io)
        .await
        .context("timed out handling ping")??;
    let _ = tokio::time::timeout(IO_TIMEOUT, conn.closed()).await;
    Ok(())
}

/// The live friends-presence listener as an iroh [`ProtocolHandler`], registered
/// once on the app's single persistent `Router` for [`FRIENDS_ALPN`]. Because the
/// Router owns `endpoint.accept()`, the listener can't be the standalone [`serve`]
/// loop (that would compete for accepts); this delegates each inbound connection to
/// the same [`exchange`] body, with the reply policy injected as a [`Handler`]
/// (which wraps [`crate::presence::should_answer`] + builds the Pong). Mirrors the
/// `AudioRouter` pattern from the B1 persistent-endpoint refactor.
#[derive(Clone)]
pub struct FriendsProtocol {
    handler: Handler,
}

impl FriendsProtocol {
    pub fn new(handler: Handler) -> Self {
        Self { handler }
    }
}

impl std::fmt::Debug for FriendsProtocol {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FriendsProtocol").finish_non_exhaustive()
    }
}

impl iroh::protocol::ProtocolHandler for FriendsProtocol {
    fn accept(
        &self,
        connection: iroh::endpoint::Connection,
    ) -> impl std::future::Future<Output = Result<(), iroh::protocol::AcceptError>> + Send {
        let handler = self.handler.clone();
        async move {
            // A failed exchange (malformed ping, timeout, etc.) is logged, not
            // surfaced as an accept error — one bad prober shouldn't disturb the
            // listener. Returning Ok keeps the router loop healthy.
            if let Err(e) = exchange(&connection, &handler).await {
                crate::log_msg(&format!("presence: inbound friends exchange failed: {e:#}"));
            }
            Ok(())
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::presence::{ControlMsg, RoomPresence};

    async fn bind() -> Endpoint {
        iroh::Endpoint::builder(iroh::endpoint::presets::N0)
            .secret_key(iroh::SecretKey::generate())
            .alpns(vec![FRIENDS_ALPN.to_vec()])
            .bind()
            .await
            .unwrap()
    }

    /// End-to-end over two real iroh endpoints on this machine. Ignored by
    /// default (binds endpoints + waits on the relay → slow, network-dependent).
    /// Run with: cargo test -- --ignored presence_net
    #[tokio::test]
    #[ignore = "binds real iroh endpoints; run on demand"]
    async fn loopback_ping_pong_and_friends_only() {
        let server = bind().await;
        let prober = bind().await;
        server.online().await;
        prober.online().await;
        let server_addr = server.addr();
        let prober_id = prober.addr().id;

        // The server answers ONLY the prober's id, reporting a room.
        let allowed = prober_id;
        let handler: Handler = Arc::new(move |from| {
            if from == allowed {
                Some(ControlMsg::Pong {
                    room: Some(RoomPresence { name: "HangOut".into(), ticket: "t".into() }),
                })
            } else {
                None // stranger -> no reply
            }
        });
        let server_ep = server.clone();
        let serve_task = tokio::spawn(async move { serve(server_ep, handler).await });

        // The allowed prober gets a Pong with the room.
        let (from, pong) = tokio::time::timeout(Duration::from_secs(15), probe(&prober, server_addr.clone()))
            .await
            .expect("probe timed out")
            .expect("probe failed");
        assert_eq!(from, server_addr.id);
        match pong {
            ControlMsg::Pong { room: Some(r) } => assert_eq!(r.name, "HangOut"),
            other => panic!("expected Pong with a room, got {other:?}"),
        }

        // A stranger (fresh id) gets an empty reply -> decode fails -> Err.
        let stranger = bind().await;
        stranger.online().await;
        let res = tokio::time::timeout(Duration::from_secs(15), probe(&stranger, server_addr))
            .await
            .expect("stranger probe timed out");
        assert!(res.is_err(), "a non-friend must not get a usable reply");

        server.close().await;
        prober.close().await;
        stranger.close().await;
        serve_task.abort();
    }
}