From 87296632bcc9e0d0641650349c2eda3c9432fad8 Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 22:54:12 +0200 Subject: [PATCH] feat(sync): add typed broadcast primitive req: sync/004 --- PLAN.md | 2 +- examples/kanban/src/main.rs | 20 ++++++ examples/kanban/templates/board.heml | 2 +- examples/kanban/tests/browser_e2e.rs | 48 ++++++++++++++ hemx-sync/src/lib.rs | 93 +++++++++++++++++++++++++++- 5 files changed, 162 insertions(+), 3 deletions(-) diff --git a/PLAN.md b/PLAN.md index 9aaa085..9bc4f26 100644 --- a/PLAN.md +++ b/PLAN.md @@ -46,7 +46,7 @@ encryption, retention, backup, and deployment policy remain host concerns. - **Build:** materialize `hemx-sync` over an integration transport with idempotent server command processing, snapshot/change cursor, durable acknowledgements, bounded ordered replay, current auth checks, rejection/conflict results, canonical replacement, reconnect jitter/backoff, multi-tab coordination, and redacted diagnostics. - **Refusals:** no default CRDT, transport in core, cached enqueue-time permission, unbounded queue, or silent last-write-wins policy. - **Requirements:** `sync/001-023`, `operations/001-005`, `security/002-005`, `performance/004-005`. -- **Proof:** `cargo test -p hemx-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` proves `sync/002` and `sync/003` through generated real WASM: `SyncEffect::send_patch` travels beside ordinary DOM effects, the strict flat scalar schema is accepted, the offline interaction performs zero patch requests while remaining durable, and reconnect uploads with identical opaque operation/idempotency identity before matching acknowledgement removes it. `cargo test -p hemx-wasm --test browser kanban_public_api_offline_sync_journey_converges_without_duplicate_replay -- --exact` proves `sync/023` through the public generated client handler and real WASM: offline durable mutation, unavailable-origin reload projection, reconnect upload, canonical convergence, duplicate idempotency without a second event, conflicting duplicate rejection, invalid-command rejection, and an empty local queue. `cargo test -p hemx-kanban-example --test browser_e2e idempotent_server_command_is_acknowledged_after_reconnect -- --exact` proves duplicate POST delivery yields one identical canonical acknowledgement/sequence, conflicting id reuse is rejected, EventSource reconnects after a server-closed first stream, the acknowledgement is delivered once with its sequence as event id, and a page reload shows the authoritative card in the canonical column. `cargo test -p hemx-kanban-example --test browser_e2e pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack -- --exact` proves automatic platform-store upload, one explicit transient failure, bounded exponential backoff plus jitter, reconnect acknowledgement, pending-before-ack ordering, acknowledged removal, canonical board convergence, and non-retried 409 rejection remaining durable with a visible reason. `cargo test -p hemx-kanban-example --test browser_e2e canonical_acknowledgement_survives_server_restart -- --exact` proves the versioned store is materialized before success, a real process restart reloads the same idempotent acknowledgement/sequence, EventSource replays it by id, and canonical board state is rebuilt. `cargo test -p hemx-kanban-example --test browser_e2e exhausted_offline_retries_keep_command_until_later_reconnect -- --exact` proves three bounded retries exhaust into visible offline/manual-recovery state while the command remains durable, then a later retry acknowledges/removes it and converges canonically. `cargo test -p hemx-kanban-example --test browser_e2e missing_history_rebase_and_user_conflict_resolution_preserve_suffix -- --exact` proves retained-history gap detection, typed/versioned snapshot fallback, deterministic already-canonical convergence, atomic snapshot/cursor commit, then a divergent canonical update producing explicit conflict with a two-command queue; the user’s canonical-state action removes only the conflicted head, retains and resumes the suffix, and reaches canonical convergence with an empty queue. `cargo test -p hemx-kanban-example --test browser_e2e keep_local_retry_preserves_conflicted_command_and_suffix_order -- --exact` proves the keep-local action derives one resolution identity; a permanent rejection preserves order and re-enables the choice; three later transport failures still preserve the original conflicted head plus suffix and expose manual recovery; final acceptance removes only the head and converges the untouched suffix without duplication or loss. `cargo test -p hemx-kanban-example --test browser_e2e redacted_sync_diagnostics_are_bounded_and_leak_no_sensitive_material -- --exact` proves a 15-second-old queue is bucketed rather than timestamped, cursor and acknowledgement latency advance after recovery, diagnostics stay under a fixed small readout, and command id/payload/actor/session/cookie material never reaches the diagnostics surface; the conflict and keep-local tests additionally assert conflict/rejection counters at their actual failure boundaries. `cargo test -p hemx-kanban-example --test browser_e2e two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_application -- --exact` proves one retry-exhausted leader/one explicit standby, zero follower upload before takeover, lease takeover after the leader closes, one canonical sequence/event, one queue removal, and one board application. `cargo test -p hemx-kanban-example --test browser_e2e upload_backpressure_keeps_pending_work_visible_and_recoverable -- --exact` proves one in-flight upload, a two-acknowledgement activation limit, one retained durable command with visible recovery state, and explicit retry draining the final command without loss. `cargo test -p hemx-kanban-example --test browser_e2e mixed_queue_removes_accepted_prefix_and_retains_rejected_tail -- --exact` proves an accepted prefix is canonically applied and removed once before a permanent rejection stops processing, exposes its typed HTTP/server cause, disables blind retry, and leaves both the rejected command and untouched suffix durable. `cargo test -p hemx-kanban-example --test browser_e2e schema_upgrade_preserves_queued_order_and_local_intent -- --exact` proves a three-command v1 queue migrates atomically to the explicit-target v2 schema, remains byte-for-intent ordered after interrupted upload, then receives canonical sequences 1–3 in original order and drains without loss. `cargo test -p hemx-kanban-example --test browser_e2e account_partition_hides_replay_and_export_until_owner_returns -- --exact` proves a beta editor and alpha viewer enumerate zero commands and issue no replay for an alpha owner queue, signed-out startup cannot open a partition, no foreign id/export surface leaks, and only the returning alpha owner can export the intact command then resume it exactly once. `cargo test -p hemx-kanban-example --test browser_e2e canonical_snapshot_and_history_are_tenant_scoped -- --exact` proves alpha and beta sessions receive only their own canonical cards and acknowledgement events with independent sequence 1 cursors, while signed-out snapshot/history reads receive 401 and expose no stream. `cargo test -p hemx-kanban-example --test browser_e2e identical_sync_inputs_reconcile_deterministically -- --exact` proves `sync/022` through a real browser and live sync endpoints: one accepted server result and canonical snapshot plus an ordered command sequence produce byte-equivalent structured reconciliation records on independent cloned inputs, do not mutate inputs, and select the explicit server-authoritative model. `cargo test -p hemx-kanban-example --test browser_e2e sync_requests_timeout_and_cancel_on_pagehide -- --exact` proves the sync request boundary enforces a named finite timeout and reports `TimeoutError`, while page teardown cancels a pending request with `AbortError`, closes the acknowledgement stream, and clears retry/lease timers. Per-requirement audit: every `sync/001-023` row now has concrete browser proof; existing live-browser proofs also cover `operations/001-002`, `operations/005`, `security/002`, `security/004`, `performance/004`, and `performance/005`, while this slice closes the sync request/stream leg of `operations/003`. `cargo test -p hemx-kanban-example --test browser_e2e adversarial_wire_inputs_are_rejected_before_partial_application -- --exact` proves malformed, oversized, truncated, trailing, unknown-version, and invalid-kind batch/state/command inputs fail without UI mutation; `cargo test -p hemx-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` additionally proves unknown-version and oversized event/state inputs are rejected at the generated real-WASM handler boundary before application code. The runtime now bounds total payloads, fields, vectors, varints, UTF-8, enums, and option discriminants before allocation or dispatch, and the durable queue validates every loaded command before replay, closing `security/005`. `cargo test -p hemx-axum --test response interaction_boundary_honors_media_type_and_host_body_limit -- --exact` proves the mutation extractor rejects missing/unexpected media types with 415 and an Axum host-configured oversized body with 413 before handler dispatch, while a valid bounded form dispatches exactly once; the extractor documentation names `DefaultBodyLimit`, and CSRF remains the existing documented application/middleware policy, closing `security/003`. `cargo test -p hemx-kanban-example --test browser_e2e acknowledgement_stream_bounds_reconnect_buffering_heartbeat_and_cancellation -- --exact` closes `operations/004`: the live stream exposes a finite 15-second production heartbeat (25 ms under proof configuration), exactly three bounded reconnect delays (100/250/500 ms), converts 65 pending events over its 64-event transport budget into one canonical snapshot recovery event, then emits heartbeats after recovery; both proof streams are explicitly closed, and the page teardown path closes its owned stream and reports cancellation. `cargo test -p hemx-kanban-example --bin hemx-kanban-example timeout -- --nocapture` completes `operations/003`: ordinary server routes are cancelled after a finite 10-second production deadline and return 504, startup store read/replay is cancelled after five seconds, oversized startup state is rejected before decoding, and focused tests prove dropped in-flight work plus named timeout diagnostics. `cargo test -p hemx-kanban-example --test browser_e2e ordinary_browser_request_exposes_deadline_and_cancels_on_pagehide -- --exact` proves the ordinary browser request boundary exposes its finite 10-second deadline, aborts the real form request with `AbortError` on page teardown, and clears pending UI state; stream routes remain governed by the `operations/004` lifecycle rather than the ordinary-handler timeout. The complete Slice 4 browser audit now passes (`cargo test -p hemx-kanban-example --test browser_e2e -- --test-threads=1` and `cargo test -p hemx-wasm --test browser -- --test-threads=1`). The audit also fixed three stale v1 command fixtures to exercise the real v2 durable schema, preserved an offline leader lease without retrying while connectivity is absent, and made finite acknowledgement responses close unless the caller explicitly requests a reconnecting stream. Slice 4 cannot close yet: citations labeled `sync/001` and `sync/004-006` prove app-local equivalents, not the required public `hemx-sync` presence tracking, `SyncEffect::broadcast(channel, effect_batch)`, `#[hemx_sync::presence]`, or `SyncEffect::ack(atom)` APIs; those symbols do not exist in `hemx-sync`. The next coherent slice is `sync/004`: add the public typed broadcast primitive and prove one generated `EffectBatch` crosses the real SSE/browser path before implementing presence or acknowledgement wrappers. +- **Proof:** `cargo test -p hemx-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` proves `sync/002` and `sync/003` through generated real WASM: `SyncEffect::send_patch` travels beside ordinary DOM effects, the strict flat scalar schema is accepted, the offline interaction performs zero patch requests while remaining durable, and reconnect uploads with identical opaque operation/idempotency identity before matching acknowledgement removes it. `cargo test -p hemx-wasm --test browser kanban_public_api_offline_sync_journey_converges_without_duplicate_replay -- --exact` proves `sync/023` through the public generated client handler and real WASM: offline durable mutation, unavailable-origin reload projection, reconnect upload, canonical convergence, duplicate idempotency without a second event, conflicting duplicate rejection, invalid-command rejection, and an empty local queue. `cargo test -p hemx-kanban-example --test browser_e2e idempotent_server_command_is_acknowledged_after_reconnect -- --exact` proves duplicate POST delivery yields one identical canonical acknowledgement/sequence, conflicting id reuse is rejected, EventSource reconnects after a server-closed first stream, the acknowledgement is delivered once with its sequence as event id, and a page reload shows the authoritative card in the canonical column. `cargo test -p hemx-kanban-example --test browser_e2e pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack -- --exact` proves automatic platform-store upload, one explicit transient failure, bounded exponential backoff plus jitter, reconnect acknowledgement, pending-before-ack ordering, acknowledged removal, canonical board convergence, and non-retried 409 rejection remaining durable with a visible reason. `cargo test -p hemx-kanban-example --test browser_e2e canonical_acknowledgement_survives_server_restart -- --exact` proves the versioned store is materialized before success, a real process restart reloads the same idempotent acknowledgement/sequence, EventSource replays it by id, and canonical board state is rebuilt. `cargo test -p hemx-kanban-example --test browser_e2e exhausted_offline_retries_keep_command_until_later_reconnect -- --exact` proves three bounded retries exhaust into visible offline/manual-recovery state while the command remains durable, then a later retry acknowledges/removes it and converges canonically. `cargo test -p hemx-kanban-example --test browser_e2e missing_history_rebase_and_user_conflict_resolution_preserve_suffix -- --exact` proves retained-history gap detection, typed/versioned snapshot fallback, deterministic already-canonical convergence, atomic snapshot/cursor commit, then a divergent canonical update producing explicit conflict with a two-command queue; the user’s canonical-state action removes only the conflicted head, retains and resumes the suffix, and reaches canonical convergence with an empty queue. `cargo test -p hemx-kanban-example --test browser_e2e keep_local_retry_preserves_conflicted_command_and_suffix_order -- --exact` proves the keep-local action derives one resolution identity; a permanent rejection preserves order and re-enables the choice; three later transport failures still preserve the original conflicted head plus suffix and expose manual recovery; final acceptance removes only the head and converges the untouched suffix without duplication or loss. `cargo test -p hemx-kanban-example --test browser_e2e redacted_sync_diagnostics_are_bounded_and_leak_no_sensitive_material -- --exact` proves a 15-second-old queue is bucketed rather than timestamped, cursor and acknowledgement latency advance after recovery, diagnostics stay under a fixed small readout, and command id/payload/actor/session/cookie material never reaches the diagnostics surface; the conflict and keep-local tests additionally assert conflict/rejection counters at their actual failure boundaries. `cargo test -p hemx-kanban-example --test browser_e2e two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_application -- --exact` proves one retry-exhausted leader/one explicit standby, zero follower upload before takeover, lease takeover after the leader closes, one canonical sequence/event, one queue removal, and one board application. `cargo test -p hemx-kanban-example --test browser_e2e upload_backpressure_keeps_pending_work_visible_and_recoverable -- --exact` proves one in-flight upload, a two-acknowledgement activation limit, one retained durable command with visible recovery state, and explicit retry draining the final command without loss. `cargo test -p hemx-kanban-example --test browser_e2e mixed_queue_removes_accepted_prefix_and_retains_rejected_tail -- --exact` proves an accepted prefix is canonically applied and removed once before a permanent rejection stops processing, exposes its typed HTTP/server cause, disables blind retry, and leaves both the rejected command and untouched suffix durable. `cargo test -p hemx-kanban-example --test browser_e2e schema_upgrade_preserves_queued_order_and_local_intent -- --exact` proves a three-command v1 queue migrates atomically to the explicit-target v2 schema, remains byte-for-intent ordered after interrupted upload, then receives canonical sequences 1–3 in original order and drains without loss. `cargo test -p hemx-kanban-example --test browser_e2e account_partition_hides_replay_and_export_until_owner_returns -- --exact` proves a beta editor and alpha viewer enumerate zero commands and issue no replay for an alpha owner queue, signed-out startup cannot open a partition, no foreign id/export surface leaks, and only the returning alpha owner can export the intact command then resume it exactly once. `cargo test -p hemx-kanban-example --test browser_e2e canonical_snapshot_and_history_are_tenant_scoped -- --exact` proves alpha and beta sessions receive only their own canonical cards and acknowledgement events with independent sequence 1 cursors, while signed-out snapshot/history reads receive 401 and expose no stream. `cargo test -p hemx-kanban-example --test browser_e2e identical_sync_inputs_reconcile_deterministically -- --exact` proves `sync/022` through a real browser and live sync endpoints: one accepted server result and canonical snapshot plus an ordered command sequence produce byte-equivalent structured reconciliation records on independent cloned inputs, do not mutate inputs, and select the explicit server-authoritative model. `cargo test -p hemx-kanban-example --test browser_e2e sync_requests_timeout_and_cancel_on_pagehide -- --exact` proves the sync request boundary enforces a named finite timeout and reports `TimeoutError`, while page teardown cancels a pending request with `AbortError`, closes the acknowledgement stream, and clears retry/lease timers. Per-requirement audit: every `sync/001-023` row now has concrete browser proof; existing live-browser proofs also cover `operations/001-002`, `operations/005`, `security/002`, `security/004`, `performance/004`, and `performance/005`, while this slice closes the sync request/stream leg of `operations/003`. `cargo test -p hemx-kanban-example --test browser_e2e adversarial_wire_inputs_are_rejected_before_partial_application -- --exact` proves malformed, oversized, truncated, trailing, unknown-version, and invalid-kind batch/state/command inputs fail without UI mutation; `cargo test -p hemx-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` additionally proves unknown-version and oversized event/state inputs are rejected at the generated real-WASM handler boundary before application code. The runtime now bounds total payloads, fields, vectors, varints, UTF-8, enums, and option discriminants before allocation or dispatch, and the durable queue validates every loaded command before replay, closing `security/005`. `cargo test -p hemx-axum --test response interaction_boundary_honors_media_type_and_host_body_limit -- --exact` proves the mutation extractor rejects missing/unexpected media types with 415 and an Axum host-configured oversized body with 413 before handler dispatch, while a valid bounded form dispatches exactly once; the extractor documentation names `DefaultBodyLimit`, and CSRF remains the existing documented application/middleware policy, closing `security/003`. `cargo test -p hemx-kanban-example --test browser_e2e acknowledgement_stream_bounds_reconnect_buffering_heartbeat_and_cancellation -- --exact` closes `operations/004`: the live stream exposes a finite 15-second production heartbeat (25 ms under proof configuration), exactly three bounded reconnect delays (100/250/500 ms), converts 65 pending events over its 64-event transport budget into one canonical snapshot recovery event, then emits heartbeats after recovery; both proof streams are explicitly closed, and the page teardown path closes its owned stream and reports cancellation. `cargo test -p hemx-kanban-example --bin hemx-kanban-example timeout -- --nocapture` completes `operations/003`: ordinary server routes are cancelled after a finite 10-second production deadline and return 504, startup store read/replay is cancelled after five seconds, oversized startup state is rejected before decoding, and focused tests prove dropped in-flight work plus named timeout diagnostics. `cargo test -p hemx-kanban-example --test browser_e2e ordinary_browser_request_exposes_deadline_and_cancels_on_pagehide -- --exact` proves the ordinary browser request boundary exposes its finite 10-second deadline, aborts the real form request with `AbortError` on page teardown, and clears pending UI state; stream routes remain governed by the `operations/004` lifecycle rather than the ordinary-handler timeout. The complete Slice 4 browser audit now passes (`cargo test -p hemx-kanban-example --test browser_e2e -- --test-threads=1` and `cargo test -p hemx-wasm --test browser -- --test-threads=1`). The audit also fixed three stale v1 command fixtures to exercise the real v2 durable schema, preserved an offline leader lease without retrying while connectivity is absent, and made finite acknowledgement responses close unless the caller explicitly requests a reconnecting stream. `cargo test -p hemx-kanban-example --test browser_e2e typed_broadcast_applies_generated_batch_over_sse -- --exact` closes `sync/004`: public `SyncEffect::broadcast(Channel, EffectBatch)` preserves an ordinary typed batch, the board subscription selects the validated typed channel, and a generated-resource presence batch crosses the real SSE runtime to update browser DOM. Slice 4 cannot close yet: citations labeled `sync/001` and `sync/005-006` prove app-local equivalents, not the required public `hemx-sync` presence tracking, `#[hemx_sync::presence]`, or `SyncEffect::ack(atom)` APIs. The next coherent slice is `sync/001` plus `sync/005`: add one public presence projection primitive and the presence declaration macro, then prove join/leave updates a generated presence atom over the typed broadcast path. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/src/main.rs b/examples/kanban/src/main.rs index 5d1c442..cddbddc 100644 --- a/examples/kanban/src/main.rs +++ b/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>) -> 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>) -> impl IntoResponse { if params.contains_key("once") { let effect = board::presence.put(&Presence { count: 1 }); diff --git a/examples/kanban/templates/board.heml b/examples/kanban/templates/board.heml index b0db8e5..78bd781 100644 --- a/examples/kanban/templates/board.heml +++ b/examples/kanban/templates/board.heml @@ -1,4 +1,4 @@ -
+

hemx Kanban

diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index 9ed372a..a6c1ea3 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/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<()> { diff --git a/hemx-sync/src/lib.rs b/hemx-sync/src/lib.rs index 915345d..8b91924 100644 --- a/hemx-sync/src/lib.rs +++ b/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) -> Result { + 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(