feat(sync): add typed broadcast primitive

req: sync/004

examples/kanban/src/main.rs

@@ -15,6 +15,7 @@ use hemx_axum::{
 use hemx_kanban_example::ui::board::{self as board};
 use hemx_kanban_example::ui::board_card as card_board;
 use hemx_kanban_example::ui::{self, board as board_ui};
+use hemx_sync::{Channel, SyncEffect};
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
 use std::convert::Infallible;
@@ -498,6 +499,7 @@ async fn main() {
     let ordinary_routes = Router::new()
         .route("/", get(home).post(interact))
         .route("/events", get(events))
+        .route("/sync/broadcast", get(sync_broadcast))
         .route("/sync-demo", get(sync_demo))
         .route("/sync.js", get(sync_js))
         .route("/sync/context", get(sync_context))
@@ -579,6 +581,24 @@ async fn interact(
 }
 
 // req: push/001 req: push/003 req: examples/001
+async fn sync_broadcast(Query(params): Query<BTreeMap<String, String>>) -> Response {
+    let Some(channel) = params
+        .get("channel")
+        .and_then(|channel| Channel::new(channel).ok())
+    else {
+        return (StatusCode::BAD_REQUEST, "missing or invalid sync channel").into_response();
+    };
+    if channel.as_str() != "board" {
+        return StatusCode::NOT_FOUND.into_response();
+    }
+    let batch = board::presence
+        .put(&Presence { count: 7 })
+        .into_batch(ui::BUILD_FINGERPRINT);
+    let broadcast = SyncEffect::broadcast(channel, batch);
+    let (_channel, effect_batch) = broadcast.into_parts();
+    sse(stream::iter([Ok::<_, Infallible>(effect_batch)]).boxed()).into_response()
+}
+
 async fn events(Query(params): Query<BTreeMap<String, String>>) -> impl IntoResponse {
     if params.contains_key("once") {
         let effect = board::presence.put(&Presence { count: 1 });

examples/kanban/templates/board.heml

@@ -1,4 +1,4 @@
-<section data-hemx-root="kanban" data-hemx-sse="/events">
+<section data-hemx-root="kanban" data-hemx-sse="/sync/broadcast?channel=board">
   <header>
     <h1>hemx Kanban</h1>
     <form data-hemx-handle="create_card" data-hemx-form="create_card" data-hemx-disable-while-pending>

examples/kanban/tests/browser_e2e.rs

@@ -2200,6 +2200,54 @@ async fn adversarial_wire_inputs_are_rejected_before_partial_application() -> We
     result.and(quit)
 }
 
+#[tokio::test]
+async fn typed_broadcast_applies_generated_batch_over_sse() -> WebDriverResult<()> {
+    // test req: sync/004
+    let app_port = available_port();
+    let app_addr = format!("127.0.0.1:{app_port}");
+    let mut app_command = Command::new(env!("CARGO_BIN_EXE_hemx-kanban-example"));
+    app_command.env("HEMX_KANBAN_ADDR", &app_addr);
+    let _app = TestProcess::start(app_command, "hemx-kanban", &app_addr, STARTUP_TIMEOUT)
+        .expect("start hemx-kanban");
+
+    let webdriver_port = available_port();
+    let webdriver_addr = format!("127.0.0.1:{webdriver_port}");
+    let mut webdriver = Command::new("geckodriver");
+    webdriver.arg("--port").arg(webdriver_port.to_string());
+    let _webdriver = TestProcess::start(webdriver, "geckodriver", &webdriver_addr, STARTUP_TIMEOUT)
+        .expect("start ready geckodriver");
+    let mut caps = DesiredCapabilities::firefox();
+    caps.set_headless()?;
+    let driver = WebDriver::new(&format!("http://{webdriver_addr}"), caps).await?;
+
+    let result = async {
+        driver.goto(&format!("http://{app_addr}/")).await?;
+        wait_until(
+            &driver,
+            "return document.body.textContent.includes('tick #7')",
+        )
+        .await?;
+        let resources = driver
+            .execute(
+                "return performance.getEntriesByType('resource').map(entry => entry.name).filter(name => name.includes('/sync/broadcast'))",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert!(
+            resources
+                .as_array()
+                .is_some_and(|resources| !resources.is_empty()),
+            "typed broadcast SSE request was not observed: {resources}"
+        );
+        Ok(())
+    }
+    .await;
+    let quit = driver.quit().await;
+    result.and(quit)
+}
+
 #[tokio::test]
 async fn ordinary_browser_request_exposes_deadline_and_cancels_on_pagehide() -> WebDriverResult<()>
 {

hemx-sync/src/lib.rs

@@ -1,4 +1,4 @@
-use hemx_core::{Effect, IntoEffect};
+use hemx_core::{Effect, EffectBatch, IntoEffect};
 use serde::{de, Deserialize, Deserializer, Serialize};
 use std::{error::Error, fmt};
 
@@ -7,6 +7,73 @@ pub const PATCH_EVENT: &str = "hemx:sync-patch";
 const INTERACTION_ID: &str = "$hemx-interaction";
 pub const BROWSER_RUNTIME: &str = include_str!("../runtime/hemx-sync.js");
 
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct Channel(String);
+
+impl Channel {
+    pub fn new(value: impl Into<String>) -> Result<Self, ChannelError> {
+        let value = value.into();
+        if value.is_empty() {
+            return Err(ChannelError::Empty);
+        }
+        if value.len() > 128 {
+            return Err(ChannelError::TooLong);
+        }
+        if !value
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b':' | b'_' | b'-' | b'.'))
+        {
+            return Err(ChannelError::InvalidCharacter);
+        }
+        Ok(Self(value))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ChannelError {
+    Empty,
+    TooLong,
+    InvalidCharacter,
+}
+
+impl fmt::Display for ChannelError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Empty => formatter.write_str("sync channel must not be empty"),
+            Self::TooLong => formatter.write_str("sync channel is too long"),
+            Self::InvalidCharacter => {
+                formatter.write_str("sync channel contains an invalid character")
+            }
+        }
+    }
+}
+
+impl Error for ChannelError {}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct Broadcast {
+    channel: Channel,
+    effect_batch: EffectBatch,
+}
+
+impl Broadcast {
+    pub fn channel(&self) -> &Channel {
+        &self.channel
+    }
+
+    pub fn effect_batch(&self) -> &EffectBatch {
+        &self.effect_batch
+    }
+
+    pub fn into_parts(self) -> (Channel, EffectBatch) {
+        (self.channel, self.effect_batch)
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(untagged)]
 pub enum PatchValue {
@@ -222,6 +289,14 @@ fn validate_key(key: &str) -> Result<(), PatchError> {
 pub struct SyncEffect(Effect); // req: sync/002
 
 impl SyncEffect {
+    // req: sync/004
+    pub fn broadcast(channel: Channel, effect_batch: EffectBatch) -> Broadcast {
+        Broadcast {
+            channel,
+            effect_batch,
+        }
+    }
+
     pub fn send_patch(patch: FlatPatch) -> Self {
         patch.validate().expect("FlatPatch must remain valid");
         Self(Effect::Emit {
@@ -241,6 +316,22 @@ impl IntoEffect for SyncEffect {
 mod tests {
     use super::*;
 
+    #[test]
+    fn broadcast_preserves_typed_channel_and_ordinary_batch() {
+        let channel = Channel::new("board:alpha").unwrap();
+        let batch = EffectBatch {
+            abi_version: 1,
+            fingerprint: hemx_core::BuildFingerprint(7),
+            ops: vec![],
+        };
+        let broadcast = SyncEffect::broadcast(channel.clone(), batch.clone());
+        assert_eq!(broadcast.into_parts(), (channel, batch));
+        assert_eq!(
+            Channel::new("board alpha"),
+            Err(ChannelError::InvalidCharacter)
+        );
+    }
+
     #[test]
     fn schema_is_flat_and_rejects_reserved_keys() {
         let patch = FlatPatch::new(