feat(sync): add typed acknowledgement effect

req: sync/006

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, PresenceScope, PresenceTracker, PresenceUpdate};
+use hemx_sync::{Channel, PresenceScope, PresenceTracker, PresenceUpdate, SyncEffect};
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
 use std::convert::Infallible;
@@ -502,6 +502,7 @@ async fn main() {
         .route("/", get(home).post(interact))
         .route("/events", get(events))
         .route("/sync/broadcast", get(sync_broadcast))
+        .route("/sync/ack", get(sync_ack))
         .route("/sync-demo", get(sync_demo))
         .route("/sync.js", get(sync_js))
         .route("/sync/context", get(sync_context))
@@ -601,6 +602,39 @@ fn presence_changed(signal: PresenceSignal) -> impl hemx::IntoEffect {
     })
 }
 
+async fn sync_ack(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    Query(params): Query<BTreeMap<String, String>>,
+) -> Result<Response, SyncRejection> {
+    let principal = current_sync_principal(&headers, &state.sync_sessions)?;
+    let command_id = CommandId::parse(params.get("command_id"))?;
+    let acknowledgement = {
+        let sync = state.sync.lock().unwrap();
+        let acknowledgement =
+            sync.acknowledgements
+                .get(&command_id)
+                .ok_or(SyncRejection::Conflict(
+                    "command has no canonical acknowledgement",
+                ))?;
+        if !visible_acknowledgement(principal, acknowledgement) {
+            return Err(SyncRejection::Forbidden(
+                "current tenant cannot access command acknowledgement",
+            ));
+        }
+        acknowledgement.clone()
+    };
+    let batch = (
+        SyncEffect::ack(board::atoms::sync_ack),
+        board::sync_status.text(format!(
+            "Canonical acknowledgement {} at server sequence {}",
+            acknowledgement.command_id, acknowledgement.server_sequence
+        )),
+    )
+        .into_batch(ui::BUILD_FINGERPRINT);
+    Ok(sse(stream::iter([Ok::<_, Infallible>(batch)]).boxed()).into_response())
+}
+
 async fn sync_broadcast(
     State(state): State<Arc<AppState>>,
     Query(params): Query<BTreeMap<String, String>>,

examples/kanban/templates/board.heml

@@ -11,5 +11,7 @@
 
   <div data-hemx-slot="board">{+= self.board =+}</div>
   <aside data-hemx-slot="presence">Waiting for presence…</aside>
+  <output id="sync-ack" data-hemx-atom="sync_ack" aria-live="polite">pending</output>
+  <output data-hemx-slot="sync_status" aria-live="polite">Waiting for acknowledgement…</output>
 
 </section>

examples/kanban/tests/browser_e2e.rs

@@ -2200,6 +2200,97 @@ async fn adversarial_wire_inputs_are_rejected_before_partial_application() -> We
     result.and(quit)
 }
 
+#[tokio::test]
+async fn canonical_acknowledgement_updates_generated_atom_over_ordinary_batch(
+) -> WebDriverResult<()> {
+    // test req: sync/006
+    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.querySelector('#sync-ack')?.textContent.trim() === 'pending'",
+        )
+        .await?;
+        let proof = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                (async () => {
+                  const commandId = 'ack-proof:1';
+                  const accepted = await fetch(`/sync/commands?command_id=${encodeURIComponent(commandId)}&card_id=1&column=done`, { method: 'POST' });
+                  const canonical = await accepted.json();
+                  const root = document.querySelector('[data-hemx-root]');
+                  let acknowledgementEvent;
+                  root.addEventListener('hemx:sync-ack', (event) => { acknowledgementEvent = event.detail; }, { once: true });
+                  const source = new EventSource(`/sync/ack?command_id=${encodeURIComponent(commandId)}`);
+                  await new Promise((resolve, reject) => {
+                    const timeout = setTimeout(() => {
+                      source.close();
+                      reject(new Error('ack batch timed out'));
+                    }, 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();
+                    });
+                    source.onerror = () => {
+                      clearTimeout(timeout);
+                      source.close();
+                      reject(new Error('ack batch failed'));
+                    };
+                  });
+                  done({
+                    acceptedStatus: accepted.status,
+                    canonical,
+                    atom: document.querySelector('#sync-ack').textContent.trim(),
+                    status: document.body.textContent,
+                    acknowledgementEvent,
+                  });
+                })().catch((error) => done({ error: String(error), stack: error?.stack }));
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert!(proof["error"].is_null(), "typed acknowledgement failed: {proof}");
+        assert_eq!(proof["acceptedStatus"], 200);
+        assert_eq!(proof["canonical"]["commandId"], "ack-proof:1");
+        assert_eq!(proof["canonical"]["serverSequence"], 1);
+        assert_eq!(proof["atom"], "acknowledged");
+        assert!(proof["status"]
+            .as_str()
+            .is_some_and(|status| status.contains("ack-proof:1 at server sequence 1")));
+        assert!(proof["acknowledgementEvent"].as_str().is_some_and(|payload| payload.contains("atomId")));
+        Ok(())
+    }
+    .await;
+    let quit = driver.quit().await;
+    result.and(quit)
+}
+
 #[tokio::test]
 async fn typed_presence_join_leave_updates_generated_atom_over_sse() -> WebDriverResult<()> {
     // test req: sync/001 req: sync/004 req: sync/005

hemx-js/runtime/hemx.js

@@ -459,6 +459,11 @@
     if (op.kind === "put") {
       if (isAtom(op.target)) {
         atomStore(scope).set(String(op.target.resource.id), op.payload.value);
+        forEachElement(scope, (element) => {
+          if (element.getAttribute("data-aid") === String(op.target.resource.id)) {
+            putPayload(element, op.payload);
+          }
+        });
         return true;
       }
       const target = targetFor(scope, op.target);

hemx-sync/src/lib.rs

@@ -1,4 +1,4 @@
-use hemx_core::{Effect, EffectBatch, IntoEffect};
+use hemx_core::{Atom, Effect, EffectBatch, IntoEffect};
 use serde::{de, Deserialize, Deserializer, Serialize};
 use std::{
     collections::{HashMap, HashSet},
@@ -12,6 +12,7 @@ pub use hemx_sync_macros::presence;
 
 pub const PATCH_SCHEMA_VERSION: u16 = 1;
 pub const PATCH_EVENT: &str = "hemx:sync-patch";
+pub const ACK_EVENT: &str = "hemx:sync-ack";
 const INTERACTION_ID: &str = "$hemx-interaction";
 pub const BROWSER_RUNTIME: &str = include_str!("../runtime/hemx-sync.js");
 
@@ -387,7 +388,7 @@ fn validate_key(key: &str) -> Result<(), PatchError> {
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
-pub struct SyncEffect(Effect); // req: sync/002
+pub struct SyncEffect(Vec<Effect>); // req: sync/002
 
 impl SyncEffect {
     // req: sync/004
@@ -398,18 +399,29 @@ impl SyncEffect {
         }
     }
 
+    // req: sync/006
+    pub fn ack<T>(atom: Atom<T>) -> Self {
+        Self(vec![
+            atom.set("acknowledged"),
+            Effect::Emit {
+                name: ACK_EVENT.to_owned(),
+                payload: format!(r#"{{"atomId":{}}}"#, atom.id().id),
+            },
+        ])
+    }
+
     pub fn send_patch(patch: FlatPatch) -> Self {
         patch.validate().expect("FlatPatch must remain valid");
-        Self(Effect::Emit {
+        Self(vec![Effect::Emit {
             name: PATCH_EVENT.to_owned(),
             payload: patch.payload(),
-        })
+        }])
     }
 }
 
 impl IntoEffect for SyncEffect {
     fn append_to(self, ops: &mut Vec<Effect>) {
-        self.0.append_to(ops);
+        ops.extend(self.0);
     }
 }
 
@@ -417,6 +429,18 @@ impl IntoEffect for SyncEffect {
 mod tests {
     use super::*;
 
+    #[test]
+    fn acknowledgement_updates_atom_and_emits_queue_signal() {
+        let atom = Atom::<String>::new(17);
+        let batch = SyncEffect::ack(atom).into_batch(hemx_core::BuildFingerprint(3));
+        assert!(
+            matches!(&batch.ops[0], Effect::Put { target, payload: hemx_core::Payload::Text(payload) } if target.resource == atom.id() && payload == "acknowledged")
+        );
+        assert!(
+            matches!(&batch.ops[1], Effect::Emit { name, payload } if name == ACK_EVENT && payload == r#"{"atomId":17}"#)
+        );
+    }
+
     #[test]
     fn presence_macro_projects_an_ordinary_effect_on_its_channel() {
         struct Signal(Channel);