teamcraft/
delivery.rs

1//! Delivery of notifications by e-mail and to Mattermost.
2//!
3//! The `notifications` table is the outbox: a notification is created with
4//! `email_state` / `chat_state` = `pending` when the user wants that channel.
5//! A background task ([`spawn`]) sends the pending ones every few seconds:
6//!
7//! * **E-mail** via SMTP ([lettre](https://lettre.rs), MIT) when
8//!   `TEAMCRAFT_SMTP_URL` and `TEAMCRAFT_MAIL_FROM` are set; plain text with a
9//!   link to the object and to the notification settings.
10//! * **Mattermost** as a direct message from a bot account through the REST API
11//!   ([`crate::mattermost`], `TEAMCRAFT_MATTERMOST_URL` + `…_BOT_TOKEN`); never a
12//!   post in a channel. An unknown or deactivated user fails at once with a
13//!   reason the user sees. Deprecated: an incoming webhook
14//!   (`TEAMCRAFT_MATTERMOST_WEBHOOK`): a post in the webhook's channel that
15//!   @-mentions the user, or a direct message (`TEAMCRAFT_MATTERMOST_DIRECT=true`).
16//!   The Mattermost user is the TeamCraft username unless the user set another one.
17//!
18//! Without configuration the pending rows are marked `skipped`. A failed
19//! delivery is retried on the next runs and marked `failed` after
20//! [`MAX_ATTEMPTS`]. One TeamCraft instance delivers (no row locking).
21//!
22//! A run serves [`PARALLEL_RECIPIENTS`] recipients at the same time; the
23//! notifications of one recipient go out one after the other, in the order
24//! they were queued, e-mail and chat of a notification side by side. A mail
25//! server that answers slowly (the SMTP timeout is 20 s) holds up the
26//! recipient it is serving, not the whole run and not the chat messages.
27
28use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc, time::Duration};
29
30use lettre::{
31    message::{header::ContentType, Mailbox},
32    AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor,
33};
34use serde_json::{json, Value};
35use sqlx::PgPool;
36use uuid::Uuid;
37
38use crate::{
39    config::Config,
40    error::{AppError, AppResult},
41    mattermost::{BotClient, ChatFailure, DirectMessenger},
42};
43
44/// Delivery attempts per channel before a notification is marked `failed`.
45pub const MAX_ATTEMPTS: i32 = 5;
46/// Pause between two runs of the background task.
47const INTERVAL: Duration = Duration::from_secs(15);
48/// Recipients served at the same time, i.e. the connections one run opens
49/// to the mail server at most. Hosted mail services limit the simultaneous
50/// connections of one sender to a handful (three is common), so this stays
51/// small: enough that one slow delivery does not hold up the others.
52pub const PARALLEL_RECIPIENTS: usize = 3;
53
54/// A notification with a pending channel, and what its delivery needs.
55#[derive(sqlx::FromRow)]
56struct Pending {
57    id: Uuid,
58    user_id: Uuid,
59    title: String,
60    body: String,
61    link: String,
62    email_state: String,
63    chat_state: String,
64    delivery_attempts: i32,
65    email: Option<String>,
66    display_name: String,
67    chat_user: String,
68    confidential: bool,
69    generic: bool,
70}
71
72pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
73
74/// Sends one e-mail (SMTP in production, a recorder in tests).
75pub trait MailSender: Send + Sync {
76    fn send(&self, message: Message) -> BoxFuture<'_, anyhow::Result<()>>;
77}
78
79struct SmtpSender(AsyncSmtpTransport<Tokio1Executor>);
80
81impl MailSender for SmtpSender {
82    fn send(&self, message: Message) -> BoxFuture<'_, anyhow::Result<()>> {
83        Box::pin(async move {
84            self.0.send(message).await?;
85            Ok(())
86        })
87    }
88}
89
90/// Posts to Mattermost (an incoming webhook in production).
91pub trait ChatSender: Send + Sync {
92    fn post(&self, payload: Value) -> BoxFuture<'_, anyhow::Result<()>>;
93}
94
95struct WebhookSender {
96    http: reqwest::Client,
97    url: String,
98}
99
100impl ChatSender for WebhookSender {
101    fn post(&self, payload: Value) -> BoxFuture<'_, anyhow::Result<()>> {
102        Box::pin(async move {
103            self.http
104                .post(&self.url)
105                .json(&payload)
106                .send()
107                .await?
108                .error_for_status()?;
109            Ok(())
110        })
111    }
112}
113
114/// The configured channels.
115#[derive(Clone)]
116pub struct Delivery {
117    public_url: String,
118    instance_name: String,
119    mail: Option<(Arc<dyn MailSender>, Mailbox)>,
120    chat: Option<Chat>,
121}
122
123/// How Mattermost is reached.
124#[derive(Clone)]
125enum Chat {
126    /// Bot account, REST API: direct messages only.
127    Bot(Arc<dyn DirectMessenger>),
128    /// Deprecated incoming webhook (`direct`: `"channel": "@user"`).
129    Webhook(Arc<dyn ChatSender>, bool),
130}
131
132/// What one run delivered.
133#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
134pub struct Report {
135    pub emails: u64,
136    pub chats: u64,
137    pub failures: u64,
138}
139
140impl Delivery {
141    pub fn from_config(config: &Config) -> anyhow::Result<Delivery> {
142        let mail = match &config.smtp {
143            Some(smtp) => {
144                let transport = AsyncSmtpTransport::<Tokio1Executor>::from_url(&smtp.url)
145                    .map_err(|e| anyhow::anyhow!("TEAMCRAFT_SMTP_URL is not usable: {e}"))?
146                    .timeout(Some(Duration::from_secs(20)))
147                    .build();
148                let from: Mailbox = smtp
149                    .from
150                    .parse()
151                    .map_err(|e| anyhow::anyhow!("TEAMCRAFT_MAIL_FROM: {e}"))?;
152                Some((Arc::new(SmtpSender(transport)) as Arc<dyn MailSender>, from))
153            }
154            None => None,
155        };
156        let chat = match &config.mattermost {
157            Some(m) => match (&m.bot, &m.webhook) {
158                (Some(bot), webhook) => {
159                    if webhook.is_some() {
160                        tracing::warn!(
161                            "TEAMCRAFT_MATTERMOST_WEBHOOK is ignored: notifications go out as direct messages of the bot"
162                        );
163                    }
164                    let client = BotClient::new(&bot.url, &bot.token)?;
165                    Some(Chat::Bot(Arc::new(client) as Arc<dyn DirectMessenger>))
166                }
167                (None, Some(webhook)) => {
168                    tracing::warn!(
169                        "TEAMCRAFT_MATTERMOST_WEBHOOK is deprecated: it posts into a channel or as the webhook's creator. Create a bot account and set TEAMCRAFT_MATTERMOST_URL and TEAMCRAFT_MATTERMOST_BOT_TOKEN (docs: admin/upgrade)"
170                    );
171                    let http = reqwest::Client::builder()
172                        .timeout(Duration::from_secs(10))
173                        .build()?;
174                    let sender = WebhookSender {
175                        http,
176                        url: webhook.clone(),
177                    };
178                    Some(Chat::Webhook(
179                        Arc::new(sender) as Arc<dyn ChatSender>,
180                        m.direct,
181                    ))
182                }
183                (None, None) => None,
184            },
185            None => None,
186        };
187        Ok(Delivery {
188            public_url: config.public_url.clone(),
189            instance_name: config.instance_name.clone(),
190            mail,
191            chat,
192        })
193    }
194
195    /// Delivery with explicit senders (tests).
196    pub fn with_senders(
197        public_url: &str,
198        instance_name: &str,
199        mail: Option<(Arc<dyn MailSender>, &str)>,
200        chat: Option<(Arc<dyn ChatSender>, bool)>,
201    ) -> anyhow::Result<Delivery> {
202        let mail = match mail {
203            Some((sender, from)) => Some((sender, from.parse()?)),
204            None => None,
205        };
206        Ok(Delivery {
207            public_url: public_url.trim_end_matches('/').to_string(),
208            instance_name: instance_name.to_string(),
209            mail,
210            chat: chat.map(|(sender, direct)| Chat::Webhook(sender, direct)),
211        })
212    }
213
214    /// Delivery whose Mattermost is a bot with a fake API (tests).
215    pub fn with_bot(
216        public_url: &str,
217        instance_name: &str,
218        bot: Arc<dyn DirectMessenger>,
219    ) -> Delivery {
220        Delivery {
221            public_url: public_url.trim_end_matches('/').to_string(),
222            instance_name: instance_name.to_string(),
223            mail: None,
224            chat: Some(Chat::Bot(bot)),
225        }
226    }
227
228    pub fn email_enabled(&self) -> bool {
229        self.mail.is_some()
230    }
231
232    pub fn chat_enabled(&self) -> bool {
233        self.chat.is_some()
234    }
235
236    /// Delivers up to 100 pending notifications.
237    pub async fn run_once(&self, pool: &PgPool) -> AppResult<Report> {
238        // Channels that are not configured: nothing will ever send them.
239        if self.mail.is_none() {
240            sqlx::query(
241                "UPDATE notifications SET email_state = 'skipped' WHERE email_state = 'pending'",
242            )
243            .execute(pool)
244            .await?;
245        }
246        if self.chat.is_none() {
247            sqlx::query(
248                "UPDATE notifications SET chat_state = 'skipped' WHERE chat_state = 'pending'",
249            )
250            .execute(pool)
251            .await?;
252        }
253        // A group restricted (or a membership lost) since the notification was
254        // queued: the recipient may no longer read it, so it is not sent.
255        // Neither to a recipient deactivated since (deactivation keeps the
256        // memberships, but the account reads nothing any more).
257        sqlx::query(&format!(
258            "UPDATE notifications n
259             SET email_state = CASE WHEN n.email_state = 'pending' THEN 'skipped' ELSE n.email_state END,
260                 chat_state = CASE WHEN n.chat_state = 'pending' THEN 'skipped' ELSE n.chat_state END
261             FROM users u
262             WHERE u.id = n.user_id AND (n.email_state = 'pending' OR n.chat_state = 'pending')
263               AND (NOT u.active OR NOT {})",
264            crate::services::notifications::notification_visible(
265                "n",
266                "u.id",
267                &crate::services::access::hidden_of_user("u")
268            )
269        ))
270        .execute(pool)
271        .await?;
272        // `confidential` is evaluated now: a group restricted after the row
273        // was queued keeps the text out of a shared channel.
274        let rows = sqlx::query_as::<_, Pending>(&format!(
275            "SELECT n.id, n.user_id, n.title, n.body, n.link, n.email_state, n.chat_state,
276                    n.delivery_attempts,
277                    u.email, u.display_name, COALESCE(u.mattermost_username, u.username) AS chat_user,
278                    {} AS confidential, {} AS generic
279             FROM notifications n JOIN users u ON u.id = n.user_id
280             WHERE (n.email_state = 'pending' OR n.chat_state = 'pending') AND u.active
281             ORDER BY n.created_at LIMIT 100",
282            crate::services::notifications::restricted_now_sql("n"),
283            crate::services::notifications::generic_text_sql("n")
284        ))
285        .fetch_all(pool)
286        .await?;
287        let name = crate::services::branding::current(pool)
288            .await?
289            .name(&self.instance_name)
290            .to_string();
291        // One line per recipient, in the order the notifications were queued:
292        // a user reads theirs in order, and the recipients are served side by
293        // side, so a slow mail server holds up one line, not all of them.
294        let mut lines: Vec<Vec<Pending>> = Vec::new();
295        let mut line_of: HashMap<Uuid, usize> = HashMap::new();
296        for row in rows {
297            let at = *line_of.entry(row.user_id).or_insert(lines.len());
298            match lines.get_mut(at) {
299                Some(line) => line.push(row),
300                None => lines.push(vec![row]),
301            }
302        }
303        let mut lines = lines.into_iter();
304        let mut tasks = tokio::task::JoinSet::new();
305        let mut report = Report::default();
306        let mut failed: Option<AppError> = None;
307        loop {
308            // after a failure (the database) no further line is started
309            while failed.is_none() && tasks.len() < PARALLEL_RECIPIENTS {
310                let Some(line) = lines.next() else { break };
311                let (delivery, pool, name) = (self.clone(), pool.clone(), name.clone());
312                tasks.spawn(async move {
313                    let mut report = Report::default();
314                    for row in line {
315                        delivery.deliver(&pool, &name, row, &mut report).await?;
316                    }
317                    Ok::<_, AppError>(report)
318                });
319            }
320            // a line that was started is finished: what it sent is recorded
321            let Some(done) = tasks.join_next().await else {
322                break;
323            };
324            match done {
325                Ok(Ok(line)) => {
326                    report.emails += line.emails;
327                    report.chats += line.chats;
328                    report.failures += line.failures;
329                }
330                Ok(Err(e)) => failed = failed.or(Some(e)),
331                Err(e) => {
332                    failed = failed.or(Some(AppError::Other(anyhow::anyhow!(
333                        "notification delivery task failed: {e}"
334                    ))));
335                }
336            }
337        }
338        match failed {
339            Some(e) => Err(e),
340            None => Ok(report),
341        }
342    }
343
344    /// Sends one notification on its pending channels – e-mail and chat at
345    /// the same time, they do not depend on each other – and records the
346    /// outcome.
347    async fn deliver(
348        &self,
349        pool: &PgPool,
350        name: &str,
351        mut row: Pending,
352        report: &mut Report,
353    ) -> AppResult<()> {
354        // a group that switched `notification_content` off: generic text
355        // and the link only, in e-mail and chat (direct messages too)
356        if row.generic {
357            (row.title, row.body) = generic_text(&self.instance_name);
358        }
359        let url = format!("{}{}", self.public_url, row.link);
360        let mail = async {
361            if row.email_state != "pending" {
362                return None;
363            }
364            Some(
365                self.send_mail(
366                    name,
367                    &row.title,
368                    &row.body,
369                    &url,
370                    row.email.as_deref(),
371                    &row.display_name,
372                )
373                .await,
374            )
375        };
376        let chat = async {
377            if row.chat_state != "pending" {
378                return None;
379            }
380            Some(
381                self.post_chat(
382                    name,
383                    &row.title,
384                    &row.body,
385                    &url,
386                    &row.chat_user,
387                    row.confidential,
388                )
389                .await,
390            )
391        };
392        let (mailed, posted) = tokio::join!(mail, chat);
393        let mut errors = Vec::new();
394        let mut email_state = row.email_state.clone();
395        match mailed {
396            Some(Ok(())) => {
397                email_state = "sent".into();
398                report.emails += 1;
399            }
400            Some(Err(e)) => errors.push(format!("e-mail: {e}")),
401            None => {}
402        }
403        let mut chat_state = row.chat_state.clone();
404        let mut permanent_chat = false;
405        match posted {
406            Some(Ok(())) => {
407                chat_state = "sent".into();
408                report.chats += 1;
409            }
410            Some(Err(e)) => {
411                // an unknown or deactivated user does not appear by retrying
412                permanent_chat = e.permanent;
413                errors.push(format!("Mattermost: {e}"));
414            }
415            None => {}
416        }
417        let attempts = row.delivery_attempts + i32::from(!errors.is_empty());
418        if !errors.is_empty() {
419            report.failures += 1;
420            tracing::warn!(notification = %row.id, attempts, errors = %errors.join("; "), "notification delivery failed");
421            if attempts >= MAX_ATTEMPTS {
422                if email_state == "pending" {
423                    email_state = "failed".into();
424                }
425                if chat_state == "pending" {
426                    chat_state = "failed".into();
427                }
428            }
429            if permanent_chat && chat_state == "pending" {
430                chat_state = "failed".into();
431            }
432        }
433        sqlx::query(
434            "UPDATE notifications SET email_state = $2, chat_state = $3, delivery_attempts = $4,
435                    delivery_error = $5
436             WHERE id = $1",
437        )
438        .bind(row.id)
439        .bind(&email_state)
440        .bind(&chat_state)
441        .bind(attempts)
442        .bind(errors.join("; "))
443        .execute(pool)
444        .await?;
445        Ok(())
446    }
447
448    async fn send_mail(
449        &self,
450        instance_name: &str,
451        title: &str,
452        body: &str,
453        url: &str,
454        to: Option<&str>,
455        display_name: &str,
456    ) -> anyhow::Result<()> {
457        let (sender, from) = self
458            .mail
459            .as_ref()
460            .ok_or_else(|| anyhow::anyhow!("SMTP is not configured"))?;
461        let address = to
462            .filter(|a| !a.is_empty())
463            .ok_or_else(|| anyhow::anyhow!("the user has no e-mail address"))?;
464        // lettre quotes and encodes the name; control characters are dropped.
465        let name = sanitize_display_name(display_name);
466        let to = Mailbox::new((!name.is_empty()).then_some(name), address.parse()?);
467        let message = Message::builder()
468            .from(from.clone())
469            .to(to)
470            .subject(format!("[{}] {title}", instance_name))
471            .header(ContentType::TEXT_PLAIN)
472            .body(email_text(
473                instance_name,
474                &self.public_url,
475                title,
476                body,
477                url,
478            ))?;
479        sender.send(message).await
480    }
481
482    async fn post_chat(
483        &self,
484        instance_name: &str,
485        title: &str,
486        body: &str,
487        url: &str,
488        user: &str,
489        confidential: bool,
490    ) -> Result<(), ChatFailure> {
491        match self.chat.as_ref() {
492            None => Err(ChatFailure::transient("Mattermost is not configured")),
493            // a direct message is read by its recipient only: full text
494            Some(Chat::Bot(bot)) => {
495                bot.send(user, &crate::mattermost::dm_text(title, body, url))
496                    .await
497            }
498            Some(Chat::Webhook(sender, direct)) => {
499                let (title, body) = chat_text(instance_name, title, body, confidential, *direct);
500                sender
501                    .post(chat_payload(
502                        instance_name,
503                        &title,
504                        &body,
505                        url,
506                        user,
507                        *direct,
508                    ))
509                    .await?;
510                Ok(())
511            }
512        }
513    }
514}
515
516/// Title and body of a notification whose group switched
517/// `notification_content` off: nothing about the object, only that there is
518/// something and (added by the sender) the link.
519pub fn generic_text(instance: &str) -> (String, String) {
520    (
521        format!("new {instance} notification"),
522        "Open the link to read it (sign-in required).".to_string(),
523    )
524}
525
526/// What a chat message may show. A notification about a restricted project
527/// group posted into a shared channel (not a direct message) shows only a
528/// generic text and the link; everybody in the channel could read it.
529pub fn chat_text(
530    instance: &str,
531    title: &str,
532    body: &str,
533    confidential: bool,
534    direct: bool,
535) -> (String, String) {
536    if confidential && !direct {
537        (format!("new {instance} notification"), String::new())
538    } else {
539        (title.to_string(), body.to_string())
540    }
541}
542
543/// Display names go into the `To:` header: no quotes, brackets or line breaks.
544fn sanitize_display_name(name: &str) -> String {
545    name.chars()
546        .filter(|c| !c.is_control() && !matches!(c, '"' | '<' | '>' | ',' | ';' | '\\'))
547        .collect::<String>()
548        .trim()
549        .to_string()
550}
551
552/// Plain-text body of a notification e-mail.
553pub fn email_text(instance: &str, public_url: &str, title: &str, body: &str, url: &str) -> String {
554    format!(
555        "{title}\n\n{body}\n\nOpen in {instance}: {url}\n\n-- \nYou receive this e-mail because of your notification \
556         settings in {instance}. Change them on your account page: {public_url}/account#notifications\n"
557    )
558}
559
560/// Mattermost webhook payload: a mention in the channel or a direct message.
561pub fn chat_payload(
562    instance: &str,
563    title: &str,
564    body: &str,
565    url: &str,
566    user: &str,
567    direct: bool,
568) -> Value {
569    let text = if direct {
570        format!("**[{title}]({url})**\n{body}")
571    } else {
572        format!("@{user} **[{title}]({url})**\n{body}")
573    };
574    let mut payload = json!({ "username": instance, "text": text });
575    if direct {
576        payload["channel"] = json!(format!("@{user}"));
577    }
578    payload
579}
580
581/// Starts the background delivery (does nothing if no channel is configured,
582/// apart from marking the pending notifications as skipped).
583pub fn spawn(pool: PgPool, delivery: Delivery) {
584    tokio::spawn(async move {
585        let mut interval = tokio::time::interval(INTERVAL);
586        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
587        loop {
588            interval.tick().await;
589            match delivery.run_once(&pool).await {
590                Ok(r) if r.emails + r.chats > 0 => {
591                    tracing::info!(
592                        emails = r.emails,
593                        mattermost = r.chats,
594                        "delivered notifications"
595                    )
596                }
597                Ok(_) => {}
598                Err(e) => tracing::warn!(error = %e, "notification delivery failed"),
599            }
600        }
601    });
602}
603
604#[cfg(test)]
605mod tests {
606    use super::*;
607
608    #[test]
609    fn email_body_links_object_and_settings() {
610        let text = email_text(
611            "TeamCraft",
612            "https://plm.example.org",
613            "Review requested: CHG-000004",
614            "Erika asks you to review PRT-000003/B.",
615            "https://plm.example.org/changes/1",
616        );
617        assert!(text.starts_with("Review requested: CHG-000004\n\nErika asks"));
618        assert!(text.contains("Open in TeamCraft: https://plm.example.org/changes/1"));
619        assert!(text.contains("https://plm.example.org/account#notifications"));
620    }
621
622    #[test]
623    fn mattermost_mentions_or_direct_messages() {
624        let p = chat_payload("TeamCraft", "T", "B", "https://x/y", "erika", false);
625        assert_eq!(p["text"], "@erika **[T](https://x/y)**\nB");
626        assert!(p.get("channel").is_none());
627        let p = chat_payload("TeamCraft", "T", "B", "https://x/y", "erika", true);
628        assert_eq!(p["channel"], "@erika");
629        assert!(!p["text"].as_str().unwrap().contains("@erika"));
630        assert_eq!(p["username"], "TeamCraft");
631    }
632
633    #[test]
634    fn confidential_notifications_are_generic_in_channels() {
635        let (t, b) = chat_text("TeamCraft", "Secret PRT", "body", true, false);
636        assert_eq!(t, "new TeamCraft notification");
637        assert!(b.is_empty());
638        let (t, b) = chat_text("TeamCraft", "Secret PRT", "body", true, true);
639        assert_eq!((t.as_str(), b.as_str()), ("Secret PRT", "body"));
640        let (t, _) = chat_text("TeamCraft", "Open PRT", "body", false, false);
641        assert_eq!(t, "Open PRT");
642    }
643
644    #[test]
645    fn display_names_are_header_safe() {
646        assert_eq!(
647            sanitize_display_name("Erika \"E\" <x>\r\nBcc: y"),
648            "Erika E xBcc: y"
649        );
650    }
651}
teamcraft/delivery.rs

teamcraft/
delivery.rs
