feat(presence): presence control transport (W7 P4 wire layer)

The I/O edge of the friends-only idle listener: bind/probe/serve a ping->pong
over a dedicated ALPN (peerspeak/friends/0), adapted from pixelpass's proven
control plane. Request/response, one exchange per connection: probe sends a
Ping and reads the Pong; serve accepts, authenticates the remote id, and asks
an injected handler (which wraps presence::should_answer + builds the pong)
what to reply -- None for a stranger/invisible, so the listener reveals
nothing to non-friends.

Verified by a loopback integration test over two real iroh endpoints (ignored
by default): the allowed prober gets a Pong with the room; a fresh stranger id
gets an empty, unusable reply. 256 lib tests + the loopback (run with
--ignored) green, clippy clean (incl --all-targets), release builds.

DEFERRED (next session, needs care + 2 machines): spawning serve on a
persistent endpoint OUTSIDE the per-join room session, and the ping scheduler.
Real fork noted in the module: a second always-on endpoint shares our node id
with the room endpoint (possible relay collision) vs refactoring to one
persistent endpoint -- intentionally not decided at 5am.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/lib.rs

@@ -8,6 +8,7 @@ pub mod config;
 pub mod identity;
 pub mod friends;
 pub mod presence;
+pub mod presence_net;
 pub mod theme;
 pub mod notify;
 pub mod screenshare;

src/presence_net.rs

@@ -0,0 +1,194 @@
+//! 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. There's a real fork there — a second always-on
+//! endpoint would share our node id with the per-join room endpoint (possible
+//! relay/identity collision), vs. refactoring to a single persistent endpoint.
+//! That decision wants care + a 2-machine check, so it's intentionally NOT made
+//! here; this module 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] = b"peerspeak/friends/0";
+
+/// 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 `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<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 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
+}
+
+/// 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")?;
+    // The authenticated remote id — NOT anything the peer puts in the payload.
+    let from = conn.remote_id();
+
+    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:?}"),
+        }
+        // Ask the policy what to send. None -> answer nothing (stranger / invisible):
+        // finish the stream with no bytes so the prober sees an empty (unusable) reply.
+        if let Some(reply) = handler(from) {
+            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(())
+}
+
+#[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 pong = tokio::time::timeout(Duration::from_secs(15), probe(&prober, server_addr.clone()))
+            .await
+            .expect("probe timed out")
+            .expect("probe failed");
+        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();
+    }
+}