tether/core/examples/iroh_gossip.rs

//! Iroh spike, milestone 3 — discovery via iroh-gossip (replaces the SSE presence bus).
//!   cargo run --example iroh_gossip
//!
//! Proves `account → {EndpointId}` the P2P way: two endpoints derive the SAME
//! gossip topic from a shared account secret (exactly like room_for_account
//! today), join it, and learn each other's public keys by broadcasting presence
//! — no hardcoded peer addresses in the steady state.
//!
//! The ONE thing gossip can't do for itself is bootstrap: to join topic T you
//! must reach ≥1 peer already in T. Here we hand the joiner the opener's address
//! directly — in production that single fact comes from a tiny rendezvous
//! supernode (the homelab), which is all the server shrinks to. Everything after
//! the first contact is peer-to-peer gossip.
//!
//! Maps onto the engine: a `GossipPresence` source would emit `MeshEvent::Present`
//! / `MeshEvent::Discovered` from received presence, replacing rtc.rs's chirps.

use std::time::Duration;

use futures_util::TryStreamExt;
use iroh::{protocol::Router, Endpoint, endpoint::presets};
use iroh_gossip::{api::Event, net::Gossip, net::GOSSIP_ALPN, proto::TopicId};
use serde::{Deserialize, Serialize};

fn ae<E: std::fmt::Display>(e: E) -> anyhow::Error {
    anyhow::anyhow!("{e}")
}

/// account → 32-byte gossip topic (same idea as room_for_account, full width).
fn topic_for_account(account: &str) -> TopicId {
    use sha2::{Digest, Sha256};
    let mut id = [0u8; 32];
    id.copy_from_slice(&Sha256::digest(account.as_bytes()));
    TopicId::from_bytes(id)
}

#[derive(Serialize, Deserialize)]
struct Presence {
    id: String,   // the broadcaster's EndpointId (author, carried in payload —
    name: String, // gossip relays, so don't trust delivered_from for identity)
}

/// Build an endpoint + a gossip instance registered on a Router.
async fn node() -> anyhow::Result<(Endpoint, Gossip, Router)> {
    let endpoint = Endpoint::bind(presets::N0).await.map_err(ae)?;
    endpoint.online().await;
    let gossip = Gossip::builder().spawn(endpoint.clone());
    let router = Router::builder(endpoint.clone())
        .accept(GOSSIP_ALPN, gossip.clone())
        .spawn();
    Ok((endpoint, gossip, router))
}

/// Join the topic, broadcast our presence, and wait until we hear the peer's.
async fn discover(
    label: &'static str,
    endpoint: Endpoint,
    gossip: Gossip,
    topic: TopicId,
    bootstrap: Vec<iroh::EndpointId>,
) -> anyhow::Result<String> {
    let me = Presence { id: endpoint.id().to_string(), name: label.to_string() };
    let payload = serde_json::to_vec(&me)?;

    let (sender, mut receiver) = gossip
        .subscribe_and_join(topic, bootstrap)
        .await
        .map_err(ae)?
        .split();
    println!("[gossip] {label} joined topic — id={}", me.id);

    // Re-broadcast on a ticker so a late-joining neighbor still hears us.
    let ticker = {
        let sender = sender.clone();
        let payload = payload.clone();
        tokio::spawn(async move {
            loop {
                let _ = sender.broadcast(payload.clone().into()).await;
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
        })
    };

    let found = async {
        while let Some(event) = receiver.try_next().await.map_err(ae)? {
            if let Event::Received(msg) = event {
                if let Ok(p) = serde_json::from_slice::<Presence>(&msg.content) {
                    if p.id != me.id {
                        return Ok::<String, anyhow::Error>(format!("{} ({})", p.name, p.id));
                    }
                }
            }
        }
        anyhow::bail!("{label}: gossip stream ended before discovering a peer")
    };

    let peer = tokio::time::timeout(Duration::from_secs(20), found)
        .await
        .map_err(|_| anyhow::anyhow!("{label}: timed out waiting to discover a peer"))??;
    ticker.abort();
    println!("[gossip] {label} discovered peer via gossip: {peer}");
    Ok(peer)
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let account = "pecord@gmail.com"; // both derive the same topic from this
    let topic = topic_for_account(account);
    println!("[gossip] account {account:?} → topic {topic}");

    // Opener and joiner.
    let (a_ep, a_gossip, a_router) = node().await?;
    let (b_ep, b_gossip, b_router) = node().await?;

    // Bootstrap: the joiner only needs the opener's EndpointId; iroh's own
    // discovery (n0 DNS/pkarr here, a rendezvous supernode in production) maps
    // that id → address. Everything after first contact is peer gossip.
    let a_id = a_ep.id();

    let a = discover("node-A", a_ep, a_gossip, topic, vec![]);
    let b = discover("node-B", b_ep, b_gossip, topic, vec![a_id]);
    let (a_found, b_found) = tokio::try_join!(a, b)?;

    // Graceful teardown so background gossip/accept tasks stop cleanly.
    a_router.shutdown().await.map_err(ae)?;
    b_router.shutdown().await.map_err(ae)?;

    println!(
        "✅ gossip discovery: both endpoints found each other in the \
         account-derived topic with zero hardcoded addresses\n   A↔{a_found}\n   B↔{b_found}"
    );
    Ok(())
}