soothe_client/
protocol.rs

1//! Protocol-1 wire envelope encode/decode.
2
3use serde::{Deserialize, Serialize};
4use serde_json::{json, Map, Value};
5use uuid::Uuid;
6
7use crate::VERSION;
8
9/// Protocol version string.
10pub const PROTO_VERSION: &str = "1";
11
12/// Client version reported in `connection_init` (crate version).
13pub const CLIENT_VERSION: &str = VERSION;
14
15/// Default capabilities declared in the handshake.
16pub const DEFAULT_CLIENT_CAPABILITIES: &[&str] = &["streaming", "batch", "heartbeat", "receipts"];
17
18/// Message class values for the envelope `type` field.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
20#[serde(rename_all = "snake_case")]
21pub enum MessageType {
22    /// Client→server handshake.
23    ConnectionInit,
24    /// Server→client handshake ack.
25    ConnectionAck,
26    /// RPC request.
27    Request,
28    /// RPC success response.
29    Response,
30    /// Fire-and-forget notification.
31    Notification,
32    /// Start a subscription.
33    Subscribe,
34    /// Stream event.
35    Next,
36    /// Structured error.
37    Error,
38    /// Stream completion.
39    Complete,
40    /// Cancel subscription.
41    Unsubscribe,
42    /// Heartbeat ping.
43    Ping,
44    /// Heartbeat pong.
45    Pong,
46    /// Receipt confirmation.
47    ReceiptResponse,
48    /// Graceful disconnect.
49    Disconnect,
50    /// Status frame (often top-level).
51    Status,
52}
53
54impl MessageType {
55    /// Wire string.
56    pub fn as_str(self) -> &'static str {
57        match self {
58            Self::ConnectionInit => "connection_init",
59            Self::ConnectionAck => "connection_ack",
60            Self::Request => "request",
61            Self::Response => "response",
62            Self::Notification => "notification",
63            Self::Subscribe => "subscribe",
64            Self::Next => "next",
65            Self::Error => "error",
66            Self::Complete => "complete",
67            Self::Unsubscribe => "unsubscribe",
68            Self::Ping => "ping",
69            Self::Pong => "pong",
70            Self::ReceiptResponse => "receipt_response",
71            Self::Disconnect => "disconnect",
72            Self::Status => "status",
73        }
74    }
75}
76
77/// Structured error nested under envelope.error.
78#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
79pub struct ErrorObject {
80    /// Numeric code.
81    pub code: i64,
82    /// Message.
83    pub message: String,
84    /// Optional data.
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub data: Option<Value>,
87}
88
89/// Unified `{proto, type, method, params, id}` envelope.
90#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
91pub struct Envelope {
92    /// Protocol version.
93    pub proto: String,
94    /// Message class.
95    #[serde(rename = "type")]
96    pub msg_type: String,
97    /// RPC / subscription method.
98    #[serde(default, skip_serializing_if = "Option::is_none")]
99    pub method: Option<String>,
100    /// Structured params.
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub params: Option<Map<String, Value>>,
103    /// Correlation id.
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub id: Option<String>,
106    /// Success result.
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub result: Option<Map<String, Value>>,
109    /// Structured error.
110    #[serde(default, skip_serializing_if = "Option::is_none")]
111    pub error: Option<ErrorObject>,
112    /// Stream payload (`next`).
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub payload: Option<Value>,
115    /// Optional receipt id.
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub receipt: Option<String>,
118    /// Extra fields (status frames, etc.).
119    #[serde(flatten)]
120    pub extra: Map<String, Value>,
121}
122
123impl Envelope {
124    /// Compact JSON text (no spaces).
125    pub fn to_wire_json(&self) -> Result<String, serde_json::Error> {
126        serde_json::to_string(self)
127    }
128
129    /// Convert to a generic JSON object.
130    pub fn to_value(&self) -> Value {
131        serde_json::to_value(self).unwrap_or(Value::Null)
132    }
133}
134
135/// Generate a 32-hex request id (UUID without dashes).
136pub fn new_request_id() -> String {
137    Uuid::new_v4().simple().to_string()
138}
139
140/// Build a request envelope.
141pub fn new_request(method: impl Into<String>, params: Map<String, Value>) -> Envelope {
142    Envelope {
143        proto: PROTO_VERSION.to_string(),
144        msg_type: "request".into(),
145        method: Some(method.into()),
146        params: if params.is_empty() {
147            None
148        } else {
149            Some(params)
150        },
151        id: Some(new_request_id()),
152        result: None,
153        error: None,
154        payload: None,
155        receipt: None,
156        extra: Map::new(),
157    }
158}
159
160/// Build a notification envelope (no id).
161pub fn new_notification(method: impl Into<String>, params: Map<String, Value>) -> Envelope {
162    Envelope {
163        proto: PROTO_VERSION.to_string(),
164        msg_type: "notification".into(),
165        method: Some(method.into()),
166        params: if params.is_empty() {
167            None
168        } else {
169            Some(params)
170        },
171        id: None,
172        result: None,
173        error: None,
174        payload: None,
175        receipt: None,
176        extra: Map::new(),
177    }
178}
179
180/// Build a subscribe envelope.
181pub fn new_subscribe(method: impl Into<String>, params: Map<String, Value>) -> Envelope {
182    Envelope {
183        proto: PROTO_VERSION.to_string(),
184        msg_type: "subscribe".into(),
185        method: Some(method.into()),
186        params: if params.is_empty() {
187            None
188        } else {
189            Some(params)
190        },
191        id: Some(new_request_id()),
192        result: None,
193        error: None,
194        payload: None,
195        receipt: None,
196        extra: Map::new(),
197    }
198}
199
200/// Build an unsubscribe envelope.
201pub fn new_unsubscribe(id: impl Into<String>) -> Envelope {
202    Envelope {
203        proto: PROTO_VERSION.to_string(),
204        msg_type: "unsubscribe".into(),
205        method: None,
206        params: None,
207        id: Some(id.into()),
208        result: None,
209        error: None,
210        payload: None,
211        receipt: None,
212        extra: Map::new(),
213    }
214}
215
216/// Build connection_init.
217pub fn new_connection_init() -> Envelope {
218    let mut params = Map::new();
219    params.insert("client_version".into(), json!(CLIENT_VERSION));
220    params.insert("client_name".into(), json!("soothe-client-rust"));
221    params.insert("accept_proto".into(), json!([PROTO_VERSION]));
222    params.insert("capabilities".into(), json!(DEFAULT_CLIENT_CAPABILITIES));
223    Envelope {
224        proto: PROTO_VERSION.to_string(),
225        msg_type: "connection_init".into(),
226        method: None,
227        params: Some(params),
228        id: None,
229        result: None,
230        error: None,
231        payload: None,
232        receipt: None,
233        extra: Map::new(),
234    }
235}
236
237/// Build ping.
238pub fn new_ping() -> Envelope {
239    Envelope {
240        proto: PROTO_VERSION.to_string(),
241        msg_type: "ping".into(),
242        method: None,
243        params: None,
244        id: None,
245        result: None,
246        error: None,
247        payload: None,
248        receipt: None,
249        extra: Map::new(),
250    }
251}
252
253/// Build pong.
254pub fn new_pong() -> Envelope {
255    Envelope {
256        proto: PROTO_VERSION.to_string(),
257        msg_type: "pong".into(),
258        method: None,
259        params: None,
260        id: None,
261        result: None,
262        error: None,
263        payload: None,
264        receipt: None,
265        extra: Map::new(),
266    }
267}
268
269/// Build disconnect notification.
270pub fn new_disconnect() -> Envelope {
271    new_notification("disconnect", Map::new())
272}
273
274/// Decode a WebSocket text frame into a JSON value.
275pub fn decode_message(text: &str) -> Result<Value, serde_json::Error> {
276    // Support rare NDJSON frames: take first non-empty line if multiple.
277    let trimmed = text.trim();
278    if trimmed.contains('\n') {
279        for line in trimmed.lines() {
280            let line = line.trim();
281            if !line.is_empty() {
282                return serde_json::from_str(line);
283            }
284        }
285    }
286    serde_json::from_str(trimmed)
287}
288
289/// Expand `event_batch` frames into individual events.
290pub fn expand_wire_messages(msg: Value) -> Vec<Value> {
291    let Some(obj) = msg.as_object() else {
292        return vec![msg];
293    };
294    if obj.get("type").and_then(|v| v.as_str()) != Some("event_batch") {
295        return vec![msg];
296    }
297    match obj.get("events").and_then(|v| v.as_array()) {
298        Some(events) if !events.is_empty() => events.clone(),
299        _ => vec![msg],
300    }
301}
302
303/// Unwrap a `next` envelope to its inner frame when possible.
304///
305/// Matches Go `appkit.UnwrapNext`: when `type==next` and `payload.data` is an
306/// object, return that object; otherwise return the original message.
307pub fn unwrap_next(msg: &Value) -> Value {
308    let Some(obj) = msg.as_object() else {
309        return msg.clone();
310    };
311    if obj.get("type").and_then(|v| v.as_str()) != Some("next") {
312        return msg.clone();
313    }
314    let Some(payload) = obj.get("payload").and_then(|p| p.as_object()) else {
315        return msg.clone();
316    };
317    if let Some(data) = payload.get("data") {
318        if data.is_object() {
319            return data.clone();
320        }
321    }
322    msg.clone()
323}
324
325/// Coerce JSON value to string.
326pub fn as_str(v: &Value) -> &str {
327    v.as_str().unwrap_or("")
328}
329
330/// Map helper: insert string.
331pub fn params_map(pairs: &[(&str, Value)]) -> Map<String, Value> {
332    let mut m = Map::new();
333    for (k, v) in pairs {
334        m.insert((*k).to_string(), v.clone());
335    }
336    m
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342
343    #[test]
344    fn request_id_is_32_hex() {
345        let id = new_request_id();
346        assert_eq!(id.len(), 32);
347        assert!(id.chars().all(|c| c.is_ascii_hexdigit()));
348    }
349
350    #[test]
351    fn connection_init_shape() {
352        let env = new_connection_init();
353        assert_eq!(env.msg_type, "connection_init");
354        assert_eq!(env.proto, "1");
355        let params = env.params.unwrap();
356        assert_eq!(params["client_name"], json!("soothe-client-rust"));
357    }
358
359    #[test]
360    fn expand_event_batch() {
361        let batch = json!({
362            "type": "event_batch",
363            "events": [{"type":"event","mode":"messages"}, {"type":"status","state":"idle"}]
364        });
365        let expanded = expand_wire_messages(batch);
366        assert_eq!(expanded.len(), 2);
367    }
368
369    #[test]
370    fn unwrap_next_returns_payload_data() {
371        let frame = json!({
372            "type": "next",
373            "payload": {
374                "mode": "event",
375                "data": {"type": "event", "mode": "messages", "data": [{"content": "hi"}]}
376            }
377        });
378        let inner = unwrap_next(&frame);
379        assert_eq!(inner["mode"], json!("messages"));
380        assert_eq!(inner["type"], json!("event"));
381    }
382}
soothe_client/protocol.rs

soothe_client/
protocol.rs
