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

arete_server/websocket/
frame.rs
