teamcraft/services/
request_digests.rs

1//! E-mail digest of the change requests that wait for triage, per project
2//! group.
3//!
4//! Every CAD editor (and admin) of a project group gets – besides the
5//! immediate notification of every new request – a summary every
6//! `TEAMCRAFT_REQUEST_DIGEST_HOURS` (default: daily) of the requests of the
7//! group that are still *new*: the ones raised since the previous digest,
8//! and how many older ones still wait. No digest when nothing new came in.
9//!
10//! A digest is a notification of the kind `request_digest` (muted on the
11//! account page like every other kind), queued for e-mail only and stored as
12//! read, so it does not count in the bell: [`crate::delivery`] sends it with
13//! its retries and checks again at sending time that the user may still read
14//! the group and every item the text names (`mentioned_projects`,
15//! `mentioned_items`). The table `request_digests` remembers per user and
16//! group up to when the digests covered the requests; it is updated in the
17//! transaction that queues the digest, so a restart or a second server never
18//! sends a period twice.
19
20use std::time::Duration;
21
22use chrono::{DateTime, Utc};
23use sqlx::{PgPool, Postgres, Transaction};
24use uuid::Uuid;
25
26use crate::{domain, error::AppResult, services::access};
27
28/// How often the background task looks for due digests.
29const TICK: Duration = Duration::from_secs(600);
30/// Request lines listed in one digest (the rest is counted).
31const MAX_LINES: usize = 50;
32/// Numbers of older waiting requests listed (the rest is counted).
33const MAX_OLDER: usize = 20;
34
35/// A request in a digest.
36#[derive(Clone, Debug, sqlx::FromRow)]
37pub struct DigestLine {
38    pub id: Uuid,
39    pub number: String,
40    pub title: String,
41    pub priority: String,
42    pub requested_by_name: String,
43    pub created_at: DateTime<Utc>,
44    /// Raised since the previous digest.
45    pub fresh: bool,
46}
47
48/// Title and body of the digest of `project` (`None` = nothing new).
49pub fn digest_text(project: &str, lines: &[DigestLine]) -> Option<(String, String)> {
50    let fresh: Vec<&DigestLine> = lines.iter().filter(|l| l.fresh).collect();
51    if fresh.is_empty() {
52        return None;
53    }
54    let older: Vec<&DigestLine> = lines.iter().filter(|l| !l.fresh).collect();
55    let plural = |n: usize| if n == 1 { "" } else { "s" };
56    let title = format!(
57        "{} new change request{} in {project} wait{} for triage",
58        fresh.len(),
59        plural(fresh.len()),
60        if fresh.len() == 1 { "s" } else { "" }
61    );
62    let mut body =
63        format!("Raised in project group {project} since the last digest and not triaged yet:\n\n");
64    for l in fresh.iter().take(MAX_LINES) {
65        let priority = if l.priority == "normal" {
66            String::new()
67        } else {
68            format!("{}, ", l.priority)
69        };
70        body.push_str(&format!(
71            "* {} {} ({priority}raised by {} on {})\n",
72            l.number,
73            l.title,
74            l.requested_by_name,
75            l.created_at.format("%Y-%m-%d")
76        ));
77    }
78    if fresh.len() > MAX_LINES {
79        body.push_str(&format!("* … and {} more\n", fresh.len() - MAX_LINES));
80    }
81    if !older.is_empty() {
82        let numbers: Vec<&str> = older
83            .iter()
84            .take(MAX_OLDER)
85            .map(|l| l.number.as_str())
86            .collect();
87        let more = if older.len() > MAX_OLDER {
88            format!(" and {} more", older.len() - MAX_OLDER)
89        } else {
90            String::new()
91        };
92        body.push_str(&format!(
93            "\n{} older request{} still wait{} for triage: {}{more}.\n",
94            older.len(),
95            plural(older.len()),
96            if older.len() == 1 { "s" } else { "" },
97            numbers.join(", ")
98        ));
99    }
100    body.push_str(
101        "\nTriage them in the change request list: accept, reject or defer them, \
102         or start the triage so they leave this digest.",
103    );
104    Some((title, body))
105}
106
107/// A user and a project group whose digest is due.
108#[derive(sqlx::FromRow)]
109struct Due {
110    user_id: Uuid,
111    project_id: Uuid,
112    project_code: String,
113}
114
115/// Queues the digests that are due (every `period_hours`; `0` = never).
116/// Returns how many were queued.
117///
118/// # Invariants
119///
120/// * **Permission:** none; a background task, not an actor. Only active
121///   editors and administrators with e-mail on, the kind not muted, who are
122///   members of an active group with new requests.
123/// * **Locks:** the digest row of the user and group (`FOR UPDATE`), so an
124///   overlapping run (or another server) finds the period covered.
125/// * **Data:** the digest lists only requests the user may read; the covered
126///   period moves on even when no digest text results.
127/// * **Audit:** none.
128/// * **Notifications:** one `request_digest` notification per due user and
129///   group, e-mail queued; **webhooks:** none.
130/// * **Atomic:** one transaction per user and group
131///   (`tests/request_digests.rs`).
132pub async fn run_once(pool: &PgPool, period_hours: u64) -> AppResult<u64> {
133    if period_hours == 0 {
134        return Ok(0);
135    }
136    let period = format!("{period_hours} hours");
137    // CAD editors of active groups with e-mail on and the kind not muted,
138    // whose last digest of the group is a period ago, in groups with
139    // requests waiting for triage. Members of a restricted group see it.
140    let due = sqlx::query_as::<_, Due>(
141        "SELECT u.id AS user_id, p.id AS project_id, p.code AS project_code
142         FROM users u
143         JOIN project_members pm ON pm.user_id = u.id
144         JOIN projects p ON p.id = pm.project_id
145         LEFT JOIN request_digests d ON d.user_id = u.id AND d.project_id = p.id
146         WHERE u.active AND u.role IN ('editor', 'admin') AND p.active
147           AND u.notify_email AND COALESCE(u.email, '') <> ''
148           AND NOT ('request_digest' = ANY(u.notify_muted))
149           AND (d.covered_until IS NULL OR d.covered_until <= now() - $1::interval)
150           AND EXISTS (SELECT 1 FROM change_requests r
151                       JOIN change_request_items cri ON cri.request_id = r.id
152                       JOIN items i ON i.id = cri.item_id
153                       WHERE i.project_id = p.id AND r.status = 'new')
154         ORDER BY u.id, p.code",
155    )
156    .bind(&period)
157    .fetch_all(pool)
158    .await?;
159    let mut queued = 0;
160    for d in due {
161        let mut tx = pool.begin().await?;
162        if queue_one(&mut tx, &d, &period).await? {
163            queued += 1;
164        }
165        tx.commit().await?;
166    }
167    Ok(queued)
168}
169
170/// Queues the digest of one user and group if it is (still) due, and moves
171/// the covered period on.
172async fn queue_one(tx: &mut Transaction<'_, Postgres>, d: &Due, period: &str) -> AppResult<bool> {
173    // The first digest covers the last period.
174    sqlx::query(
175        "INSERT INTO request_digests (user_id, project_id, covered_until)
176         VALUES ($1, $2, now() - $3::interval) ON CONFLICT DO NOTHING",
177    )
178    .bind(d.user_id)
179    .bind(d.project_id)
180    .bind(period)
181    .execute(&mut **tx)
182    .await?;
183    // Locked: another server (or an overlapping run) waits here and then
184    // finds the period covered.
185    let (since, due): (DateTime<Utc>, bool) = sqlx::query_as(
186        "SELECT covered_until, covered_until <= now() - $3::interval
187         FROM request_digests WHERE user_id = $1 AND project_id = $2 FOR UPDATE",
188    )
189    .bind(d.user_id)
190    .bind(d.project_id)
191    .bind(period)
192    .fetch_one(&mut **tx)
193    .await?;
194    if !due {
195        return Ok(false);
196    }
197    // The requests of the group the user may read (a request that also
198    // touches a group hidden from the user stays out).
199    let lines = sqlx::query_as::<_, DigestLine>(&format!(
200        "SELECT DISTINCT r.id, r.number, r.title, r.priority,
201                COALESCE(ru.display_name, '') AS requested_by_name, r.created_at,
202                r.created_at > $3 AS fresh
203         FROM change_requests r
204         JOIN change_request_items cri ON cri.request_id = r.id
205         JOIN items i ON i.id = cri.item_id
206         JOIN users u ON u.id = $1
207         LEFT JOIN users ru ON ru.id = r.requested_by
208         WHERE i.project_id = $2 AND r.status = 'new' AND {}
209         ORDER BY r.created_at, r.number",
210        access::request_visible_for("r", "u.id", &access::hidden_of_user("u"))
211    ))
212    .bind(d.user_id)
213    .bind(d.project_id)
214    .bind(since)
215    .fetch_all(&mut **tx)
216    .await?;
217    let queued = match digest_text(&d.project_code, &lines) {
218        None => false,
219        Some((title, body)) => {
220            let ids: Vec<Uuid> = lines.iter().map(|l| l.id).collect();
221            sqlx::query(
222                "INSERT INTO notifications (id, user_id, kind, title, body, link, actor_name,
223                                            read_at, email_state, chat_state, confidential,
224                                            mentioned_items, mentioned_projects)
225                 SELECT gen_random_uuid(), $1, 'request_digest', $3, $4, '/requests?status=new',
226                        'TeamCraft', now(), 'pending', 'none',
227                        EXISTS (SELECT 1 FROM projects WHERE id = $2 AND visibility = 'restricted'),
228                        ARRAY(SELECT DISTINCT item_id FROM change_request_items
229                              WHERE request_id = ANY($5)),
230                        ARRAY[$2]::uuid[]",
231            )
232            .bind(d.user_id)
233            .bind(d.project_id)
234            .bind(domain::truncate_chars(&title, 300))
235            .bind(domain::truncate_chars(&body, 4000))
236            .bind(&ids)
237            .execute(&mut **tx)
238            .await?;
239            true
240        }
241    };
242    sqlx::query(
243        "UPDATE request_digests SET covered_until = now() WHERE user_id = $1 AND project_id = $2",
244    )
245    .bind(d.user_id)
246    .bind(d.project_id)
247    .execute(&mut **tx)
248    .await?;
249    Ok(queued)
250}
251
252/// Starts the background task (nothing when `period_hours` is `0`).
253pub fn spawn(pool: PgPool, period_hours: u64) {
254    if period_hours == 0 {
255        return;
256    }
257    tokio::spawn(async move {
258        let mut interval = tokio::time::interval(TICK);
259        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
260        loop {
261            interval.tick().await;
262            match run_once(&pool, period_hours).await {
263                Ok(n) if n > 0 => tracing::info!(digests = n, "queued change request digests"),
264                Ok(_) => {}
265                Err(e) => tracing::warn!(error = %e, "change request digests failed"),
266            }
267        }
268    });
269}
270
271#[cfg(test)]
272mod tests {
273    use super::*;
274
275    fn line(number: &str, fresh: bool, priority: &str) -> DigestLine {
276        DigestLine {
277            id: Uuid::new_v4(),
278            number: number.into(),
279            title: format!("Title of {number}"),
280            priority: priority.into(),
281            requested_by_name: "Erika Muster".into(),
282            created_at: DateTime::parse_from_rfc3339("2026-11-05T08:00:00Z")
283                .unwrap()
284                .with_timezone(&Utc),
285            fresh,
286        }
287    }
288
289    #[test]
290    fn nothing_new_no_digest() {
291        assert!(digest_text("ROVER", &[]).is_none());
292        assert!(digest_text("ROVER", &[line("CR-000001", false, "normal")]).is_none());
293    }
294
295    #[test]
296    fn new_and_older_requests() {
297        let (title, body) = digest_text(
298            "ROVER",
299            &[
300                line("CR-000001", false, "normal"),
301                line("CR-000004", true, "high"),
302                line("CR-000005", true, "normal"),
303            ],
304        )
305        .unwrap();
306        assert_eq!(title, "2 new change requests in ROVER wait for triage");
307        assert!(body.contains(
308            "* CR-000004 Title of CR-000004 (high, raised by Erika Muster on 2026-11-05)"
309        ));
310        assert!(
311            body.contains("* CR-000005 Title of CR-000005 (raised by Erika Muster on 2026-11-05)")
312        );
313        assert!(body.contains("1 older request still waits for triage: CR-000001."));
314        let (title, _) = digest_text("ROVER", &[line("CR-000009", true, "normal")]).unwrap();
315        assert_eq!(title, "1 new change request in ROVER waits for triage");
316    }
317
318    #[test]
319    fn long_digests_are_cut() {
320        let mut lines: Vec<DigestLine> = (0..60)
321            .map(|i| line(&format!("CR-{i:06}"), true, "normal"))
322            .collect();
323        lines.extend((0..25).map(|i| line(&format!("CR-9{i:05}"), false, "normal")));
324        let (_, body) = digest_text("ROVER", &lines).unwrap();
325        assert!(body.contains("* … and 10 more"));
326        assert!(body.contains("25 older requests still wait for triage:"));
327        assert!(body.contains(" and 5 more."));
328        assert!(!body.contains("CR-000055"));
329    }
330}
teamcraft/services/request_digests.rs

teamcraft/services/
request_digests.rs
