af_notify/
lib.rs

1//! Transport-neutral notification commands and the durable delivery worker.
2
3#![deny(missing_docs)]
4#![deny(rustdoc::broken_intra_doc_links)]
5
6use std::collections::HashMap;
7use std::sync::Arc;
8use std::time::Duration;
9
10use af_context::{NotificationAttemptId, NotificationId, RequestContext, SubjectId, TenantId};
11use async_trait::async_trait;
12use chrono::{DateTime, Utc};
13use serde::{Deserialize, Serialize};
14use tokio::time::Instant;
15use tokio_util::sync::CancellationToken;
16
17/// One transport-neutral piece of notification content.
18#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
19pub enum Block {
20    /// A heading.
21    Heading(String),
22    /// A paragraph.
23    Text(String),
24    /// Key/value fields.
25    Fields(Vec<(String, String)>),
26    /// A separator.
27    Divider,
28    /// A labelled destination.
29    Action {
30        /// Link label.
31        label: String,
32        /// Link destination.
33        url: String,
34    },
35}
36
37impl Block {
38    /// Builds a heading block.
39    pub fn heading(value: impl Into<String>) -> Self {
40        Self::Heading(value.into())
41    }
42    /// Builds a text block.
43    pub fn text(value: impl Into<String>) -> Self {
44        Self::Text(value.into())
45    }
46    /// Builds a field block.
47    pub fn fields(value: Vec<(String, String)>) -> Self {
48        Self::Fields(value)
49    }
50    /// Builds an action block.
51    pub fn action(label: impl Into<String>, url: impl Into<String>) -> Self {
52        Self::Action {
53            label: label.into(),
54            url: url.into(),
55        }
56    }
57}
58
59/// Transport-neutral notification payload.
60#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
61pub struct Notification {
62    /// Optional display title.
63    pub title: Option<String>,
64    /// Ordered content blocks.
65    pub blocks: Vec<Block>,
66}
67
68impl Notification {
69    /// Builds an empty notification.
70    pub fn new() -> Self {
71        Self::default()
72    }
73    /// Builds a localized title and text body from shared catalog keys.
74    pub fn from_template(
75        catalog: &af_i18n::I18n,
76        locale: &str,
77        title_key: &str,
78        body_key: &str,
79        args: &[(&str, &str)],
80    ) -> Self {
81        Self::new()
82            .title(catalog.t(locale, title_key, args))
83            .block(Block::text(catalog.t(locale, body_key, args)))
84    }
85    /// Sets the title.
86    pub fn title(mut self, value: impl Into<String>) -> Self {
87        self.title = Some(value.into());
88        self
89    }
90    /// Appends a content block.
91    pub fn block(mut self, value: Block) -> Self {
92        self.blocks.push(value);
93        self
94    }
95    /// Renders a conservative plain-text representation.
96    pub fn to_plain_text(&self) -> String {
97        let mut out = String::new();
98        if let Some(title) = &self.title {
99            out.push_str(title);
100            out.push_str("\n\n");
101        }
102        for block in &self.blocks {
103            match block {
104                Block::Heading(value) | Block::Text(value) => {
105                    out.push_str(value);
106                    out.push('\n');
107                }
108                Block::Fields(fields) => {
109                    for (key, value) in fields {
110                        out.push_str(&format!("{key}: {value}\n"));
111                    }
112                }
113                Block::Divider => out.push_str("---\n"),
114                Block::Action { label, url } => out.push_str(&format!("{label}: {url}\n")),
115            }
116        }
117        out.trim_end().to_owned()
118    }
119}
120
121/// Durable notification state.
122#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
123#[serde(rename_all = "snake_case")]
124pub enum NotificationStatus {
125    /// Waiting for a worker.
126    Pending,
127    /// Leased to a worker.
128    Sending,
129    /// Waiting for a retry deadline.
130    RetryScheduled,
131    /// Delivered successfully.
132    Sent,
133    /// Exhausted or permanently rejected.
134    DeadLetter,
135}
136
137impl NotificationStatus {
138    /// Stable database/wire value.
139    pub fn as_str(self) -> &'static str {
140        match self {
141            Self::Pending => "pending",
142            Self::Sending => "sending",
143            Self::RetryScheduled => "retry_scheduled",
144            Self::Sent => "sent",
145            Self::DeadLetter => "dead_letter",
146        }
147    }
148}
149
150/// Transport failure retry classification.
151#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
152#[serde(rename_all = "snake_case")]
153pub enum DeliveryClass {
154    /// A later attempt may succeed.
155    Retryable,
156    /// Repeating the same request cannot succeed.
157    Permanent,
158    /// The provider did not expose enough information.
159    Unknown,
160}
161
162/// Notification boundary failure.
163#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
164pub enum NotifyError {
165    /// No sender is registered for a channel.
166    #[error("no sender registered for channel '{0}'")]
167    UnknownChannel(String),
168    /// Retryable transport failure.
169    #[error("transport temporarily unavailable")]
170    Retryable,
171    /// Permanent transport rejection.
172    #[error("transport rejected the notification")]
173    Permanent,
174    /// Transport result cannot be classified safely.
175    #[error("transport outcome is unknown")]
176    Unknown,
177    /// Delivery was cancelled before completion.
178    #[error("notification delivery cancelled")]
179    Cancelled,
180    /// Delivery exceeded its deadline.
181    #[error("notification delivery deadline exceeded")]
182    DeadlineExceeded,
183}
184
185impl NotifyError {
186    /// Classification persisted in attempt history.
187    pub fn class(&self) -> DeliveryClass {
188        match self {
189            Self::Retryable | Self::DeadlineExceeded => DeliveryClass::Retryable,
190            Self::Permanent | Self::UnknownChannel(_) => DeliveryClass::Permanent,
191            Self::Unknown | Self::Cancelled => DeliveryClass::Unknown,
192        }
193    }
194}
195
196/// Durable repository failure.
197#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
198pub enum NotifyStoreError {
199    /// Input failed validation.
200    #[error("invalid notification request: {0}")]
201    Invalid(String),
202    /// Notification does not exist in the tenant.
203    #[error("notification not found")]
204    NotFound,
205    /// One idempotency key was reused for different content.
206    #[error("idempotency key was reused with different notification content")]
207    IdempotencyConflict,
208    /// A stale worker attempted to settle a newer lease.
209    #[error("notification lease was lost")]
210    LeaseLost,
211    /// The caller lacks dead-letter retry authority.
212    #[error("dead-letter retry is not authorized")]
213    Unauthorized,
214    /// Storage is unavailable.
215    #[error("notification store unavailable: {0}")]
216    Unavailable(String),
217}
218
219/// Cancellation and deadline supplied to one transport request.
220#[derive(Clone)]
221pub struct DeliveryContext {
222    /// Cancellation shared with the durable worker.
223    pub cancellation: CancellationToken,
224    /// Absolute request deadline.
225    pub deadline: Instant,
226}
227
228/// New durable notification command.
229#[derive(Debug, Clone, PartialEq)]
230pub struct NewOutboxItem {
231    /// Owning tenant.
232    pub tenant_id: TenantId,
233    /// Acting subject.
234    pub subject_id: SubjectId,
235    /// Caller idempotency key.
236    pub idempotency_key: String,
237    /// Registered transport channel.
238    pub channel: String,
239    /// Channel-specific recipient.
240    pub recipient: String,
241    /// Content to deliver.
242    pub notification: Notification,
243    /// Maximum delivery attempts.
244    pub max_attempts: u32,
245}
246
247/// Durable notification projection.
248#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
249pub struct OutboxItem {
250    /// Stable identity.
251    pub id: NotificationId,
252    /// Owning tenant.
253    pub tenant_id: TenantId,
254    /// Acting subject.
255    pub subject_id: SubjectId,
256    /// Registered transport channel.
257    pub channel: String,
258    /// Channel-specific recipient.
259    pub recipient: String,
260    /// Content to deliver.
261    pub notification: Notification,
262    /// Current durable state.
263    pub status: NotificationStatus,
264    /// Attempts claimed so far.
265    pub attempts: u32,
266    /// Maximum attempts before dead-lettering.
267    pub max_attempts: u32,
268    /// Current lease fence.
269    pub lease_version: i64,
270    /// Last redacted failure class, if any.
271    pub last_error: Option<String>,
272    /// Creation time.
273    pub created_at: DateTime<Utc>,
274    /// Last projection update.
275    pub updated_at: DateTime<Utc>,
276}
277
278/// One immutable delivery-attempt event.
279#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
280pub struct AttemptEvent {
281    /// Event identity.
282    pub id: NotificationAttemptId,
283    /// Parent notification.
284    pub notification_id: NotificationId,
285    /// Attempt number.
286    pub attempt: u32,
287    /// Lease fence that authored the event.
288    pub lease_version: i64,
289    /// Stable event kind.
290    pub kind: String,
291    /// Optional failure classification.
292    pub class: Option<DeliveryClass>,
293    /// Event time.
294    pub created_at: DateTime<Utc>,
295}
296
297/// Store-owned durable queue and query contract.
298#[async_trait]
299pub trait DurableOutbox: Send + Sync {
300    /// Subscribes to in-process projection changes; consumers always read the
301    /// durable row after receiving an id.
302    fn subscribe(&self) -> tokio::sync::broadcast::Receiver<NotificationId>;
303    /// Enqueues idempotently, rejecting payload mismatches.
304    async fn enqueue(&self, item: NewOutboxItem) -> Result<OutboxItem, NotifyStoreError>;
305    /// Reads one tenant-owned notification.
306    async fn get(
307        &self,
308        context: &RequestContext,
309        id: &NotificationId,
310    ) -> Result<OutboxItem, NotifyStoreError>;
311    /// Lists tenant-owned notifications after an optional cursor.
312    async fn list(
313        &self,
314        context: &RequestContext,
315        status: Option<NotificationStatus>,
316        limit: usize,
317        after: Option<&NotificationId>,
318    ) -> Result<Vec<OutboxItem>, NotifyStoreError>;
319    /// Lists immutable attempt history.
320    async fn attempts(
321        &self,
322        context: &RequestContext,
323        id: &NotificationId,
324    ) -> Result<Vec<AttemptEvent>, NotifyStoreError>;
325    /// Claims due work across tenants from a BYPASSRLS worker connection.
326    async fn claim(
327        &self,
328        worker_id: &str,
329        lease_secs: i64,
330        batch: usize,
331    ) -> Result<Vec<OutboxItem>, NotifyStoreError>;
332    /// Records successful delivery under the current lease fence.
333    async fn mark_sent(
334        &self,
335        id: &NotificationId,
336        lease_version: i64,
337    ) -> Result<(), NotifyStoreError>;
338    /// Records a classified failure, scheduling retry or dead-lettering atomically.
339    async fn record_failure(
340        &self,
341        id: &NotificationId,
342        lease_version: i64,
343        class: DeliveryClass,
344        delay: Duration,
345    ) -> Result<NotificationStatus, NotifyStoreError>;
346    /// Requeues a tenant-owned dead letter after API authorization.
347    async fn retry_dead_letter(
348        &self,
349        context: &RequestContext,
350        id: &NotificationId,
351    ) -> Result<OutboxItem, NotifyStoreError>;
352}
353
354/// A single-attempt transport. Durable retry belongs to [`Dispatcher`].
355#[async_trait]
356pub trait Sender: Send + Sync {
357    /// Registered channel name.
358    fn name(&self) -> &str;
359    /// Performs exactly one bounded delivery attempt.
360    async fn send(
361        &self,
362        context: &DeliveryContext,
363        recipient: &str,
364        notification: &Notification,
365    ) -> Result<(), NotifyError>;
366}
367
368/// Bounded retry policy with deterministic jitter.
369#[derive(Debug, Clone, Copy)]
370pub struct RetryPolicy {
371    /// Initial delay.
372    pub base: Duration,
373    /// Maximum delay.
374    pub cap: Duration,
375}
376
377/// One durable worker's bounded execution settings.
378#[derive(Debug, Clone)]
379pub struct WorkerConfig {
380    /// Stable worker identity used in leases.
381    pub worker_id: String,
382    /// Lease duration in seconds.
383    pub lease_secs: i64,
384    /// Maximum records claimed in one pass.
385    pub batch: usize,
386    /// Deadline for each single transport attempt.
387    pub request_timeout: Duration,
388    /// Durable retry delay policy.
389    pub retry: RetryPolicy,
390}
391
392impl RetryPolicy {
393    /// Computes capped exponential backoff with stable id-derived jitter.
394    pub fn delay(self, id: &NotificationId, attempt: u32) -> Duration {
395        let exponent = attempt.saturating_sub(1).min(20);
396        let raw = self
397            .base
398            .as_millis()
399            .saturating_mul(1u128 << exponent)
400            .min(self.cap.as_millis());
401        let hash = id.as_bytes().iter().fold(0u64, |value, byte| {
402            value.wrapping_mul(31).wrapping_add(u64::from(*byte))
403        });
404        let jitter = 90 + hash % 21;
405        Duration::from_millis(
406            u64::try_from(raw.saturating_mul(u128::from(jitter)) / 100).unwrap_or(u64::MAX),
407        )
408    }
409}
410
411/// Routes claimed durable notifications to registered transports.
412#[derive(Default, Clone)]
413pub struct Dispatcher {
414    senders: HashMap<String, Arc<dyn Sender>>,
415}
416
417impl Dispatcher {
418    /// Builds an empty dispatcher.
419    pub fn new() -> Self {
420        Self::default()
421    }
422    /// Registers or replaces one channel sender.
423    pub fn register(&mut self, sender: Arc<dyn Sender>) -> &mut Self {
424        self.senders.insert(sender.name().to_owned(), sender);
425        self
426    }
427    /// Returns registered channel names.
428    pub fn channels(&self) -> impl Iterator<Item = &str> {
429        self.senders.keys().map(String::as_str)
430    }
431    /// Performs one transport attempt.
432    pub async fn dispatch(
433        &self,
434        context: &DeliveryContext,
435        channel: &str,
436        recipient: &str,
437        notification: &Notification,
438    ) -> Result<(), NotifyError> {
439        let sender = self
440            .senders
441            .get(channel)
442            .ok_or_else(|| NotifyError::UnknownChannel(channel.to_owned()))?;
443        sender.send(context, recipient, notification).await
444    }
445    /// Claims and settles one bounded worker batch.
446    pub async fn drain(
447        &self,
448        outbox: &dyn DurableOutbox,
449        config: &WorkerConfig,
450        cancellation: &CancellationToken,
451    ) -> Result<usize, NotifyStoreError> {
452        let items = outbox
453            .claim(
454                &config.worker_id,
455                config.lease_secs.max(
456                    i64::try_from(
457                        config
458                            .request_timeout
459                            .as_secs()
460                            .saturating_mul(config.batch.clamp(1, 100) as u64)
461                            .saturating_add(1),
462                    )
463                    .unwrap_or(i64::MAX),
464                ),
465                config.batch.clamp(1, 100),
466            )
467            .await?;
468        for item in &items {
469            if cancellation.is_cancelled() {
470                break;
471            }
472            let context = DeliveryContext {
473                cancellation: cancellation.child_token(),
474                deadline: Instant::now() + config.request_timeout,
475            };
476            match self
477                .dispatch(&context, &item.channel, &item.recipient, &item.notification)
478                .await
479            {
480                Ok(()) => outbox.mark_sent(&item.id, item.lease_version).await?,
481                Err(error) => {
482                    outbox
483                        .record_failure(
484                            &item.id,
485                            item.lease_version,
486                            error.class(),
487                            config.retry.delay(&item.id, item.attempts),
488                        )
489                        .await?;
490                }
491            }
492        }
493        Ok(items.len())
494    }
495}
496
497/// Development sender that logs rendered content.
498pub struct LogSender;
499
500#[async_trait]
501impl Sender for LogSender {
502    fn name(&self) -> &str {
503        "log"
504    }
505    async fn send(
506        &self,
507        _context: &DeliveryContext,
508        recipient: &str,
509        notification: &Notification,
510    ) -> Result<(), NotifyError> {
511        tracing::info!(target: "notify", recipient, body = %notification.to_plain_text(), "notification");
512        Ok(())
513    }
514}
515
516/// In-memory fakes for host and failure-injection tests.
517pub mod testing;
518
519#[cfg(test)]
520mod tests {
521    use super::*;
522
523    #[test]
524    fn rendering_and_backoff_are_stable() {
525        let notification = Notification::new()
526            .title("Completed")
527            .block(Block::fields(vec![("duration".into(), "1s".into())]));
528        assert_eq!(notification.to_plain_text(), "Completed\n\nduration: 1s");
529        let id = NotificationId::parse("n-1").unwrap();
530        let policy = RetryPolicy {
531            base: Duration::from_secs(1),
532            cap: Duration::from_secs(60),
533        };
534        assert_eq!(policy.delay(&id, 2), policy.delay(&id, 2));
535        assert!(policy.delay(&id, 3) > policy.delay(&id, 2));
536        let mut catalog = af_i18n::I18n::new("en").unwrap();
537        catalog
538            .load_locale(
539                "en",
540                serde_json::json!({"title":"Done", "body":"Hello {name}"}),
541            )
542            .unwrap();
543        assert_eq!(
544            Notification::from_template(&catalog, "en", "title", "body", &[("name", "Ada")])
545                .to_plain_text(),
546            "Done\n\nHello Ada"
547        );
548    }
549}
af_notify/lib.rs

af_notify/
lib.rs
