systemprompt_models/notifications/
declarations.rs

1//! Declarative notification config: the YAML shapes under
2//! `services/notifications/` and their cross-reference validation.
3//!
4//! Copyright (c) systemprompt.io — Business Source License 1.1.
5//! See <https://systemprompt.io> for licensing details.
6
7use std::collections::HashSet;
8
9use chrono::{NaiveTime, Weekday};
10use serde::{Deserialize, Serialize};
11use systemprompt_identifiers::{NotificationRuleId, NotificationStreamId};
12
13use super::attention::Attention;
14use super::subject::{AudienceSubject, SubjectRef};
15
16/// A declared stream: who reads it, who may publish, and its attention band.
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(deny_unknown_fields)]
19pub struct StreamDecl {
20    pub id: NotificationStreamId,
21    pub name: String,
22    #[serde(default)]
23    pub description: Option<String>,
24    pub audience: Vec<AudienceSubject>,
25    pub publishers: Vec<AudienceSubject>,
26    #[serde(default)]
27    pub attention: StreamAttention,
28    #[serde(default)]
29    pub retention_days: Option<u32>,
30    #[serde(default)]
31    pub coalesce_window_secs: Option<u32>,
32}
33
34/// A stream's default attention and the band every recipient is clamped to.
35#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(deny_unknown_fields, default)]
37pub struct StreamAttention {
38    pub default: Attention,
39    pub floor: Option<Attention>,
40    pub ceiling: Option<Attention>,
41}
42
43/// A declared subscription of an audience to a stream or a resource.
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45#[serde(deny_unknown_fields)]
46pub struct SubscriptionDecl {
47    pub subject: AudienceSubject,
48    pub target: SubscriptionTarget,
49    #[serde(default)]
50    pub attention_ceiling: Option<Attention>,
51    #[serde(default)]
52    pub quiet_hours: Option<QuietHours>,
53    #[serde(default)]
54    pub digest: Option<DigestWindow>,
55}
56
57/// What a subscription follows: `{stream: <id>}` or `{resource: {kind, id}}`.
58///
59/// It reads and writes as a one-key map rather than a tagged enum, so YAML
60/// declarations use the plain map form.
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(try_from = "TargetMap", into = "TargetMap")]
63pub enum SubscriptionTarget {
64    Stream(NotificationStreamId),
65    Resource(SubjectRef),
66}
67
68#[derive(Debug, Clone, Serialize, Deserialize)]
69#[serde(deny_unknown_fields)]
70struct TargetMap {
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    stream: Option<NotificationStreamId>,
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    resource: Option<SubjectRef>,
75}
76
77#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
78#[error("a subscription target names exactly one of `stream` or `resource`")]
79pub struct SubscriptionTargetError;
80
81impl TryFrom<TargetMap> for SubscriptionTarget {
82    type Error = SubscriptionTargetError;
83
84    fn try_from(map: TargetMap) -> Result<Self, Self::Error> {
85        match (map.stream, map.resource) {
86            (Some(stream), None) => Ok(Self::Stream(stream)),
87            (None, Some(resource)) => Ok(Self::Resource(resource)),
88            _ => Err(SubscriptionTargetError),
89        }
90    }
91}
92
93impl From<SubscriptionTarget> for TargetMap {
94    fn from(target: SubscriptionTarget) -> Self {
95        match target {
96            SubscriptionTarget::Stream(stream) => Self {
97                stream: Some(stream),
98                resource: None,
99            },
100            SubscriptionTarget::Resource(resource) => Self {
101                stream: None,
102                resource: Some(resource),
103            },
104        }
105    }
106}
107
108/// A daily window, in a named time zone, during which delivery is held.
109#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
110#[serde(deny_unknown_fields)]
111pub struct QuietHours {
112    pub start: NaiveTime,
113    pub end: NaiveTime,
114    pub tz: String,
115    #[serde(default)]
116    pub weekdays: Option<Vec<Weekday>>,
117}
118
119/// Batch delivery into one digest per window.
120#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
121#[serde(deny_unknown_fields)]
122pub struct DigestWindow {
123    pub every_secs: u32,
124}
125
126/// A declared rule: when its match holds, its action adjusts delivery.
127#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
128#[serde(deny_unknown_fields)]
129pub struct RuleDecl {
130    pub id: NotificationRuleId,
131    pub name: String,
132    #[serde(default)]
133    pub priority: i32,
134    #[serde(default)]
135    pub r#match: RuleMatch,
136    #[serde(default)]
137    pub action: RuleAction,
138}
139
140/// Rule predicates; an empty or absent field matches everything.
141#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
142#[serde(deny_unknown_fields, default)]
143pub struct RuleMatch {
144    pub kinds: Vec<String>,
145    pub streams: Vec<NotificationStreamId>,
146    pub resources: Vec<SubjectRef>,
147    pub actors: Vec<AudienceSubject>,
148    pub actor_is_agent: Option<bool>,
149    pub recipients: Vec<AudienceSubject>,
150}
151
152/// What a matching rule does to a recipient's delivery.
153#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(deny_unknown_fields, default)]
155pub struct RuleAction {
156    pub set: Option<Attention>,
157    pub cap: Option<Attention>,
158    pub floor: Option<Attention>,
159    pub mute: bool,
160    pub coalesce_window_secs: Option<u32>,
161    pub digest: Option<DigestWindow>,
162    pub quiet_hours: Option<QuietHours>,
163}
164
165/// One notifications YAML file, and the merge of all of them.
166#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
167#[serde(deny_unknown_fields, default)]
168pub struct NotificationsDoc {
169    pub streams: Vec<StreamDecl>,
170    pub rules: Vec<RuleDecl>,
171    pub subscriptions: Vec<SubscriptionDecl>,
172}
173
174impl NotificationsDoc {
175    pub fn merge(&mut self, other: Self) {
176        self.streams.extend(other.streams);
177        self.rules.extend(other.rules);
178        self.subscriptions.extend(other.subscriptions);
179    }
180}
181
182/// A declaration set that is internally inconsistent.
183#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
184pub enum DeclError {
185    #[error("stream {0} is declared more than once")]
186    DuplicateStream(NotificationStreamId),
187    #[error("rule {0} is declared more than once")]
188    DuplicateRule(NotificationRuleId),
189    #[error("stream {0} has an attention floor above its ceiling")]
190    StreamFloorAboveCeiling(NotificationStreamId),
191    #[error("rule {0} has an attention floor above its cap")]
192    RuleFloorAboveCap(NotificationRuleId),
193    #[error("stream {0} declares no audience")]
194    EmptyAudience(NotificationStreamId),
195    #[error("stream {0} declares no publishers")]
196    EmptyPublishers(NotificationStreamId),
197    #[error("rule {rule} references undeclared stream {stream}")]
198    UnknownRuleStream {
199        rule: NotificationRuleId,
200        stream: NotificationStreamId,
201    },
202    #[error("subscription references undeclared stream {0}")]
203    UnknownSubscriptionStream(NotificationStreamId),
204}
205
206pub fn validate(doc: &NotificationsDoc) -> Result<(), DeclError> {
207    let mut streams = HashSet::new();
208    for stream in &doc.streams {
209        if !streams.insert(&stream.id) {
210            return Err(DeclError::DuplicateStream(stream.id.clone()));
211        }
212        if let (Some(floor), Some(ceiling)) = (stream.attention.floor, stream.attention.ceiling)
213            && floor > ceiling
214        {
215            return Err(DeclError::StreamFloorAboveCeiling(stream.id.clone()));
216        }
217        if stream.audience.is_empty() {
218            return Err(DeclError::EmptyAudience(stream.id.clone()));
219        }
220        if stream.publishers.is_empty() {
221            return Err(DeclError::EmptyPublishers(stream.id.clone()));
222        }
223    }
224    let mut rules = HashSet::new();
225    for rule in &doc.rules {
226        if !rules.insert(&rule.id) {
227            return Err(DeclError::DuplicateRule(rule.id.clone()));
228        }
229        if let (Some(floor), Some(cap)) = (rule.action.floor, rule.action.cap)
230            && floor > cap
231        {
232            return Err(DeclError::RuleFloorAboveCap(rule.id.clone()));
233        }
234        if let Some(stream) = rule.r#match.streams.iter().find(|s| !streams.contains(s)) {
235            return Err(DeclError::UnknownRuleStream {
236                rule: rule.id.clone(),
237                stream: stream.clone(),
238            });
239        }
240    }
241    for subscription in &doc.subscriptions {
242        if let SubscriptionTarget::Stream(stream) = &subscription.target
243            && !streams.contains(stream)
244        {
245            return Err(DeclError::UnknownSubscriptionStream(stream.clone()));
246        }
247    }
248    Ok(())
249}
systemprompt_models/notifications/declarations.rs

systemprompt_models/notifications/
declarations.rs
