1use std::collections::BTreeSet;
2use std::fmt;
3
4use anyhow::{Result, bail};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema, Default)]
9pub struct EventsSection {
10 #[serde(default)]
11 pub providers: Vec<EventProviderSpec>,
12}
13
14impl EventsSection {
15 pub fn validate(&self) -> Result<()> {
16 let mut seen = BTreeSet::new();
17 for provider in &self.providers {
18 provider.validate()?;
19 if !seen.insert(provider.name.clone()) {
20 bail!("duplicate events provider name: {}", provider.name);
21 }
22 }
23 Ok(())
24 }
25}
26
27#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
28pub struct EventProviderSpec {
29 pub name: String,
30 pub kind: EventProviderKind,
31 pub component: String,
32 #[serde(default, skip_serializing_if = "Option::is_none")]
33 pub default_flow: Option<String>,
34 #[serde(default, skip_serializing_if = "Option::is_none")]
35 pub custom_flow: Option<String>,
36 #[serde(default)]
37 pub capabilities: EventProviderCapabilities,
38}
39
40impl EventProviderSpec {
41 fn validate(&self) -> Result<()> {
42 if self.name.trim().is_empty() {
43 bail!("events.providers[].name is required");
44 }
45 if self.component.trim().is_empty() {
46 bail!(
47 "events.providers[{}].component must not be empty",
48 self.name
49 );
50 }
51 for topic in &self.capabilities.topics {
52 if topic.trim().is_empty() {
53 bail!(
54 "events.providers[{}].capabilities.topics may not contain empty entries",
55 self.name
56 );
57 }
58 }
59 Ok(())
60 }
61}
62
63#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema, Default)]
64pub struct EventProviderCapabilities {
65 #[serde(default, skip_serializing_if = "Option::is_none")]
66 pub transport: Option<TransportKind>,
67 #[serde(default, skip_serializing_if = "Option::is_none")]
68 pub reliability: Option<ReliabilityKind>,
69 #[serde(default, skip_serializing_if = "Option::is_none")]
70 pub ordering: Option<OrderingKind>,
71 #[serde(default, skip_serializing_if = "Vec::is_empty")]
72 pub topics: Vec<String>,
73}
74
75#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
76#[serde(rename_all = "snake_case")]
77pub enum EventProviderKind {
78 Broker,
79 Source,
80 Sink,
81 Bridge,
82}
83
84impl fmt::Display for EventProviderKind {
85 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86 let value = match self {
87 Self::Broker => "broker",
88 Self::Source => "source",
89 Self::Sink => "sink",
90 Self::Bridge => "bridge",
91 };
92 f.write_str(value)
93 }
94}
95
96#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
97#[serde(rename_all = "snake_case")]
98#[serde(untagged)]
99pub enum TransportKind {
100 Nats,
101 Kafka,
102 Sqs,
103 Webhook,
104 Email,
105 Other(String),
106}
107
108impl fmt::Display for TransportKind {
109 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
110 match self {
111 Self::Nats => f.write_str("nats"),
112 Self::Kafka => f.write_str("kafka"),
113 Self::Sqs => f.write_str("sqs"),
114 Self::Webhook => f.write_str("webhook"),
115 Self::Email => f.write_str("email"),
116 Self::Other(value) => f.write_str(value),
117 }
118 }
119}
120
121#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
122#[serde(rename_all = "snake_case")]
123pub enum ReliabilityKind {
124 AtMostOnce,
125 AtLeastOnce,
126 EffectivelyOnce,
127}
128
129impl fmt::Display for ReliabilityKind {
130 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
131 let value = match self {
132 Self::AtMostOnce => "at_most_once",
133 Self::AtLeastOnce => "at_least_once",
134 Self::EffectivelyOnce => "effectively_once",
135 };
136 f.write_str(value)
137 }
138}
139
140#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
141#[serde(rename_all = "snake_case")]
142pub enum OrderingKind {
143 None,
144 PerKey,
145 Global,
146}
147
148impl fmt::Display for OrderingKind {
149 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
150 let value = match self {
151 Self::None => "none",
152 Self::PerKey => "per_key",
153 Self::Global => "global",
154 };
155 f.write_str(value)
156 }
157}
158
159#[cfg(test)]
160mod tests {
161 use super::*;
162
163 fn valid_provider() -> EventProviderSpec {
164 EventProviderSpec {
165 name: "orders".to_string(),
166 kind: EventProviderKind::Broker,
167 component: "broker.component".to_string(),
168 default_flow: Some("flow.default".to_string()),
169 custom_flow: None,
170 capabilities: EventProviderCapabilities {
171 transport: Some(TransportKind::Kafka),
172 reliability: Some(ReliabilityKind::AtLeastOnce),
173 ordering: Some(OrderingKind::PerKey),
174 topics: vec!["orders.created".to_string()],
175 },
176 }
177 }
178
179 #[test]
180 fn validate_accepts_unique_provider_with_topics() {
181 let section = EventsSection {
182 providers: vec![valid_provider()],
183 };
184
185 section.validate().expect("valid events section");
186 }
187
188 #[test]
189 fn validate_rejects_duplicate_provider_names() {
190 let provider = valid_provider();
191 let section = EventsSection {
192 providers: vec![provider.clone(), provider],
193 };
194
195 let err = section.validate().expect_err("duplicate names should fail");
196 assert!(err.to_string().contains("duplicate events provider name"));
197 }
198
199 #[test]
200 fn validate_rejects_blank_topic_entries() {
201 let mut provider = valid_provider();
202 provider.capabilities.topics.push(" ".to_string());
203
204 let err = EventsSection {
205 providers: vec![provider],
206 }
207 .validate()
208 .expect_err("blank topics should fail");
209
210 assert!(
211 err.to_string()
212 .contains("topics may not contain empty entries")
213 );
214 }
215
216 #[test]
217 fn display_formats_enum_values() {
218 assert_eq!(EventProviderKind::Sink.to_string(), "sink");
219 assert_eq!(TransportKind::Other("sns".to_string()).to_string(), "sns");
220 assert_eq!(
221 ReliabilityKind::EffectivelyOnce.to_string(),
222 "effectively_once"
223 );
224 assert_eq!(OrderingKind::Global.to_string(), "global");
225 }
226}