diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md index bd730303..aab0be0f 100644 --- a/docs/plan/message-bus.md +++ b/docs/plan/message-bus.md @@ -603,6 +603,46 @@ Land with their producer PRs; each is a small addition to polymorphic `RtFolderEventNotification` with `oneOf`. Better codegen for typed clients. Refactor when we generate an FE SDK. +### TypeScript client codegen via `@asyncapi/modelina` (Phase-A polish) + +AsyncAPI has the same "spec → typed FE SDK" story OpenAPI has. Wire +it once, avoid hand-maintaining a growing catalog of message types. + +- **Tool:** `@asyncapi/modelina` — the AsyncAPI-native model + generator. Reads `resources/gen/asyncapi.json`, emits TypeScript + interfaces + tagged unions for every message and schema. Actively + maintained, produces idiomatic TS. +- **Not** `@asyncapi/generator`'s WebSocket TEMPLATE — that generates + a full client SDK on assumptions (fetch shape, subscription model) + that don't match our `useTopic` singleton store. Custom composable + stays; only the message DTOs come from codegen. +- **Wiring:** + - `frontend/package.json` dev-dep: `@asyncapi/modelina`. + - Script `frontend/scripts/gen-realtime-types.mjs` invokes Modelina, + writes to `frontend/src/lib/generated/realtime/`. + - `just asyncapi-ts` recipe alongside `just asyncapi`. + - CI dirty-tree check — regenerate on every build, fail if `git + diff` on the generated folder is non-empty. Same discipline as + OpenAPI's check. + - Generated files carry a `// AUTO-GENERATED — do not edit; run + `just asyncapi-ts` to regenerate` header. +- **What the FE gets:** + - `type RtEvent = FileCreatedData | FileRenamedData | ...` — a + tagged union keyed on the `event` discriminator, so the folder + view's `switch (evt.event)` is exhaustive at compile time. + - `RtSubscribeRequestBody`, `RtErrorResponseBody`, error-code enum, + `RtPongResponseBody.result.pong === true` narrowed by type. + - No divergence between wire spec and FE types — the CI check + catches drift. +- **Also worth:** if we ever want a typed WS client for other + languages (Rust sync client, Python integration), the AsyncAPI + spec is the source; Modelina supports 8+ target languages. +- **Timing:** the current spec covers 8 event variants + 6 message + envelopes. Marginal savings today; substantial as Phase B adds + ~15 more event variants (comments, mentions, presence, share + events). Set up now so the discipline is in place BEFORE the + surface grows. + ## AuthZ model (audit rules per AGENTS.md) ### The subscribe gate diff --git a/src/application/ports/realtime_ports.rs b/src/application/ports/realtime_ports.rs index 85a10019..3e0e2c39 100644 --- a/src/application/ports/realtime_ports.rs +++ b/src/application/ports/realtime_ports.rs @@ -245,6 +245,18 @@ pub enum RealtimeEvent { parent_id: Uuid, actor: Uuid, }, + /// A user's authorization changed — publishes on + /// [`Topic::UserAuthz`]. The WS handler auto-subscribes each + /// session to its own `user:{caller}:authz` topic; on receipt it + /// walks the session's active subscriptions and evicts any whose + /// resource is in `affected_folders`, emitting a `rt.revoked` + /// notification per evicted topic. + /// + /// MVP carries folder UUIDs only (the only resource-scoped topic + /// that ships in Phase A). When file/drive/calendar topics land, + /// the payload extends with additional resource classes — see the + /// plan's Phase-B roadmap. + AuthzChanged { affected_folders: Vec }, } // ════════════════════════════════════════════════════════════════════════════ @@ -529,6 +541,12 @@ mod tests { }, "folder_deleted", ), + ( + RealtimeEvent::AuthzChanged { + affected_folders: vec![Uuid::nil()], + }, + "authz_changed", + ), ]; for (ev, expected) in cases { let json = serde_json::to_value(ev).unwrap(); diff --git a/src/bin/generate-asyncapi.rs b/src/bin/generate-asyncapi.rs index 8da99838..8824e4ef 100644 --- a/src/bin/generate-asyncapi.rs +++ b/src/bin/generate-asyncapi.rs @@ -116,6 +116,7 @@ fn channels() -> Value { "SubscribedResponse": { "$ref": "#/components/messages/RtSubscribedResponse" }, "ErrorResponse": { "$ref": "#/components/messages/RtErrorResponse" }, "FolderEvent": { "$ref": "#/components/messages/RtFolderEventNotification" }, + "RevokedNotification": { "$ref": "#/components/messages/RtRevokedNotification" }, } }, "UserAuthz": { @@ -165,6 +166,14 @@ fn operations() -> Value { { "$ref": "#/channels/Folder/messages/FolderEvent" } ] }, + "receiveRevoked": { + "action": "receive", + "channel": { "$ref": "#/channels/Folder" }, + "summary": "Server-initiated eviction of a subscription (grant revoked, resource deleted, etc.). Client stops rendering the topic.", + "messages": [ + { "$ref": "#/channels/Folder/messages/RevokedNotification" } + ] + }, // Application-layer keepalive. Separate from the RFC 6455 Ping // control frame the server sends on `OXICLOUD_RT_WS_KEEPALIVE_SECONDS` // (which is transport-level and not modelled in AsyncAPI). This @@ -235,6 +244,12 @@ fn components() -> Value { "title": "Folder mutation event", "contentType": "application/json", "payload": { "$ref": "#/components/schemas/RtFolderEventBody" }, + }, + "RtRevokedNotification": { + "name": "rt.revoked", + "title": "Subscription evicted", + "contentType": "application/json", + "payload": { "$ref": "#/components/schemas/RtRevokedBody" }, } }, "schemas": { @@ -245,6 +260,7 @@ fn components() -> Value { "RtPongResponseBody": rpc_pong_response_schema(), "RtErrorResponseBody": rpc_error_response_schema(), "RtFolderEventBody": folder_event_notification_schema(), + "RtRevokedBody": revoked_notification_schema(), "FileCreatedData": file_created_schema(), "FileRenamedData": file_renamed_schema(), "FileMovedData": file_moved_schema(), @@ -526,3 +542,36 @@ fn folder_deleted_schema() -> Value { } }) } + +/// `rt.revoked` notification body — server tells the client that a +/// specific subscription has been evicted. `topic` is the wire-form +/// string the client originally subscribed to. `reason` is the stable +/// eviction vocabulary — never repurpose an existing value (matches +/// the AuthZ audit-line convention). +fn revoked_notification_schema() -> Value { + json!({ + "type": "object", + "description": "JSON-RPC notification (no `id`). `method = \"rt.revoked\"`.", + "required": ["jsonrpc", "method", "params"], + "properties": { + "jsonrpc": { "type": "string", "const": "2.0" }, + "method": { "type": "string", "const": "rt.revoked" }, + "params": { + "type": "object", + "required": ["topic", "reason"], + "properties": { + "topic": { "type": "string" }, + "reason": { + "type": "string", + "enum": [ + "grant_revoked", + "resource_deleted", + "group_membership_lost", + "admin_kick", + ] + } + } + } + } + }) +} diff --git a/src/bin/rt-hurl-helper.rs b/src/bin/rt-hurl-helper.rs index 5eac6e96..36bc446b 100644 --- a/src/bin/rt-hurl-helper.rs +++ b/src/bin/rt-hurl-helper.rs @@ -261,6 +261,11 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { } let mut events: Vec = Vec::new(); + // Server-initiated eviction notifications (`rt.revoked`) — captured + // separately from `rt.event` so scenarios can assert on eviction + // scoping (evicted topic vs. surviving topic) without conflating + // them with real content events. + let mut revoked: Vec = Vec::new(); // Count server-initiated protocol Pings so scenarios can assert the // keepalive fires. tokio-tungstenite queues an auto-Pong on the next // write path, so we don't need to send one ourselves; we just observe @@ -327,14 +332,25 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { // Notification (id-less)? let method = value.get("method").and_then(|v| v.as_str()).unwrap_or(""); - if method == "rt.event" - && let Some(params) = value.get("params") - { - events.push(params.clone()); + match method { + "rt.event" => { + if let Some(params) = value.get("params") { + events.push(params.clone()); + } + } + "rt.revoked" => { + // Server evicted one of our subscriptions. Record for + // the shell to assert on; do NOT increment `events` — + // eviction is orthogonal to content delivery. + if let Some(params) = value.get("params") { + revoked.push(params.clone()); + } + } + _ => { + // Unknown notification method — ignored. `rt.pong` and + // future server-pushed methods land here silently. + } } - // Other notifications (`rt.revoked`, `rt.pong`) — ignored for - // subscribe-and-collect. They can be added to the output - // schema when scenarios need them. } // Assertion: at least `expect_events` collected before timeout. @@ -345,6 +361,7 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { let summary = json!({ "subscribed": subscribed, "events": events, + "revoked": revoked, "pings_received": pings_received, "timed_out": timed_out, }); diff --git a/src/interfaces/api/handlers/grant_handler.rs b/src/interfaces/api/handlers/grant_handler.rs index feb6f17e..93dda8f1 100644 --- a/src/interfaces/api/handlers/grant_handler.rs +++ b/src/interfaces/api/handlers/grant_handler.rs @@ -504,6 +504,26 @@ pub async fn revoke_grant( self_revoke = (granter == caller_id), "🗑️ grant revoked", ); + + // Realtime eviction cascade — the revoke committed, so any WS + // session that had the affected user auto-subscribed to + // `user:{u}:authz` gets an AuthzChanged event and drops any live + // subscriptions to the affected resource. Silent no-op when the + // subject isn't a User (Group / Token subjects don't have live + // sessions to notify — group cascade is Phase-B once group + // membership expansion ships). Folder resources only for MVP; + // File/Drive topics don't exist yet. + if let (Subject::User(target_user), Resource::Folder(folder_id)) = (subject, resource) { + use crate::application::ports::realtime_ports::{RealtimeBus, RealtimeEvent, Topic}; + RealtimeBus::publish( + state.bus.as_ref(), + &Topic::UserAuthz(target_user), + RealtimeEvent::AuthzChanged { + affected_folders: vec![folder_id], + }, + ); + } + StatusCode::NO_CONTENT.into_response() } diff --git a/src/interfaces/api/handlers/rt_ws.rs b/src/interfaces/api/handlers/rt_ws.rs index 510bf88f..d96ee7ad 100644 --- a/src/interfaces/api/handlers/rt_ws.rs +++ b/src/interfaces/api/handlers/rt_ws.rs @@ -184,16 +184,45 @@ impl Drop for Sub { } } +/// Messages the per-topic reader tasks send to the session's main +/// loop. Two shapes: +/// +/// - `Frame` — a client-bound text frame (`rt.event` notification, +/// `rt.revoked` notification, whatever). Main loop writes it to +/// the socket. +/// - `EvictFolders` — internal control signal. The reader for the +/// session's auto-subscribed `user:{caller}:authz` topic translates +/// inbound [`RealtimeEvent::AuthzChanged`] events into this rather +/// than a client-visible frame. Main loop walks its subs, drops any +/// whose resource is in the list, and emits one `rt.revoked` frame +/// per evicted topic. +enum SessionOut { + Frame(String), + EvictFolders(Vec), +} + async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc) { - // Outbound queue — every path that produces a text frame for the - // client enqueues here; the writer half of the select drains. - let (out_tx, mut out_rx) = mpsc::channel::(OUTBOUND_CHANNEL_CAPACITY); + // Outbound queue — every path that produces a client-bound frame + // enqueues here; the writer half of the select drains. Also + // carries internal `EvictFolders` control signals from the + // authz reader — the main loop reacts to those without them + // hitting the socket. + let (out_tx, mut out_rx) = mpsc::channel::(OUTBOUND_CHANNEL_CAPACITY); // Active subscriptions on this session. Keyed by the wire-form topic // string so an incoming `rt.unsubscribe` with the same string is // recognised without re-parsing. let mut subs: HashMap = HashMap::new(); + // Auto-subscribe to the caller's private authz-change topic. + // No AuthZ check (identity-scoped: caller_id == user_id by + // construction), no client `rt.subscribe` frame. The reader for + // this topic translates `AuthzChanged` events into + // `SessionOut::EvictFolders` signals instead of pushing an + // `rt.event` notification the client can see — client-visible + // effect is the `rt.revoked` per evicted sub. + install_subscription(Topic::UserAuthz(caller_id), &mut subs, &out_tx, &state); + // Server-initiated protocol Ping ticker — prevents intermediate // proxies (Traefik, nginx, Cloudflare) and NAT boxes from reaping // the TCP session as idle. Browsers can't send Ping control frames @@ -238,11 +267,36 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc { match outbound { - Some(text) => { + Some(SessionOut::Frame(text)) => { if socket.send(Message::Text(text.into())).await.is_err() { break; } } + Some(SessionOut::EvictFolders(folders)) => { + // Grant-revocation cascade. Walk the sub set; + // drop any Folder(id) whose id is in the list; + // emit one `rt.revoked` frame per eviction so + // the client knows to stop rendering that + // resource. Idempotent: re-evicting an + // already-gone topic is a no-op. + for folder_uuid in folders { + let wire = Topic::Folder(folder_uuid).to_wire_key(); + if subs.remove(&wire).is_some() { + let frame = revoked_notification( + &wire, + "grant_revoked", + ); + if socket + .send(Message::Text(frame.into())) + .await + .is_err() + { + return; // session dead + } + audit_evicted(caller_id, &wire, "grant_revoked"); + } + } + } None => break, // out_tx dropped — unreachable but safe } } @@ -299,7 +353,7 @@ async fn handle_text_frame( caller_id: Uuid, state: &Arc, subs: &mut HashMap, - out_tx: &mpsc::Sender, + out_tx: &mpsc::Sender, ) -> Option { // Parse envelope. On malformed JSON: reply with an id-less error per // JSON-RPC 2.0 (id = null when the request couldn't be parsed). @@ -346,7 +400,7 @@ async fn handle_subscribe( caller_id: Uuid, state: &Arc, subs: &mut HashMap, - out_tx: &mpsc::Sender, + out_tx: &mpsc::Sender, ) -> String { // Extract topic. let topic_str = match params.get("topic").and_then(Value::as_str) { @@ -437,20 +491,7 @@ async fn handle_subscribe( // AuthZ passed — install the subscription and spawn a reader task // that forwards bus events to the outbound channel as `rt.event` // notifications. - let stream = RealtimeBus::subscribe(state.bus.as_ref(), &topic); - let topic_wire = topic_str.clone(); - let out_tx_task = out_tx.clone(); - let reader = tokio::spawn(async move { - let mut stream = stream; - while let Some(event) = stream.next().await { - let notification = event_notification(&topic_wire, &event); - if out_tx_task.send(notification).await.is_err() { - // Session's outbound channel closed — receiver dropped. - break; - } - } - }); - subs.insert(topic_str.clone(), Sub { reader }); + install_subscription(topic, subs, out_tx, state); success_response(id, serde_json::json!({ "subscribed": topic_str })) } @@ -470,6 +511,62 @@ fn handle_unsubscribe(id: Value, params: Value, subs: &mut HashMap) success_response(id, serde_json::json!({ "unsubscribed": topic_str })) } +// ════════════════════════════════════════════════════════════════════════════ +// Subscription installer +// ════════════════════════════════════════════════════════════════════════════ + +/// Spawn a reader task for `topic` and insert it into `subs`. No AuthZ +/// check — the caller is responsible for gating (either via +/// `handle_subscribe`'s explicit dispatch, or via identity-by- +/// construction for the auto-subscribed `Topic::UserAuthz(caller)`). +/// +/// The reader interprets bus events differently by topic class: +/// +/// - For `Topic::UserAuthz(_)`: an incoming `RealtimeEvent::AuthzChanged` +/// is translated to `SessionOut::EvictFolders(affected)` — the main +/// loop then walks the sub set and drops matching topics. Any other +/// event kind on this topic is ignored (defensive; shouldn't happen +/// in MVP). +/// - For every other topic: bus events are wrapped into a client- +/// visible `rt.event` notification and pushed as `SessionOut::Frame`. +fn install_subscription( + topic: Topic, + subs: &mut HashMap, + out_tx: &mpsc::Sender, + state: &Arc, +) { + let topic_wire = topic.to_wire_key(); + let mut stream = RealtimeBus::subscribe(state.bus.as_ref(), &topic); + let out_tx_task = out_tx.clone(); + let translate_authz = matches!(topic, Topic::UserAuthz(_)); + // Clone for the reader closure; keep the original to key `subs`. + let topic_wire_reader = topic_wire.clone(); + + let reader = tokio::spawn(async move { + while let Some(event) = stream.next().await { + let message = if translate_authz { + match event { + RealtimeEvent::AuthzChanged { affected_folders } => { + SessionOut::EvictFolders(affected_folders) + } + // The authz topic only carries AuthzChanged in + // MVP; other variants would be a producer bug — + // drop them silently so a mis-wired publish + // doesn't spam the client. + _ => continue, + } + } else { + SessionOut::Frame(event_notification(&topic_wire_reader, &event)) + }; + if out_tx_task.send(message).await.is_err() { + // Session's outbound channel closed — receiver dropped. + break; + } + } + }); + subs.insert(topic_wire, Sub { reader }); +} + // ════════════════════════════════════════════════════════════════════════════ // Envelope helpers // ════════════════════════════════════════════════════════════════════════════ @@ -538,6 +635,22 @@ fn split_event_discriminator(mut event_json: Value) -> (String, Value) { ("unknown".to_owned(), event_json) } +/// Build the server-initiated `rt.revoked` JSON-RPC notification. +/// Emitted when a subscription is evicted mid-session (grant revoked, +/// resource deleted, etc.). Not tied to a request id — client sees +/// this as a signal to stop rendering the topic. +fn revoked_notification(topic_wire: &str, reason: &'static str) -> String { + serde_json::to_string(&RpcNotification { + jsonrpc: JSONRPC_V2, + method: "rt.revoked", + params: serde_json::json!({ + "topic": topic_wire, + "reason": reason, + }), + }) + .expect("RpcNotification always serializes") +} + // ════════════════════════════════════════════════════════════════════════════ // Audit // ════════════════════════════════════════════════════════════════════════════ @@ -553,6 +666,20 @@ fn audit_denied(caller_id: Uuid, topic: &str, reason: &'static str) { ); } +/// Audit line for server-initiated eviction — every `rt.revoked` +/// frame we send should also have a durable trail. Stable `reason` +/// vocabulary matches the WS wire's `reason` field. +fn audit_evicted(caller_id: Uuid, topic: &str, reason: &'static str) { + tracing::info!( + target: "audit", + event = "realtime.subscription_evicted", + reason = reason, + caller_id = %caller_id, + topic = %topic, + "🚫 realtime subscription evicted", + ); +} + // ════════════════════════════════════════════════════════════════════════════ // Tests // ════════════════════════════════════════════════════════════════════════════ diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index 153f2684..350d6eb1 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -28,6 +28,13 @@ # one session → observe TWO `file_moved` # events (one via the A topic, one via B). # Same file_id/from/to on both. +# S8 Grant-revoke eviction — user2 subscribes to A + B (both granted), +# user1 revokes only A → user2 sees +# `rt.revoked` for folder:A AND an event +# on folder:B (upload after revoke). +# Locks in three invariants: eviction +# fires, scoping is per-topic, session +# survives. # # Exit non-zero on any failure — run.sh treats that as a suite failure. # ───────────────────────────────────────────────────────────────────────────── @@ -123,26 +130,42 @@ user2_login=$(c_post "$base_url/api/auth/login" "" \ "$(printf '{"username":"%s","password":"%s"}' "$user2_name" "$user2_pass")") user2_token=$(printf '%s' "$user2_login" | jq -r '.access_token') [[ -n "$user2_token" && "$user2_token" != "null" ]] || die "no user2 token: $user2_login" +# S8 needs user2's UUID to target them as the grant subject. +user2_id=$(printf '%s' "$user2_login" | jq -r '.user.full.user.id') +[[ -n "$user2_id" && "$user2_id" != "null" ]] || die "no user2 id: $user2_login" # ── Helper: create a small file inside a folder via the byte-upload path. # Not delta / instant-upload; keeps the wire simple and hits the same # `upload_file_streaming` publish hook. mkfile_in() { local folder_id="$1" name="$2" token="$3" - local tmpfile + local tmpfile respfile status tmpfile="$(mktemp -t rtbus_body.XXXXXX)" + respfile="$(mktemp -t rtbus_resp.XXXXXX)" printf 'rt-bus-test-payload' > "$tmpfile" # Multipart-upload path used by the frontend for byte uploads. # NOTE: `folder_id` MUST come BEFORE the `file` part — file_handler.rs # streams the parts in order and the fail-fast folder-required check # fires the moment it sees the file bytes; a folder_id sent after the # file arrives too late (returns 400 "folder_id is required"). - curl -sS -X POST \ + # + # Capture body + status so a silent 4xx doesn't look like a timing + # bug in the bus. A stale server binary that lost the publish hook, + # or a schema change that broke the endpoint, would otherwise + # present as "subscribe works, no event, timeout" — exactly the + # shape of a real regression but a completely different root cause. + status=$(curl -sS -o "$respfile" -w "%{http_code}" -X POST \ -H "Authorization: Bearer $token" \ -F "folder_id=$folder_id" \ -F "file=@$tmpfile;filename=$name" \ - "$base_url/api/files/upload" > /dev/null + "$base_url/api/files/upload") rm -f "$tmpfile" + if [[ "$status" -lt 200 || "$status" -ge 300 ]]; then + printf 'mkfile_in FAIL: HTTP %s\nbody: %s\n' "$status" "$(cat "$respfile")" >&2 + rm -f "$respfile" + return 1 + fi + rm -f "$respfile" } # ── Scenario 1 — Positive delivery ────────────────────────────────────────── @@ -371,4 +394,106 @@ if ! jq -e --arg fid "$s7_file_id" --arg from "$folder_a" --arg to "$folder_b" \ fi log "S7 OK" -log "All seven realtime-bus scenarios passed." +# ── Scenario 8 — Grant-revocation eviction (scoped, session survives) ─────── +# The strong version of "eviction fires": prove that revoking one grant +# affects ONLY the corresponding subscription — the session stays alive, +# unrelated subs keep delivering events, and only the revoked topic +# gets `rt.revoked`. +# +# Setup: +# - user1 grants user2 `viewer` on folders A AND B (two independent +# grants; user2 has no prior access). +# - user2 subscribes to BOTH folders on one WS session. +# Action: +# - user1 revokes the grant on folder A only. +# - user1 uploads a file to folder B (the surviving sub). +# Invariants: +# (a) helper output records exactly ONE `rt.revoked` for `folder:A` +# with reason `grant_revoked` — the eviction fired. +# (b) helper output records exactly ONE `file_created` event for +# `folder:B` — the unrelated sub is still delivering. Regression +# that mass-drops subs on any AuthzChanged would surface as 0 +# events. +# (c) `subscribed` contains BOTH folder:A and folder:B — both +# original subs were installed (regression that failed the initial +# subscribe under AuthZ would fail here). +# (d) `timed_out == false` — session actor kept running through the +# revoke + subsequent event. Regression that killed the whole +# session on AuthzChanged would surface as a timeout or the +# helper's `wait` failing. +log "S8: revoke user2's grant on folder A; unrelated sub on B still delivers." +# 8.1 Grant user2 viewer role on folder A + folder B. +grant_a=$(c_post "$base_url/api/grants" "$user1_token" \ + "$(printf '{"subject":{"type":"user","id":"%s"},"resource":{"type":"folder","id":"%s"},"role":"viewer"}' \ + "$user2_id" "$folder_a")") +grant_a_id=$(printf '%s' "$grant_a" | jq -r '.grants[0].id') +[[ -n "$grant_a_id" && "$grant_a_id" != "null" ]] \ + || die "S8: grant on folder A failed: $grant_a" + +grant_b=$(c_post "$base_url/api/grants" "$user1_token" \ + "$(printf '{"subject":{"type":"user","id":"%s"},"resource":{"type":"folder","id":"%s"},"role":"viewer"}' \ + "$user2_id" "$folder_b")") +grant_b_id=$(printf '%s' "$grant_b" | jq -r '.grants[0].id') +[[ -n "$grant_b_id" && "$grant_b_id" != "null" ]] \ + || die "S8: grant on folder B failed: $grant_b" + +# 8.2 user2 subscribes to BOTH folder topics; --expect-events 1 exits +# when the post-revoke upload lands on the SURVIVING sub. +out_s8="$(mktemp -t rtbus_s8.XXXXXX)" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --token "$user2_token" \ + --subscribe "folder:$folder_a" \ + --subscribe "folder:$folder_b" \ + --expect-events 1 \ + --timeout 6s \ + --output "$out_s8" & +helper_pid=$! +sleep 0.4 # let both subscribes install + +# 8.3 user1 revokes only the folder-A grant. +curl -sS -X DELETE \ + -H "Authorization: Bearer $user1_token" \ + "$base_url/api/grants/$grant_a_id" > /dev/null +sleep 0.3 # let AuthzChanged propagate + +# 8.4 user1 uploads to folder B → triggers file_created on the +# surviving sub. +mkfile_in "$folder_b" "s8.txt" "$user1_token" + +if ! wait "$helper_pid"; then + cat "$out_s8" >&2 || true + die "S8: helper did not observe the post-revoke event on folder B" +fi + +# 8.5 Assert on the four invariants. +# (a) One rt.revoked for folder:A with reason grant_revoked. +revoked_count=$(jq -r '.revoked | length' "$out_s8") +[[ "$revoked_count" == "1" ]] \ + || { cat "$out_s8"; die "S8: expected 1 rt.revoked, got $revoked_count"; } +[[ "$(jq -r '.revoked[0].topic' "$out_s8")" == "folder:$folder_a" ]] \ + || die "S8: revoked wrong topic: $(jq -r '.revoked[0].topic' "$out_s8")" +[[ "$(jq -r '.revoked[0].reason' "$out_s8")" == "grant_revoked" ]] \ + || die "S8: revoked wrong reason: $(jq -r '.revoked[0].reason' "$out_s8")" + +# (b) One file_created for folder:B — surviving sub delivered. +event_count=$(jq -r '.events | length' "$out_s8") +[[ "$event_count" == "1" ]] \ + || { cat "$out_s8"; die "S8: expected 1 event on surviving sub, got $event_count (regression: mass eviction?)"; } +[[ "$(jq -r '.events[0].event' "$out_s8")" == "file_created" ]] \ + || die "S8: wrong event kind on surviving sub" +[[ "$(jq -r '.events[0].data.parent_id' "$out_s8")" == "$folder_b" ]] \ + || die "S8: wrong parent_id on surviving-sub event" + +# (c) Both original subs were installed. +sub_count=$(jq -r '.subscribed | length' "$out_s8") +[[ "$sub_count" == "2" ]] \ + || { cat "$out_s8"; die "S8: expected both subs installed, got $sub_count"; } + +# (d) Session did not time out — main loop kept running. +[[ "$(jq -r '.timed_out' "$out_s8")" == "false" ]] \ + || die "S8: session timed out (regression: session died on AuthzChanged?)" + +log "S8 OK" + +log "All eight realtime-bus scenarios passed."