Menu
Public repository / 8d41f1e10aceed387d425b43038f2070e399a349
AkurAI-Build / src/notifications.rs
Latest change b3f9a218e1ff51ec87a0dd7c07ff6b98fdb47905 - OLI-275: notifications feed, page and mail for reviews, merges and failed runs by Ólafur Búi Ólafsson
//! Notifications (OLI-275): a per-user feed (review requests, submitted
//! reviews, merged pull requests, failed runs) that is also mailed once through
//! a command (`AKURAI_BUILD_MAIL_CMD`), like the chat's handover mail.
use anyhow::{Context, Result, bail, ensure};
use rusqlite::params;
use serde::Serialize;
use std::process::{Command, Stdio};
use crate::access;
use crate::db::Database;
pub const KINDS: [&str; 4] = [
"review_requested",
"review_submitted",
"run_failed",
"pull_merged",
];
const MAX_FEED: usize = 200;
#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
pub struct Notification {
pub id: i64,
pub subject: String,
pub kind: String,
pub repository: String,
pub title: String,
pub body: String,
pub link: String,
pub created_at: i64,
pub read_at: Option<i64>,
pub mailed_at: Option<i64>,
}
pub struct New<'a> {
pub kind: &'a str,
pub repository: &'a str,
pub title: &'a str,
pub body: &'a str,
pub link: &'a str,
}
/// Notify one subject (an email); returns the id, or `None` for an empty subject.
pub fn notify(database: &Database, subject: &str, new: &New<'_>) -> Result<Option<i64>> {
let subject = subject.trim().to_lowercase();
if subject.is_empty() {
return Ok(None);
}
ensure!(
KINDS.contains(&new.kind),
"unknown notification kind {:?}",
new.kind
);
let title: String = new.title.chars().take(300).collect();
let connection = database.connection()?;
connection.execute(
"INSERT INTO notifications (subject, kind, repository, title, body, link) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![subject, new.kind, new.repository, title, new.body, new.link],
)?;
Ok(Some(connection.last_insert_rowid()))
}
/// Notify several subjects, each once.
pub fn notify_all(database: &Database, subjects: &[String], new: &New<'_>) -> Result<usize> {
let mut seen: Vec<String> = Vec::new();
let mut count = 0;
for subject in subjects {
let key = subject.trim().to_lowercase();
if key.is_empty() || seen.contains(&key) {
continue;
}
seen.push(key.clone());
if notify(database, &key, new)?.is_some() {
count += 1;
}
}
Ok(count)
}
/// Who hears about a repository's failed runs: collaborators with write or
/// admin, plus the forge admins.
pub fn run_watchers(
database: &Database,
repository: &str,
admins: &[String],
) -> Result<Vec<String>> {
let mut subjects: Vec<String> = access::collaborators(database, repository)?
.into_iter()
.filter(|c| c.role.allows(access::Role::Write))
.map(|c| c.subject)
.collect();
subjects.extend(admins.iter().cloned());
Ok(subjects)
}
fn map_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Notification> {
Ok(Notification {
id: row.get(0)?,
subject: row.get(1)?,
kind: row.get(2)?,
repository: row.get(3)?,
title: row.get(4)?,
body: row.get(5)?,
link: row.get(6)?,
created_at: row.get(7)?,
read_at: row.get(8)?,
mailed_at: row.get(9)?,
})
}
const SELECT: &str = "SELECT id, subject, kind, repository, title, body, link, created_at, read_at, mailed_at FROM notifications";
/// The subject's feed, newest first; `unread_only` hides what was read.
pub fn feed(database: &Database, subject: &str, unread_only: bool) -> Result<Vec<Notification>> {
let connection = database.connection()?;
let mut statement = connection.prepare(&format!(
"{SELECT} WHERE subject = ?1 AND (?2 = 0 OR read_at IS NULL) ORDER BY id DESC LIMIT ?3"
))?;
let rows = statement.query_map(
params![
subject.trim().to_lowercase(),
unread_only as i64,
MAX_FEED as i64
],
map_row,
)?;
Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
}
pub fn unread_count(database: &Database, subject: &str) -> Result<i64> {
Ok(database.connection()?.query_row(
"SELECT COUNT(*) FROM notifications WHERE subject = ?1 AND read_at IS NULL",
[subject.trim().to_lowercase()],
|row| row.get(0),
)?)
}
/// Mark one of the subject's notifications read; `false` when it is not theirs.
pub fn mark_read(database: &Database, subject: &str, id: i64, now: i64) -> Result<bool> {
Ok(database.connection()?.execute(
"UPDATE notifications SET read_at = COALESCE(read_at, ?3) WHERE id = ?1 AND subject = ?2",
params![id, subject.trim().to_lowercase(), now],
)? > 0)
}
pub fn mark_all_read(database: &Database, subject: &str, now: i64) -> Result<usize> {
Ok(database.connection()?.execute(
"UPDATE notifications SET read_at = ?2 WHERE subject = ?1 AND read_at IS NULL",
params![subject.trim().to_lowercase(), now],
)?)
}
/// How mail goes out: a command reading the message on stdin (the real one)
/// or a recorder (tests).
pub trait Mailer: Send + Sync {
fn send(&self, to: &str, subject: &str, body: &str) -> Result<()>;
}
/// Runs `AKURAI_BUILD_MAIL_CMD` with the RFC 822 message on stdin and
/// `AKURAI_BUILD_MAIL_FROM` as the sender.
pub struct CommandMailer {
from: String,
program: String,
args: Vec<String>,
}
impl CommandMailer {
pub fn new(from: &str, command: &str) -> Result<CommandMailer> {
let mut parts = command.split_whitespace().map(String::from);
let program = parts.next().context("mail command is empty")?;
ensure!(from.contains('@'), "mail sender must be an address");
Ok(CommandMailer {
from: from.to_string(),
program,
args: parts.collect(),
})
}
}
impl Mailer for CommandMailer {
fn send(&self, to: &str, subject: &str, body: &str) -> Result<()> {
let clean = |s: &str| s.replace(['\r', '\n'], " ");
let message = format!(
"From: {}\r\nTo: {}\r\nSubject: {}\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n{}\r\n",
clean(&self.from),
clean(to),
clean(subject),
body
);
let output = Command::new(&self.program)
.args(&self.args)
.stdin(Stdio::piped())
.stdout(Stdio::null())
.stderr(Stdio::piped())
.spawn()
.and_then(|mut child| {
use std::io::Write;
if let Some(mut stdin) = child.stdin.take() {
stdin.write_all(message.as_bytes())?;
}
child.wait_with_output()
})
.with_context(|| format!("could not run mail command {}", self.program))?;
if !output.status.success() {
bail!(
"mail command failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}
}
/// Mail every notification not yet mailed (at most `limit`), newest last so the
/// inbox reads in order. Returns (mailed, failed).
pub fn mail_pending(
database: &Database,
mailer: &dyn Mailer,
now: i64,
limit: usize,
) -> Result<(usize, usize)> {
let pending: Vec<Notification> = {
let connection = database.connection()?;
let mut statement = connection.prepare(&format!(
"{SELECT} WHERE mailed_at IS NULL ORDER BY id LIMIT ?1"
))?;
let rows = statement.query_map([limit as i64], map_row)?;
rows.collect::<std::result::Result<Vec<_>, _>>()?
};
let (mut mailed, mut failed) = (0, 0);
for n in pending {
let body = if n.link.is_empty() {
format!("{}\n\n{}\n", n.title, n.body)
} else {
format!("{}\n\n{}\n\n{}\n", n.title, n.body, n.link)
};
match mailer.send(&n.subject, &format!("[AkurAI Build] {}", n.title), &body) {
Ok(()) => {
database.connection()?.execute(
"UPDATE notifications SET mailed_at = ?2 WHERE id = ?1",
params![n.id, now],
)?;
mailed += 1;
}
Err(error) => {
tracing::warn!(id = n.id, %error, "notification mail failed");
// Give up on this one so a dead address cannot block the queue.
database.connection()?.execute(
"UPDATE notifications SET mailed_at = 0 WHERE id = ?1",
[n.id],
)?;
failed += 1;
}
}
}
Ok((mailed, failed))
}
#[cfg(test)]
pub mod testing {
use super::*;
use std::sync::Mutex;
#[derive(Default)]
pub struct RecordingMailer {
pub sent: Mutex<Vec<(String, String, String)>>,
pub fail_for: Mutex<Option<String>>,
}
impl Mailer for RecordingMailer {
fn send(&self, to: &str, subject: &str, body: &str) -> Result<()> {
if self.fail_for.lock().expect("lock").as_deref() == Some(to) {
bail!("dead address");
}
self.sent.lock().expect("lock").push((
to.to_string(),
subject.to_string(),
body.to_string(),
));
Ok(())
}
}
}
#[cfg(test)]
mod tests {
use super::testing::RecordingMailer;
use super::*;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
fn database() -> Database {
let db = Database::memory(KEY).expect("db");
db.add_repository("demo", "https://example.com/demo.git", "main")
.expect("repo");
db
}
fn review(title: &str) -> New<'_> {
New {
kind: "review_requested",
repository: "demo",
title,
body: "Please look.",
link: "https://build.example/app/repositories/demo/pulls/1",
}
}
#[test]
fn feed_is_per_subject_newest_first_with_unread_counts_and_marking() {
let db = database();
notify(&db, "Ada@Example.is", &review("Review #1")).expect("ok");
notify(&db, "ada@example.is", &review("Review #2")).expect("ok");
notify(&db, "bob@example.is", &review("Review #1")).expect("ok");
assert!(notify(&db, " ", &review("x")).expect("ok").is_none());
assert!(
notify(
&db,
"ada@example.is",
&New {
kind: "bogus",
..review("x")
}
)
.is_err()
);
let ada = feed(&db, "ADA@example.is", false).expect("ok");
assert_eq!(ada.len(), 2);
assert_eq!(ada[0].title, "Review #2");
assert_eq!(unread_count(&db, "ada@example.is").expect("ok"), 2);
assert!(mark_read(&db, "ada@example.is", ada[0].id, 50).expect("ok"));
assert!(
!mark_read(&db, "bob@example.is", ada[0].id, 50).expect("ok"),
"not bob's"
);
assert_eq!(feed(&db, "ada@example.is", true).expect("ok").len(), 1);
assert_eq!(mark_all_read(&db, "ada@example.is", 60).expect("ok"), 1);
assert_eq!(unread_count(&db, "ada@example.is").expect("ok"), 0);
}
#[test]
fn run_watchers_are_writers_admins_and_forge_admins_once_each() {
let db = database();
access::set_collaborator(&db, "demo", "writer@example.is", access::Role::Write)
.expect("ok");
access::set_collaborator(&db, "demo", "reader@example.is", access::Role::Read).expect("ok");
access::set_collaborator(&db, "demo", "boss@example.is", access::Role::Admin).expect("ok");
let watchers = run_watchers(
&db,
"demo",
&["boss@example.is".into(), "root@example.is".into()],
)
.expect("ok");
let n = notify_all(
&db,
&watchers,
&New {
kind: "run_failed",
repository: "demo",
title: "Run #7 failed",
body: "",
link: "",
},
)
.expect("ok");
assert_eq!(n, 3, "writer, boss (once) and root");
assert!(
feed(&db, "reader@example.is", false)
.expect("ok")
.is_empty()
);
}
#[test]
fn pending_notifications_are_mailed_once_and_a_dead_address_does_not_block_the_rest() {
let db = database();
notify(&db, "dead@example.is", &review("Review #1")).expect("ok");
notify(&db, "ada@example.is", &review("Review #2")).expect("ok");
let mailer = RecordingMailer::default();
*mailer.fail_for.lock().expect("lock") = Some("dead@example.is".into());
assert_eq!(mail_pending(&db, &mailer, 100, 10).expect("ok"), (1, 1));
assert_eq!(
mail_pending(&db, &mailer, 101, 10).expect("ok"),
(0, 0),
"nothing twice"
);
let sent = mailer.sent.lock().expect("lock").clone();
assert_eq!(sent.len(), 1);
assert_eq!(sent[0].0, "ada@example.is");
assert!(
sent[0].1.contains("Review #2")
&& sent[0]
.2
.contains("https://build.example/app/repositories/demo/pulls/1")
);
}
#[test]
fn command_mailer_needs_a_command_and_an_address() {
assert!(CommandMailer::new("build@example.is", "").is_err());
assert!(CommandMailer::new("not-an-address", "cat").is_err());
let mailer = CommandMailer::new("build@example.is", "sh -c cat").expect("ok");
assert!(mailer.send("ada@example.is", "Hi\nthere", "body").is_ok());
let failing = CommandMailer::new("build@example.is", "sh -c false").expect("ok");
assert!(failing.send("ada@example.is", "x", "y").is_err());
}
}