pixelpass/src/gui/presence.rs

//! The always-on friends presence service.
//!
//! A control-plane iroh endpoint ([`endpoint::bind_control`]) that lives for the
//! whole GUI session on its own thread with a current-thread tokio runtime — the
//! GUI is a synchronous winit/egui loop, so iroh's async work can't run on it
//! (the same reason [`super::tray`] has its own thread + runtime).
//!
//! Inbound control messages are forwarded over a std mpsc channel the UI drains
//! each [`super::PixelPassApp::tick`]; the [`Waker`] is pinged on arrival so a
//! message wakes the loop even while the window is hidden to the tray — the same
//! trick the headless-child reader uses.

use std::sync::Arc;
use std::sync::mpsc::{self, Receiver};
use std::thread;

use iroh::{Endpoint, EndpointId};
use tokio::sync::mpsc as tmpsc;

use super::Waker;
use crate::common::{
    control::{self, ControlMsg, Inbound},
    endpoint, identity,
};

/// A command the UI hands the presence service over [`PresenceHandle`].
enum Command {
    /// Deliver one message, once, fire-and-forget (friend request/accept/decline
    /// and the presence `Hello`). A failure is logged, not retried.
    Send { peer: EndpointId, msg: ControlMsg },
    /// Begin — or replace — a share campaign: push `msg` (a
    /// [`ControlMsg::ShareCode`]) to every peer in `peers`, retrying the ones
    /// that are offline until they're reached or the campaign is stopped. Each
    /// success emits a [`PresenceEvent::ShareDelivered`]. Replaces any campaign
    /// already running (a fresh host session supersedes the previous code).
    StartShare {
        msg: ControlMsg,
        peers: Vec<EndpointId>,
    },
    /// Stop the active share campaign — the host stopped or left the screen, so
    /// the perishable code is no longer valid and offline friends shouldn't keep
    /// being chased.
    StopShare,
}

/// Something the service surfaces to the UI, drained each tick.
pub enum PresenceEvent {
    /// A control message arrived from a peer.
    Message(Inbound),
    /// A share-campaign code reached `peer` (its ACK came back). Lets the host
    /// screen flip that friend's row from "retrying" to "delivered."
    ShareDelivered { peer: EndpointId },
}

/// How long to wait before re-attempting delivery to friends who were offline
/// on the previous round of a share campaign.
const SHARE_RETRY: std::time::Duration = std::time::Duration::from_secs(5);

/// Handle the GUI holds for the presence service. Dropping it doesn't stop the
/// service (the thread is detached; the endpoint closes when the process exits)
/// — it just stops the UI from draining inbound messages.
pub struct PresenceHandle {
    /// Our stable control-plane id — what friends know us by, and what we embed
    /// in a wrapped share code so a viewer can find us.
    id: EndpointId,
    /// Service events (inbound messages + share receipts), drained by
    /// [`PresenceHandle::drain`] each tick.
    rx: Receiver<PresenceEvent>,
    /// Commands handed to the service thread. Unbounded tokio sender so the sync
    /// UI can enqueue without blocking or being inside the runtime.
    out_tx: tmpsc::UnboundedSender<Command>,
}

impl PresenceHandle {
    /// Our stable control-plane id.
    pub fn id(&self) -> EndpointId {
        self.id
    }

    /// Pull every service event received since the last call. Collected by the
    /// caller so it can take `&mut self` while handling them.
    pub fn drain(&self) -> Vec<PresenceEvent> {
        std::iter::from_fn(|| self.rx.try_recv().ok()).collect()
    }

    /// Enqueue a one-shot message for delivery to `peer`. Fire-and-forget from
    /// the UI's view; the service connects, delivers, and logs a failure. A send
    /// error here only means the service thread is gone.
    pub fn send(&self, peer: EndpointId, msg: ControlMsg) {
        self.command(Command::Send { peer, msg });
    }

    /// Begin (or replace) a share campaign pushing `msg` to `peers`, retrying
    /// offline friends until [`PresenceHandle::stop_share`] or the next call.
    pub fn start_share(&self, msg: ControlMsg, peers: Vec<EndpointId>) {
        self.command(Command::StartShare { msg, peers });
    }

    /// Stop the active share campaign (host stopped — the code is now stale).
    pub fn stop_share(&self) {
        self.command(Command::StopShare);
    }

    fn command(&self, cmd: Command) {
        if self.out_tx.send(cmd).is_err() {
            tracing::warn!("presence: service thread gone; dropping command");
        }
    }
}

/// Start the presence service. Returns `None` if the persistent identity can't
/// be loaded — the GUI then simply runs without friends features rather than
/// refusing to start. The endpoint binds asynchronously on the spawned thread;
/// our id is known immediately because it derives from the saved key, so we can
/// fail-fast and log it without waiting on the relay handshake.
pub fn start(waker: Waker, relay: Option<String>) -> Option<PresenceHandle> {
    let id: EndpointId = match identity::load_or_create() {
        Ok(key) => key.public(),
        Err(e) => {
            tracing::warn!("presence: no identity, friends features disabled: {e:#}");
            return None;
        }
    };
    tracing::info!(%id, "presence: starting control service");

    let (tx, rx) = mpsc::channel::<PresenceEvent>();
    let (out_tx, out_rx) = tmpsc::unbounded_channel::<Command>();
    thread::Builder::new()
        .name("pixelpass-presence".into())
        .spawn(move || run(relay, id, tx, out_rx, waker))
        .map_err(|e| tracing::warn!("presence: could not spawn service thread: {e}"))
        .ok()?;

    Some(PresenceHandle { id, rx, out_tx })
}

/// Thread body: a current-thread tokio runtime that binds the control endpoint,
/// runs the accept loop, bridges inbound messages to the UI channel, and
/// delivers outbound messages the UI enqueues.
fn run(
    relay: Option<String>,
    id: EndpointId,
    tx: mpsc::Sender<PresenceEvent>,
    mut out_rx: tmpsc::UnboundedReceiver<Command>,
    waker: Waker,
) {
    let rt = match tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
    {
        Ok(rt) => rt,
        Err(e) => {
            tracing::error!("presence: failed to build runtime: {e}");
            return;
        }
    };

    rt.block_on(async move {
        let ep = match endpoint::bind_control(relay.as_deref()).await {
            Ok(ep) => ep,
            Err(e) => {
                tracing::error!("presence: failed to bind control endpoint: {e:#}");
                return;
            }
        };
        tracing::info!(%id, "presence: control endpoint online");

        // One async→sync bridge for *everything* the UI sees: every producer
        // (the accept loop and the share campaign) pushes a `PresenceEvent` into
        // `ui_tx`; this task drains it onto the std channel and wakes the loop so
        // the event lands even while the window is hidden to the tray.
        let (ui_tx, mut ui_rx) = tmpsc::channel::<PresenceEvent>(64);
        let forward = tokio::spawn(async move {
            while let Some(event) = ui_rx.recv().await {
                if tx.send(event).is_err() {
                    break; // UI gone
                }
                waker.wake();
            }
        });

        // Wrap inbound control messages as events and feed the bridge.
        let (itx, mut irx) = tmpsc::channel::<Inbound>(32);
        let inbound_ui = ui_tx.clone();
        let inbound = tokio::spawn(async move {
            while let Some(msg) = irx.recv().await {
                if inbound_ui.send(PresenceEvent::Message(msg)).await.is_err() {
                    break;
                }
            }
        });

        // Handle UI commands: one-shot sends each on their own task, and a single
        // abortable share campaign (StartShare replaces it, StopShare cancels it).
        let cmd_ep = ep.clone();
        let commands = tokio::spawn(async move {
            let mut share: Option<tokio::task::JoinHandle<()>> = None;
            while let Some(cmd) = out_rx.recv().await {
                match cmd {
                    Command::Send { peer, msg } => {
                        let ep = cmd_ep.clone();
                        tokio::spawn(async move {
                            if let Err(e) = control::send(&ep, peer, &msg).await {
                                tracing::warn!(%peer, "presence: outbound send failed: {e:#}");
                            }
                        });
                    }
                    Command::StartShare { msg, peers } => {
                        if let Some(t) = share.take() {
                            t.abort();
                        }
                        let ep = cmd_ep.clone();
                        let ui = ui_tx.clone();
                        share = Some(tokio::spawn(run_share(ep, msg, peers, ui)));
                    }
                    Command::StopShare => {
                        if let Some(t) = share.take() {
                            t.abort();
                        }
                    }
                }
            }
        });

        control::serve(ep, itx).await;
        forward.abort();
        inbound.abort();
        commands.abort();
    });
}

/// Push `msg` to every peer in `peers`, retrying the ones that are offline every
/// [`SHARE_RETRY`] until all are delivered (or the task is aborted by a
/// StartShare/StopShare). Emits one [`PresenceEvent::ShareDelivered`] per peer
/// the moment its ACK comes back — that ACK *is* the delivery signal.
///
/// Each round fires all still-pending peers **concurrently**, so a single
/// offline friend's ~10s connect timeout doesn't serialise the whole round
/// (which it did when peers were tried one at a time).
async fn run_share(
    ep: Endpoint,
    msg: ControlMsg,
    mut pending: Vec<EndpointId>,
    ui: tmpsc::Sender<PresenceEvent>,
) {
    // The code is immutable for the campaign's life; share it across the
    // per-peer tasks via an `Arc` rather than re-cloning the payload each round.
    let msg = Arc::new(msg);
    while !pending.is_empty() {
        let mut round = tokio::task::JoinSet::new();
        for peer in pending {
            let ep = ep.clone();
            let msg = Arc::clone(&msg);
            round.spawn(async move {
                match control::send(&ep, peer, &msg).await {
                    Ok(()) => (peer, true),
                    Err(e) => {
                        tracing::debug!(%peer, "presence: share not yet delivered: {e:#}");
                        (peer, false)
                    }
                }
            });
        }

        let mut still = Vec::new();
        while let Some(joined) = round.join_next().await {
            let (peer, delivered) = match joined {
                Ok(outcome) => outcome,
                // A send task panicking is unexpected; log and drop that peer
                // from the campaign rather than abort the whole round. (A
                // campaign-level abort drops this future entirely — we never
                // observe that as a JoinError here.)
                Err(e) => {
                    tracing::warn!("presence: share task failed: {e}");
                    continue;
                }
            };
            if delivered {
                tracing::info!(%peer, "presence: shared code delivered");
                if ui
                    .send(PresenceEvent::ShareDelivered { peer })
                    .await
                    .is_err()
                {
                    return; // UI gone — nothing left to report to
                }
            } else {
                still.push(peer);
            }
        }

        if still.is_empty() {
            break;
        }
        pending = still;
        tokio::time::sleep(SHARE_RETRY).await;
    }
    tracing::info!("presence: share campaign complete");
}