feat(sync): replay durable projections from Rust

req: ms/001

req: ms/002

req: ms/003

hemx-sync/runtime/hemx-sync.js

@@ -62,6 +62,28 @@ function validatePatch(patch) {
   return patch;
 }
 
+function validateProjection(projection) {
+  if (!Array.isArray(projection)
+      || projection.length > 1024 * 1024
+      || projection.some((byte) => !Number.isInteger(byte) || byte < 0 || byte > 255)) {
+    throw new Error("invalid durable projection");
+  }
+  return projection;
+}
+
+function normalizeEvent(payload) {
+  if (payload && Object.getPrototypeOf(payload) === Object.prototype && "patch" in payload) {
+    const keys = Object.keys(payload).sort();
+    if (keys.length !== 2 || keys[0] !== "patch" || keys[1] !== "projection") {
+      throw new Error("durable patch fields do not match schema");
+    }
+    const patch = validatePatch(payload.patch);
+    return { idempotencyKey: patch.idempotencyKey, patch, projection: validateProjection(payload.projection) };
+  }
+  const patch = validatePatch(payload);
+  return { idempotencyKey: patch.idempotencyKey, patch, projection: null };
+}
+
 async function allPatches() {
   const transaction = database.transaction(STORE, "readonly");
   const done = transactionDone(transaction);
@@ -70,10 +92,21 @@ async function allPatches() {
   return patches.sort((left, right) => left.queuedAt - right.queuedAt || left.idempotencyKey.localeCompare(right.idempotencyKey));
 }
 
-async function persist(patch) {
+function normalizeStoredRecord(stored) {
+  if (stored.patch) {
+    return {
+      patch: validatePatch(stored.patch),
+      projection: stored.projection === null ? null : validateProjection(stored.projection),
+    };
+  }
+  const { queuedAt: _queuedAt, ...legacyPatch } = stored;
+  return { patch: validatePatch(legacyPatch), projection: null };
+}
+
+async function persist(record) {
   const transaction = database.transaction(STORE, "readwrite");
   const done = transactionDone(transaction);
-  transaction.objectStore(STORE).add({ ...patch, queuedAt: Date.now() });
+  transaction.objectStore(STORE).add({ ...record, queuedAt: Date.now() });
   await done;
   root?.setAttribute("data-hemx-sync-pending", String((await allPatches()).length));
 }
@@ -90,7 +123,7 @@ async function pump() {
   pumping = true;
   try {
     for (const stored of await allPatches()) {
-      const { queuedAt: _queuedAt, ...patch } = stored;
+      const { patch } = normalizeStoredRecord(stored);
       const endpoint = root?.getAttribute("data-sync-endpoint") || "/sync/patches";
       const response = await fetch(endpoint, {
         method: "POST",
@@ -122,18 +155,25 @@ async function pump() {
 async function start() {
   if (!root) return;
   database = await openDatabase();
-  root.setAttribute("data-hemx-sync-ready", "");
-  root.setAttribute("data-hemx-sync-pending", String((await allPatches()).length));
+  const pending = await allPatches();
+  for (const stored of pending) {
+    const { projection } = normalizeStoredRecord(stored);
+    if (projection) {
+      window.hemx?.applyBatch(Uint8Array.from(projection).buffer, root);
+    }
+  }
   document.addEventListener(EVENT, async (event) => {
     try {
-      const patch = validatePatch(JSON.parse(event.detail));
-      await persist(patch);
+      const record = normalizeEvent(JSON.parse(event.detail));
+      await persist(record);
       await pump();
     } catch (error) {
       root.setAttribute("data-hemx-sync-error", error instanceof Error ? error.message : String(error));
     }
   });
   window.addEventListener("online", () => pump());
+  root.setAttribute("data-hemx-sync-pending", String(pending.length));
+  root.setAttribute("data-hemx-sync-ready", "");
   await pump();
 }
 

hemx-sync/src/lib.rs

@@ -417,6 +417,29 @@ impl SyncEffect {
             payload: patch.payload(),
         }])
     }
+
+    /// Apply an optimistic projection now and carry the same ordinary batch in
+    /// the durable patch event so the framework sync runtime can replay it
+    /// after reload before acknowledgement.
+    pub fn durable(patch: FlatPatch, projection: EffectBatch) -> Self {
+        patch.validate().expect("FlatPatch must remain valid");
+        let projection_wire = projection
+            .to_wire()
+            .iter()
+            .map(u8::to_string)
+            .collect::<Vec<_>>()
+            .join(",");
+        let payload = format!(
+            r#"{{"patch":{},"projection":[{projection_wire}]}}"#,
+            patch.payload()
+        );
+        let mut ops = projection.ops;
+        ops.push(Effect::Emit {
+            name: PATCH_EVENT.to_owned(),
+            payload,
+        });
+        Self(ops)
+    }
 }
 
 impl IntoEffect for SyncEffect {
@@ -429,6 +452,24 @@ impl IntoEffect for SyncEffect {
 mod tests {
     use super::*;
 
+    #[test]
+    fn durable_patch_carries_and_applies_ordinary_projection_batch() {
+        let patch =
+            FlatPatch::for_interaction("cardColumn", PatchValue::String("done".into())).unwrap();
+        let projection = Effect::Emit {
+            name: "projected".into(),
+            payload: "card:1".into(),
+        }
+        .into_batch(hemx_core::BuildFingerprint(7));
+        let batch =
+            SyncEffect::durable(patch, projection).into_batch(hemx_core::BuildFingerprint(7));
+        assert_eq!(batch.ops.len(), 2);
+        assert!(matches!(&batch.ops[0], Effect::Emit { name, .. } if name == "projected"));
+        assert!(
+            matches!(&batch.ops[1], Effect::Emit { name, payload } if name == PATCH_EVENT && payload.contains("\"projection\":[") && payload.contains("$hemx-interaction"))
+        );
+    }
+
     #[test]
     fn acknowledgement_updates_atom_and_emits_queue_signal() {
         let atom = Atom::<String>::new(17);