diff --git a/Cargo.lock b/Cargo.lock index a5f29a1..aac8eec 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index bb4ad6a..7b11d30 100644 --- a/Cargo.toml +++ b/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" diff --git a/PLAN.md b/PLAN.md index 9bc4f26..cc08de5 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. `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. +- **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. `cargo test -p hemx-kanban-example --test browser_e2e typed_presence_join_leave_updates_generated_atom_over_sse -- --exact` closes `sync/001` and `sync/005`: public channel-scoped `PresenceTracker` makes duplicate joins and absent leaves idempotent, `#[hemx_sync::presence]` turns one synchronous typed projection into a `PresenceUpdate`, and real join/duplicate-join/join/leave requests produce counts 1/1/2/1 in a generated presence atom over typed SSE broadcast. The macro rejects arguments, async functions, missing return types, wrong arity, and destructured inputs with local diagnostics. Slice 4 cannot close yet: `sync/006` still lacks the required public `SyncEffect::ack(atom)` acknowledgement primitive; the next coherent slice is to add that typed acknowledgement wrapper and prove the canonical acknowledgement updates a generated atom through the ordinary browser batch path. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/src/main.rs b/examples/kanban/src/main.rs index cddbddc..5ed846e 100644 --- a/examples/kanban/src/main.rs +++ b/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, sync_sessions: SyncSessionTokens, acknowledgement_heartbeat_interval: Duration, + presence: Mutex>, } #[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>) -> 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>, + Query(params): Query>, +) -> 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>) -> 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() } diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index a6c1ea3..0530764 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/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; diff --git a/hemx-sync-macros/Cargo.toml b/hemx-sync-macros/Cargo.toml new file mode 100644 index 0000000..cb337cd --- /dev/null +++ b/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"] } diff --git a/hemx-sync-macros/src/lib.rs b/hemx-sync-macros/src/lib.rs new file mode 100644 index 0000000..0548a9a --- /dev/null +++ b/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() +} diff --git a/hemx-sync/Cargo.toml b/hemx-sync/Cargo.toml index 8625f11..3d14898 100644 --- a/hemx-sync/Cargo.toml +++ b/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] diff --git a/hemx-sync/src/lib.rs b/hemx-sync/src/lib.rs index 8b91924..2f048b4 100644 --- a/hemx-sync/src/lib.rs +++ b/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 { + members: HashMap>, +} + +impl Default for PresenceTracker { + fn default() -> Self { + Self { + members: HashMap::new(), + } + } +} + +impl PresenceTracker +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 { + channel: Channel, + effect: Effect, +} + +impl PresenceProjection { + 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 PresenceUpdate for PresenceProjection +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 IntoEffect for PresenceProjection +where + Effect: IntoEffect, +{ + fn append_to(self, ops: &mut Vec) { + 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();