feat(sync): add typed presence projection

req: sync/001

req: sync/005

Cargo.lock

@@ -616,10 +616,20 @@ name = "hemx-sync"
 version = "0.1.0"
 dependencies = [
  "hemx-core",
+ "hemx-sync-macros",
  "serde",
  "serde_json",
 ]
 
+[[package]]
+name = "hemx-sync-macros"
+version = "0.1.0"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "hemx-techdemo"
 version = "0.1.0"

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["hemx", "hemx-core", "hemx-host", "hemx-derive", "hemx-js", "hemx-axum", "hemx-build", "hemx-test", "hemx-sync", "hemx-wasm", "hemx-lsp", "hemx-xtask", "examples/v0", "examples/html_examples", "examples/kanban", "examples/client_local", "examples/techdemo", "examples/saas", "examples/workout"]
+members = ["hemx", "hemx-core", "hemx-host", "hemx-derive", "hemx-js", "hemx-axum", "hemx-build", "hemx-test", "hemx-sync", "hemx-sync-macros", "hemx-wasm", "hemx-lsp", "hemx-xtask", "examples/v0", "examples/html_examples", "examples/kanban", "examples/client_local", "examples/techdemo", "examples/saas", "examples/workout"]
 
 [workspace.package]
 version = "0.1.0"

examples/kanban/src/main.rs

@@ -15,7 +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 hemx_sync::{Channel, PresenceScope, PresenceTracker, PresenceUpdate};
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
 use std::convert::Infallible;
@@ -46,6 +46,7 @@ struct AppState {
     sync_store: Option<SyncStore>,
     sync_sessions: SyncSessionTokens,
     acknowledgement_heartbeat_interval: Duration,
+    presence: Mutex<PresenceTracker<String>>,
 }
 
 #[derive(Default)]
@@ -494,6 +495,7 @@ async fn main() {
             .filter(|milliseconds| *milliseconds > 0)
             .map(Duration::from_millis)
             .unwrap_or(ACKNOWLEDGEMENT_HEARTBEAT_INTERVAL),
+        presence: Mutex::new(PresenceTracker::default()),
     });
 
     let ordinary_routes = Router::new()
@@ -581,7 +583,28 @@ async fn interact(
 }
 
 // req: push/001 req: push/003 req: examples/001
-async fn sync_broadcast(Query(params): Query<BTreeMap<String, String>>) -> Response {
+struct PresenceSignal {
+    channel: Channel,
+    count: usize,
+}
+
+impl PresenceScope for PresenceSignal {
+    fn presence_channel(&self) -> Channel {
+        self.channel.clone()
+    }
+}
+
+#[hemx_sync::presence]
+fn presence_changed(signal: PresenceSignal) -> impl hemx::IntoEffect {
+    board::presence.put(&Presence {
+        count: u64::try_from(signal.count).expect("presence count fits u64"),
+    })
+}
+
+async fn sync_broadcast(
+    State(state): State<Arc<AppState>>,
+    Query(params): Query<BTreeMap<String, String>>,
+) -> Response {
     let Some(channel) = params
         .get("channel")
         .and_then(|channel| Channel::new(channel).ok())
@@ -591,10 +614,21 @@ async fn sync_broadcast(Query(params): Query<BTreeMap<String, String>>) -> Respo
     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 member = params.get("member").cloned();
+    let count = {
+        let mut presence = state.presence.lock().unwrap();
+        match (params.get("action").map(String::as_str), member.as_ref()) {
+            (Some("join"), Some(member)) => presence.join(channel.clone(), member.clone()).count,
+            (Some("leave"), Some(member)) => presence.leave(&channel, member).count,
+            (None | Some("snapshot"), None) => presence.count(&channel),
+            _ => {
+                return (StatusCode::BAD_REQUEST, "invalid presence action or member")
+                    .into_response();
+            }
+        }
+    };
+    let broadcast =
+        presence_changed(PresenceSignal { channel, count }).into_broadcast(ui::BUILD_FINGERPRINT);
     let (_channel, effect_batch) = broadcast.into_parts();
     sse(stream::iter([Ok::<_, Infallible>(effect_batch)]).boxed()).into_response()
 }

examples/kanban/tests/browser_e2e.rs

@@ -2201,8 +2201,8 @@ async fn adversarial_wire_inputs_are_rejected_before_partial_application() -> We
 }
 
 #[tokio::test]
-async fn typed_broadcast_applies_generated_batch_over_sse() -> WebDriverResult<()> {
-    // test req: sync/004
+async fn typed_presence_join_leave_updates_generated_atom_over_sse() -> WebDriverResult<()> {
+    // test req: sync/001 req: sync/004 req: sync/005
     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"));
@@ -2222,25 +2222,57 @@ async fn typed_broadcast_applies_generated_batch_over_sse() -> WebDriverResult<(
 
     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'))",
+        wait_until(&driver, "return document.body.textContent.includes('tick #0')").await?;
+        let proof = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                (async () => {
+                  const root = document.querySelector('[data-hemx-root]');
+                  const apply = (url) => new Promise((resolve, reject) => {
+                    const source = new EventSource(url);
+                    const timeout = setTimeout(() => {
+                      source.close();
+                      reject(new Error(`presence event timed out: ${url}`));
+                    }, 5000);
+                    source.addEventListener('hemx', (event) => {
+                      clearTimeout(timeout);
+                      const normalized = event.data.replace(/-/g, '+').replace(/_/g, '/');
+                      const padded = normalized + '='.repeat((4 - normalized.length % 4) % 4);
+                      const raw = atob(padded);
+                      const bytes = Uint8Array.from(raw, (character) => character.charCodeAt(0));
+                      window.hemx.applyBatch(bytes.buffer, root);
+                      source.close();
+                      resolve(document.body.textContent);
+                    });
+                    source.onerror = () => {
+                      clearTimeout(timeout);
+                      source.close();
+                      reject(new Error(`presence event failed: ${url}`));
+                    };
+                  });
+                  const joinedAda = await apply('/sync/broadcast?channel=board&action=join&member=ada');
+                  const duplicateAda = await apply('/sync/broadcast?channel=board&action=join&member=ada');
+                  const joinedGrace = await apply('/sync/broadcast?channel=board&action=join&member=grace');
+                  const leftAda = await apply('/sync/broadcast?channel=board&action=leave&member=ada');
+                  done({
+                    joinedAda: joinedAda.includes('tick #1'),
+                    duplicateAda: duplicateAda.includes('tick #1'),
+                    joinedGrace: joinedGrace.includes('tick #2'),
+                    leftAda: leftAda.includes('tick #1'),
+                  });
+                })().catch((error) => done({ error: String(error), stack: error?.stack }));
+                "#,
                 Vec::new(),
             )
             .await?
             .json()
             .clone();
-        assert!(
-            resources
-                .as_array()
-                .is_some_and(|resources| !resources.is_empty()),
-            "typed broadcast SSE request was not observed: {resources}"
-        );
+        assert!(proof["error"].is_null(), "typed presence failed: {proof}");
+        assert_eq!(proof["joinedAda"], true, "{proof}");
+        assert_eq!(proof["duplicateAda"], true, "{proof}");
+        assert_eq!(proof["joinedGrace"], true, "{proof}");
+        assert_eq!(proof["leftAda"], true, "{proof}");
         Ok(())
     }
     .await;

hemx-sync-macros/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "hemx-sync-macros"
+version.workspace = true
+edition.workspace = true
+
+[lib]
+proc-macro = true
+
+[dependencies]
+proc-macro2 = "1"
+quote = "1"
+syn = { version = "2", features = ["full"] }

hemx-sync-macros/src/lib.rs

@@ -0,0 +1,65 @@
+use proc_macro::TokenStream;
+use quote::quote;
+use syn::{parse_macro_input, Error, FnArg, ItemFn, Pat, ReturnType};
+
+#[proc_macro_attribute]
+pub fn presence(attributes: TokenStream, item: TokenStream) -> TokenStream {
+    if !attributes.is_empty() {
+        return Error::new(
+            proc_macro2::Span::call_site(),
+            "#[hemx_sync::presence] does not accept arguments",
+        )
+        .to_compile_error()
+        .into();
+    }
+
+    let mut function = parse_macro_input!(item as ItemFn);
+    if function.sig.asyncness.is_some() {
+        return Error::new_spanned(
+            function.sig.asyncness,
+            "presence projections must be synchronous",
+        )
+        .to_compile_error()
+        .into();
+    }
+    if matches!(function.sig.output, ReturnType::Default) {
+        return Error::new_spanned(
+            &function.sig,
+            "presence projections must return impl IntoEffect",
+        )
+        .to_compile_error()
+        .into();
+    }
+    let argument = match function.sig.inputs.first() {
+        Some(FnArg::Typed(argument)) if function.sig.inputs.len() == 1 => argument,
+        _ => {
+            return Error::new_spanned(
+                &function.sig.inputs,
+                "presence projections require exactly one typed presence argument",
+            )
+            .to_compile_error()
+            .into();
+        }
+    };
+    let argument_name = match argument.pat.as_ref() {
+        Pat::Ident(argument) => argument.ident.clone(),
+        pattern => {
+            return Error::new_spanned(
+                pattern,
+                "presence projection argument must be a simple identifier",
+            )
+            .to_compile_error()
+            .into();
+        }
+    };
+
+    let body = function.block;
+    function.sig.output = syn::parse_quote!(-> impl ::hemx_sync::PresenceUpdate);
+    function.block = Box::new(syn::parse_quote!({
+        let __hemx_sync_channel =
+            ::hemx_sync::PresenceScope::presence_channel(&#argument_name);
+        let __hemx_sync_effect = (|| #body)();
+        ::hemx_sync::PresenceProjection::new(__hemx_sync_channel, __hemx_sync_effect)
+    }));
+    quote!(#function).into()
+}

hemx-sync/Cargo.toml

@@ -5,6 +5,7 @@ edition.workspace = true
 
 [dependencies]
 hemx-core = { path = "../hemx-core" }
+hemx-sync-macros = { path = "../hemx-sync-macros" }
 serde = { version = "1", features = ["derive"] }
 
 [dev-dependencies]

hemx-sync/src/lib.rs

@@ -1,6 +1,14 @@
 use hemx_core::{Effect, EffectBatch, IntoEffect};
 use serde::{de, Deserialize, Deserializer, Serialize};
-use std::{error::Error, fmt};
+use std::{
+    collections::{HashMap, HashSet},
+    error::Error,
+    fmt,
+    hash::Hash,
+};
+
+extern crate self as hemx_sync;
+pub use hemx_sync_macros::presence;
 
 pub const PATCH_SCHEMA_VERSION: u16 = 1;
 pub const PATCH_EVENT: &str = "hemx:sync-patch";
@@ -54,6 +62,99 @@ impl fmt::Display for ChannelError {
 
 impl Error for ChannelError {}
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PresenceChange {
+    pub changed: bool,
+    pub count: usize,
+}
+
+#[derive(Clone, Debug)]
+pub struct PresenceTracker<Member> {
+    members: HashMap<Channel, HashSet<Member>>,
+}
+
+impl<Member> Default for PresenceTracker<Member> {
+    fn default() -> Self {
+        Self {
+            members: HashMap::new(),
+        }
+    }
+}
+
+impl<Member> PresenceTracker<Member>
+where
+    Member: Eq + Hash,
+{
+    pub fn join(&mut self, channel: Channel, member: Member) -> PresenceChange {
+        let members = self.members.entry(channel).or_default();
+        PresenceChange {
+            changed: members.insert(member),
+            count: members.len(),
+        }
+    }
+
+    pub fn leave(&mut self, channel: &Channel, member: &Member) -> PresenceChange {
+        let Some(members) = self.members.get_mut(channel) else {
+            return PresenceChange {
+                changed: false,
+                count: 0,
+            };
+        };
+        let changed = members.remove(member);
+        let count = members.len();
+        if members.is_empty() {
+            self.members.remove(channel);
+        }
+        PresenceChange { changed, count }
+    }
+
+    pub fn count(&self, channel: &Channel) -> usize {
+        self.members.get(channel).map_or(0, HashSet::len)
+    }
+}
+
+pub trait PresenceScope {
+    fn presence_channel(&self) -> Channel;
+}
+
+pub struct PresenceProjection<Effect> {
+    channel: Channel,
+    effect: Effect,
+}
+
+impl<Effect> PresenceProjection<Effect> {
+    pub fn new(channel: Channel, effect: Effect) -> Self {
+        Self { channel, effect }
+    }
+}
+
+pub trait PresenceUpdate: IntoEffect + Sized {
+    fn presence_channel(&self) -> &Channel;
+    fn into_broadcast(self, fingerprint: hemx_core::BuildFingerprint) -> Broadcast;
+}
+
+impl<Effect> PresenceUpdate for PresenceProjection<Effect>
+where
+    Effect: IntoEffect,
+{
+    fn presence_channel(&self) -> &Channel {
+        &self.channel
+    }
+
+    fn into_broadcast(self, fingerprint: hemx_core::BuildFingerprint) -> Broadcast {
+        SyncEffect::broadcast(self.channel, self.effect.into_batch(fingerprint))
+    }
+}
+
+impl<Effect> IntoEffect for PresenceProjection<Effect>
+where
+    Effect: IntoEffect,
+{
+    fn append_to(self, ops: &mut Vec<hemx_core::Effect>) {
+        self.effect.append_to(ops);
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct Broadcast {
     channel: Channel,
@@ -316,6 +417,53 @@ impl IntoEffect for SyncEffect {
 mod tests {
     use super::*;
 
+    #[test]
+    fn presence_macro_projects_an_ordinary_effect_on_its_channel() {
+        struct Signal(Channel);
+        impl PresenceScope for Signal {
+            fn presence_channel(&self) -> Channel {
+                self.0.clone()
+            }
+        }
+        #[presence]
+        fn project(signal: Signal) -> impl IntoEffect {
+            Effect::Emit {
+                name: "presence".to_owned(),
+                payload: signal.0.as_str().to_owned(),
+            }
+        }
+
+        let update = project(Signal(Channel::new("board").unwrap()));
+        assert_eq!(update.presence_channel().as_str(), "board");
+        let broadcast = update.into_broadcast(hemx_core::BuildFingerprint(9));
+        assert_eq!(broadcast.channel().as_str(), "board");
+        assert_eq!(broadcast.effect_batch().ops.len(), 1);
+    }
+
+    #[test]
+    fn presence_tracker_is_idempotent_and_channel_scoped() {
+        let alpha = Channel::new("board:alpha").unwrap();
+        let beta = Channel::new("board:beta").unwrap();
+        let mut tracker = PresenceTracker::default();
+        assert_eq!(
+            tracker.join(alpha.clone(), "ada"),
+            PresenceChange {
+                changed: true,
+                count: 1
+            }
+        );
+        assert_eq!(
+            tracker.join(alpha.clone(), "ada"),
+            PresenceChange {
+                changed: false,
+                count: 1
+            }
+        );
+        assert_eq!(tracker.join(beta.clone(), "ada").count, 1);
+        assert_eq!(tracker.leave(&alpha, &"ada").count, 0);
+        assert_eq!(tracker.count(&beta), 1);
+    }
+
     #[test]
     fn broadcast_preserves_typed_channel_and_ordinary_batch() {
         let channel = Channel::new("board:alpha").unwrap();