SSE streaming with long-poll fallback; encrypted unicast profile; registry enc keys

src/commands.rs

@@ -148,6 +148,15 @@ pub fn key_show(config_path: &Path) -> Result<()> {
     Ok(())
 }
 
+pub fn key_show_enc(config_path: &Path) -> Result<()> {
+    let config = Config::load(config_path)?;
+    let secret = config
+        .load_enc_key()?
+        .ok_or_else(|| anyhow!("no encryption key at {}", config.enc_key_path().display()))?;
+    println!("{}", crate::crypto::enc_public_from_secret(&secret)?);
+    Ok(())
+}
+
 pub fn key_rotate(config_path: &Path) -> Result<()> {
     let config = Config::load(config_path)?;
     let old = config.load_key()?;
@@ -155,8 +164,11 @@ pub fn key_rotate(config_path: &Path) -> Result<()> {
     std::fs::copy(config.key_path(), &backup)?;
     let new_key = Keypair::generate();
     config.save_key(&new_key)?;
+    let (enc_secret, enc_public) = crate::crypto::generate_enc_keypair();
+    config.save_enc_key(&enc_secret)?;
     println!("old pubkey {}", old.public_hex());
     println!("new pubkey {}", new_key.public_hex());
+    println!("new enc pubkey {enc_public}");
     println!("old key saved to {}", backup.display());
     println!(
         "peers accept the new key via: frxd member add <name> {} --previous {}",
@@ -293,8 +305,10 @@ pub fn registry_add(
     class: &str,
     not_before: Option<u64>,
     not_after: Option<u64>,
+    enc_key: Option<String>,
 ) -> Result<()> {
     let pubkey = normalize_key(pubkey)?;
+    let enc_key = enc_key.map(|key| normalize_key(&key)).transpose()?;
     let class = if class == CLASS_ENRICHMENT {
         CLASS_ENRICHMENT
     } else {
@@ -312,6 +326,7 @@ pub fn registry_add(
                 not_before: not_before.unwrap_or_else(now_ts),
                 not_after,
             }],
+            enc_key: enc_key.clone(),
         });
         Ok(())
     })?;
@@ -319,6 +334,19 @@ pub fn registry_add(
     Ok(())
 }
 
+pub fn registry_set_enc_key(dir: &Path, id: &str, enc_key: &str) -> Result<()> {
+    let enc_key = normalize_key(enc_key)?;
+    mutate_registry(dir, |doc| {
+        let Some(member) = doc.members.iter_mut().find(|member| member.id == id) else {
+            return Err(anyhow!("no member named {id}"));
+        };
+        member.enc_key = Some(enc_key.clone());
+        Ok(())
+    })?;
+    println!("set enc key for {id}");
+    Ok(())
+}
+
 pub fn registry_add_key(
     dir: &Path,
     id: &str,
@@ -387,6 +415,9 @@ pub fn registry_list(dir: &Path) -> Result<()> {
             };
             println!("    {} ({window})", entry.key);
         }
+        if let Some(enc_key) = &member.enc_key {
+            println!("    enc {enc_key}");
+        }
     }
     Ok(())
 }

src/config.rs

@@ -123,13 +123,13 @@ fn default_data_dir() -> String {
 }
 
 impl Config {
-    pub fn new(name: &str, listen: &str, relay: &str, data_dir: &str) -> Self {
+    pub fn new(name: &str, listen: &str, relays: Vec<String>, data_dir: &str) -> Self {
         Self {
             node: NodeSection {
                 name: name.to_string(),
                 id: None,
                 listen: listen.to_string(),
-                relays: vec![relay.to_string()],
+                relays,
                 registry: None,
                 ma_key: None,
                 dev_bootstrap: false,
@@ -182,6 +182,27 @@ impl Config {
         self.data_dir().join("registry-cache.json")
     }
 
+    pub fn enc_key_path(&self) -> PathBuf {
+        self.data_dir().join("enc-key.hex")
+    }
+
+    pub fn load_enc_key(&self) -> Result<Option<String>> {
+        let path = self.enc_key_path();
+        if !path.exists() {
+            return Ok(None);
+        }
+        let raw = fs::read_to_string(&path).context("reading enc key")?;
+        Ok(Some(raw.trim().to_string()))
+    }
+
+    pub fn save_enc_key(&self, secret_hex: &str) -> Result<()> {
+        fs::create_dir_all(self.data_dir())?;
+        let path = self.enc_key_path();
+        fs::write(&path, secret_hex)?;
+        set_private_permissions(&path)?;
+        Ok(())
+    }
+
     pub fn load_key(&self) -> Result<Keypair> {
         let raw = fs::read_to_string(self.key_path())
             .with_context(|| format!("reading key {}", self.key_path().display()))?;

src/crypto.rs

@@ -1,9 +1,17 @@
 use crate::PROTOCOL;
 use crate::message::Envelope;
 use anyhow::{Context, Result, anyhow};
+use chacha20poly1305::aead::{Aead, KeyInit};
+use chacha20poly1305::{ChaCha20Poly1305, Key, Nonce};
 use ed25519_dalek::{Signature, Signer, SigningKey, VerifyingKey};
+use hkdf::Hkdf;
 use rand::RngCore;
+use rand::rngs::OsRng;
 use serde_json::Value;
+use sha2::Sha256;
+use x25519_dalek::{EphemeralSecret, PublicKey as X25519PublicKey, StaticSecret};
+
+pub const ENC_ALG: &str = "x25519-hkdf-sha256-chacha20poly1305";
 
 pub struct Keypair {
     signing: SigningKey,
@@ -136,6 +144,102 @@ pub fn verify_envelope(envelope: &Envelope) -> Result<()> {
     verify_signature(&envelope.key, &signing_bytes(envelope), &envelope.sig)
 }
 
+pub fn generate_enc_keypair() -> (String, String) {
+    let secret = StaticSecret::random_from_rng(OsRng);
+    let public = X25519PublicKey::from(&secret);
+    (
+        hex::encode(secret.to_bytes()),
+        hex::encode(public.to_bytes()),
+    )
+}
+
+pub fn enc_public_from_secret(secret_hex: &str) -> Result<String> {
+    let secret = enc_secret_from_hex(secret_hex)?;
+    Ok(hex::encode(X25519PublicKey::from(&secret).to_bytes()))
+}
+
+fn enc_secret_from_hex(secret_hex: &str) -> Result<StaticSecret> {
+    let bytes = hex::decode(secret_hex.trim()).context("enc key is not hex")?;
+    let bytes: [u8; 32] = bytes
+        .as_slice()
+        .try_into()
+        .map_err(|_| anyhow!("enc key must be 32 bytes"))?;
+    Ok(StaticSecret::from(bytes))
+}
+
+fn enc_public_from_hex(public_hex: &str) -> Result<X25519PublicKey> {
+    let bytes = hex::decode(public_hex.trim()).context("enc public key is not hex")?;
+    let bytes: [u8; 32] = bytes
+        .as_slice()
+        .try_into()
+        .map_err(|_| anyhow!("enc public key must be 32 bytes"))?;
+    Ok(X25519PublicKey::from(bytes))
+}
+
+pub fn unicast_context(from: &str, recipient_enc_public: &str) -> String {
+    format!("{} unicast {} {}", PROTOCOL, from, recipient_enc_public)
+}
+
+pub fn encrypt_unicast(recipient_enc_hex: &str, context: &str, plaintext: &[u8]) -> Result<Value> {
+    let recipient = enc_public_from_hex(recipient_enc_hex)?;
+    let ephemeral = EphemeralSecret::random_from_rng(OsRng);
+    let epk = X25519PublicKey::from(&ephemeral);
+    let shared = ephemeral.diffie_hellman(&recipient);
+    let mut key = [0u8; 32];
+    Hkdf::<Sha256>::new(None, shared.as_bytes())
+        .expand(context.as_bytes(), &mut key)
+        .map_err(|_| anyhow!("hkdf expand failed"))?;
+    let cipher = ChaCha20Poly1305::new(Key::from_slice(&key));
+    let mut nonce_bytes = [0u8; 12];
+    rand::thread_rng().fill_bytes(&mut nonce_bytes);
+    let ciphertext = cipher
+        .encrypt(Nonce::from_slice(&nonce_bytes), plaintext)
+        .map_err(|_| anyhow!("encryption failed"))?;
+    let mut payload = nonce_bytes.to_vec();
+    payload.extend_from_slice(&ciphertext);
+    Ok(serde_json::json!({
+        "enc": {
+            "alg": ENC_ALG,
+            "epk": hex::encode(epk.to_bytes()),
+            "ct": hex::encode(payload),
+        }
+    }))
+}
+
+pub fn decrypt_unicast(secret_hex: &str, context: &str, enc: &Value) -> Result<Vec<u8>> {
+    let secret = enc_secret_from_hex(secret_hex)?;
+    let alg = enc
+        .get("alg")
+        .and_then(Value::as_str)
+        .ok_or_else(|| anyhow!("missing algorithm"))?;
+    if alg != ENC_ALG {
+        return Err(anyhow!("unsupported encryption algorithm"));
+    }
+    let epk = enc
+        .get("epk")
+        .and_then(Value::as_str)
+        .ok_or_else(|| anyhow!("missing ephemeral key"))?;
+    let epk = enc_public_from_hex(epk)?;
+    let ct = enc
+        .get("ct")
+        .and_then(Value::as_str)
+        .ok_or_else(|| anyhow!("missing ciphertext"))?;
+    let payload = hex::decode(ct).context("ciphertext is not hex")?;
+    if payload.len() < 12 {
+        return Err(anyhow!("ciphertext too short"));
+    }
+    let (nonce_bytes, ciphertext) = payload.split_at(12);
+    let shared = secret.diffie_hellman(&epk);
+    let mut key = [0u8; 32];
+    Hkdf::<Sha256>::new(None, shared.as_bytes())
+        .expand(context.as_bytes(), &mut key)
+        .map_err(|_| anyhow!("hkdf expand failed"))?;
+    let cipher = ChaCha20Poly1305::new(Key::from_slice(&key));
+    cipher
+        .decrypt(Nonce::from_slice(nonce_bytes), ciphertext)
+        .map_err(|_| anyhow!("decryption failed"))
+}
+
 pub fn now_ts() -> u64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -168,6 +272,31 @@ mod tests {
         verify_envelope(&env).unwrap();
     }
 
+    #[test]
+    fn unicast_encryption_roundtrips_and_rejects_tampering() {
+        let (secret, public) = generate_enc_keypair();
+        let context = unicast_context("alice.example", &public);
+        let plaintext = b"{\"qid\":\"q1\",\"results\":[]}";
+        let envelope_body = encrypt_unicast(&public, &context, plaintext).unwrap();
+        assert_eq!(
+            envelope_body.pointer("/enc/alg").and_then(Value::as_str),
+            Some(ENC_ALG)
+        );
+        let enc = envelope_body.get("enc").unwrap();
+        let decrypted = decrypt_unicast(&secret, &context, enc).unwrap();
+        assert_eq!(decrypted, plaintext);
+
+        let (other_secret, _) = generate_enc_keypair();
+        assert!(decrypt_unicast(&other_secret, &context, enc).is_err());
+
+        let mut tampered = enc.clone();
+        let ct = tampered.get("ct").and_then(Value::as_str).unwrap();
+        let mut bytes = hex::decode(ct).unwrap();
+        bytes[13] ^= 0xff;
+        tampered["ct"] = json!(hex::encode(bytes));
+        assert!(decrypt_unicast(&secret, &context, &tampered).is_err());
+    }
+
     #[test]
     fn tampered_body_fails_verification() {
         let key = Keypair::generate();

src/main.rs

@@ -28,7 +28,7 @@ enum Command {
         #[arg(long, default_value = "127.0.0.1:7701")]
         listen: String,
         #[arg(long, default_value = "http://127.0.0.1:7700")]
-        relay: String,
+        relay: Vec<String>,
         #[arg(long, default_value = "./frx-data")]
         data_dir: String,
         #[arg(long)]
@@ -125,6 +125,7 @@ enum MemberCommand {
 #[derive(Subcommand)]
 enum KeyCommand {
     Show,
+    ShowEnc,
     Rotate,
 }
 
@@ -140,6 +141,12 @@ enum RegistryCommand {
         not_before: Option<u64>,
         #[arg(long)]
         not_after: Option<u64>,
+        #[arg(long)]
+        enc_key: Option<String>,
+    },
+    SetEncKey {
+        id: String,
+        enc_key: String,
     },
     AddKey {
         id: String,
@@ -197,7 +204,7 @@ async fn main() -> Result<()> {
             if registry.is_some() && ma_key.is_none() {
                 bail!("--ma-key is required with --registry");
             }
-            let mut config = Config::new(&name, &listen, &relay, &data_dir);
+            let mut config = Config::new(&name, &listen, relay, &data_dir);
             config.node.id = id;
             config.node.registry = registry;
             config.node.ma_key = ma_key;
@@ -209,10 +216,13 @@ async fn main() -> Result<()> {
             }
             let key = Keypair::generate();
             config.save_key(&key)?;
+            let (enc_secret, enc_public) = frxd::crypto::generate_enc_keypair();
+            config.save_enc_key(&enc_secret)?;
             config.save(&cli.config)?;
             println!("wrote config {}", cli.config.display());
             println!("wrote key {}", config.key_path().display());
             println!("pubkey {}", key.public_hex());
+            println!("enc pubkey {enc_public}");
             println!("data dir {}", config.data_dir().display());
         }
         Command::Add {
@@ -279,6 +289,7 @@ async fn main() -> Result<()> {
         },
         Command::Key { command } => match command {
             KeyCommand::Show => commands::key_show(&cli.config)?,
+            KeyCommand::ShowEnc => commands::key_show_enc(&cli.config)?,
             KeyCommand::Rotate => commands::key_rotate(&cli.config)?,
         },
         Command::Registry { dir, command } => match command {
@@ -289,7 +300,13 @@ async fn main() -> Result<()> {
                 class,
                 not_before,
                 not_after,
-            } => commands::registry_add(&dir, &id, &pubkey, &class, not_before, not_after)?,
+                enc_key,
+            } => {
+                commands::registry_add(&dir, &id, &pubkey, &class, not_before, not_after, enc_key)?
+            }
+            RegistryCommand::SetEncKey { id, enc_key } => {
+                commands::registry_set_enc_key(&dir, &id, &enc_key)?
+            }
             RegistryCommand::AddKey {
                 id,
                 pubkey,

src/node.rs

@@ -38,6 +38,8 @@ pub struct Node {
     members: RwLock<Vec<Member>>,
     members_mtime: Mutex<Option<SystemTime>>,
     registry: Option<Arc<Watcher>>,
+    enc_secret: Option<String>,
+    enc_public: Option<String>,
     aggregates: Mutex<Aggregates>,
     seen: Mutex<HashSet<String>>,
     pending: Mutex<HashMap<String, Vec<(String, ResponseBody)>>>,
@@ -136,6 +138,11 @@ impl Node {
         if let Some(watcher) = &registry {
             watcher.load_initial();
         }
+        let enc_secret = config.load_enc_key()?;
+        let enc_public = match &enc_secret {
+            Some(secret) => Some(crate::crypto::enc_public_from_secret(secret)?),
+            None => None,
+        };
         Ok(Arc::new(Self {
             config,
             key,
@@ -143,6 +150,8 @@ impl Node {
             members: RwLock::new(members),
             members_mtime: Mutex::new(members_mtime),
             registry,
+            enc_secret,
+            enc_public,
             aggregates: Mutex::new(Aggregates::default()),
             seen: Mutex::new(HashSet::new()),
             pending: Mutex::new(HashMap::new()),
@@ -386,7 +395,52 @@ impl Node {
         delivered
     }
 
-    async fn dispatch(self: &Arc<Self>, envelope: Envelope, relay: &str) {
+    fn encrypt_for(&self, recipient_key: &str, body: &Value) -> Value {
+        let Some(watcher) = &self.registry else {
+            return body.clone();
+        };
+        let Some(recipient_enc) = watcher.enc_key(recipient_key) else {
+            return body.clone();
+        };
+        let context = crate::crypto::unicast_context(&self.identifier(), &recipient_enc);
+        match serde_json::to_vec(body) {
+            Ok(plaintext) => crate::crypto::encrypt_unicast(&recipient_enc, &context, &plaintext)
+                .unwrap_or_else(|_| body.clone()),
+            Err(_) => body.clone(),
+        }
+    }
+
+    fn decrypt_body(&self, envelope: &mut Envelope) -> Result<()> {
+        let Some(enc) = envelope.body.get("enc").cloned() else {
+            return Ok(());
+        };
+        let secret = self
+            .enc_secret
+            .as_deref()
+            .ok_or_else(|| anyhow!("encrypted message received but no enc key configured"))?;
+        let public = self
+            .enc_public
+            .as_deref()
+            .ok_or_else(|| anyhow!("encrypted message received but no enc key configured"))?;
+        let context = crate::crypto::unicast_context(&envelope.from, public);
+        let plaintext = crate::crypto::decrypt_unicast(secret, &context, &enc)?;
+        envelope.body = serde_json::from_slice(&plaintext).context("decrypted body is not json")?;
+        Ok(())
+    }
+
+    async fn send_payload(
+        &self,
+        to_key: &str,
+        msg_type: &str,
+        body: Value,
+        relay: &str,
+    ) -> Result<()> {
+        let body = self.encrypt_for(to_key, &body);
+        let envelope = Envelope::new(&self.key, &self.identifier(), msg_type, body);
+        self.send_unicast(to_key, envelope, relay).await
+    }
+
+    async fn dispatch(self: &Arc<Self>, mut envelope: Envelope, relay: &str) {
         if envelope.verify().is_err() {
             return;
         }
@@ -413,6 +467,10 @@ impl Node {
         if !listed {
             return;
         }
+        if let Err(error) = self.decrypt_body(&mut envelope) {
+            eprintln!("dropped encrypted message: {error}");
+            return;
+        }
         match envelope.msg_type.as_str() {
             TYPE_QUERY => {
                 if envelope.from == self.identifier() || !self.config.node.responder {
@@ -498,13 +556,8 @@ impl Node {
             return Ok(());
         }
         let body = build_response(&query.qid, response_items(&hits), total, max);
-        let envelope = Envelope::new(
-            &self.key,
-            &self.identifier(),
-            TYPE_RESPONSE,
-            serde_json::to_value(&body)?,
-        );
-        self.send_unicast(querier, envelope, relay).await
+        self.send_payload(querier, TYPE_RESPONSE, serde_json::to_value(&body)?, relay)
+            .await
     }
 
     pub async fn request_aggregate(
@@ -561,8 +614,10 @@ impl Node {
         let Ok(value) = serde_json::to_value(&body) else {
             return;
         };
-        let envelope = Envelope::new(&self.key, &self.identifier(), TYPE_AGGREGATE, value);
-        if let Err(error) = self.send_unicast(requester_key, envelope, relay).await {
+        if let Err(error) = self
+            .send_payload(requester_key, TYPE_AGGREGATE, value, relay)
+            .await
+        {
             eprintln!("aggregate reply failed: {error}");
         }
     }
@@ -618,6 +673,82 @@ async fn poll_relay(node: Arc<Node>, relay: String) {
             continue;
         };
         let signature = node.key.sign(&poll_signing_bytes(&member, &nonce));
+        let url = format!(
+            "{}/v1/stream?member={}&nonce={}&sig={}",
+            base, member, nonce, signature
+        );
+        match node.client.get(&url).send().await {
+            Ok(response) if response.status().is_success() => {
+                if let Err(error) = consume_stream(&node, &base, response).await {
+                    eprintln!("stream from {base} ended: {error}");
+                }
+                tokio::time::sleep(Duration::from_millis(200)).await;
+            }
+            Ok(response)
+                if response.status() == reqwest::StatusCode::NOT_FOUND
+                    || response.status() == reqwest::StatusCode::METHOD_NOT_ALLOWED =>
+            {
+                long_poll_relay(&node, &base).await;
+                return;
+            }
+            Ok(response) => {
+                eprintln!("stream from {base} rejected: {}", response.status());
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+            Err(_) => {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+        }
+    }
+}
+
+async fn consume_stream(
+    node: &Arc<Node>,
+    base: &str,
+    response: reqwest::Response,
+) -> anyhow::Result<()> {
+    use futures_util::StreamExt;
+    let mut stream = response.bytes_stream();
+    let mut buffer = String::new();
+    let mut event = String::new();
+    let mut data = String::new();
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk?;
+        buffer.push_str(&String::from_utf8_lossy(&chunk));
+        while let Some(newline) = buffer.find('\n') {
+            let line = buffer[..newline].trim_end_matches('\r').to_string();
+            buffer.drain(..=newline);
+            if line.is_empty() {
+                if event == "envelope" && !data.is_empty() {
+                    if let Ok(envelope) = serde_json::from_str::<Envelope>(&data) {
+                        node.dispatch(envelope, base).await;
+                    }
+                } else if event == "lag" {
+                    eprintln!("relay {base} reports lag: {data}");
+                }
+                event.clear();
+                data.clear();
+            } else if let Some(rest) = line.strip_prefix("event:") {
+                event = rest.trim().to_string();
+            } else if let Some(rest) = line.strip_prefix("data:") {
+                if !data.is_empty() {
+                    data.push('\n');
+                }
+                data.push_str(rest.strip_prefix(' ').unwrap_or(rest));
+            }
+        }
+    }
+    Ok(())
+}
+
+async fn long_poll_relay(node: &Arc<Node>, base: &str) {
+    let member = node.key.public_hex();
+    loop {
+        let Some(nonce) = fetch_challenge(node, base, &member).await else {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            continue;
+        };
+        let signature = node.key.sign(&poll_signing_bytes(&member, &nonce));
         let url = format!(
             "{}/v1/poll?member={}&nonce={}&sig={}&timeout_ms=20000",
             base, member, nonce, signature
@@ -632,7 +763,7 @@ async fn poll_relay(node: Arc<Node>, relay: String) {
                                 if let Ok(envelope) =
                                     serde_json::from_value::<Envelope>(message.clone())
                                 {
-                                    node.dispatch(envelope, &base).await;
+                                    node.dispatch(envelope, base).await;
                                 }
                             }
                         }
@@ -696,6 +827,7 @@ async fn local_status(State(node): State<Arc<Node>>) -> Response {
     Json(json!({
         "name": node.config.node.name,
         "id": node.config.node.id,
+        "enc_key": node.enc_public,
         "registry_version": registry_version,
         "pubkey": node.key.public_hex(),
         "listen": node.config.node.listen,

src/registry.rs

@@ -8,9 +8,7 @@ use anyhow::{Context, Result, anyhow};
 use serde::{Deserialize, Serialize};
 
 use crate::PROTOCOL;
-#[cfg(test)]
-use crate::crypto::now_ts;
-use crate::crypto::{Keypair, canonical_json, verify_signature};
+use crate::crypto::{Keypair, canonical_json, now_ts, verify_signature};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct KeyEntry {
@@ -28,6 +26,8 @@ pub struct RegistryMember {
     pub class: String,
     #[serde(default)]
     pub keys: Vec<KeyEntry>,
+    #[serde(default)]
+    pub enc_key: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -182,6 +182,18 @@ impl Watcher {
             .and_then(|signed| authorized_keys(signed, now).remove(key))
     }
 
+    pub fn enc_key(&self, member_key: &str) -> Option<String> {
+        let current = self.current.lock().expect("registry lock");
+        let signed = current.as_ref()?;
+        let (id, _) = authorized_keys(signed, now_ts()).remove(member_key)?;
+        signed
+            .doc
+            .members
+            .iter()
+            .find(|member| member.id == id)
+            .and_then(|member| member.enc_key.clone())
+    }
+
     pub fn version(&self) -> Option<u64> {
         self.current
             .lock()
@@ -234,6 +246,7 @@ mod tests {
                     not_before,
                     not_after,
                 }],
+                enc_key: None,
             })
             .collect();
         sign_registry(
@@ -292,6 +305,7 @@ mod tests {
                 not_before: 0,
                 not_after: Some(now.saturating_sub(1)),
             }],
+            enc_key: None,
         });
         registry.doc.members = expired;
         let registry = sign_registry(registry.doc, &ma);

src/relay.rs

@@ -1,4 +1,5 @@
 use std::collections::{HashMap, VecDeque};
+use std::convert::Infallible;
 use std::net::SocketAddr;
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
@@ -6,12 +7,14 @@ use std::time::{Duration, Instant};
 use anyhow::{Result, anyhow};
 use axum::extract::{Query, State};
 use axum::http::{StatusCode, header};
+use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::{IntoResponse, Response};
 use axum::routing::{get, post};
 use axum::{Json, Router};
 use serde::Deserialize;
 use serde_json::json;
 use tokio::net::TcpListener;
+use tokio::sync::Notify;
 
 use crate::crypto::{now_ts, poll_signing_bytes, random_nonce, verify_signature};
 use crate::message::{Envelope, TYPE_AGGREGATE, TYPE_QUERY, TYPE_RESPONSE, timestamp_is_fresh};
@@ -68,6 +71,7 @@ pub struct Relay {
     options: RelayOptions,
     registry: Option<Arc<Watcher>>,
     client: reqwest::Client,
+    notify: Notify,
 }
 
 impl Relay {
@@ -92,6 +96,7 @@ impl Relay {
             client: reqwest::Client::builder()
                 .timeout(Duration::from_secs(5))
                 .build()?,
+            notify: Notify::new(),
         });
         if let Some(watcher) = &relay.registry {
             watcher.load_initial();
@@ -123,38 +128,45 @@ impl Relay {
     }
 
     fn push_local(&self, target: Option<&str>, envelope: Envelope) -> Result<usize, StatusCode> {
-        let mut inner = self.inner.lock().expect("relay lock");
-        inner.seq += 1;
-        let seq = inner.seq;
-        match target {
-            Some(member) => {
-                let Some(queue) = inner.members.get_mut(member) else {
-                    return Err(StatusCode::NOT_FOUND);
-                };
-                if queue.items.len() >= self.options.capacity {
-                    queue.lagging = true;
-                    queue.missed += 1;
-                    return Err(StatusCode::TOO_MANY_REQUESTS);
-                }
-                queue.items.push_back((seq, envelope));
-                Ok(1)
-            }
-            None => {
-                let publisher = envelope.key.clone();
-                inner.members.entry(publisher).or_default();
-                let mut delivered = 0;
-                for queue in inner.members.values_mut() {
-                    if queue.items.len() >= self.options.capacity {
-                        queue.lagging = true;
-                        queue.missed += 1;
-                    } else {
-                        queue.items.push_back((seq, envelope.clone()));
-                        delivered += 1;
+        let outcome = {
+            let mut inner = self.inner.lock().expect("relay lock");
+            inner.seq += 1;
+            let seq = inner.seq;
+            match target {
+                Some(member) => match inner.members.get_mut(member) {
+                    None => Err(StatusCode::NOT_FOUND),
+                    Some(queue) => {
+                        if queue.items.len() >= self.options.capacity {
+                            queue.lagging = true;
+                            queue.missed += 1;
+                            Err(StatusCode::TOO_MANY_REQUESTS)
+                        } else {
+                            queue.items.push_back((seq, envelope));
+                            Ok(1)
+                        }
                     }
+                },
+                None => {
+                    let publisher = envelope.key.clone();
+                    inner.members.entry(publisher).or_default();
+                    let mut delivered = 0;
+                    for queue in inner.members.values_mut() {
+                        if queue.items.len() >= self.options.capacity {
+                            queue.lagging = true;
+                            queue.missed += 1;
+                        } else {
+                            queue.items.push_back((seq, envelope.clone()));
+                            delivered += 1;
+                        }
+                    }
+                    Ok(delivered)
                 }
-                Ok(delivered)
             }
+        };
+        if outcome.is_ok() {
+            self.notify.notify_waiters();
         }
+        outcome
     }
 
     fn forward(&self, envelope: Envelope, origin: Option<&str>, hops: usize) {
@@ -199,6 +211,7 @@ pub fn router(relay: Arc<Relay>) -> Router {
         .route("/v1/federation", post(federation))
         .route("/v1/unicast", post(unicast))
         .route("/v1/poll", get(poll))
+        .route("/v1/stream", get(stream))
         .with_state(relay)
 }
 
@@ -377,9 +390,9 @@ struct PollParams {
     timeout_ms: Option<u64>,
 }
 
-async fn poll(State(relay): State<Arc<Relay>>, Query(params): Query<PollParams>) -> Response {
+fn authenticate(relay: &Relay, params: &PollParams) -> Result<(), Response> {
     if !valid_key(&params.member) {
-        return bad_request("valid member key required");
+        return Err(bad_request("valid member key required"));
     }
     if verify_signature(
         &params.member,
@@ -388,16 +401,72 @@ async fn poll(State(relay): State<Arc<Relay>>, Query(params): Query<PollParams>)
     )
     .is_err()
     {
-        return unauthorized("invalid poll signature");
+        return Err(unauthorized("invalid poll signature"));
     }
-    {
-        let mut inner = relay.inner.lock().expect("relay lock");
-        match inner.challenges.remove(&params.nonce) {
-            Some(challenge)
-                if challenge.member == params.member
-                    && challenge.created.elapsed() < CHALLENGE_TTL => {}
-            _ => return unauthorized("unknown, expired, or reused challenge"),
+    let mut inner = relay.inner.lock().expect("relay lock");
+    match inner.challenges.remove(&params.nonce) {
+        Some(challenge)
+            if challenge.member == params.member && challenge.created.elapsed() < CHALLENGE_TTL =>
+        {
+            Ok(())
         }
+        _ => Err(unauthorized("unknown, expired, or reused challenge")),
+    }
+}
+
+async fn stream(State(relay): State<Arc<Relay>>, Query(params): Query<PollParams>) -> Response {
+    if let Err(response) = authenticate(&relay, &params) {
+        return response;
+    }
+    let relay_for_stream = relay.clone();
+    let member = params.member.clone();
+    let events = async_stream::stream! {
+        loop {
+            let (batch, lagged): (Vec<Envelope>, Option<u64>) = {
+                let mut inner = relay_for_stream.inner.lock().expect("relay lock");
+                let queue = inner.members.entry(member.clone()).or_default();
+                if queue.lagging {
+                    let missed = queue.missed;
+                    queue.lagging = false;
+                    queue.missed = 0;
+                    queue.items.clear();
+                    (Vec::new(), Some(missed))
+                } else {
+                    (
+                        queue
+                            .items
+                            .drain(..)
+                            .map(|(_, envelope)| envelope)
+                            .collect(),
+                        None,
+                    )
+                }
+            };
+            if let Some(missed) = lagged {
+                yield Ok::<Event, Infallible>(
+                    Event::default()
+                        .event("lag")
+                        .data(json!({ "missed": missed }).to_string()),
+                );
+                continue;
+            }
+            for envelope in batch {
+                if let Ok(data) = serde_json::to_string(&envelope) {
+                    yield Ok::<Event, Infallible>(Event::default().event("envelope").data(data));
+                }
+            }
+            let _ = tokio::time::timeout(Duration::from_secs(15), relay_for_stream.notify.notified())
+                .await;
+        }
+    };
+    Sse::new(events)
+        .keep_alive(KeepAlive::default())
+        .into_response()
+}
+
+async fn poll(State(relay): State<Arc<Relay>>, Query(params): Query<PollParams>) -> Response {
+    if let Err(response) = authenticate(&relay, &params) {
+        return response;
     }
     let timeout = Duration::from_millis(params.timeout_ms.unwrap_or(25_000).min(60_000));
     let deadline = Instant::now() + timeout;