arete_server/websocket/
frame.rs

1use serde::ser::{SerializeMap as _, SerializeSeq as _};
2use serde::{Deserialize, Serialize, Serializer};
3use serde_json::value::RawValue;
4use smallvec::SmallVec;
5
6use crate::shared_entity::{maps_sort_keys, SharedEntity};
7use crate::websocket::subscription::{SubscriptionQuery, PROTOCOL_VERSION};
8
9#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
10#[serde(rename_all = "lowercase")]
11pub enum Mode {
12    State,
13    Append,
14    List,
15}
16
17#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
18#[serde(rename_all = "lowercase")]
19pub enum SortOrder {
20    Asc,
21    Desc,
22}
23
24#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
25pub struct SortConfig {
26    pub field: Vec<String>,
27    pub order: SortOrder,
28}
29
30#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
31pub struct WireFormat {
32    pub wide_int_paths: Vec<Vec<String>>,
33}
34
35#[derive(Debug, Clone, Serialize, Deserialize)]
36#[serde(rename_all = "camelCase")]
37pub struct SubscribedFrame {
38    pub protocol_version: u8,
39    pub subscription_id: String,
40    pub op: &'static str,
41    pub query: SubscriptionQuery,
42    pub mode: Mode,
43    #[serde(skip_serializing_if = "Option::is_none")]
44    pub sort: Option<SortConfig>,
45    /// Offsets this view can replay, for append views backed by the event
46    /// journal. Absent when the view has no retained tape.
47    #[serde(skip_serializing_if = "Option::is_none")]
48    pub replay_window: Option<crate::journal::ReplayWindow>,
49    /// The server sends a key whole (`upsert` or a snapshot row) before any
50    /// `patch` for it, so a client may drop a patch for a key it does not
51    /// hold. Older servers omit it, and clients then keep such a patch. Tape
52    /// records (frames with an `offset`) are events and not covered.
53    #[serde(default)]
54    pub whole_entities: bool,
55}
56
57impl SubscribedFrame {
58    pub fn new(
59        subscription_id: String,
60        query: SubscriptionQuery,
61        mode: Mode,
62        sort: Option<SortConfig>,
63    ) -> Self {
64        Self {
65            protocol_version: PROTOCOL_VERSION,
66            subscription_id,
67            op: "subscribed",
68            query,
69            mode,
70            sort,
71            replay_window: None,
72            whole_entities: true,
73        }
74    }
75
76    /// Advertise the offsets a resuming consumer can still ask for.
77    pub fn with_replay_window(mut self, window: crate::journal::ReplayWindow) -> Self {
78        self.replay_window = Some(window);
79        self
80    }
81}
82
83#[derive(Debug, Clone, Serialize, Deserialize)]
84#[serde(rename_all = "camelCase")]
85pub struct UnsubscribedFrame {
86    pub protocol_version: u8,
87    pub subscription_id: String,
88    pub op: &'static str,
89}
90
91impl UnsubscribedFrame {
92    pub fn new(subscription_id: String) -> Self {
93        Self {
94            protocol_version: PROTOCOL_VERSION,
95            subscription_id,
96            op: "unsubscribed",
97        }
98    }
99}
100
101#[derive(Debug, Clone, Serialize, Deserialize)]
102#[serde(rename_all = "camelCase")]
103pub struct Frame {
104    pub protocol_version: u8,
105    pub subscription_id: String,
106    pub mode: Mode,
107    #[serde(rename = "entity")]
108    pub export: String,
109    pub op: String,
110    pub key: String,
111    pub data: serde_json::Value,
112    #[serde(skip_serializing_if = "Vec::is_empty", default)]
113    pub append: Vec<String>,
114    #[serde(skip_serializing_if = "Option::is_none")]
115    pub seq: Option<String>,
116}
117
118impl Frame {
119    pub fn scoped(
120        subscription_id: impl Into<String>,
121        mode: Mode,
122        export: impl Into<String>,
123        op: impl Into<String>,
124        key: impl Into<String>,
125        data: serde_json::Value,
126        seq: Option<String>,
127    ) -> Self {
128        Self {
129            protocol_version: PROTOCOL_VERSION,
130            subscription_id: subscription_id.into(),
131            mode,
132            export: export.into(),
133            op: op.into(),
134            key: key.into(),
135            data,
136            append: vec![],
137            seq,
138        }
139    }
140
141    pub fn entity(&self) -> &str {
142        &self.export
143    }
144
145    pub fn key(&self) -> &str {
146        &self.key
147    }
148}
149
150/// Unscoped payload published by the projector and never sent directly to clients.
151#[derive(Debug, Clone, Serialize)]
152pub(crate) struct SourceFrame {
153    pub mode: Mode,
154    #[serde(rename = "entity")]
155    pub export: String,
156    pub op: &'static str,
157    pub key: String,
158    pub data: serde_json::Value,
159    #[serde(skip_serializing_if = "Vec::is_empty", default)]
160    pub append: Vec<String>,
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub seq: Option<String>,
163    /// Journal offset for replayable append views. Absent for latest-state
164    /// modes, which have no per-event identity to resume from.
165    #[serde(skip_serializing_if = "Option::is_none")]
166    pub offset: Option<u64>,
167}
168
169#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
170pub struct SnapshotEntity {
171    pub key: String,
172    pub data: serde_json::Value,
173}
174
175#[derive(Debug, Clone, Serialize, Deserialize)]
176#[serde(rename_all = "camelCase")]
177pub struct SnapshotFrame {
178    pub protocol_version: u8,
179    pub subscription_id: String,
180    pub snapshot_id: String,
181    pub authoritative: bool,
182    pub mode: Mode,
183    #[serde(rename = "entity")]
184    pub export: String,
185    pub op: &'static str,
186    #[serde(skip_serializing_if = "Option::is_none")]
187    pub key: Option<String>,
188    pub data: Vec<SnapshotEntity>,
189    pub complete: bool,
190}
191
192/// An entity as a view puts it on the wire: serialized with the view's
193/// [`WireFormat`] applied, byte for byte as [`apply_wire_format`] on a copy of
194/// it would serialize, without making the copy.
195pub(crate) struct WireEntity<'a> {
196    pub entity: &'a SharedEntity,
197    pub wire_format: &'a WireFormat,
198}
199
200impl Serialize for WireEntity<'_> {
201    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
202        let wire = WirePaths::new(self.wire_format.wide_int_paths.iter().map(Vec::as_slice));
203        if wire.is_noop() {
204            return self.entity.serialize(serializer);
205        }
206        self.entity.serialize_with(
207            serializer,
208            |key, value| FieldWire {
209                value,
210                wire: wire.field(key),
211            },
212            |value| WireValue { value, wire: &wire },
213        )
214    }
215}
216
217/// A top-level field and the paths that reach it.
218struct FieldWire<'a> {
219    value: &'a serde_json::Value,
220    wire: WirePaths<'a>,
221}
222
223impl Serialize for FieldWire<'_> {
224    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
225        WireValue {
226            value: self.value,
227            wire: &self.wire,
228        }
229        .serialize(serializer)
230    }
231}
232
233/// The wide-int paths that reach one node of a value, as
234/// [`stringify_value_at_path`] walks them: `terminal` when one ends here.
235#[derive(Clone)]
236struct WirePaths<'a> {
237    terminal: bool,
238    /// The rest of every path that continues below this node.
239    paths: SmallVec<[&'a [String]; 4]>,
240}
241
242impl<'a> WirePaths<'a> {
243    fn new(paths: impl Iterator<Item = &'a [String]>) -> Self {
244        let mut terminal = false;
245        let mut rest = SmallVec::new();
246        for path in paths {
247            if path.is_empty() {
248                terminal = true;
249            } else {
250                rest.push(path);
251            }
252        }
253        Self {
254            terminal,
255            paths: rest,
256        }
257    }
258
259    /// The paths reaching an object's field. A path ending at the object does
260    /// not reach into its fields.
261    fn field(&self, key: &str) -> Self {
262        Self::new(
263            self.paths
264                .iter()
265                .filter(|path| path[0] == key)
266                .map(|path| &path[1..]),
267        )
268    }
269
270    fn is_noop(&self) -> bool {
271        !self.terminal && self.paths.is_empty()
272    }
273}
274
275struct WireValue<'a, 'w> {
276    value: &'a serde_json::Value,
277    wire: &'w WirePaths<'a>,
278}
279
280impl Serialize for WireValue<'_, '_> {
281    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
282        use serde_json::Value;
283        if self.wire.is_noop() {
284            return self.value.serialize(serializer);
285        }
286        match self.value {
287            Value::Number(number) if self.wire.terminal => {
288                if let Some(unsigned) = number.as_u64() {
289                    serializer.serialize_str(&unsigned.to_string())
290                } else if let Some(signed) = number.as_i64() {
291                    serializer.serialize_str(&signed.to_string())
292                } else {
293                    number.serialize(serializer)
294                }
295            }
296            // Paths and a path's end both reach every element of an array.
297            Value::Array(values) => {
298                let mut sequence = serializer.serialize_seq(Some(values.len()))?;
299                for value in values {
300                    sequence.serialize_element(&WireValue {
301                        value,
302                        wire: self.wire,
303                    })?;
304                }
305                sequence.end()
306            }
307            Value::Object(fields) => {
308                let mut map = serializer.serialize_map(Some(fields.len()))?;
309                for (key, value) in fields {
310                    let wire = self.wire.field(key);
311                    if wire.is_noop() {
312                        map.serialize_entry(key, value)?;
313                    } else {
314                        map.serialize_entry(key, &WireValue { value, wire: &wire })?;
315                    }
316                }
317                map.end()
318            }
319            other => other.serialize(serializer),
320        }
321    }
322}
323
324/// A source frame's top-level fields, each kept as its JSON text, in the
325/// order parsing the frame into a [`serde_json::Value`] would keep them.
326/// Reading one field parses only that field: a frame's `data` (a whole
327/// entity, for an `upsert`) is never built unless asked for.
328pub(crate) struct SourceFields<'a>(Vec<(String, &'a RawValue)>);
329
330impl<'a> SourceFields<'a> {
331    /// The fields of `payload`, or `None` if it is not a JSON object.
332    pub fn parse(payload: &'a [u8]) -> Option<Self> {
333        serde_json::from_slice(payload).ok()
334    }
335
336    /// One field, parsed.
337    pub fn value(&self, key: &str) -> Option<serde_json::Value> {
338        let (_, raw) = self.0.iter().find(|(field, _)| field == key)?;
339        serde_json::from_str(raw.get()).ok()
340    }
341
342    /// Set a field, as inserting into a parsed object does: a key already
343    /// there keeps its place.
344    pub fn insert(&mut self, key: &str, value: &'a RawValue) {
345        match self.0.iter_mut().find(|(field, _)| field == key) {
346            Some((_, held)) => *held = value,
347            None => self.0.push((key.to_string(), value)),
348        }
349    }
350
351    /// The frame scoped to one subscription: these fields plus
352    /// `protocolVersion` and `subscriptionId`, byte for byte as serializing
353    /// the parsed frame with them inserted would write it. Each field's text
354    /// is copied as it is, which is what reserializing it would write: a
355    /// source frame is itself `serde_json` output.
356    pub fn scoped(payload: &[u8], subscription_id: &str) -> serde_json::Result<Vec<u8>> {
357        let version = RawValue::from_string(PROTOCOL_VERSION.to_string())?;
358        let subscription = serde_json::value::to_raw_value(subscription_id)?;
359        let mut fields: SourceFields<'_> = serde_json::from_slice(payload)?;
360        fields.insert("protocolVersion", &version);
361        fields.insert("subscriptionId", &subscription);
362        if maps_sort_keys() {
363            fields.0.sort_by(|(left, _), (right, _)| left.cmp(right));
364        }
365        serde_json::to_vec(&fields)
366    }
367}
368
369impl<'de> Deserialize<'de> for SourceFields<'de> {
370    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
371        struct Fields;
372
373        impl<'de> serde::de::Visitor<'de> for Fields {
374            type Value = SourceFields<'de>;
375
376            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
377                formatter.write_str("a source frame object")
378            }
379
380            fn visit_map<A: serde::de::MapAccess<'de>>(
381                self,
382                mut map: A,
383            ) -> Result<Self::Value, A::Error> {
384                let mut fields = SourceFields(Vec::new());
385                while let Some((key, value)) = map.next_entry::<String, &'de RawValue>()? {
386                    fields.insert(&key, value);
387                }
388                Ok(fields)
389            }
390        }
391
392        deserializer.deserialize_map(Fields)
393    }
394}
395
396impl Serialize for SourceFields<'_> {
397    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
398        let mut map = serializer.serialize_map(Some(self.0.len()))?;
399        for (key, value) in &self.0 {
400            map.serialize_entry(key, value)?;
401        }
402        map.end()
403    }
404}
405
406pub fn apply_wire_format(value: &mut serde_json::Value, wire_format: &WireFormat) {
407    for path in &wire_format.wide_int_paths {
408        stringify_value_at_path(value, path);
409    }
410}
411
412fn stringify_value_at_path(value: &mut serde_json::Value, path: &[String]) {
413    if path.is_empty() {
414        stringify_wide_int_value(value);
415        return;
416    }
417
418    match value {
419        serde_json::Value::Object(map) => {
420            if let Some(child) = map.get_mut(&path[0]) {
421                stringify_value_at_path(child, &path[1..]);
422            }
423        }
424        serde_json::Value::Array(values) => {
425            for child in values {
426                stringify_value_at_path(child, path);
427            }
428        }
429        _ => {}
430    }
431}
432
433fn stringify_wide_int_value(value: &mut serde_json::Value) {
434    match value {
435        serde_json::Value::Number(number) => {
436            if let Some(unsigned) = number.as_u64() {
437                *value = serde_json::Value::String(unsigned.to_string());
438            } else if let Some(signed) = number.as_i64() {
439                *value = serde_json::Value::String(signed.to_string());
440            }
441        }
442        serde_json::Value::Array(values) => {
443            for child in values {
444                stringify_wide_int_value(child);
445            }
446        }
447        _ => {}
448    }
449}
450
451#[cfg(test)]
452mod tests {
453    use super::*;
454    use serde_json::json;
455
456    #[test]
457    fn scoped_live_frames_carry_v2_identity() {
458        let frame = Frame::scoped(
459            "rounds:1",
460            Mode::List,
461            "OreRound/latest",
462            "upsert",
463            "123",
464            json!({"id": 123}),
465            Some("10:000000000001".to_string()),
466        );
467        let value = serde_json::to_value(frame).unwrap();
468        assert_eq!(value["protocolVersion"], 2);
469        assert_eq!(value["subscriptionId"], "rounds:1");
470        assert_eq!(value["seq"], "10:000000000001");
471    }
472
473    #[test]
474    fn snapshot_serializes_conformance_metadata() {
475        let frame = SnapshotFrame {
476            protocol_version: PROTOCOL_VERSION,
477            subscription_id: "rounds:1".to_string(),
478            snapshot_id: "snapshot-1".to_string(),
479            authoritative: true,
480            mode: Mode::List,
481            export: "OreRound/latest".to_string(),
482            op: "snapshot",
483            key: None,
484            data: vec![],
485            complete: true,
486        };
487        let value = serde_json::to_value(frame).unwrap();
488        assert_eq!(value["snapshotId"], "snapshot-1");
489        assert_eq!(value["authoritative"], true);
490        assert_eq!(value["complete"], true);
491    }
492
493    /// Serializing a shared entity with a wire format writes the bytes that
494    /// applying the format to a copy and serializing the copy writes.
495    #[test]
496    fn a_wire_entity_serializes_as_the_formatted_copy() {
497        let entity = json!({
498            "_version": "e:7",
499            "_seq": "10:000000000001",
500            "amount": 42,
501            "negative": -42,
502            "max": u64::MAX,
503            "ratio": 1.5,
504            "label": "7",
505            "small": 5,
506            "nested": {"amount": 9, "deeper": {"amount": [1, -2, [3, 4.5]]}},
507            "positions": [{"liquidity": 9}, {"liquidity": [11, 12]}, 13, {"other": 1}],
508            "matrix": [[1, 2], [3, {"x": 4}]],
509            "empty": {},
510        });
511        let paths = |paths: &[&[&str]]| WireFormat {
512            wide_int_paths: paths
513                .iter()
514                .map(|path| path.iter().map(|segment| segment.to_string()).collect())
515                .collect(),
516        };
517        let formats = [
518            paths(&[]),
519            paths(&[&["amount"]]),
520            paths(&[&["negative"], &["max"], &["ratio"], &["label"]]),
521            paths(&[&["positions", "liquidity"]]),
522            paths(&[&["positions"]]),
523            paths(&[
524                &["nested"],
525                &["nested", "amount"],
526                &["nested", "deeper", "amount"],
527            ]),
528            paths(&[&["matrix"], &["matrix", "x"]]),
529            paths(&[
530                &["_version"],
531                &["_seq"],
532                &["missing", "path"],
533                &["empty", "x"],
534            ]),
535            paths(&[&["amount"], &["amount"]]),
536            paths(&[&[]]),
537        ];
538        for format in formats {
539            for whole in [
540                entity.clone(),
541                json!([1, {"amount": 2}]),
542                json!(3),
543                json!(null),
544            ] {
545                let shared = SharedEntity::new(whole.clone());
546                let mut copy = whole.clone();
547                apply_wire_format(&mut copy, &format);
548                assert_eq!(
549                    serde_json::to_string(&WireEntity {
550                        entity: &shared,
551                        wire_format: &format,
552                    })
553                    .unwrap(),
554                    serde_json::to_string(&copy).unwrap(),
555                    "{whole} with {:?}",
556                    format.wide_int_paths
557                );
558            }
559        }
560    }
561
562    /// The old scoping: parse the frame, insert the two fields, serialize.
563    fn scoped_by_parsing(payload: &[u8], subscription_id: &str) -> Option<Vec<u8>> {
564        let mut value: serde_json::Value = serde_json::from_slice(payload).ok()?;
565        let object = value.as_object_mut()?;
566        object.insert("protocolVersion".to_string(), PROTOCOL_VERSION.into());
567        object.insert(
568            "subscriptionId".to_string(),
569            serde_json::Value::String(subscription_id.to_string()),
570        );
571        serde_json::to_vec(&value).ok()
572    }
573
574    /// Scoping a source frame without parsing its data writes the bytes that
575    /// parsing and reserializing it wrote, for any frame `serde_json` writes.
576    #[test]
577    fn scoped_source_frames_match_reserialized_ones() {
578        let data = json!({
579            "_seq": "10:000000000001",
580            "_version": "e:1",
581            "float": 1.5,
582            "tiny": 1e-7,
583            "whole_float": 3.0,
584            "negative_zero": -0.0,
585            "big": u64::MAX,
586            "negative": i64::MIN,
587            "text": "π \"quoted\" \\ \u{0001} \n\t / \u{2028} \u{1F600}",
588            "nested": {"z": [1, {"b": null, "a": true}], "a": []},
589        });
590        let frames = [
591            json!({"mode": "list", "entity": "Thing/list", "op": "patch", "key": "1", "data": data.clone(), "seq": "10:1"}),
592            json!({"mode": "state", "entity": "Thing/state", "op": "upsert", "key": "k\"ey", "data": data, "append": ["trades"], "offset": 9}),
593            json!({"op": "delete", "data": null, "protocolVersion": 1, "subscriptionId": "old"}),
594            json!({}),
595        ];
596        for frame in frames {
597            for payload in [
598                serde_json::to_vec(&frame).unwrap(),
599                frame.to_string().into_bytes(),
600            ] {
601                assert_eq!(
602                    SourceFields::scoped(&payload, "sub \"1\"").ok(),
603                    scoped_by_parsing(&payload, "sub \"1\""),
604                    "{frame}"
605                );
606            }
607        }
608        // A frame with a key twice reads like the parsed one: the last value
609        // wins.
610        let twice = br#"{"op":"patch","data":{"a":1},"data":{"b":2},"key":"1"}"#;
611        assert_eq!(
612            SourceFields::scoped(twice, "s").ok(),
613            scoped_by_parsing(twice, "s")
614        );
615        for invalid in [&b"[1,2]"[..], b"\"text\"", b"{\"op\":", b""] {
616            assert!(SourceFields::scoped(invalid, "s").is_err());
617            assert!(SourceFields::parse(invalid).is_none());
618        }
619    }
620
621    #[test]
622    fn source_fields_parse_one_field_at_a_time() {
623        let payload = br#"{"op":"patch","seq":"10:1","offset":4,"data":{"a":[1,2]}}"#;
624        let fields = SourceFields::parse(payload).unwrap();
625        assert_eq!(fields.value("op"), Some(json!("patch")));
626        assert_eq!(fields.value("offset"), Some(json!(4)));
627        assert_eq!(fields.value("data"), Some(json!({"a": [1, 2]})));
628        assert_eq!(fields.value("missing"), None);
629    }
630
631    #[test]
632    fn wire_format_stringifies_marked_wide_int_paths() {
633        let wire_format = WireFormat {
634            wide_int_paths: vec![
635                vec!["amount".to_string()],
636                vec!["positions".to_string(), "liquidity".to_string()],
637            ],
638        };
639        let mut value = json!({
640            "amount": 42,
641            "positions": [{"liquidity": 9}, {"liquidity": 11}],
642            "small": 5,
643        });
644        apply_wire_format(&mut value, &wire_format);
645        assert_eq!(value["amount"], "42");
646        assert_eq!(value["positions"][1]["liquidity"], "11");
647        assert_eq!(value["small"], 5);
648    }
649}
arete_server/websocket/frame.rs

arete_server/websocket/
frame.rs
