cratestack_axum/rpc/
subscribe_query.rs

1//! A model feed's input (`docs/design/subscriptions-websocket.md` §3.1, §5.3,
2//! slice S3), whichever binding carried it: the REST query of
3//! `GET /<plural>/$subscribe`, or an `RpcSubscribeInput` (RPC `POST`,
4//! WebTransport and WebSocket request frames), which becomes the same query.
5//!
6//! `operations` is split out here; the rest is `list`'s filter syntax, parsed
7//! by the generated `list` parser so a feed filters exactly as a read does.
8//! `list`'s paging, ordering and inclusion parameters mean nothing for a
9//! stream of single events and are refused rather than ignored.
10
11use cratestack_core::rpc::RpcSubscribeInput;
12use cratestack_core::{CratestackError, ModelEventKind, SubscriptionStream};
13use futures_util::StreamExt;
14
15/// The operations a feed wants (`None`: all the model emits) and the rest of
16/// its query, for the `list` parser.
17#[derive(Debug, Default, PartialEq, Eq)]
18pub struct SubscribeQuery {
19    pub operations: Option<Vec<ModelEventKind>>,
20    pub rest: Option<String>,
21}
22
23const LIST_ONLY: &[&str] = &[
24    "limit",
25    "offset",
26    "sort",
27    "orderBy",
28    "include",
29    "computedParams",
30];
31
32pub fn split_subscribe_query(raw: Option<&str>) -> Result<SubscribeQuery, CratestackError> {
33    let mut operations = None;
34    let mut rest = Vec::new();
35    for (key, value) in crate::parse_query_pairs(raw)? {
36        if key == "operations" {
37            let parsed = value
38                .split(',')
39                .map(str::trim)
40                .filter(|op| !op.is_empty())
41                .map(|op| {
42                    ModelEventKind::parse(op).map_err(|_| {
43                        CratestackError::BadRequest(format!(
44                            "unknown operation `{op}`: expected created, updated or deleted"
45                        ))
46                    })
47                })
48                .collect::<Result<Vec<_>, _>>()?;
49            operations = Some(parsed);
50        } else if LIST_ONLY.contains(&key.as_str()) || key.starts_with("includeFields[") {
51            return Err(CratestackError::BadRequest(format!(
52                "`{key}` does not apply to a subscription"
53            )));
54        } else {
55            rest.push((key, value));
56        }
57    }
58    let rest = (!rest.is_empty()).then(|| {
59        url::form_urlencoded::Serializer::new(String::new())
60            .extend_pairs(rest.iter().map(|(k, v)| (k.as_str(), v.as_str())))
61            .finish()
62    });
63    Ok(SubscribeQuery { operations, rest })
64}
65
66/// A request frame's input as the REST query: `null` or `{}` is none.
67pub fn subscribe_input_query(input: serde_json::Value) -> Result<Option<String>, CratestackError> {
68    if input.is_null() {
69        return Ok(None);
70    }
71    let input: RpcSubscribeInput = serde_json::from_value(input).map_err(|error| {
72        CratestackError::BadRequest(format!("invalid subscription input: {error}"))
73    })?;
74    Ok(input.to_query())
75}
76
77/// `items` with each event's record narrowed to `fields` (`None`: all).
78pub fn project_event_fields(
79    items: SubscriptionStream<serde_json::Value>,
80    fields: Option<Vec<String>>,
81) -> SubscriptionStream<serde_json::Value> {
82    let Some(fields) = fields else {
83        return items;
84    };
85    Box::pin(items.map(move |item| {
86        item.map(|mut event| {
87            if let Some(serde_json::Value::Object(data)) = event.get_mut("data") {
88                data.retain(|key, _| fields.iter().any(|field| field == key));
89            }
90            event
91        })
92    }))
93}
94
95#[cfg(test)]
96mod tests;
cratestack_axum/rpc/subscribe_query.rs

cratestack_axum/rpc/
subscribe_query.rs
