diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 917e9a5..a7cfc24 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -422,8 +422,18 @@ multi-modal payloads (text inline, audio/photo/data via `iroh-blobs`). Receiving A `GossipPresence` source would emit `MeshEvent::Present`/`Discovered`, retiring the SSE presence bus. Confirmed: gossip compiles to WASM, so the web client uses the same discovery unchanged (§6.3). - - [ ] Milestone 4 — wrap as `IrohTransport` behind the `Transport`/ - `MeshEvent` contract; A/B against RTC+LAN on latency & connect rate. + - [~] **Milestone 4 — `IrohTransport` landed** (`core/src/iroh.rs`, behind + the `iroh-transport` feature, off by default so the UniFFI/xcframework/ + agent builds are untouched). Implements the `Transport` trait, emits + `MeshEvent`s; gossip carries clipboard `Message`s + presence over the + room-derived topic. Verified: `cargo run --features iroh-transport + --example iroh_engine` → the **real Engine** (unchanged public API) + syncs a clipboard A→B, E2E-decrypted, **no SSE/RTC/LAN/relay**. The + seam paid off — zero engine/client changes. + - [ ] Remaining: rendezvous for zero-config bootstrap (today via + `TETHER_IROH_BOOTSTRAP` env stopgap); file transfer via + `iroh-blobs`; make webrtc/mdns/reqwest optional so the iroh/wasm + build is lean; A/B vs RTC+LAN; flip the feature on by default. - [ ] Spike **Automerge** as the Sync tier; re-express the clipboard as a tiny app over replicated state to validate the App API. - [ ] Durable device keypairs + authenticated handshake (Noise/`snow`, or diff --git a/core/Cargo.toml b/core/Cargo.toml index 792aedf..6fbba14 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -29,10 +29,24 @@ bytes = "1.12.0" chacha20poly1305 = "0.10.1" base64 = "0.22.1" getrandom = "0.4.3" +# M4: the iroh Mesh, behind the `iroh-transport` feature (off by default, so the +# UniFFI/xcframework/agent builds are untouched). When on, IrohTransport replaces +# the hand-rolled SSE/RTC/LAN stack. +iroh = { version = "1", optional = true } +iroh-gossip = { version = "0.101", optional = true } + +[features] +default = [] +iroh-transport = ["dep:iroh", "dep:iroh-gossip"] [dev-dependencies] -# Iroh spike (examples/iroh_spike.rs): evaluate iroh as an alternative Mesh — -# QUIC P2P + relay holepunch + key-based identity. Dev-only; not in the shipped lib. +# Iroh spikes (examples/iroh_*.rs) link iroh directly regardless of the feature. iroh = "1" iroh-gossip = "0.101" anyhow = "1" + +# The M4 integration example drives the real Engine over IrohTransport, so it +# needs the feature (and Engine::iroh_id, which only exists under it). +[[example]] +name = "iroh_engine" +required-features = ["iroh-transport"] diff --git a/core/examples/iroh_engine.rs b/core/examples/iroh_engine.rs new file mode 100644 index 0000000..7041ecb --- /dev/null +++ b/core/examples/iroh_engine.rs @@ -0,0 +1,88 @@ +//! M4 integration test — the real `Engine` syncing a clipboard over iroh. +//! cargo run --features iroh-transport --example iroh_engine +//! +//! Two engines, same account → same room → same gossip topic. No SSE/RTC/LAN: +//! with the iroh-transport feature the engine's only transport is IrohTransport. +//! Proves the public API (Engine::new/start/send + MessageHandler) is unchanged +//! while the entire Mesh underneath is iroh — the payoff of the MeshEvent seam. +//! +//! Bootstrap is wired in-process via the TETHER_IROH_BOOTSTRAP env var (the v1 +//! stopgap until the rendezvous supernode): start A, read its EndpointId, hand +//! it to B. + +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +use tethercore::{Engine, Message, MessageHandler}; + +struct Noop; +impl MessageHandler for Noop { + fn on_message(&self, _m: Message) {} + fn on_status(&self, _c: bool) {} + fn on_file(&self, _n: String, _m: String, _d: Vec) {} +} + +struct Collector(mpsc::Sender); +impl MessageHandler for Collector { + fn on_message(&self, m: Message) { + let _ = self.0.send(m.text); + } + fn on_status(&self, _c: bool) {} + fn on_file(&self, _n: String, _m: String, _d: Vec) {} +} + +fn main() { + let account = "pecord@gmail.com"; // same account → same room → same topic + key + let room = "iroh-engine-test"; + let server = "unused"; // iroh ignores cfg.server; it uses n0 infra + + // Opener (no bootstrap yet). + let a = Engine::new( + server.into(), room.into(), "iroh-a".into(), "macos".into(), "A".into(), account.into(), + ); + a.start(Box::new(Noop)); + + // Wait for A's iroh endpoint to come online, then bootstrap B off it. + let a_id = { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if let Some(id) = a.iroh_id() { + break id; + } + if Instant::now() > deadline { + eprintln!("❌ A never came online"); + std::process::exit(1); + } + std::thread::sleep(Duration::from_millis(200)); + } + }; + eprintln!("A online, id={a_id}"); + std::env::set_var("TETHER_IROH_BOOTSTRAP", &a_id); + + // Joiner (bootstraps to A via the env var its IrohTransport reads at start). + let (tx, rx) = mpsc::channel(); + let b = Engine::new( + server.into(), room.into(), "iroh-b".into(), "ios".into(), "B".into(), account.into(), + ); + b.start(Box::new(Collector(tx))); + + eprintln!("waiting for gossip topic to form…"); + std::thread::sleep(Duration::from_secs(8)); + + let payload = "clipboard synced over iroh 🎉"; + eprintln!("A sending: {payload:?}"); + a.send(payload.to_string()); + + match rx.recv_timeout(Duration::from_secs(10)) { + Ok(t) if t == payload => { + println!("✅ engine-over-iroh: B received {t:?} (E2E-decrypted, via gossip, no relay/SSE/RTC/LAN)"); + } + Ok(t) => println!("⚠️ B got {t:?} (expected {payload:?})"), + Err(_) => { + println!("❌ B received nothing — gossip topic may not have formed"); + std::process::exit(1); + } + } + a.stop(); + b.stop(); +} diff --git a/core/src/iroh.rs b/core/src/iroh.rs new file mode 100644 index 0000000..d18c8f8 --- /dev/null +++ b/core/src/iroh.rs @@ -0,0 +1,150 @@ +//! IrohTransport — the Mesh, on iroh. The end state of M4: this one transport +//! subsumes the hand-rolled SSE/RTC/LAN stack. iroh handles key-addressing, +//! holepunch, and relay; `iroh-gossip` carries the clipboard `Message`s and +//! presence over the account-derived topic. +//! +//! It implements the same `Transport` trait and emits the same `MeshEvent`s as +//! the legacy transports, so the engine event loop and every client are +//! unchanged — the whole point of the MeshEvent seam. +//! +//! Bootstrap note (v1): joining the gossip topic needs ≥1 peer id. Until the +//! rendezvous supernode lands (see ARCHITECTURE §M3), bootstrap ids come from +//! the `TETHER_IROH_BOOTSTRAP` env var (comma-separated). That's the one +//! stopgap; everything else is production-shaped. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex as StdMutex}; + +use futures_util::TryStreamExt; +use iroh::{endpoint::presets, protocol::Router, Endpoint, EndpointId}; +use iroh_gossip::{ + api::{Event, GossipSender}, + net::{Gossip, GOSSIP_ALPN}, + proto::TopicId, +}; +use tokio::runtime::Handle; + +use crate::{Config, EventTx, MeshEvent, Message, Transport}; + +pub(crate) struct IrohTransport { + cfg: Config, + events: EventTx, + /// Set once the gossip topic is joined; used by `publish`. + sender: Arc>>, + /// This node's iroh EndpointId, surfaced once online (for bootstrap wiring). + id: Arc>>, +} + +impl IrohTransport { + pub(crate) fn new(cfg: Config, events: EventTx, id: Arc>>) -> Self { + Self { cfg, events, sender: Arc::new(StdMutex::new(None)), id } + } +} + +/// account/room → 32-byte gossip topic (consistent with room_for_account: the +/// room is already sha256(account)[..4], this is the full-width derivation). +fn topic_for(room: &str) -> TopicId { + use sha2::{Digest, Sha256}; + let mut id = [0u8; 32]; + id.copy_from_slice(&Sha256::digest(format!("tether:{room}").as_bytes())); + TopicId::from_bytes(id) +} + +fn bootstrap_ids() -> Vec { + std::env::var("TETHER_IROH_BOOTSTRAP") + .ok() + .into_iter() + .flat_map(|s| { + s.split(',') + .filter_map(|x| x.trim().parse::().ok()) + .collect::>() + }) + .collect() +} + +impl Transport for IrohTransport { + fn name(&self) -> &'static str { + "iroh" + } + + fn direct(&self) -> bool { + true // iroh chooses direct-vs-relay itself; treat as a direct peer channel + } + + fn start(self: Arc, rt: Handle, running: Arc) { + let this = self; + rt.spawn(async move { + let endpoint = match Endpoint::bind(presets::N0).await { + Ok(e) => e, + Err(e) => { + eprintln!("tethercore[iroh]: bind failed: {e}"); + return; + } + }; + endpoint.online().await; + let my_id = endpoint.id().to_string(); + *this.id.lock().unwrap() = Some(my_id.clone()); + eprintln!("tethercore[iroh]: online id={my_id}"); + + let gossip = Gossip::builder().spawn(endpoint.clone()); + let router = Router::builder(endpoint.clone()) + .accept(GOSSIP_ALPN, gossip.clone()) + .spawn(); + + let topic = topic_for(&this.cfg.room); + let sub = match gossip.subscribe_and_join(topic, bootstrap_ids()).await { + Ok(s) => s, + Err(e) => { + eprintln!("tethercore[iroh]: topic join failed: {e}"); + return; + } + }; + let (sender, mut receiver) = sub.split(); + *this.sender.lock().unwrap() = Some(sender); + let _ = this.events.send(MeshEvent::Status(true)); + eprintln!("tethercore[iroh]: joined topic {topic}"); + + while running.load(Ordering::SeqCst) { + match receiver.try_next().await { + Ok(Some(Event::Received(msg))) => { + if let Ok(m) = serde_json::from_slice::(&msg.content) { + if m.from != this.cfg.from { + let _ = this.events.send(MeshEvent::Message(m)); + } + } + } + Ok(Some(Event::NeighborUp(id))) => { + let _ = this.events.send(MeshEvent::DirectUp(id.to_string())); + } + Ok(Some(Event::NeighborDown(id))) => { + let _ = this.events.send(MeshEvent::DirectDown(id.to_string())); + } + Ok(Some(_)) => {} // Lagged, etc. + Ok(None) => break, // stream ended + Err(e) => { + eprintln!("tethercore[iroh]: recv error: {e}"); + break; + } + } + } + drop(router); + }); + } + + fn publish(&self, rt: &Handle, msg: Message) { + // Clipboard + receipts ride gossip; other kinds are local-only. + if !(msg.kind.is_empty() || msg.kind == "clipboard" || msg.kind == "receipt") { + return; + } + let sender = self.sender.lock().unwrap().clone(); + if let Some(sender) = sender { + let body = match serde_json::to_vec(&msg) { + Ok(b) => b, + Err(_) => return, + }; + rt.spawn(async move { + let _ = sender.broadcast(body.into()).await; + }); + } + } +} diff --git a/core/src/lib.rs b/core/src/lib.rs index c547f49..200daa9 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -13,6 +13,12 @@ //! across all transports and de-duplicates inbound so the same clipboard //! arriving over two channels surfaces once. +// With the iroh-transport feature, the legacy SSE/RTC/LAN transports aren't +// constructed (IrohTransport replaces them); they still compile, so quiet the +// expected dead-code noise. Fully dropping their heavy deps (webrtc/mdns) for +// the iroh/wasm build is a follow-up. +#![cfg_attr(feature = "iroh-transport", allow(dead_code))] + use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -24,6 +30,8 @@ use tokio::runtime::Handle; use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender}; mod crypto; +#[cfg(feature = "iroh-transport")] +mod iroh; mod lan; mod rtc; @@ -333,6 +341,9 @@ pub struct Engine { /// Current message handler, swapped in on each `start()` (latest wins, as /// before) and read by the event loop per event. handler: Arc>>>, + /// This node's iroh EndpointId once online (for bootstrap wiring in tests). + #[cfg(feature = "iroh-transport")] + iroh_id: Arc>>, } #[uniffi::export] @@ -367,6 +378,18 @@ impl Engine { // all state and reacts. Transports get a sender at construction; the // engine keeps one clone alive so the loop never sees the channel close. let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel::(); + + // M4: with the iroh-transport feature, one iroh-backed transport replaces + // the hand-rolled SSE/RTC/LAN stack. Default build is unchanged. + #[cfg(feature = "iroh-transport")] + let iroh_id: Arc>> = Arc::new(Mutex::new(None)); + #[cfg(feature = "iroh-transport")] + let transports: Vec> = vec![Arc::new(iroh::IrohTransport::new( + cfg.clone(), + events_tx.clone(), + iroh_id.clone(), + ))]; + #[cfg(not(feature = "iroh-transport"))] let transports: Vec> = vec![ Arc::new(SseTransport { cfg: cfg.clone(), @@ -393,6 +416,8 @@ impl Engine { events_rx: Mutex::new(Some(events_rx)), loop_started: AtomicBool::new(false), handler: Arc::new(Mutex::new(None)), + #[cfg(feature = "iroh-transport")] + iroh_id, }) } @@ -634,6 +659,16 @@ impl Engine { } } +/// Non-FFI helpers (iroh feature only) — used by the M4 integration example to +/// wire bootstrap between two in-process engines. +#[cfg(feature = "iroh-transport")] +impl Engine { + /// This node's iroh EndpointId once the transport is online, else None. + pub fn iroh_id(&self) -> Option { + self.iroh_id.lock().unwrap().clone() + } +} + /// Short-window de-dup so a clipboard echoed over both SSE and the RTC data /// channel surfaces once. Keyed on (from, text) — NOT ts: the SSE copy carries /// the server's stamp while the data-channel copy has none, so ts must be