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

arete_server/websocket/
frame.rs
