distributed 4.0.2

CQRS/ES framework for Rust using Plain Old Rust Structs — append-only events, replay, snapshots, outbox, service bus, and pluggable infrastructure
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

use super::*;

const MAX_LIVE_RESUME_PROJECTION_BYTES: usize = 512;

/// Parse the private request extension used by generated live operations.
/// Invalid input is a conservative reset signal, never trusted cursor state.
pub(crate) fn parse_requested_live_resume(request: &Request) -> RequestedLiveResume {
    let Some(distributed) = request.extensions.get("distributed") else {
        return RequestedLiveResume::Absent;
    };
    let Ok(distributed) = distributed.clone().into_json() else {
        return RequestedLiveResume::Invalid;
    };
    let Some(distributed) = distributed.as_object() else {
        return RequestedLiveResume::Invalid;
    };
    let Some(resume) = distributed.get("resume") else {
        return RequestedLiveResume::Absent;
    };
    let Some(cursors) = resume
        .as_object()
        .and_then(|resume| resume.get("cursors"))
        .and_then(serde_json::Value::as_array)
    else {
        return RequestedLiveResume::Invalid;
    };
    if cursors.len() > MAX_LIVE_RESUME_CURSORS {
        return RequestedLiveResume::Invalid;
    }

    let mut parsed = Vec::with_capacity(cursors.len());
    for cursor in cursors {
        let Some(cursor) = cursor.as_object() else {
            return RequestedLiveResume::Invalid;
        };
        let Some(projection) = cursor.get("projection").and_then(serde_json::Value::as_str) else {
            return RequestedLiveResume::Invalid;
        };
        if projection.is_empty() || projection.len() > MAX_LIVE_RESUME_PROJECTION_BYTES {
            return RequestedLiveResume::Invalid;
        }
        let Some(position) = cursor.get("position").and_then(serde_json::Value::as_str) else {
            return RequestedLiveResume::Invalid;
        };
        let Ok(parsed_position) = position.parse::<u64>() else {
            return RequestedLiveResume::Invalid;
        };
        if parsed_position.to_string() != position {
            return RequestedLiveResume::Invalid;
        }
        let Some(token) = cursor.get("token").and_then(serde_json::Value::as_str) else {
            return RequestedLiveResume::Invalid;
        };
        let Ok(token) = OpaqueProtocolToken::parse(token) else {
            return RequestedLiveResume::Invalid;
        };
        parsed.push(DistributedLiveCursor {
            projection: projection.to_string(),
            position: position.to_string(),
            token,
        });
    }
    RequestedLiveResume::Cursors(parsed)
}

#[cfg(test)]
mod live_resume_request_tests {
    use super::*;

    fn request_with_resume(value: serde_json::Value) -> Request {
        serde_json::from_value(serde_json::json!({
            "query": "subscription Watch { todos { id } }",
            "extensions": { "distributed": { "resume": value } }
        }))
        .expect("GraphQL request")
    }

    #[test]
    fn live_resume_request_is_bounded_and_canonical() {
        let token = ProtocolTokenCodec::new([9; 32])
            .issue(ProtocolTokenPurpose::LiveResume, &("bounded-test", 7_u64))
            .unwrap();
        let request = request_with_resume(serde_json::json!({
            "cursors": [{
                "projection": "todos",
                "position": "7",
                "token": token.as_str()
            }]
        }));
        let RequestedLiveResume::Cursors(cursors) = parse_requested_live_resume(&request) else {
            panic!("valid cursor must parse")
        };
        assert_eq!(cursors.len(), 1);
        assert_eq!(cursors[0].projection, "todos");
        assert_eq!(cursors[0].position, "7");

        for invalid in [
            serde_json::json!({"cursors": [{
                "projection": "todos", "position": "07", "token": token.as_str()
            }]}),
            serde_json::json!({"cursors": [{
                "projection": "todos", "position": "7", "token": "not-a-token"
            }]}),
            serde_json::json!({"cursors": "not-an-array"}),
        ] {
            assert_eq!(
                parse_requested_live_resume(&request_with_resume(invalid)),
                RequestedLiveResume::Invalid
            );
        }

        let too_many = vec![
            serde_json::json!({
                "projection": "todos",
                "position": "7",
                "token": token.as_str()
            });
            MAX_LIVE_RESUME_CURSORS + 1
        ];
        assert_eq!(
            parse_requested_live_resume(&request_with_resume(
                serde_json::json!({"cursors": too_many})
            )),
            RequestedLiveResume::Invalid
        );
    }

    #[test]
    fn request_without_resume_remains_a_fresh_subscription() {
        let request: Request = serde_json::from_value(serde_json::json!({
            "query": "subscription Watch { todos { id } }",
            "extensions": { "distributed": {} }
        }))
        .unwrap();
        assert_eq!(
            parse_requested_live_resume(&request),
            RequestedLiveResume::Absent
        );
    }
}