arete_server/websocket/
frame.rs

1use serde::{Deserialize, Serialize};
2
3use crate::websocket::subscription::{SubscriptionQuery, PROTOCOL_VERSION};
4
5#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
6#[serde(rename_all = "lowercase")]
7pub enum Mode {
8    State,
9    Append,
10    List,
11}
12
13#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
14#[serde(rename_all = "lowercase")]
15pub enum SortOrder {
16    Asc,
17    Desc,
18}
19
20#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
21pub struct SortConfig {
22    pub field: Vec<String>,
23    pub order: SortOrder,
24}
25
26#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
27pub struct WireFormat {
28    pub wide_int_paths: Vec<Vec<String>>,
29}
30
31#[derive(Debug, Clone, Serialize, Deserialize)]
32#[serde(rename_all = "camelCase")]
33pub struct SubscribedFrame {
34    pub protocol_version: u8,
35    pub subscription_id: String,
36    pub op: &'static str,
37    pub query: SubscriptionQuery,
38    pub mode: Mode,
39    #[serde(skip_serializing_if = "Option::is_none")]
40    pub sort: Option<SortConfig>,
41    /// Offsets this view can replay, for append views backed by the event
42    /// journal. Absent when the view has no retained tape.
43    #[serde(skip_serializing_if = "Option::is_none")]
44    pub replay_window: Option<crate::journal::ReplayWindow>,
45    /// The server sends a key whole (`upsert` or a snapshot row) before any
46    /// `patch` for it, so a client may drop a patch for a key it does not
47    /// hold. Older servers omit it, and clients then keep such a patch. Tape
48    /// records (frames with an `offset`) are events and not covered.
49    #[serde(default)]
50    pub whole_entities: bool,
51}
52
53impl SubscribedFrame {
54    pub fn new(
55        subscription_id: String,
56        query: SubscriptionQuery,
57        mode: Mode,
58        sort: Option<SortConfig>,
59    ) -> Self {
60        Self {
61            protocol_version: PROTOCOL_VERSION,
62            subscription_id,
63            op: "subscribed",
64            query,
65            mode,
66            sort,
67            replay_window: None,
68            whole_entities: true,
69        }
70    }
71
72    /// Advertise the offsets a resuming consumer can still ask for.
73    pub fn with_replay_window(mut self, window: crate::journal::ReplayWindow) -> Self {
74        self.replay_window = Some(window);
75        self
76    }
77}
78
79#[derive(Debug, Clone, Serialize, Deserialize)]
80#[serde(rename_all = "camelCase")]
81pub struct UnsubscribedFrame {
82    pub protocol_version: u8,
83    pub subscription_id: String,
84    pub op: &'static str,
85}
86
87impl UnsubscribedFrame {
88    pub fn new(subscription_id: String) -> Self {
89        Self {
90            protocol_version: PROTOCOL_VERSION,
91            subscription_id,
92            op: "unsubscribed",
93        }
94    }
95}
96
97#[derive(Debug, Clone, Serialize, Deserialize)]
98#[serde(rename_all = "camelCase")]
99pub struct Frame {
100    pub protocol_version: u8,
101    pub subscription_id: String,
102    pub mode: Mode,
103    #[serde(rename = "entity")]
104    pub export: String,
105    pub op: String,
106    pub key: String,
107    pub data: serde_json::Value,
108    #[serde(skip_serializing_if = "Vec::is_empty", default)]
109    pub append: Vec<String>,
110    #[serde(skip_serializing_if = "Option::is_none")]
111    pub seq: Option<String>,
112}
113
114impl Frame {
115    pub fn scoped(
116        subscription_id: impl Into<String>,
117        mode: Mode,
118        export: impl Into<String>,
119        op: impl Into<String>,
120        key: impl Into<String>,
121        data: serde_json::Value,
122        seq: Option<String>,
123    ) -> Self {
124        Self {
125            protocol_version: PROTOCOL_VERSION,
126            subscription_id: subscription_id.into(),
127            mode,
128            export: export.into(),
129            op: op.into(),
130            key: key.into(),
131            data,
132            append: vec![],
133            seq,
134        }
135    }
136
137    pub fn entity(&self) -> &str {
138        &self.export
139    }
140
141    pub fn key(&self) -> &str {
142        &self.key
143    }
144}
145
146/// Unscoped payload published by the projector and never sent directly to clients.
147#[derive(Debug, Clone, Serialize)]
148pub(crate) struct SourceFrame {
149    pub mode: Mode,
150    #[serde(rename = "entity")]
151    pub export: String,
152    pub op: &'static str,
153    pub key: String,
154    pub data: serde_json::Value,
155    #[serde(skip_serializing_if = "Vec::is_empty", default)]
156    pub append: Vec<String>,
157    #[serde(skip_serializing_if = "Option::is_none")]
158    pub seq: Option<String>,
159    /// Journal offset for replayable append views. Absent for latest-state
160    /// modes, which have no per-event identity to resume from.
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub offset: Option<u64>,
163}
164
165#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
166pub struct SnapshotEntity {
167    pub key: String,
168    pub data: serde_json::Value,
169}
170
171#[derive(Debug, Clone, Serialize, Deserialize)]
172#[serde(rename_all = "camelCase")]
173pub struct SnapshotFrame {
174    pub protocol_version: u8,
175    pub subscription_id: String,
176    pub snapshot_id: String,
177    pub authoritative: bool,
178    pub mode: Mode,
179    #[serde(rename = "entity")]
180    pub export: String,
181    pub op: &'static str,
182    #[serde(skip_serializing_if = "Option::is_none")]
183    pub key: Option<String>,
184    pub data: Vec<SnapshotEntity>,
185    pub complete: bool,
186}
187
188pub fn apply_wire_format(value: &mut serde_json::Value, wire_format: &WireFormat) {
189    for path in &wire_format.wide_int_paths {
190        stringify_value_at_path(value, path);
191    }
192}
193
194fn stringify_value_at_path(value: &mut serde_json::Value, path: &[String]) {
195    if path.is_empty() {
196        stringify_wide_int_value(value);
197        return;
198    }
199
200    match value {
201        serde_json::Value::Object(map) => {
202            if let Some(child) = map.get_mut(&path[0]) {
203                stringify_value_at_path(child, &path[1..]);
204            }
205        }
206        serde_json::Value::Array(values) => {
207            for child in values {
208                stringify_value_at_path(child, path);
209            }
210        }
211        _ => {}
212    }
213}
214
215fn stringify_wide_int_value(value: &mut serde_json::Value) {
216    match value {
217        serde_json::Value::Number(number) => {
218            if let Some(unsigned) = number.as_u64() {
219                *value = serde_json::Value::String(unsigned.to_string());
220            } else if let Some(signed) = number.as_i64() {
221                *value = serde_json::Value::String(signed.to_string());
222            }
223        }
224        serde_json::Value::Array(values) => {
225            for child in values {
226                stringify_wide_int_value(child);
227            }
228        }
229        _ => {}
230    }
231}
232
233#[cfg(test)]
234mod tests {
235    use super::*;
236    use serde_json::json;
237
238    #[test]
239    fn scoped_live_frames_carry_v2_identity() {
240        let frame = Frame::scoped(
241            "rounds:1",
242            Mode::List,
243            "OreRound/latest",
244            "upsert",
245            "123",
246            json!({"id": 123}),
247            Some("10:000000000001".to_string()),
248        );
249        let value = serde_json::to_value(frame).unwrap();
250        assert_eq!(value["protocolVersion"], 2);
251        assert_eq!(value["subscriptionId"], "rounds:1");
252        assert_eq!(value["seq"], "10:000000000001");
253    }
254
255    #[test]
256    fn snapshot_serializes_conformance_metadata() {
257        let frame = SnapshotFrame {
258            protocol_version: PROTOCOL_VERSION,
259            subscription_id: "rounds:1".to_string(),
260            snapshot_id: "snapshot-1".to_string(),
261            authoritative: true,
262            mode: Mode::List,
263            export: "OreRound/latest".to_string(),
264            op: "snapshot",
265            key: None,
266            data: vec![],
267            complete: true,
268        };
269        let value = serde_json::to_value(frame).unwrap();
270        assert_eq!(value["snapshotId"], "snapshot-1");
271        assert_eq!(value["authoritative"], true);
272        assert_eq!(value["complete"], true);
273    }
274
275    #[test]
276    fn wire_format_stringifies_marked_wide_int_paths() {
277        let wire_format = WireFormat {
278            wide_int_paths: vec![
279                vec!["amount".to_string()],
280                vec!["positions".to_string(), "liquidity".to_string()],
281            ],
282        };
283        let mut value = json!({
284            "amount": 42,
285            "positions": [{"liquidity": 9}, {"liquidity": 11}],
286            "small": 5,
287        });
288        apply_wire_format(&mut value, &wire_format);
289        assert_eq!(value["amount"], "42");
290        assert_eq!(value["positions"][1]["liquidity"], "11");
291        assert_eq!(value["small"], 5);
292    }
293}
arete_server/websocket/frame.rs

arete_server/websocket/
frame.rs
