feat(message-bus): add ping/keepalive on WS + root declaraiton on AsyncAPI

- plan also eviction in case of permison revoked

src/bin/generate-asyncapi.rs

@@ -59,12 +59,16 @@ Phase C (sync-client push, album live) extend the same channels — see
 "#.trim(),
             "license": { "name": "AGPL-3.0-or-later" },
         },
+        // Applied to every message that doesn't set its own — the JSON-RPC
+        // control frames are all `application/json`. Binary Yjs frames
+        // stay out of AsyncAPI (see the Server description for pointers).
+        "defaultContentType": "application/json",
         "servers": {
             "default": {
                 "host": "{host}",
                 "pathname": "/api/rt/ws",
                 "protocol": "wss",
-                "description": "OxiCloud realtime bus WebSocket endpoint",
+                "description": "OxiCloud realtime bus WebSocket endpoint. Text frames are JSON-RPC 2.0. Binary frames (out of AsyncAPI scope) are Yjs sync protocol for the collab editor — see `docs/plan/markdown-collab.md`.",
                 "variables": {
                     "host": {
                         "description": "Server host — replace with the deployment domain",
@@ -78,6 +82,16 @@ Phase C (sync-client push, album live) extend the same channels — see
                 "bindings": {
                     "ws": { "subProtocol": "oxi.rt.v1" }
                 },
+                // Every request MUST be authenticated. Programmatic
+                // clients set `Authorization: Bearer <jwt>` on the WS
+                // upgrade (same header the REST API uses); browser
+                // clients — which can't set headers on `new WebSocket()`
+                // — will use the deferred ticket flow (a plain HTTP
+                // POST issues a short-lived one-shot ticket bound to
+                // the WS URL, see the plan's DPoP-gap section).
+                "security": [
+                    { "$ref": "#/components/securitySchemes/bearerAuth" }
+                ],
             }
         },
         "channels": channels(),
@@ -98,6 +112,7 @@ fn channels() -> Value {
                 "SubscribeRequest":   { "$ref": "#/components/messages/RtSubscribeRequest" },
                 "UnsubscribeRequest": { "$ref": "#/components/messages/RtUnsubscribeRequest" },
                 "PingRequest":        { "$ref": "#/components/messages/RtPingRequest" },
+                "PongResponse":       { "$ref": "#/components/messages/RtPongResponse" },
                 "SubscribedResponse": { "$ref": "#/components/messages/RtSubscribedResponse" },
                 "ErrorResponse":      { "$ref": "#/components/messages/RtErrorResponse" },
                 "FolderEvent":        { "$ref": "#/components/messages/RtFolderEventNotification" },
@@ -149,6 +164,26 @@ fn operations() -> Value {
             "messages": [
                 { "$ref": "#/channels/Folder/messages/FolderEvent" }
             ]
+        },
+        // 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
+        // operation lets a client actively confirm the socket is
+        // end-to-end alive when transport-level Pings alone can't rule
+        // out a proxy black-hole.
+        "ping": {
+            "action": "send",
+            "channel": { "$ref": "#/channels/Folder" },
+            "summary": "Application-level keepalive; `rt.pong` reply confirms end-to-end liveness",
+            "messages": [
+                { "$ref": "#/channels/Folder/messages/PingRequest" }
+            ],
+            "reply": {
+                "channel": { "$ref": "#/channels/Folder" },
+                "messages": [
+                    { "$ref": "#/channels/Folder/messages/PongResponse" }
+                ]
+            }
         }
     })
 }
@@ -188,6 +223,12 @@ fn components() -> Value {
                 "contentType": "application/json",
                 "payload": { "$ref": "#/components/schemas/RtErrorResponseBody" },
             },
+            "RtPongResponse": {
+                "name": "rt.pong",
+                "title": "Reply to rt.ping — `result.pong == true`",
+                "contentType": "application/json",
+                "payload": { "$ref": "#/components/schemas/RtPongResponseBody" },
+            },
             // ── Notifications (server → client) ─────────────────────
             "RtFolderEventNotification": {
                 "name": "rt.event",
@@ -201,10 +242,22 @@ fn components() -> Value {
             "RtUnsubscribeRequestBody": rpc_request_schema("rt.unsubscribe", topic_params_schema()),
             "RtPingRequestBody": rpc_request_schema("rt.ping", json!({ "type": "null" })),
             "RtSuccessResponseBody": rpc_success_response_schema(),
+            "RtPongResponseBody": rpc_pong_response_schema(),
             "RtErrorResponseBody": rpc_error_response_schema(),
             "RtFolderEventBody": folder_event_notification_schema(),
             "FileCreatedData": file_created_schema(),
             "FolderCreatedData": folder_created_schema(),
+        },
+        // How the client authenticates. Handler side is `auth_middleware`
+        // — the same middleware every `/api/*` request goes through, so
+        // any JWT valid for REST is valid for WS.
+        "securitySchemes": {
+            "bearerAuth": {
+                "type": "http",
+                "scheme": "bearer",
+                "bearerFormat": "JWT",
+                "description": "OxiCloud JWT — same access_token minted by `POST /api/auth/login` (or the OPAQUE handshake). Programmatic clients set `Authorization: Bearer <jwt>` on the WS upgrade request. Browsers, which cannot set headers on `new WebSocket()`, will use the deferred ticket flow (`POST /api/rt/ticket` → short-lived one-shot ticket in the WS URL); see the plan's DPoP-gap section.",
+            }
         }
     })
 }
@@ -250,6 +303,26 @@ fn rpc_success_response_schema() -> Value {
     })
 }
 
+/// Reply to `rt.ping` — the shape pins `result.pong == true` so
+/// contract tests can assert on it directly.
+fn rpc_pong_response_schema() -> Value {
+    json!({
+        "type": "object",
+        "required": ["jsonrpc", "id", "result"],
+        "properties": {
+            "jsonrpc": { "type": "string", "const": "2.0" },
+            "id":      { "type": ["integer", "string", "null"] },
+            "result": {
+                "type": "object",
+                "required": ["pong"],
+                "properties": {
+                    "pong": { "type": "boolean", "const": true }
+                }
+            },
+        }
+    })
+}
+
 fn rpc_error_response_schema() -> Value {
     // The `code`/`message` catalog is the stable public vocabulary —
     // any change here IS a wire break. Every entry mirrors

src/bin/rt-hurl-helper.rs

@@ -261,6 +261,11 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> {
     }
 
     let mut events: Vec<Value> = 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
+    // the frame.
+    let mut pings_received: usize = 0;
     let mut timed_out = false;
 
     let deadline = tokio::time::Instant::now() + args.timeout;
@@ -290,9 +295,16 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> {
             }
         };
 
-        let Message::Text(text) = msg else {
-            // Ignore ping/pong/binary; server may send close later.
-            continue;
+        let text = match msg {
+            Message::Text(t) => t,
+            Message::Ping(_) => {
+                // Server-initiated keepalive — observable proof that the
+                // interval is firing. tokio-tungstenite queues an
+                // auto-Pong on the next flush; nothing to do here.
+                pings_received += 1;
+                continue;
+            }
+            _ => continue, // pong/binary/close — not asserted on
         };
         let value: Value = serde_json::from_str(&text)
             .map_err(|e| HelperError::Protocol(format!("bad frame: {e}: {text}")))?;
@@ -333,6 +345,7 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> {
         let summary = json!({
             "subscribed": subscribed,
             "events": events,
+            "pings_received": pings_received,
             "timed_out": timed_out,
         });
         std::fs::write(path, serde_json::to_vec_pretty(&summary).unwrap())

src/interfaces/api/handlers/rt_ws.rs

@@ -35,7 +35,9 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
+use axum::body::Bytes;
 use axum::extract::State;
 use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
 use axum::response::Response;
@@ -44,6 +46,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
+use tokio::time::MissedTickBehavior;
 use uuid::Uuid;
 
 use crate::application::ports::authorization_ports::AuthorizationEngine;
@@ -64,6 +67,31 @@ const MAX_SUBSCRIPTIONS_PER_CONNECTION: usize = 128;
 /// socket layer doesn't back-pressure into the bus's broadcast ring.
 const OUTBOUND_CHANNEL_CAPACITY: usize = 512;
 
+/// Default server-initiated protocol Ping interval. Keeps intermediate
+/// proxies (Traefik, nginx, Cloudflare) and NAT boxes from reaping the
+/// TCP session as idle. 30 s sits comfortably under nginx's 60 s
+/// default and Cloudflare's 100 s hard limit; behind Traefik we
+/// document a much longer `idleTimeout` anyway.
+///
+/// Overridable at server start via `OXICLOUD_RT_WS_KEEPALIVE_SECONDS`
+/// — test suites drop it to a low value to exercise the keepalive path
+/// within a bounded wall-clock.
+const DEFAULT_KEEPALIVE_SECONDS: u64 = 30;
+
+/// Read the keepalive interval from env at connection time. Kept as a
+/// function rather than a `LazyLock` so a running server with the env
+/// var flipped picks it up on the NEXT connection without a restart —
+/// useful for smoke tests that toggle the value on the fly.
+fn keepalive_interval() -> Duration {
+    Duration::from_secs(
+        std::env::var("OXICLOUD_RT_WS_KEEPALIVE_SECONDS")
+            .ok()
+            .and_then(|s| s.parse().ok())
+            .filter(|&n: &u64| n > 0)
+            .unwrap_or(DEFAULT_KEEPALIVE_SECONDS),
+    )
+}
+
 // ════════════════════════════════════════════════════════════════════════════
 // JSON-RPC 2.0 envelope types
 // ════════════════════════════════════════════════════════════════════════════
@@ -166,6 +194,42 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc<AppSt
     // recognised without re-parsing.
     let mut subs: HashMap<String, Sub> = HashMap::new();
 
+    // 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
+    // (the JS `WebSocket` API doesn't expose them), so the server owns
+    // this responsibility; the client's WS layer auto-Pongs. A truly
+    // dead peer surfaces on the next `socket.send` and breaks out of
+    // the loop the same way any WS error does — no pong-timeout
+    // tracking needed for MVP.
+    //
+    // ─────────────────────── Scaling note ────────────────────────────
+    // This is a `tokio::time::interval` PER connection — not a thread.
+    // The tokio timer wheel handles arbitrary N intervals in O(1) and
+    // each Sleep future is ~150 bytes of state. Per-session task
+    // memory dominates at any interesting N (~1 KB stack), which is
+    // still trivial: 10 000 clients ≈ 12 MB total + ~333 Pings/sec
+    // spread across the worker pool.
+    //
+    // If a deployment ever hits 100 000+ concurrent WS AND the
+    // per-connection interval becomes a measurable cost, the swap is:
+    //   1. one global `tokio::spawn(async { interval.tick().await; ... })`
+    //      task that scans a `DashMap<SessionId, mpsc::Sender<()>>`
+    //      registry and pings each session's mailbox on tick,
+    //   2. session tasks receive the mailbox signal in their `select!`
+    //      and send `Message::Ping` from there (still per-session, so
+    //      one slow socket doesn't block the whole fleet).
+    // Neither pattern change would touch the wire; both are same-file
+    // refactors. Don't do this until N genuinely warrants it — until
+    // then, per-connection is the standard tokio idiom for a reason.
+    let mut keepalive = tokio::time::interval(keepalive_interval());
+    // Coalesce backlog if the runtime pauses (e.g. under heavy load)
+    // rather than firing a burst of Pings when it recovers.
+    keepalive.set_missed_tick_behavior(MissedTickBehavior::Delay);
+    // Discard the immediate first tick — the socket just opened; a
+    // client sending its opening `rt.subscribe` shouldn't race a Ping.
+    keepalive.tick().await;
+
     loop {
         tokio::select! {
             // biased: process outbound before inbound so an event burst
@@ -183,6 +247,15 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc<AppSt
                 }
             }
 
+            _ = keepalive.tick() => {
+                // RFC 6455 Ping control frame. 0-byte payload is
+                // spec-legal and the smallest wire footprint. Client
+                // auto-Pongs; nothing to observe here on that.
+                if socket.send(Message::Ping(Bytes::new())).await.is_err() {
+                    break;
+                }
+            }
+
             incoming = socket.recv() => {
                 match incoming {
                     Some(Ok(Message::Text(txt))) => {
@@ -199,7 +272,10 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc<AppSt
                         // frames on the same connection isn't rejected.
                     }
                     Some(Ok(Message::Ping(_) | Message::Pong(_))) => {
-                        // Handled by axum's WebSocket state machine.
+                        // Client Ping → axum auto-Pongs. Client Pong is
+                        // the response to OUR keepalive Ping — nothing
+                        // to do at the app layer; TCP + WS keep the
+                        // pipe warm regardless.
                     }
                     Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break,
                 }