lhumina_code/hero_archipelagos
1
0
Fork 0
Code Issues 50 Pull requests Projects Releases Packages Wiki Activity Actions

messaging: no live updates — recipient must reload to see new messages #182

New issue
Closed
opened 2026-04-29 13:23:15 +00:00 by rawdaGastan · 2 comments
rawdaGastan commented 2026-04-29 13:23:15 +00:00
Member
Copy link

Problem

The messaging archipelago is pull-only — there is no SSE / WebSocket / poll loop. After mount the conversation list and active-chat messages are fetched once; they are never refetched on their own. A user who sends a message can see it (optimistic update from the closed #45 patch), but the recipient sees nothing until they reload the page or navigate away and back.

Grep for subscribe|interval|poll|watch|stream|sse under archipelagos/messaging/src/ returns nothing.

Repro

  1. Open the same DM in two browser tabs (currently both see the same data thanks to hero_osis #40 — separate contexts will work the same way once that lands).
  2. Send from tab A.
  3. Tab B: nothing happens. Refresh → message appears.

Why this is worse than it sounds

Combined with hero_archipelagos #62, on the next reload tab B sees the new message and its own historical messages all rendered as bubble-other. Effectively no one can tell who sent what without auth.

Suggested fix

Short-term: poll conversation.list + conversation.list_messages(active_sid) every N seconds while the island is focused.

Medium-term: add an SSE/WS subscription on the hero_osis_communication socket and stream chatmessage.new / conversation.updated events to subscribed clients.

Found via QA session 2026-04-29 (qa/messaging_2026-04-29/FINDINGS.md).

## Problem The messaging archipelago is pull-only — there is no SSE / WebSocket / poll loop. After mount the conversation list and active-chat messages are fetched once; they are never refetched on their own. A user who sends a message can see it (optimistic update from the closed #45 patch), but the *recipient* sees nothing until they reload the page or navigate away and back. Grep for `subscribe|interval|poll|watch|stream|sse` under `archipelagos/messaging/src/` returns nothing. ## Repro 1. Open the same DM in two browser tabs (currently both see the same data thanks to hero_osis #40 — separate contexts will work the same way once that lands). 2. Send from tab A. 3. Tab B: nothing happens. Refresh → message appears. ## Why this is worse than it sounds Combined with hero_archipelagos #62, on the next reload tab B sees the new message *and* its own historical messages all rendered as `bubble-other`. Effectively no one can tell who sent what without auth. ## Suggested fix Short-term: poll `conversation.list` + `conversation.list_messages(active_sid)` every N seconds while the island is focused. Medium-term: add an SSE/WS subscription on the `hero_osis_communication` socket and stream `chatmessage.new` / `conversation.updated` events to subscribed clients. Found via QA session 2026-04-29 (`qa/messaging_2026-04-29/FINDINGS.md`).
rawdaGastan commented 2026-05-03 11:38:33 +00:00
Author
Member
Copy link

Implementation Spec for Issue #182

Objective

Replace the messaging archipelago's pull-only data flow with a server-pushed event stream so that a recipient sees new messages without reloading the page. The medium-term path is chosen: hero_osis exposes a per-domain Server-Sent Events (SSE) endpoint at /events, the hero_osis_sdk::communication::CommunicationClient gains a subscribe() method that yields a Stream<Item = Result<ChatEvent, ClientError>>, and the messaging archipelago consumes that stream after the existing initial fetch.

Requirements

  1. Server-side broadcaster lives inside hero_osis_server's OsisCommunication domain. Events are emitted by the existing chatmessage_trigger_save_post and conversation_trigger_save_post hooks (already hand-written in rpc.rs, lines 1501 and 1534), so no *_generated.rs file is touched and the codegen does not need a new feature.
  2. SSE is the chosen transport. JSON-RPC 2.0 over HTTP POST is already the per-domain transport (hero_rpc/crates/server/src/server/domain_server.rs) and SSE rides over the same axum router as a GET /events route — the simplest coexistence with the existing OpenRpcTransport. WebSocket would require a second protocol upgrade path that hero_router already proxies generically but the SDK does not. gRPC streaming is rejected because nothing in the stack speaks it.
  3. The route is added by extending the per-domain router. Because hero_rpc::server::DomainServer::spawn (in hero_rpc, not in hero_osis) hard-codes the four routes (/rpc, /health, /openrpc.json, /.well-known/heroservice.json) and there is no extra_routes hook today, we add one to the generator: an optional domain_router_extension configured via OschemaBuildConfig::with_domain_router_ext("hero_osis_server::communication::sse_router"). The extension function returns an axum::Router<()> that is merged in. This generator change is the only change required to hero_rpc and follows the same pattern already used by bin_ui.
  4. Events are scoped per (context_name, conversation participation). The route handler reads X-Hero-Context from RequestContext::from_headers (already done in domain_server.rs), then for every event filters by conversation.participant_keys.contains(&caller_user_key). With hero_osis #40 closed, X-Hero-Context is now respected by the data layer end-to-end; the SSE filter sits on top of that as a delivery-time check.
  5. CommunicationClient::subscribe() is hand-written in crates/hero_osis_sdk/src/communication/mod.rs (the existing pub mod communication { ... } already alongside generated code). It is not generated. Codegen today produces only request-response RPCs (see osis_client_generated.rs's uniform rpc_call shape) and adding stream support there would touch the generator for a single domain.
  6. UI subscribes after initial fetch and merges incoming events into existing signals, deduplicating by message SID (the existing dedupe in archipelago.rs lines 297–301 is the canonical rule).
  7. A reconnect-with-backoff loop in the SDK ensures the UI recovers from transient disconnects without manual reload.

Files to Modify/Create

Repo hero_osis

  • crates/hero_osis_server/src/communication/server/rpc.rs — modify: emit events from the existing chatmessage_trigger_save_post and conversation_trigger_save_post lifecycle triggers (lines 1501, 1534).
  • crates/hero_osis_server/src/communication/server/mod.rs — modify: add pub mod sse; (declares the new hand-written supplementary module).
  • crates/hero_osis_server/src/communication/server/sse.rs — create: holds the per-context tokio::sync::broadcast registry, the ChatEvent enum, a pub fn sse_router() -> axum::Router<()>, and the GET /events handler.
  • crates/hero_osis_server/build.rs — modify: call .with_domain_router_ext("communication", "crate::communication::server::sse_router").
  • crates/hero_osis_server/Cargo.toml — modify: add axum (for Sse/KeepAlive) and tokio-stream features as needed; tokio = { features = [..., "sync"] } is already present transitively.
  • crates/hero_osis_sdk/src/communication/mod.rs — modify: add pub use events::*; and the hand-written impl CommunicationClient { pub async fn subscribe(...) -> impl Stream<Item = Result<ChatEvent, ClientError>> { ... } }.
  • crates/hero_osis_sdk/src/communication/events.rs — create: holds the ChatEvent JSON shape (mirrors the server's enum), the parser, and the WASM-vs-native split for eventsource-style consumption.
  • crates/hero_osis_sdk/Cargo.toml — modify: add futures = "0.3", reqwest = { features = ["stream"] } (native), gloo-net SSE / web-sys EventSource (wasm).
  • tests/communication_subscribe.rs — create: integration test that boots hero_osis_server in-process, calls subscribe(), performs a send_message, and asserts the receiver gets a chatmessage.new event.

Repo hero_archipelagos

  • archipelagos/messaging/src/services/messaging_service.rs — modify: add pub async fn subscribe_chat_events(osis_url, context_name) -> Result<impl Stream<Item = ChatEvent>, String> thin wrapper.
  • archipelagos/messaging/src/archipelago.rs — modify: after the existing fetch_chats use_effect (line 107), spawn a long-lived task that consumes the stream and updates conversations, active_messages (filtered by current Route::Chat { sid, .. }), with reconnect-with-backoff.
  • archipelagos/messaging/Cargo.toml — modify: ensure futures and a path-pinned hero_osis_sdk (during dev — see Notes).

Repo hero_rpc (one upstream change required)

  • crates/generator/src/build/build.rs — modify: add with_domain_router_ext(domain, fqfn) builder method (mirrors bin_ui), and store it in OschemaBuildConfig.
  • crates/server/src/server/domain_server.rs — modify: after the existing Router::new() ... .with_state(state); block, if a router_extension: Option<fn() -> Router<()>> was passed via the spawn API, router = router.merge(extension()).
  • generator's per-domain spawn template — modify so each domain's spawn call threads its extension through.

Implementation Plan

Step 1 — Generator: add with_domain_router_ext

  • Files: hero_rpc/crates/generator/src/build/build.rs, hero_rpc/crates/server/src/server/domain_server.rs, hero_rpc/crates/server/src/server/spawn.rs.
  • Repo: hero_rpc.
  • Subtasks:
    • Add a HashMap<String, String> field domain_router_exts on OschemaBuildConfig and a with_domain_router_ext(domain, fqfn) builder method.
    • Have the generator emit, in the per-domain spawn block, a let extra = <fqfn>(); router.merge(extra) if the domain has an entry.
    • Add a new DomainServer::spawn_with_extension(...) overload accepting Option<axum::Router>; have the existing spawn delegate to it with None.
  • Dependencies: none.

Step 2 — Server: define ChatEvent and broadcaster

  • Files: hero_osis/crates/hero_osis_server/src/communication/server/sse.rs (new), hero_osis/crates/hero_osis_server/src/communication/server/mod.rs.
  • Repo: hero_osis.
  • Subtasks:
    • Define pub enum ChatEvent { ChatmessageNew(ChatMessage), ConversationUpdated(Conversation) } deriving Serialize. Optional ConversationRead { sid: String, by: String } if mark-as-read should also stream.
    • Define pub struct CommunicationBroadcaster { senders: RwLock<HashMap<String /* context_name */, tokio::sync::broadcast::Sender<ChatEvent>>> } exposed as a OnceLock global (the trigger functions chatmessage_trigger_save_post are static fn and have no access to the domain Arc, so the broadcaster must be a process-wide singleton).
    • Implement pub fn publish(context_name: &str, event: ChatEvent) and pub fn subscribe(context_name: &str) -> broadcast::Receiver<ChatEvent>.
    • Implement pub fn sse_router() -> axum::Router<()> returning a router with GET /events that:
      1. Reads X-Hero-Context from headers via RequestContext::from_headers.
      2. Reads ?subscribe=conversation:<sid>,... query params or accepts no filter (subscribes to all events visible to the context).
      3. Subscribes to the broadcaster for that context.
      4. Filters each event server-side: for ChatmessageNew, look up the Conversation by event.conversation_sid and only forward if participant_keys.contains(&caller_self_key); for ConversationUpdated, the conversation is already in hand.
      5. Wraps the receiver in axum::response::sse::Sse with a KeepAlive heartbeat (~15s).
  • Dependencies: Step 1.

Step 3 — Server: emit events from lifecycle triggers

  • Files: hero_osis/crates/hero_osis_server/src/communication/server/rpc.rs.
  • Repo: hero_osis.
  • Subtasks:
    • In OsisCommunication::chatmessage_trigger_save_post(obj: &ChatMessage) (line 1501), emit ChatmessageNew(obj.clone()) via sse::publish(...). Because the trigger fn is static and has no RequestContext, publish on a global channel; the SSE handler does the per-context filter at delivery time using Conversation.participant_keys against the caller's self_key. With hero_osis #40 closed, the caller's self_key is correctly context-isolated upstream, so this delivery-time filter is sound.
    • Same edit in conversation_trigger_save_post(obj: &Conversation) (line 1534), publishing ConversationUpdated.
  • Dependencies: Step 2.

Step 4 — SDK: ChatEvent types and subscribe()

  • Files: hero_osis/crates/hero_osis_sdk/src/communication/events.rs (new), hero_osis/crates/hero_osis_sdk/src/communication/mod.rs, hero_osis/crates/hero_osis_sdk/Cargo.toml.
  • Repo: hero_osis.
  • Subtasks:
    • Mirror the server's ChatEvent enum (tagged JSON shape, e.g. { "type": "chatmessage.new", "data": ChatMessage }).
    • Native impl: use reqwest with text/event-stream accept, parse SSE frames into ChatEvent via serde_json, expose as impl Stream<Item = Result<ChatEvent, ClientError>>. Endpoint: {base_url}/hero_osis_communication/rpc/events (sibling to /rpc; goes through hero_router's existing per-service proxy unchanged).
    • WASM impl: use web_sys::EventSource with with_credentials = false, wrap the onmessage callback into a futures::channel::mpsc::UnboundedReceiver<ChatEvent> exposed as Stream.
    • Add a public pub async fn subscribe(&self) -> Result<impl Stream<Item = Result<ChatEvent, ClientError>> + Unpin, ClientError> on CommunicationClient.
  • Dependencies: Step 3 (so events are actually emitted to test against).

Step 5 — Server unit and SDK integration tests

  • Files: hero_osis/tests/communication_subscribe.rs (new), hero_osis/crates/hero_osis_server/src/communication/server/sse.rs (extend with #[cfg(test)] mod tests).
  • Repo: hero_osis.
  • Subtasks:
    • Unit: subscribe to broadcaster, call publish, assert the receiver gets the event.
    • Integration: spawn hero_osis server in-process on a temp socket dir, create a CommunicationClient, subscribe, then send_message from a second client, assert the first receives a ChatmessageNew event with the correct content.
    • Cross-context isolation: subscribers on context ctx_1 must not receive events for conversations whose participants do not include the ctx_1 caller key.
  • Dependencies: Steps 2–4.

Step 6 — UI: subscribe-on-mount in messaging archipelago

  • Files: hero_archipelagos/archipelagos/messaging/src/services/messaging_service.rs, hero_archipelagos/archipelagos/messaging/src/archipelago.rs.
  • Repo: hero_archipelagos.
  • Subtasks:
    • In messaging_service.rs, add pub async fn subscribe_chat_events(osis_url, context_name) -> Result<impl Stream<Item = ChatEvent>, String> that delegates to client.subscribe().
    • In archipelago.rs, after the existing fetch_chats use_effect block (around line 107–127), add a new use_effect that:
      1. Spawns one long-lived task per archipelago mount.
      2. Loops: connect → consume stream → on disconnect, exponential backoff up to 30s, reconnect.
      3. On ChatEvent::ChatmessageNew(msg):
        • Update conversations (last_message, last_message_time = msg.created_at * 1000, unread_count += 1 unless this is the active chat).
        • If the message's conversation_sid matches the current Route::Chat { sid, .. } (read from route.peek()), convert via services::chatmessage_to_messagedata and push to active_messages — but only if !active_messages.peek().iter().any(|m| m.id == msg_data.id) (mirrors the existing dedupe at lines 297–301).
      4. On ChatEvent::ConversationUpdated(conv): replace the matching preview row in conversations, or insert at the head if it's a brand-new conversation.
  • Dependencies: Step 4 must be merge-ready (or path-pinned during dev — see Notes).

Step 7 — Manual smoke test

  • Files: none.
  • Repo: hero_archipelagos.
  • Subtasks: run two browser tabs against the same dev hero_osis instance, log in as different users, send from A, observe live append on B.

Acceptance Criteria

  • hero_osis server exposes GET /events on the per-domain communication socket and emits SSE frames typed as ChatEvent.
  • chatmessage.new events are emitted whenever chatmessage_set succeeds, and conversation.updated events whenever conversation_set succeeds.
  • Subscribers receive only events for conversations where their caller key is in Conversation.participant_keys (filter is enforced server-side, not client-side).
  • hero_osis_sdk::communication::CommunicationClient::subscribe() returns a Stream<Item = Result<ChatEvent, ClientError>> on both native (reqwest) and WASM (EventSource).
  • SDK reconnects with exponential backoff (capped at 30s) when the stream closes.
  • Messaging archipelago no longer requires reload: opening tab B while user A sends a message in tab A produces a live append in B's active_messages and an updated preview row in B's conversation list within 2 seconds, with no manual refresh.
  • Existing optimistic-send dedupe still holds (sender does not see duplicate bubbles when its own message echoes back over the stream).
  • An integration test in hero_osis/tests/communication_subscribe.rs exercises send → receive end-to-end via in-process server and passes in CI.
  • Cross-context isolation test: a subscriber on context ctx_1 does not receive any event produced for ctx_2.

Notes

Multi-repo PR strategy

Three PRs, ordered: hero_rpc → hero_osis → hero_archipelagos.

Rationale: the cargo workspace dependency direction is one-way (hero_archipelagos → hero_osis_sdk → hero_rpc). Coordinating across repos with [patch] rev pins is the standard Hero pattern (see the cargo_deps skill).

Sequence:

  1. PR 1 in hero_rpc: adds with_domain_router_ext to OschemaBuildConfig and the optional router extension parameter to DomainServer::spawn. Lands first; backwards-compatible (existing callers pass None).
  2. PR 2 in hero_osis: server SSE route + broadcaster, SDK subscribe(), server+SDK tests. While PR 1 is in review, PR 2's branch carries a temporary [patch.'https://forge.ourworld.tf/lhumina_code/hero_rpc.git'] entry.
  3. PR 3 in hero_archipelagos: UI consumer. While PR 2 is in review, PR 3 carries a temporary [patch.'https://forge.ourworld.tf/lhumina_code/hero_osis.git'] entry pointing to PR 2's branch. The patch line is removed in the final commit before merge, once PR 2 is on development.

Relationship to hero_osis #40 (closed)

hero_osis #40 (server ignored X-Hero-Context for communication) was resolved. With it closed, the data layer correctly isolates by context, which means the SSE handler's delivery-time filter (participant_keys vs caller's self_key) sits on top of an already context-isolated store rather than being the only line of defense. The trigger fn signature remains &ChatMessage / &Conversation (no &RequestContext), so the broadcaster still fans out on a global channel and the SSE handler does the per-caller filter — that is intentional and correct given the closed state of #40. A future cleanup could pipe &RequestContext into triggers via a generator change and shard the broadcaster by context, but it is not required for this issue.

Why SSE and not WebSocket

  • The transport is one-directional (server → client). Subscriptions are stateless once established.
  • reqwest and EventSource give plain SSE on both targets with no extra deps; WebSocket requires tokio-tungstenite natively and the web_sys::WebSocket ceremony in WASM, plus a framing protocol.
  • hero_router's per-service proxy already streams arbitrary HTTP responses byte-for-byte, so no router change is required.

Generated vs hand-written boundary

  • Generated (untouched in this issue): osis_server_generated.rs, rpc_generated.rs, osis_client_generated.rs, types.rs.
  • Hand-written touched: rpc.rs (existing trigger fns get bodies), mod.rs (server, SDK), new sse.rs, new events.rs.
  • Generator touched once (additive only): OschemaBuildConfig::with_domain_router_ext and the DomainServer::spawn extension parameter. Pattern mirrors the existing bin_ui hook.
## Implementation Spec for Issue #182 ### Objective Replace the messaging archipelago's pull-only data flow with a server-pushed event stream so that a recipient sees new messages without reloading the page. The medium-term path is chosen: hero_osis exposes a per-domain Server-Sent Events (SSE) endpoint at `/events`, the `hero_osis_sdk::communication::CommunicationClient` gains a `subscribe()` method that yields a `Stream<Item = Result<ChatEvent, ClientError>>`, and the messaging archipelago consumes that stream after the existing initial fetch. ### Requirements 1. Server-side broadcaster lives inside `hero_osis_server`'s `OsisCommunication` domain. Events are emitted by the existing `chatmessage_trigger_save_post` and `conversation_trigger_save_post` hooks (already hand-written in `rpc.rs`, lines 1501 and 1534), so no `*_generated.rs` file is touched and the codegen does not need a new feature. 2. SSE is the chosen transport. JSON-RPC 2.0 over HTTP POST is already the per-domain transport (`hero_rpc/crates/server/src/server/domain_server.rs`) and SSE rides over the same axum router as a `GET /events` route — the simplest coexistence with the existing `OpenRpcTransport`. WebSocket would require a second protocol upgrade path that hero_router already proxies generically but the SDK does not. gRPC streaming is rejected because nothing in the stack speaks it. 3. The route is added by extending the per-domain router. Because `hero_rpc::server::DomainServer::spawn` (in `hero_rpc`, not in `hero_osis`) hard-codes the four routes (`/rpc`, `/health`, `/openrpc.json`, `/.well-known/heroservice.json`) and there is no `extra_routes` hook today, we add one to the generator: an optional `domain_router_extension` configured via `OschemaBuildConfig::with_domain_router_ext("hero_osis_server::communication::sse_router")`. The extension function returns an `axum::Router<()>` that is `merge`d in. This generator change is the **only** change required to `hero_rpc` and follows the same pattern already used by `bin_ui`. 4. Events are scoped per `(context_name, conversation participation)`. The route handler reads `X-Hero-Context` from `RequestContext::from_headers` (already done in `domain_server.rs`), then for every event filters by `conversation.participant_keys.contains(&caller_user_key)`. With hero_osis #40 closed, `X-Hero-Context` is now respected by the data layer end-to-end; the SSE filter sits on top of that as a delivery-time check. 5. `CommunicationClient::subscribe()` is hand-written in `crates/hero_osis_sdk/src/communication/mod.rs` (the existing `pub mod communication { ... }` already alongside generated code). It is **not** generated. Codegen today produces only request-response RPCs (see `osis_client_generated.rs`'s uniform `rpc_call` shape) and adding stream support there would touch the generator for a single domain. 6. UI subscribes after initial fetch and merges incoming events into existing signals, deduplicating by message SID (the existing dedupe in `archipelago.rs` lines 297–301 is the canonical rule). 7. A reconnect-with-backoff loop in the SDK ensures the UI recovers from transient disconnects without manual reload. ### Files to Modify/Create #### Repo `hero_osis` - `crates/hero_osis_server/src/communication/server/rpc.rs` — modify: emit events from the existing `chatmessage_trigger_save_post` and `conversation_trigger_save_post` lifecycle triggers (lines 1501, 1534). - `crates/hero_osis_server/src/communication/server/mod.rs` — modify: add `pub mod sse;` (declares the new hand-written supplementary module). - `crates/hero_osis_server/src/communication/server/sse.rs` — create: holds the per-context `tokio::sync::broadcast` registry, the `ChatEvent` enum, a `pub fn sse_router() -> axum::Router<()>`, and the GET `/events` handler. - `crates/hero_osis_server/build.rs` — modify: call `.with_domain_router_ext("communication", "crate::communication::server::sse_router")`. - `crates/hero_osis_server/Cargo.toml` — modify: add `axum` (for `Sse`/`KeepAlive`) and `tokio-stream` features as needed; `tokio = { features = [..., "sync"] }` is already present transitively. - `crates/hero_osis_sdk/src/communication/mod.rs` — modify: add `pub use events::*;` and the hand-written `impl CommunicationClient { pub async fn subscribe(...) -> impl Stream<Item = Result<ChatEvent, ClientError>> { ... } }`. - `crates/hero_osis_sdk/src/communication/events.rs` — create: holds the `ChatEvent` JSON shape (mirrors the server's enum), the parser, and the WASM-vs-native split for `eventsource`-style consumption. - `crates/hero_osis_sdk/Cargo.toml` — modify: add `futures = "0.3"`, `reqwest = { features = ["stream"] }` (native), `gloo-net` SSE / `web-sys` `EventSource` (wasm). - `tests/communication_subscribe.rs` — create: integration test that boots `hero_osis_server` in-process, calls `subscribe()`, performs a `send_message`, and asserts the receiver gets a `chatmessage.new` event. #### Repo `hero_archipelagos` - `archipelagos/messaging/src/services/messaging_service.rs` — modify: add `pub async fn subscribe_chat_events(osis_url, context_name) -> Result<impl Stream<Item = ChatEvent>, String>` thin wrapper. - `archipelagos/messaging/src/archipelago.rs` — modify: after the existing `fetch_chats` use_effect (line 107), spawn a long-lived task that consumes the stream and updates `conversations`, `active_messages` (filtered by current `Route::Chat { sid, .. }`), with reconnect-with-backoff. - `archipelagos/messaging/Cargo.toml` — modify: ensure `futures` and a path-pinned `hero_osis_sdk` (during dev — see Notes). #### Repo `hero_rpc` (one upstream change required) - `crates/generator/src/build/build.rs` — modify: add `with_domain_router_ext(domain, fqfn)` builder method (mirrors `bin_ui`), and store it in `OschemaBuildConfig`. - `crates/server/src/server/domain_server.rs` — modify: after the existing `Router::new() ... .with_state(state);` block, if a `router_extension: Option<fn() -> Router<()>>` was passed via the spawn API, `router = router.merge(extension())`. - generator's per-domain spawn template — modify so each domain's spawn call threads its extension through. ### Implementation Plan #### Step 1 — Generator: add `with_domain_router_ext` - **Files**: `hero_rpc/crates/generator/src/build/build.rs`, `hero_rpc/crates/server/src/server/domain_server.rs`, `hero_rpc/crates/server/src/server/spawn.rs`. - **Repo**: `hero_rpc`. - **Subtasks**: - Add a `HashMap<String, String>` field `domain_router_exts` on `OschemaBuildConfig` and a `with_domain_router_ext(domain, fqfn)` builder method. - Have the generator emit, in the per-domain spawn block, a `let extra = <fqfn>(); router.merge(extra)` if the domain has an entry. - Add a new `DomainServer::spawn_with_extension(...)` overload accepting `Option<axum::Router>`; have the existing `spawn` delegate to it with `None`. - **Dependencies**: none. #### Step 2 — Server: define `ChatEvent` and broadcaster - **Files**: `hero_osis/crates/hero_osis_server/src/communication/server/sse.rs` (new), `hero_osis/crates/hero_osis_server/src/communication/server/mod.rs`. - **Repo**: `hero_osis`. - **Subtasks**: - Define `pub enum ChatEvent { ChatmessageNew(ChatMessage), ConversationUpdated(Conversation) }` deriving `Serialize`. Optional `ConversationRead { sid: String, by: String }` if mark-as-read should also stream. - Define `pub struct CommunicationBroadcaster { senders: RwLock<HashMap<String /* context_name */, tokio::sync::broadcast::Sender<ChatEvent>>> }` exposed as a `OnceLock` global (the trigger functions `chatmessage_trigger_save_post` are static `fn` and have no access to the domain `Arc`, so the broadcaster must be a process-wide singleton). - Implement `pub fn publish(context_name: &str, event: ChatEvent)` and `pub fn subscribe(context_name: &str) -> broadcast::Receiver<ChatEvent>`. - Implement `pub fn sse_router() -> axum::Router<()>` returning a router with `GET /events` that: 1. Reads `X-Hero-Context` from headers via `RequestContext::from_headers`. 2. Reads `?subscribe=conversation:<sid>,...` query params or accepts no filter (subscribes to all events visible to the context). 3. Subscribes to the broadcaster for that context. 4. Filters each event server-side: for `ChatmessageNew`, look up the `Conversation` by `event.conversation_sid` and only forward if `participant_keys.contains(&caller_self_key)`; for `ConversationUpdated`, the conversation is already in hand. 5. Wraps the receiver in `axum::response::sse::Sse` with a `KeepAlive` heartbeat (~15s). - **Dependencies**: Step 1. #### Step 3 — Server: emit events from lifecycle triggers - **Files**: `hero_osis/crates/hero_osis_server/src/communication/server/rpc.rs`. - **Repo**: `hero_osis`. - **Subtasks**: - In `OsisCommunication::chatmessage_trigger_save_post(obj: &ChatMessage)` (line 1501), emit `ChatmessageNew(obj.clone())` via `sse::publish(...)`. Because the trigger fn is static and has no `RequestContext`, publish on a global channel; the SSE handler does the per-context filter at delivery time using `Conversation.participant_keys` against the caller's `self_key`. With hero_osis #40 closed, the caller's `self_key` is correctly context-isolated upstream, so this delivery-time filter is sound. - Same edit in `conversation_trigger_save_post(obj: &Conversation)` (line 1534), publishing `ConversationUpdated`. - **Dependencies**: Step 2. #### Step 4 — SDK: `ChatEvent` types and `subscribe()` - **Files**: `hero_osis/crates/hero_osis_sdk/src/communication/events.rs` (new), `hero_osis/crates/hero_osis_sdk/src/communication/mod.rs`, `hero_osis/crates/hero_osis_sdk/Cargo.toml`. - **Repo**: `hero_osis`. - **Subtasks**: - Mirror the server's `ChatEvent` enum (tagged JSON shape, e.g. `{ "type": "chatmessage.new", "data": ChatMessage }`). - Native impl: use `reqwest` with `text/event-stream` accept, parse SSE frames into `ChatEvent` via `serde_json`, expose as `impl Stream<Item = Result<ChatEvent, ClientError>>`. Endpoint: `{base_url}/hero_osis_communication/rpc/events` (sibling to `/rpc`; goes through hero_router's existing per-service proxy unchanged). - WASM impl: use `web_sys::EventSource` with `with_credentials = false`, wrap the `onmessage` callback into a `futures::channel::mpsc::UnboundedReceiver<ChatEvent>` exposed as `Stream`. - Add a public `pub async fn subscribe(&self) -> Result<impl Stream<Item = Result<ChatEvent, ClientError>> + Unpin, ClientError>` on `CommunicationClient`. - **Dependencies**: Step 3 (so events are actually emitted to test against). #### Step 5 — Server unit and SDK integration tests - **Files**: `hero_osis/tests/communication_subscribe.rs` (new), `hero_osis/crates/hero_osis_server/src/communication/server/sse.rs` (extend with `#[cfg(test)] mod tests`). - **Repo**: `hero_osis`. - **Subtasks**: - Unit: subscribe to broadcaster, call `publish`, assert the receiver gets the event. - Integration: spawn `hero_osis` server in-process on a temp socket dir, create a `CommunicationClient`, subscribe, then `send_message` from a second client, assert the first receives a `ChatmessageNew` event with the correct content. - Cross-context isolation: subscribers on context `ctx_1` must not receive events for conversations whose participants do not include the `ctx_1` caller key. - **Dependencies**: Steps 2–4. #### Step 6 — UI: subscribe-on-mount in messaging archipelago - **Files**: `hero_archipelagos/archipelagos/messaging/src/services/messaging_service.rs`, `hero_archipelagos/archipelagos/messaging/src/archipelago.rs`. - **Repo**: `hero_archipelagos`. - **Subtasks**: - In `messaging_service.rs`, add `pub async fn subscribe_chat_events(osis_url, context_name) -> Result<impl Stream<Item = ChatEvent>, String>` that delegates to `client.subscribe()`. - In `archipelago.rs`, after the existing `fetch_chats` `use_effect` block (around line 107–127), add a new `use_effect` that: 1. Spawns one long-lived task per archipelago mount. 2. Loops: connect → consume stream → on disconnect, exponential backoff up to 30s, reconnect. 3. On `ChatEvent::ChatmessageNew(msg)`: - Update `conversations` (`last_message`, `last_message_time = msg.created_at * 1000`, `unread_count += 1` unless this is the active chat). - If the message's `conversation_sid` matches the current `Route::Chat { sid, .. }` (read from `route.peek()`), convert via `services::chatmessage_to_messagedata` and push to `active_messages` — but only if `!active_messages.peek().iter().any(|m| m.id == msg_data.id)` (mirrors the existing dedupe at lines 297–301). 4. On `ChatEvent::ConversationUpdated(conv)`: replace the matching preview row in `conversations`, or insert at the head if it's a brand-new conversation. - **Dependencies**: Step 4 must be merge-ready (or path-pinned during dev — see Notes). #### Step 7 — Manual smoke test - **Files**: none. - **Repo**: `hero_archipelagos`. - **Subtasks**: run two browser tabs against the same dev `hero_osis` instance, log in as different users, send from A, observe live append on B. ### Acceptance Criteria - [ ] `hero_osis` server exposes `GET /events` on the per-domain `communication` socket and emits SSE frames typed as `ChatEvent`. - [ ] `chatmessage.new` events are emitted whenever `chatmessage_set` succeeds, and `conversation.updated` events whenever `conversation_set` succeeds. - [ ] Subscribers receive only events for conversations where their caller key is in `Conversation.participant_keys` (filter is enforced server-side, not client-side). - [ ] `hero_osis_sdk::communication::CommunicationClient::subscribe()` returns a `Stream<Item = Result<ChatEvent, ClientError>>` on both native (`reqwest`) and WASM (`EventSource`). - [ ] SDK reconnects with exponential backoff (capped at 30s) when the stream closes. - [ ] Messaging archipelago no longer requires reload: opening tab B while user A sends a message in tab A produces a live append in B's `active_messages` and an updated preview row in B's conversation list within 2 seconds, with no manual refresh. - [ ] Existing optimistic-send dedupe still holds (sender does not see duplicate bubbles when its own message echoes back over the stream). - [ ] An integration test in `hero_osis/tests/communication_subscribe.rs` exercises send → receive end-to-end via in-process server and passes in CI. - [ ] Cross-context isolation test: a subscriber on context `ctx_1` does not receive any event produced for `ctx_2`. ### Notes #### Multi-repo PR strategy **Three PRs, ordered: hero_rpc → hero_osis → hero_archipelagos.** Rationale: the cargo workspace dependency direction is one-way (`hero_archipelagos` → `hero_osis_sdk` → `hero_rpc`). Coordinating across repos with `[patch]` rev pins is the standard Hero pattern (see the `cargo_deps` skill). Sequence: 1. **PR 1** in `hero_rpc`: adds `with_domain_router_ext` to `OschemaBuildConfig` and the optional router extension parameter to `DomainServer::spawn`. Lands first; backwards-compatible (existing callers pass `None`). 2. **PR 2** in `hero_osis`: server SSE route + broadcaster, SDK `subscribe()`, server+SDK tests. While PR 1 is in review, PR 2's branch carries a temporary `[patch.'https://forge.ourworld.tf/lhumina_code/hero_rpc.git']` entry. 3. **PR 3** in `hero_archipelagos`: UI consumer. While PR 2 is in review, PR 3 carries a temporary `[patch.'https://forge.ourworld.tf/lhumina_code/hero_osis.git']` entry pointing to PR 2's branch. The patch line is removed in the final commit before merge, once PR 2 is on `development`. #### Relationship to hero_osis #40 (closed) hero_osis #40 (server ignored `X-Hero-Context` for communication) was resolved. With it closed, the data layer correctly isolates by context, which means the SSE handler's delivery-time filter (participant_keys vs caller's self_key) sits on top of an already context-isolated store rather than being the only line of defense. The trigger fn signature remains `&ChatMessage` / `&Conversation` (no `&RequestContext`), so the broadcaster still fans out on a global channel and the SSE handler does the per-caller filter — that is intentional and correct given the closed state of #40. A future cleanup could pipe `&RequestContext` into triggers via a generator change and shard the broadcaster by context, but it is **not** required for this issue. #### Why SSE and not WebSocket - The transport is one-directional (server → client). Subscriptions are stateless once established. - `reqwest` and `EventSource` give plain SSE on both targets with no extra deps; WebSocket requires `tokio-tungstenite` natively and the `web_sys::WebSocket` ceremony in WASM, plus a framing protocol. - hero_router's per-service proxy already streams arbitrary HTTP responses byte-for-byte, so no router change is required. #### Generated vs hand-written boundary - Generated (untouched in this issue): `osis_server_generated.rs`, `rpc_generated.rs`, `osis_client_generated.rs`, `types.rs`. - Hand-written touched: `rpc.rs` (existing trigger fns get bodies), `mod.rs` (server, SDK), new `sse.rs`, new `events.rs`. - Generator touched once (additive only): `OschemaBuildConfig::with_domain_router_ext` and the `DomainServer::spawn` extension parameter. Pattern mirrors the existing `bin_ui` hook.
rawdaGastan added this to the ACTIVE project 2026-05-04 12:43:37 +00:00
rawdaGastan commented 2026-05-04 13:56:12 +00:00
Author
Member
Copy link

Implementation complete — three PRs

The fix shipped as a coordinated three-PR change across hero_rpc, hero_osis, and hero_archipelagos. Each PR linked below.

Step Repo PR Status
1 hero_rpc #39 — feat(server): add per-domain web_events.sock router extension hook merged
2 hero_osis #44 — feat(communication): server-pushed chat events via SSE merged
3 hero_archipelagos #204 — feat(messaging): subscribe to live ChatEvent SSE updates open

Architecture summary

hero_rpc (#39) — generator-level hook so any domain server can register supplementary axum routes without modifying the per-domain spawn template.

  • OschemaBuildConfig::with_domain_router_ext(domain, fqfn) — new builder.
  • DomainService.extension: Option<Router> — when set, spawn_domain_server binds a dedicated web_events.sock alongside rpc.sock and ui.sock. hero_router auto-discovers it via the standard web_<name>.sock convention so browsers can reach it at /<service>/<name> (e.g. /hero_osis_communication/events).
  • Backwards-compatible: domains without a registered extension behave exactly as before.

hero_osis (#44) — server-side broadcaster + SSE endpoint + SDK subscribe().

  • crates/hero_osis_server/src/communication/server/sse.rs: tagged-JSON ChatEvent enum (chatmessage.new / conversation.updated), process-wide tokio::sync::broadcast with Lagged → resync SSE-frame conversion, in-memory (conversation_sid → participant_keys) cache primed by conversation_trigger_save_post and _get_post, and an axum router exposing the SSE stream at /.
  • chatmessage_trigger_save_post reads the cache to embed participants in the published event so the SSE delivery filter is O(participants) at delivery time, no storage round-trip on the hot path.
  • events_handler resolves the caller's pubkey from X-Public-Key (or query-string fallback for web_sys::EventSource which can't set headers), filters per-subscriber, wraps as axum::response::sse::Sse with 15s keepalive.
  • SDK adds a CommunicationClient::subscribe(public_key) method on both native (reqwest text/event-stream consumer with line-oriented SSE parser) and WASM (web_sys::EventSource with query-string params + Drop guard). All existing RPC methods remain available via Deref to the inner generated client.
  • 11 unit tests cover broadcaster, filter, trigger pipeline (save_post + get_post cache priming), cache-miss safe-fail, and resolve_caller_pubkey precedence.

hero_archipelagos (#204) — UI consumer.

  • After the existing fetch_chats use_effect, the messaging archipelago opens an SSE subscription with reconnect-with-backoff (1s → 30s, 2x).
  • Incoming chatmessage.new events update the conversation preview row (last_message, last_message_time, unread_count) and append to active_messages when the user is currently viewing that chat. Dedupe by message id keeps the optimistic-insert clean.
  • Incoming conversation.updated events patch the matching preview row or insert a new one at the head.
  • Resync triggers a refetch without closing the stream; Disconnected triggers reconnect; SseError is logged.
  • A small race fix in handle_send upgrades the existing entry to is_own = true when SSE wins the race against the optimistic insert, so the sender always sees their own bubble blue.

Validation

  • All 247 unit tests pass across the three repos (118 hero_rpc, 124+5 hero_osis, compile-only for hero_archipelagos UI).
  • Manual end-to-end browser smoke: two browser profiles, A sends → B sees the bubble appear in the active chat panel within ~1-2s without reload. Conversation list preview + unread count also update live. Sender's own bubble renders blue (own) at send time without flicker.

Pre-auth carve-out (TODO post-auth)

The server-side should_deliver filter in sse.rs currently delivers all events to anonymous subscribers (no X-Public-Key). This is the documented pre-auth carve-out: the messaging archipelago still runs on the SYSTEM_KEY = "system" placeholder, so without the carve-out the QA repro for this issue would silently drop everything at the SSE filter. A TODO is flagged inline to flip back to fail-closed once real authentication lands. The participant-filter path is fully exercised whenever the caller IS authenticated.

Out of scope

  • Generator regen drift in hero_osis (*_generated.rs files churn after PR #38 + #39 land in hero_rpc) was deliberately not committed in PR #44 — would have buried the actual SSE work in ~30k lines of mechanical churn. It will regenerate on first cargo build and can be committed in a follow-up if needed.
  • Issue #192 (every reloaded message renders as bubble-other because the server hardcodes sender_public_key = "system") is a separate root cause and remains open. The race fix in this PR's handle_send keeps the just-sent bubble blue at send time, but reload-time rendering is unchanged.
  • A follow-up rustfmt-on-generator improvement in hero_rpc (so *_generated.rs files are emitted formatted) is deferred to a separate PR.
## Implementation complete — three PRs The fix shipped as a coordinated three-PR change across hero_rpc, hero_osis, and hero_archipelagos. Each PR linked below. | Step | Repo | PR | Status | |---|---|---|---| | 1 | hero_rpc | [#39 — feat(server): add per-domain web_events.sock router extension hook](https://forge.ourworld.tf/lhumina_code/hero_rpc/pulls/39) | merged | | 2 | hero_osis | [#44 — feat(communication): server-pushed chat events via SSE](https://forge.ourworld.tf/lhumina_code/hero_osis/pulls/44) | merged | | 3 | hero_archipelagos | [#204 — feat(messaging): subscribe to live ChatEvent SSE updates](https://forge.ourworld.tf/lhumina_code/hero_archipelagos/pulls/204) | open | ## Architecture summary **hero_rpc (#39)** — generator-level hook so any domain server can register supplementary axum routes without modifying the per-domain spawn template. - `OschemaBuildConfig::with_domain_router_ext(domain, fqfn)` — new builder. - `DomainService.extension: Option<Router>` — when set, `spawn_domain_server` binds a dedicated `web_events.sock` alongside `rpc.sock` and `ui.sock`. `hero_router` auto-discovers it via the standard `web_<name>.sock` convention so browsers can reach it at `/<service>/<name>` (e.g. `/hero_osis_communication/events`). - Backwards-compatible: domains without a registered extension behave exactly as before. **hero_osis (#44)** — server-side broadcaster + SSE endpoint + SDK `subscribe()`. - `crates/hero_osis_server/src/communication/server/sse.rs`: tagged-JSON `ChatEvent` enum (`chatmessage.new` / `conversation.updated`), process-wide `tokio::sync::broadcast` with `Lagged → resync` SSE-frame conversion, in-memory `(conversation_sid → participant_keys)` cache primed by `conversation_trigger_save_post` and `_get_post`, and an axum router exposing the SSE stream at `/`. - `chatmessage_trigger_save_post` reads the cache to embed participants in the published event so the SSE delivery filter is O(participants) at delivery time, no storage round-trip on the hot path. - `events_handler` resolves the caller's pubkey from `X-Public-Key` (or query-string fallback for `web_sys::EventSource` which can't set headers), filters per-subscriber, wraps as `axum::response::sse::Sse` with 15s keepalive. - SDK adds a `CommunicationClient::subscribe(public_key)` method on both native (reqwest `text/event-stream` consumer with line-oriented SSE parser) and WASM (`web_sys::EventSource` with query-string params + Drop guard). All existing RPC methods remain available via `Deref` to the inner generated client. - 11 unit tests cover broadcaster, filter, trigger pipeline (save_post + get_post cache priming), cache-miss safe-fail, and `resolve_caller_pubkey` precedence. **hero_archipelagos (#204)** — UI consumer. - After the existing `fetch_chats` `use_effect`, the messaging archipelago opens an SSE subscription with reconnect-with-backoff (1s → 30s, 2x). - Incoming `chatmessage.new` events update the conversation preview row (`last_message`, `last_message_time`, `unread_count`) and append to `active_messages` when the user is currently viewing that chat. Dedupe by message id keeps the optimistic-insert clean. - Incoming `conversation.updated` events patch the matching preview row or insert a new one at the head. - `Resync` triggers a refetch without closing the stream; `Disconnected` triggers reconnect; `SseError` is logged. - A small race fix in `handle_send` upgrades the existing entry to `is_own = true` when SSE wins the race against the optimistic insert, so the sender always sees their own bubble blue. ## Validation - All 247 unit tests pass across the three repos (118 hero_rpc, 124+5 hero_osis, compile-only for hero_archipelagos UI). - Manual end-to-end browser smoke: two browser profiles, A sends → B sees the bubble appear in the active chat panel within ~1-2s without reload. Conversation list preview + unread count also update live. Sender's own bubble renders blue (own) at send time without flicker. ## Pre-auth carve-out (TODO post-auth) The server-side `should_deliver` filter in `sse.rs` currently delivers all events to anonymous subscribers (no `X-Public-Key`). This is the documented pre-auth carve-out: the messaging archipelago still runs on the `SYSTEM_KEY = "system"` placeholder, so without the carve-out the QA repro for this issue would silently drop everything at the SSE filter. A `TODO` is flagged inline to flip back to fail-closed once real authentication lands. The participant-filter path is fully exercised whenever the caller IS authenticated. ## Out of scope - **Generator regen drift** in hero_osis (`*_generated.rs` files churn after PR #38 + #39 land in hero_rpc) was deliberately not committed in PR #44 — would have buried the actual SSE work in ~30k lines of mechanical churn. It will regenerate on first `cargo build` and can be committed in a follow-up if needed. - **Issue [#192](https://forge.ourworld.tf/lhumina_code/hero_archipelagos/issues/192)** (every reloaded message renders as `bubble-other` because the server hardcodes `sender_public_key = "system"`) is a separate root cause and remains open. The race fix in this PR's `handle_send` keeps the just-sent bubble blue at send time, but reload-time rendering is unchanged. - A follow-up rustfmt-on-generator improvement in hero_rpc (so `*_generated.rs` files are emitted formatted) is deferred to a separate PR.
Sign in to join this conversation.
No Branch/Tag specified
Branches Tags
development
main
No results found.
Labels
Clear labels   
prio_critical

   
prio_low

   
type_bug

   
type_contact

   
type_issue

   
type_lead

   
type_question

   
type_story

   
type_task
No labels
prio_critical
prio_low
type_bug
type_contact
type_issue
type_lead
type_question
type_story
type_task
Milestone
Clear milestone
No items
No milestone
Projects
Clear projects
No items
No project
ACTIVE
Assignees
Clear assignees
No assignees
rawdaGastan
1 participant
Notifications
Due date
The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference
lhumina_code/hero_archipelagos#182
No description provided.