cloacina_client/
ws.rs

1/*
2 *  Copyright 2025-2026 Colliery Software
3 *
4 *  Licensed under the Apache License, Version 2.0 (the "License");
5 *  you may not use this file except in compliance with the License.
6 *  You may obtain a copy of the License at
7 *
8 *      http://www.apache.org/licenses/LICENSE-2.0
9 *
10 *  Unless required by applicable law or agreed to in writing, software
11 *  distributed under the License is distributed on an "AS IS" BASIS,
12 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 *  See the License for the specific language governing permissions and
14 *  limitations under the License.
15 */
16
17//! Substrate delivery WebSocket consumer (`GET /v1/ws/delivery/{recipient}`).
18//!
19//! Protocol reference: the WebSocket Protocol page of the docs site;
20//! JSON Schemas under `/schemas/ws/`. Delivery is at-least-once — this
21//! stream dedups on row id and acks each frame after yielding it, so a
22//! consumer crash before processing leaves the row unacked → redelivered.
23
24use std::collections::HashSet;
25use std::time::Duration;
26
27use async_stream::try_stream;
28use futures_util::{Stream, StreamExt};
29
30pub use cloacina_api_types::delivery::DELIVERY_PROTOCOL_VERSION;
31use cloacina_api_types::delivery::{ClientMessage, ServerMessage};
32
33use crate::error::ClientError;
34use crate::Client;
35
36/// The slice of WebSocket traffic the delivery protocol cares about, decoded
37/// per target by [`socket::Socket`]. Anything else (pings, binary) is `Other`.
38enum WsEvent {
39    Text(String),
40    /// Peer closed; carries the close code when the transport exposes one.
41    Close(Option<u16>),
42    Other,
43}
44
45/// Per-target socket transport (CLOACI-T-0932). The protocol loop below is
46/// target-independent; only connect/next/send/sleep differ:
47/// tokio-tungstenite natively, the browser WebSocket (gloo-net) on wasm32.
48#[cfg(not(target_arch = "wasm32"))]
49mod socket {
50    use super::WsEvent;
51    use futures_util::{SinkExt, StreamExt};
52    use tokio_tungstenite::tungstenite::Message;
53
54    pub struct Socket(
55        tokio_tungstenite::WebSocketStream<
56            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
57        >,
58    );
59
60    impl Socket {
61        pub async fn connect(url: &str) -> Result<Self, String> {
62            let (socket, _resp) = tokio_tungstenite::connect_async(url)
63                .await
64                .map_err(|e| e.to_string())?;
65            Ok(Self(socket))
66        }
67
68        pub async fn next(&mut self) -> Option<Result<WsEvent, String>> {
69            let msg = self.0.next().await?;
70            Some(match msg {
71                Ok(Message::Text(text)) => Ok(WsEvent::Text(text)),
72                Ok(Message::Close(frame)) => Ok(WsEvent::Close(frame.map(|f| f.code.into()))),
73                Ok(_) => Ok(WsEvent::Other),
74                Err(e) => Err(e.to_string()),
75            })
76        }
77
78        pub async fn send_text(&mut self, text: String) -> Result<(), String> {
79            self.0
80                .send(Message::Text(text))
81                .await
82                .map_err(|e| e.to_string())
83        }
84    }
85
86    pub async fn sleep(d: std::time::Duration) {
87        tokio::time::sleep(d).await;
88    }
89}
90
91#[cfg(target_arch = "wasm32")]
92mod socket {
93    use super::WsEvent;
94    use futures_util::{SinkExt, StreamExt};
95    use gloo_net::websocket::{futures::WebSocket, Message, WebSocketError};
96
97    pub struct Socket(WebSocket);
98
99    impl Socket {
100        pub async fn connect(url: &str) -> Result<Self, String> {
101            // gloo's open is synchronous (the browser connects in the
102            // background); failures surface on the first read.
103            WebSocket::open(url).map(Self).map_err(|e| e.to_string())
104        }
105
106        pub async fn next(&mut self) -> Option<Result<WsEvent, String>> {
107            let msg = self.0.next().await?;
108            Some(match msg {
109                Ok(Message::Text(text)) => Ok(WsEvent::Text(text)),
110                Ok(Message::Bytes(_)) => Ok(WsEvent::Other),
111                // The browser API folds close into the error path; keep the
112                // code so 4426 (protocol-version) stays terminal.
113                Err(WebSocketError::ConnectionClose(ev)) => Ok(WsEvent::Close(Some(ev.code))),
114                Err(e) => Err(e.to_string()),
115            })
116        }
117
118        pub async fn send_text(&mut self, text: String) -> Result<(), String> {
119            self.0
120                .send(Message::Text(text))
121                .await
122                .map_err(|e| e.to_string())
123        }
124    }
125
126    pub async fn sleep(d: std::time::Duration) {
127        gloo_timers::future::TimeoutFuture::new(d.as_millis() as u32).await;
128    }
129}
130
131/// One decoded delivery push.
132#[derive(Debug, Clone)]
133pub struct DeliveryPush {
134    /// Outbox row id — dedup key, already acked by the stream.
135    pub id: i64,
136    /// Producer-defined payload discriminator (e.g. `execution_event`).
137    pub kind: String,
138    pub recipient: String,
139    pub tenant_id: Option<String>,
140    /// Decoded payload bytes (base64 on the wire).
141    pub payload: Vec<u8>,
142}
143
144/// Options for [`Client::subscribe_delivery`].
145#[derive(Debug, Clone)]
146pub struct SubscribeOptions {
147    /// Reconnect on abnormal closure (default true).
148    pub reconnect: bool,
149    /// Initial reconnect backoff (default 100ms, doubles up to max).
150    pub reconnect_initial: Duration,
151    /// Max reconnect backoff (default 30s).
152    pub reconnect_max: Duration,
153}
154
155impl Default for SubscribeOptions {
156    fn default() -> Self {
157        Self {
158            reconnect: true,
159            reconnect_initial: Duration::from_millis(100),
160            reconnect_max: Duration::from_secs(30),
161        }
162    }
163}
164
165fn ws_base(server: &str) -> Result<String, ClientError> {
166    if let Some(rest) = server.strip_prefix("https://") {
167        Ok(format!("wss://{rest}"))
168    } else if let Some(rest) = server.strip_prefix("http://") {
169        Ok(format!("ws://{rest}"))
170    } else {
171        Err(ClientError::Config(format!(
172            "server must start with http:// or https:// (got {server})"
173        )))
174    }
175}
176
177pub(crate) fn subscribe_delivery(
178    client: Client,
179    recipient: String,
180    options: SubscribeOptions,
181) -> impl Stream<Item = Result<DeliveryPush, ClientError>> {
182    try_stream! {
183        let base = ws_base(client.server())?;
184        let mut seen: HashSet<i64> = HashSet::new();
185        let mut backoff = options.reconnect_initial;
186
187        loop {
188            // Tickets are single-use — mint a fresh one per connection.
189            let ticket = client.create_ws_ticket().await?.ticket;
190            let url = format!(
191                "{base}/v1/ws/delivery/{}?token={}",
192                urlencoding::encode(&recipient),
193                urlencoding::encode(&ticket),
194            );
195
196            let mut socket = socket::Socket::connect(&url)
197                .await
198                .map_err(|e| ClientError::Ws(format!("connect failed for {url}: {e}")))?;
199
200            // Declare our protocol version; an incompatible server closes
201            // with 4426, which we surface as a terminal error below.
202            let hello = serde_json::to_string(&ClientMessage::Hello {
203                protocol_version: DELIVERY_PROTOCOL_VERSION,
204                since_id: None,
205            })
206            .expect("hello serializes");
207            socket
208                .send_text(hello)
209                .await
210                .map_err(|e| ClientError::Ws(format!("hello send failed: {e}")))?;
211
212            let mut close_code: Option<u16> = None;
213
214            while let Some(msg) = socket.next().await {
215                let msg = match msg {
216                    Ok(m) => m,
217                    Err(e) => {
218                        if !options.reconnect {
219                            Err(ClientError::Ws(format!("recv error: {e}")))?;
220                        }
221                        break;
222                    }
223                };
224                match msg {
225                    WsEvent::Text(text) => {
226                        let frame: ServerMessage = match serde_json::from_str(&text) {
227                            Ok(f) => f,
228                            Err(_) => continue, // tolerate unknown frames
229                        };
230                        if let ServerMessage::Push { id, kind, recipient: r, tenant_id, .. } = &frame {
231                            let payload = frame
232                                .decode_push_payload()
233                                .map_err(|e| ClientError::Ws(format!("bad push payload: {e}")))?;
234                            let push = DeliveryPush {
235                                id: *id,
236                                kind: kind.clone(),
237                                recipient: r.clone(),
238                                tenant_id: tenant_id.clone(),
239                                payload,
240                            };
241                            let fresh = seen.insert(push.id);
242                            let ack_id = push.id;
243                            if fresh {
244                                yield push;
245                            }
246                            // Ack after yield: a consumer crash before
247                            // processing leaves the row unacked.
248                            let ack = serde_json::to_string(&ClientMessage::Ack {
249                                protocol_version: DELIVERY_PROTOCOL_VERSION,
250                                id: ack_id,
251                            })
252                            .expect("ack serializes");
253                            if socket.send_text(ack).await.is_err() {
254                                break;
255                            }
256                        }
257                        backoff = options.reconnect_initial;
258                    }
259                    WsEvent::Close(code) => {
260                        close_code = code;
261                        break;
262                    }
263                    WsEvent::Other => {}
264                }
265            }
266
267            // 4426 = unsupported protocol_version — reconnecting cannot help.
268            if close_code == Some(4426) {
269                Err(ClientError::ProtocolVersion {
270                    client_version: DELIVERY_PROTOCOL_VERSION,
271                })?;
272            }
273            if !options.reconnect {
274                break;
275            }
276            socket::sleep(backoff).await;
277            backoff = (backoff * 2).min(options.reconnect_max);
278        }
279    }
280}
281
282pub(crate) fn follow_execution_events(
283    client: Client,
284    execution_id: String,
285    options: SubscribeOptions,
286) -> impl Stream<Item = Result<serde_json::Value, ClientError>> {
287    try_stream! {
288        let recipient = format!("exec_events:{execution_id}");
289        let stream = subscribe_delivery(client, recipient, options);
290        let mut stream = std::pin::pin!(stream);
291        while let Some(push) = stream.next().await {
292            let push = push?;
293            let event: serde_json::Value = serde_json::from_slice(&push.payload)
294                .map_err(|e| ClientError::Ws(format!("push payload is not JSON: {e}")))?;
295            yield event;
296        }
297    }
298}
cloacina_client/ws.rs

cloacina_client/
ws.rs
