cufflink_types/manifest/
subscription.rs

1use crate::features;
2use serde::{Deserialize, Serialize};
3use std::collections::HashMap;
4
5#[derive(Debug, Clone, Default, Serialize, Deserialize)]
6pub struct SubscriptionDefinition {
7    pub subject: String,
8    pub handler: String,
9    pub handler_type: HandlerType,
10    #[serde(default, serialize_with = "crate::serde_util::sorted")]
11    pub config: HashMap<String, String>,
12    /// Explicit JetStream stream to host this subscription's consumer. When
13    /// `None`, the platform uses the tenant stream (`DW_{tenant}`).
14    #[serde(default, skip_serializing_if = "Option::is_none")]
15    pub stream: Option<String>,
16    /// When a delivery is acknowledged.
17    #[serde(default, skip_serializing_if = "AckMode::is_always")]
18    pub ack: AckMode,
19    /// The durable consumer's configuration; the platform's own when unset.
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub consumer: Option<ConsumerSpec>,
22    /// The handler's execution budget; 60 s when unset.
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub timeout_secs: Option<u32>,
25}
26
27impl SubscriptionDefinition {
28    pub const DEFAULT_TIMEOUT_SECS: u32 = 60;
29    /// The range `timeout_secs` must fall in.
30    pub const TIMEOUT_SECS: std::ops::RangeInclusive<u32> = 1..=300;
31
32    /// A subscription with the platform's delivery defaults.
33    pub fn new(
34        subject: impl Into<String>,
35        handler: impl Into<String>,
36        handler_type: HandlerType,
37    ) -> Self {
38        Self {
39            subject: subject.into(),
40            handler: handler.into(),
41            handler_type,
42            ..Self::default()
43        }
44    }
45
46    /// Platform features this subscription's declaration requires.
47    pub fn derived_features(&self) -> impl Iterator<Item = &'static str> {
48        (self.consumer.is_some() || !self.ack.is_always() || self.timeout_secs.is_some())
49            .then_some(features::NATS_CONSUMER_CONFIG)
50            .into_iter()
51    }
52}
53
54#[derive(Debug, Clone, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
55#[serde(rename_all = "snake_case")]
56pub enum HandlerType {
57    DeleteCascade,
58    UpdateField,
59    Webhook,
60    #[default]
61    Wasm,
62}
63
64impl HandlerType {
65    /// The name a manifest declares it by.
66    pub fn as_str(&self) -> &'static str {
67        match self {
68            HandlerType::DeleteCascade => "delete_cascade",
69            HandlerType::UpdateField => "update_field",
70            HandlerType::Webhook => "webhook",
71            HandlerType::Wasm => "wasm",
72        }
73    }
74}
75
76#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
77#[serde(rename_all = "snake_case")]
78pub enum AckMode {
79    /// Acknowledged whatever the handler returns.
80    #[default]
81    Always,
82    /// Acknowledged when the handler succeeds; redelivered, then
83    /// dead-lettered, when it does not.
84    OnSuccess,
85}
86
87impl AckMode {
88    pub fn is_always(&self) -> bool {
89        *self == AckMode::Always
90    }
91}
92
93#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
94pub struct ConsumerSpec {
95    /// Bind, or create, exactly this durable; the platform never deletes it.
96    #[serde(default, skip_serializing_if = "Option::is_none")]
97    pub durable_name: Option<String>,
98    #[serde(default, skip_serializing_if = "Option::is_none")]
99    pub ack_wait_secs: Option<u32>,
100    /// `-1` redelivers without limit.
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub max_deliver: Option<i64>,
103    /// The delay before each redelivery, the last repeating.
104    #[serde(default, skip_serializing_if = "Vec::is_empty")]
105    pub backoff_secs: Vec<u32>,
106    #[serde(default, skip_serializing_if = "Option::is_none")]
107    pub max_ack_pending: Option<i64>,
108    #[serde(default, skip_serializing_if = "Option::is_none")]
109    pub batch: Option<u32>,
110    #[serde(default, skip_serializing_if = "Option::is_none")]
111    pub deliver_policy: Option<DeliverPolicySpec>,
112    /// Filters beside the subscription's subject, which is always one.
113    #[serde(default, skip_serializing_if = "Vec::is_empty")]
114    pub filter_subjects: Vec<String>,
115    /// Messages handled at once per replica; 1 when unset.
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub concurrency: Option<u32>,
118    /// Whether a message given up on is published to the dead-letter
119    /// stream; on when unset and `ack` is `on_success`.
120    #[serde(default, skip_serializing_if = "Option::is_none")]
121    pub dead_letter: Option<bool>,
122}
123
124impl ConsumerSpec {
125    pub const ACK_WAIT_SECS: std::ops::RangeInclusive<u32> = 5..=3600;
126    pub const MAX_DELIVER: std::ops::RangeInclusive<i64> = 1..=100;
127    pub const UNLIMITED_DELIVERIES: i64 = -1;
128    pub const MAX_BACKOFF_STEPS: usize = 10;
129    pub const BACKOFF_SECS: std::ops::RangeInclusive<u32> = 1..=86_400;
130    pub const MAX_ACK_PENDING: std::ops::RangeInclusive<i64> = 1..=100_000;
131    pub const BATCH: std::ops::RangeInclusive<u32> = 1..=256;
132    pub const CONCURRENCY: std::ops::RangeInclusive<u32> = 1..=64;
133}
134
135#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
136#[serde(rename_all = "snake_case")]
137pub enum DeliverPolicySpec {
138    All,
139    New,
140    Last,
141    LastPerSubject,
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147
148    #[test]
149    fn a_handler_type_is_named_as_a_manifest_declares_it() {
150        for handler_type in [
151            HandlerType::DeleteCascade,
152            HandlerType::UpdateField,
153            HandlerType::Webhook,
154            HandlerType::Wasm,
155        ] {
156            assert_eq!(
157                serde_json::to_value(&handler_type).expect("serialize"),
158                serde_json::json!(handler_type.as_str())
159            );
160        }
161    }
162
163    #[test]
164    fn test_subscription_stream_defaults_to_none_in_legacy_json() {
165        let json = r#"{"subject":"x","handler":"h","handler_type":"wasm"}"#;
166        let sub: SubscriptionDefinition = serde_json::from_str(json).unwrap();
167        assert!(sub.stream.is_none());
168    }
169
170    #[test]
171    fn test_subscription_stream_roundtrip_when_set() {
172        let sub = SubscriptionDefinition {
173            stream: Some("CUFFLINK_TICKS".into()),
174            ..SubscriptionDefinition::new(
175                "cufflink.tick.5min",
176                "handle_tick_resync",
177                HandlerType::Wasm,
178            )
179        };
180        let json = serde_json::to_string(&sub).unwrap();
181        assert!(json.contains("\"stream\":\"CUFFLINK_TICKS\""));
182        let parsed: SubscriptionDefinition = serde_json::from_str(&json).unwrap();
183        assert_eq!(parsed.stream.as_deref(), Some("CUFFLINK_TICKS"));
184    }
185
186    #[test]
187    fn test_subscription_stream_skipped_when_none() {
188        let sub = SubscriptionDefinition::new("x", "h", HandlerType::Wasm);
189        let json = serde_json::to_string(&sub).unwrap();
190        assert!(!json.contains("stream"));
191    }
192
193    #[test]
194    fn delivery_fields_are_omitted_at_their_defaults() {
195        let sub = SubscriptionDefinition::new("dw.{tenant}.items.created", "h", HandlerType::Wasm);
196        assert_eq!(
197            serde_json::to_value(&sub).unwrap(),
198            serde_json::json!({
199                "subject": "dw.{tenant}.items.created",
200                "handler": "h",
201                "handler_type": "wasm",
202                "config": {}
203            })
204        );
205        assert_eq!(sub.derived_features().count(), 0);
206    }
207
208    #[test]
209    fn delivery_fields_round_trip_and_demand_consumer_config() {
210        let sub = SubscriptionDefinition {
211            stream: Some("BOOKS".into()),
212            ack: AckMode::OnSuccess,
213            consumer: Some(ConsumerSpec {
214                durable_name: Some("loans-service-book-consumer".into()),
215                ack_wait_secs: Some(30),
216                max_deliver: Some(3),
217                backoff_secs: vec![10, 60],
218                max_ack_pending: Some(1000),
219                batch: Some(32),
220                deliver_policy: Some(DeliverPolicySpec::LastPerSubject),
221                filter_subjects: vec!["books.deleted.loans".into()],
222                concurrency: Some(4),
223                dead_letter: Some(false),
224            }),
225            timeout_secs: Some(120),
226            ..SubscriptionDefinition::new("books.due_date.set", "on_book", HandlerType::Wasm)
227        };
228        let json = serde_json::to_value(&sub).unwrap();
229        assert_eq!(
230            json,
231            serde_json::json!({
232                "subject": "books.due_date.set",
233                "handler": "on_book",
234                "handler_type": "wasm",
235                "config": {},
236                "stream": "BOOKS",
237                "ack": "on_success",
238                "consumer": {
239                    "durable_name": "loans-service-book-consumer",
240                    "ack_wait_secs": 30,
241                    "max_deliver": 3,
242                    "backoff_secs": [10, 60],
243                    "max_ack_pending": 1000,
244                    "batch": 32,
245                    "deliver_policy": "last_per_subject",
246                    "filter_subjects": ["books.deleted.loans"],
247                    "concurrency": 4,
248                    "dead_letter": false
249                },
250                "timeout_secs": 120
251            })
252        );
253        let parsed: SubscriptionDefinition = serde_json::from_value(json).unwrap();
254        assert_eq!(parsed.ack, AckMode::OnSuccess);
255        assert_eq!(parsed.consumer, sub.consumer);
256        assert_eq!(parsed.timeout_secs, Some(120));
257    }
258
259    #[test]
260    fn a_config_is_written_in_the_order_of_its_keys() {
261        let sub = SubscriptionDefinition {
262            config: [
263                ("where_column", "customer_id"),
264                ("table", "order_stats"),
265                ("set_value", "lines + 1"),
266                ("set_column", "lines"),
267            ]
268            .into_iter()
269            .map(|(key, value)| (key.to_string(), value.to_string()))
270            .collect(),
271            ..SubscriptionDefinition::new("x", "count_lines", HandlerType::UpdateField)
272        };
273        assert_eq!(
274            serde_json::to_string(&sub).unwrap(),
275            r#"{"subject":"x","handler":"count_lines","handler_type":"update_field","config":{"set_column":"lines","set_value":"lines + 1","table":"order_stats","where_column":"customer_id"}}"#
276        );
277    }
278
279    #[test]
280    fn a_default_subscription_is_a_wasm_handler_with_the_platforms_delivery() {
281        let sub = SubscriptionDefinition::default();
282        assert!(matches!(sub.handler_type, HandlerType::Wasm));
283        assert_eq!(sub.ack, AckMode::Always);
284        assert_eq!((&sub.consumer, sub.timeout_secs), (&None, None));
285        assert_eq!(sub.derived_features().count(), 0);
286    }
287
288    #[test]
289    fn each_delivery_field_alone_demands_consumer_config() {
290        let plain = || SubscriptionDefinition::new("x", "h", HandlerType::Wasm);
291        for sub in [
292            SubscriptionDefinition {
293                ack: AckMode::OnSuccess,
294                ..plain()
295            },
296            SubscriptionDefinition {
297                consumer: Some(ConsumerSpec::default()),
298                ..plain()
299            },
300            SubscriptionDefinition {
301                timeout_secs: Some(60),
302                ..plain()
303            },
304        ] {
305            assert_eq!(
306                sub.derived_features().collect::<Vec<_>>(),
307                [features::NATS_CONSUMER_CONFIG]
308            );
309        }
310    }
311}
cufflink_types/manifest/subscription.rs

cufflink_types/manifest/
subscription.rs
