cortexkit_usage_steps/
ops.rs

1use serde::{Deserialize, Serialize};
2
3use crate::{codes, Cursor, StepRow};
4
5/// Request of `usage.steps.subscribe`: always `{}`. The provider refuses any
6/// member `malformed`, and so does this type.
7#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
8#[serde(deny_unknown_fields)]
9#[non_exhaustive]
10pub struct SubscribeRequest {}
11
12impl SubscribeRequest {
13    pub fn new() -> Self {
14        Self {}
15    }
16}
17
18/// One StreamData frame of an open subscription: new steps may exist up to
19/// `through`, the newest committed cursor.
20///
21/// Notices are coalesced (at most four a second), and one may cover many
22/// steps or none the caller may see. It carries no rows; read them with
23/// `usage.steps.read`. Unknown members are ignored.
24#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
25#[non_exhaustive]
26pub struct Notice {
27    pub through: Cursor,
28}
29
30impl Notice {
31    pub fn new(through: impl Into<Cursor>) -> Self {
32        Self {
33            through: through.into(),
34        }
35    }
36
37    /// Reports a member outside the published v1 notice list, by name. For
38    /// the provider's own tests; readers ignore unknown members instead.
39    pub fn check_closed(value: &serde_json::Value) -> Result<(), crate::NotClosed> {
40        crate::row::check_members("notice", value, &["through"])
41    }
42}
43
44/// Request of `usage.steps.read`. Both members are optional and also accept
45/// `null`; any other member is refused `malformed`, and so does this type.
46///
47/// Without `after`, the read starts at the oldest row the feed still holds
48/// (the rows committed in the last 24 hours). `limit` defaults to 200; 0 is
49/// refused `invalid_request` and above 1,000 `range_too_large`. This type
50/// does not check `limit`: the provider does.
51#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
52#[serde(deny_unknown_fields)]
53#[non_exhaustive]
54pub struct ReadRequest {
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub after: Option<Cursor>,
57    #[serde(default, skip_serializing_if = "Option::is_none")]
58    pub limit: Option<u32>,
59}
60
61impl ReadRequest {
62    /// A read from the oldest row the feed holds, with the default limit.
63    pub fn new() -> Self {
64        Self::default()
65    }
66
67    /// Reads the rows after `cursor`.
68    pub fn after(mut self, cursor: impl Into<Cursor>) -> Self {
69        self.after = Some(cursor.into());
70        self
71    }
72
73    /// Asks for at most `limit` rows.
74    pub fn with_limit(mut self, limit: u32) -> Self {
75        self.limit = Some(limit);
76        self
77    }
78}
79
80/// Reply of `usage.steps.read`. Unknown members are ignored.
81#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
82#[non_exhaustive]
83pub struct ReadReply {
84    /// Oldest first. Several rows may share an identity; the last one wins.
85    pub rows: Vec<StepRow>,
86    /// The cursor to pass as `after` for the following page, or null when
87    /// this reply reached `through`. Always sent, as null when absent.
88    #[serde(default)]
89    pub next: Option<Cursor>,
90    /// The newest committed cursor.
91    pub through: Cursor,
92}
93
94impl ReadReply {
95    pub fn new(rows: Vec<StepRow>, next: Option<Cursor>, through: impl Into<Cursor>) -> Self {
96        Self {
97            rows,
98            next,
99            through: through.into(),
100        }
101    }
102
103    /// The cursor the next read passes as `after`: `next` while more pages
104    /// remain, otherwise `through`.
105    pub fn resume_after(&self) -> &Cursor {
106        self.next.as_ref().unwrap_or(&self.through)
107    }
108
109    /// Reports a reply member, or a member of any row, outside the published
110    /// v1 lists. For the provider's own tests: whatever it emits must stay
111    /// within the published lists. Readers ignore unknown members instead.
112    pub fn check_closed(value: &serde_json::Value) -> Result<(), crate::NotClosed> {
113        crate::row::check_members("reply", value, &["rows", "next", "through"])?;
114        if let Some(serde_json::Value::Array(rows)) = value.get("rows") {
115            for row in rows {
116                StepRow::check_closed(row)?;
117            }
118        }
119        Ok(())
120    }
121}
122
123/// The body of an Error frame: a refusal of either operation, or the `lagged`
124/// or `cancelled` ending of a subscription.
125///
126/// `code` is one of [`codes`] (an unknown code is kept as received);
127/// `message` is for people and is not a stable interface. `lagged` and
128/// `cursor_expired` carry `detail` `{"through": <cursor>}`. Unknown members
129/// are ignored.
130#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
131#[non_exhaustive]
132pub struct Refusal {
133    pub code: String,
134    #[serde(default)]
135    pub message: String,
136    #[serde(default, skip_serializing_if = "Option::is_none")]
137    pub detail: Option<serde_json::Value>,
138}
139
140impl Refusal {
141    /// A refusal without detail.
142    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
143        Self {
144            code: code.into(),
145            message: message.into(),
146            detail: None,
147        }
148    }
149
150    /// The `lagged` ending: `through` is the last notified cursor.
151    pub fn lagged(through: impl Into<Cursor>, message: impl Into<String>) -> Self {
152        Self::new(codes::LAGGED, message).with_through(through)
153    }
154
155    /// The `cancelled` ending a provider may send after the caller's own
156    /// Cancel. A reader treats it, after its own Cancel, as the normal end.
157    pub fn cancelled(message: impl Into<String>) -> Self {
158        Self::new(codes::CANCELLED, message)
159    }
160
161    /// The `cursor_expired` refusal of a read: `through` is the newest cursor.
162    pub fn cursor_expired(through: impl Into<Cursor>, message: impl Into<String>) -> Self {
163        Self::new(codes::CURSOR_EXPIRED, message).with_through(through)
164    }
165
166    fn with_through(mut self, through: impl Into<Cursor>) -> Self {
167        self.detail = Some(serde_json::json!({ "through": through.into() }));
168        self
169    }
170
171    /// The cursor in `detail.through`, when the refusal carries one.
172    pub fn through(&self) -> Option<Cursor> {
173        self.detail
174            .as_ref()?
175            .get("through")?
176            .as_str()
177            .map(Cursor::new)
178    }
179}
cortexkit_usage_steps/ops.rs

cortexkit_usage_steps/
ops.rs
