285 lines
12 KiB
Rust
285 lines
12 KiB
Rust
//! Portal's side of the mail gate. Content decides when a person
|
|||
|
|
//! hears something (`mail:` on a state in aggregates.yaml, and the
|
||
|
|
//! built-in invite mail); portal renders the message the moment a
|
||
|
|
//! record enters that state and publishes it on one JetStream subject.
|
||
|
|
//! gdo (uhhm/gdo) holds a durable consumer on that subject and hands
|
||
|
|
//! each message to the host's SMTP. Portal never talks SMTP, gdo never
|
||
|
|
//! reads content, and a message survives either of them being down.
|
||
|
|
//!
|
||
|
|
//! The message is the contract between the two, `Outgoing` below,
|
||
|
|
//! versioned by `v`. Its id is the JetStream message id, so a
|
||
|
|
//! transition that fires twice (a retry, a reseed) sends once.
|
||
|
|
#![cfg(feature = "ssr")]
|
||
|
|
|
||
|
|
use serde::{Deserialize, Serialize};
|
||
|
|
use std::time::Duration;
|
||
|
|
|
||
|
|
/// The stream both sides create on boot, with the same config, so
|
||
|
|
/// whichever starts first owns nothing the other lacks.
|
||
|
|
pub const STREAM: &str = "WORMHOLE";
|
||
|
|
pub const SUBJECTS: &str = "portal.mail.>";
|
||
|
|
pub const SUBJECT: &str = "portal.mail.send";
|
||
|
|
/// Bumped when `Outgoing` changes shape; gdo refuses what it does not
|
||
|
|
/// know rather than guessing.
|
||
|
|
pub const CONTRACT: u8 = 1;
|
||
|
|
|
||
|
|
/// One mail, fully rendered: everything gdo needs and nothing it has
|
||
|
|
/// to look up.
|
||
|
|
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
|
||
|
|
pub struct Outgoing {
|
||
|
|
pub v: u8,
|
||
|
|
/// `<bucket>:<record>:<state>` or `invite:<user>:<group>:<ms>` -
|
||
|
|
/// the dedup key.
|
||
|
|
pub id: String,
|
||
|
|
pub site: String,
|
||
|
|
pub from: String,
|
||
|
|
#[serde(default)]
|
||
|
|
pub reply_to: Option<String>,
|
||
|
|
pub to: String,
|
||
|
|
pub subject: String,
|
||
|
|
pub text: String,
|
||
|
|
pub html: String,
|
||
|
|
}
|
||
|
|
|
||
|
|
/// Idempotent, called by portal at boot and by gdo. A week of
|
||
|
|
/// retention: a mail nobody delivered in a week is stale, not owed.
|
||
|
|
/// The duplicate window has to fit inside it.
|
||
|
|
pub async fn ensure_stream(js: &async_nats::jetstream::Context) -> anyhow::Result<async_nats::jetstream::stream::Stream> {
|
||
|
|
Ok(js
|
||
|
|
.get_or_create_stream(async_nats::jetstream::stream::Config {
|
||
|
|
name: STREAM.to_string(),
|
||
|
|
subjects: vec![SUBJECTS.to_string()],
|
||
|
|
max_age: Duration::from_secs(7 * 24 * 3600),
|
||
|
|
duplicate_window: Duration::from_secs(24 * 3600),
|
||
|
|
..Default::default()
|
||
|
|
})
|
||
|
|
.await?)
|
||
|
|
}
|
||
|
|
|
||
|
|
/// `{key}` in a template becomes the variable's value, an unknown key
|
||
|
|
/// becomes nothing, and a brace that opens no key stays a brace.
|
||
|
|
pub fn render(template: &str, vars: &serde_json::Map<String, serde_json::Value>) -> String {
|
||
|
|
let mut out = String::with_capacity(template.len());
|
||
|
|
let mut rest = template;
|
||
|
|
while let Some(start) = rest.find('{') {
|
||
|
|
out.push_str(&rest[..start]);
|
||
|
|
let after = &rest[start + 1..];
|
||
|
|
let key_len = after
|
||
|
|
.char_indices()
|
||
|
|
.take_while(|(_, c)| c.is_ascii_alphanumeric() || *c == '_' || *c == '-' || *c == '.')
|
||
|
|
.last()
|
||
|
|
.map(|(i, c)| i + c.len_utf8())
|
||
|
|
.unwrap_or(0);
|
||
|
|
if key_len > 0 && after[key_len..].starts_with('}') {
|
||
|
|
let key = &after[..key_len];
|
||
|
|
match vars.get(key) {
|
||
|
|
Some(serde_json::Value::String(s)) => out.push_str(s),
|
||
|
|
Some(serde_json::Value::Null) | None => {}
|
||
|
|
Some(other) => out.push_str(&other.to_string()),
|
||
|
|
}
|
||
|
|
rest = &after[key_len + 1..];
|
||
|
|
} else {
|
||
|
|
out.push('{');
|
||
|
|
rest = after;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
out.push_str(rest);
|
||
|
|
out
|
||
|
|
}
|
||
|
|
|
||
|
|
/// The HTML alternative: the markdown body as a single column with
|
||
|
|
/// inline styles, which is as much as mail clients agree on. Raw HTML
|
||
|
|
/// in the body is dropped, the same rule as page descriptions.
|
||
|
|
pub fn to_html(markdown: &str) -> String {
|
||
|
|
use pulldown_cmark::{html, Event, Options, Parser};
|
||
|
|
let parser = Parser::new_ext(markdown, Options::empty()).filter(|e| !matches!(e, Event::Html(_) | Event::InlineHtml(_)));
|
||
|
|
let mut body = String::new();
|
||
|
|
html::push_html(&mut body, parser);
|
||
|
|
format!(
|
||
|
|
"<!doctype html><html><body style=\"margin:0;padding:1.5em;background:#f4f4f2;color:#1c1c1a;font:16px/1.5 -apple-system,Segoe UI,Helvetica,Arial,sans-serif\"><div style=\"max-width:36em;margin:0 auto;background:#fff;padding:1.5em 2em;border-radius:0.5em\">{body}</div></body></html>"
|
||
|
|
)
|
||
|
|
}
|
||
|
|
|
||
|
|
fn public_url() -> String {
|
||
|
|
std::env::var("PUBLIC_URL").unwrap_or_default().trim_end_matches('/').to_string()
|
||
|
|
}
|
||
|
|
|
||
|
|
fn site_name(site: &crate::content::SiteConfig) -> String {
|
||
|
|
site.title.clone().unwrap_or_else(|| option_env!("SITE_NAME").unwrap_or("portal").to_string())
|
||
|
|
}
|
||
|
|
|
||
|
|
/// The variables a state mail may use: every field of the record,
|
||
|
|
/// then the runtime's own.
|
||
|
|
pub fn vars_for(bucket: &str, id: &str, state: &str, responses: &serde_json::Value) -> serde_json::Map<String, serde_json::Value> {
|
||
|
|
let mut vars = serde_json::Map::new();
|
||
|
|
if let Some(fields) = responses.as_object() {
|
||
|
|
for (k, v) in fields {
|
||
|
|
vars.insert(k.clone(), match v {
|
||
|
|
serde_json::Value::String(s) => serde_json::Value::String(s.clone()),
|
||
|
|
serde_json::Value::Null => serde_json::Value::String(String::new()),
|
||
|
|
other => serde_json::Value::String(other.to_string()),
|
||
|
|
});
|
||
|
|
}
|
||
|
|
}
|
||
|
|
let email = responses.get("email").and_then(|v| v.as_str()).unwrap_or_default().trim().to_string();
|
||
|
|
vars.insert("email".into(), email.into());
|
||
|
|
vars.insert("site".into(), public_url().into());
|
||
|
|
vars.insert("chain".into(), id.into());
|
||
|
|
vars.insert("bucket".into(), bucket.into());
|
||
|
|
vars.insert("state".into(), state.into());
|
||
|
|
vars
|
||
|
|
}
|
||
|
|
|
||
|
|
async fn publish(js: &async_nats::jetstream::Context, mail: &Outgoing) -> anyhow::Result<()> {
|
||
|
|
let mut headers = async_nats::HeaderMap::new();
|
||
|
|
headers.insert("Nats-Msg-Id", mail.id.as_str());
|
||
|
|
let bytes = serde_json::to_vec(mail)?;
|
||
|
|
js.publish_with_headers(SUBJECT, headers, bytes.into()).await?.await?;
|
||
|
|
Ok(())
|
||
|
|
}
|
||
|
|
|
||
|
|
/// A record entered `state` in `bucket`: send what the schema says,
|
||
|
|
/// if anything. Best-effort by design - the transition is already
|
||
|
|
/// committed, and a mail that cannot be rendered is logged, never a
|
||
|
|
/// failed request.
|
||
|
|
pub async fn on_state(state: &crate::server::AppState, bucket: &str, id: &str, to_state: &str, responses: &serde_json::Value) {
|
||
|
|
let Some(spec) = state.aggregates.load().get(bucket).and_then(|s| s.mail_for(to_state)).cloned() else {
|
||
|
|
return;
|
||
|
|
};
|
||
|
|
let site = state.site.load();
|
||
|
|
let Some(cfg) = &site.mail else {
|
||
|
|
tracing::warn!(bucket, to_state, "a mail is declared for this state, but site.yaml has no `mail.from` - not sent");
|
||
|
|
return;
|
||
|
|
};
|
||
|
|
let vars = vars_for(bucket, id, to_state, responses);
|
||
|
|
let to = if spec.to_submitter() {
|
||
|
|
match vars.get("email").and_then(|v| v.as_str()).filter(|e| e.contains('@')) {
|
||
|
|
Some(e) => e.to_string(),
|
||
|
|
None => {
|
||
|
|
tracing::warn!(bucket, id, to_state, "mail to the submitter, but the record carries no email - not sent");
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
spec.to.clone()
|
||
|
|
};
|
||
|
|
let text = render(&spec.body, &vars);
|
||
|
|
let mail = Outgoing {
|
||
|
|
v: CONTRACT,
|
||
|
|
id: format!("{bucket}:{id}:{to_state}"),
|
||
|
|
site: site_name(&site),
|
||
|
|
from: cfg.from.clone(),
|
||
|
|
reply_to: cfg.reply_to.clone(),
|
||
|
|
to,
|
||
|
|
subject: render(&spec.subject, &vars),
|
||
|
|
html: to_html(&text),
|
||
|
|
text,
|
||
|
|
};
|
||
|
|
if let Err(e) = publish(&state.jetstream, &mail).await {
|
||
|
|
tracing::error!(bucket, id, to_state, "publishing the mail failed: {e}");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
/// The invite mail: the one-time link to the person just onboarded.
|
||
|
|
/// Built in, since portal knows everything about it; `mail.invite` in
|
||
|
|
/// site.yaml rewords it. `{name}`, `{email}`, `{group}`, `{link}`,
|
||
|
|
/// `{site}`, `{expires}`.
|
||
|
|
pub async fn on_invite(state: &crate::server::AppState, invited: &crate::people::Invited) {
|
||
|
|
let site = state.site.load();
|
||
|
|
let Some(cfg) = &site.mail else {
|
||
|
|
tracing::warn!(user = %invited.username, "invite made, but site.yaml has no `mail.from` - the link is on the desk only");
|
||
|
|
return;
|
||
|
|
};
|
||
|
|
let norwegian = site.lang.as_deref().is_some_and(|l| l.starts_with("no") || l.starts_with("nb") || l.starts_with("nn"));
|
||
|
|
let (subject, body) = match (&cfg.invite, invited.reset_link.is_some(), norwegian) {
|
||
|
|
(Some(t), _, _) => (t.subject.clone(), t.body.clone()),
|
||
|
|
(None, true, true) => (
|
||
|
|
"Du er invitert inn i {group} på {site_name}".to_string(),
|
||
|
|
"Hei {name},\n\nDu har fått en konto på {site} og en plass i {group}. Velg passord eller passnøkkel her, lenken virker fram til {expires}:\n\n{link}\n\nBrukernavnet ditt er {username}.".to_string(),
|
||
|
|
),
|
||
|
|
(None, false, true) => (
|
||
|
|
"Du har fått plass i {group} på {site_name}".to_string(),
|
||
|
|
"Hei {name},\n\nKontoen din på {site} har nå plass i {group}. Logg inn som vanlig.".to_string(),
|
||
|
|
),
|
||
|
|
(None, true, false) => (
|
||
|
|
"You are invited into {group} at {site_name}".to_string(),
|
||
|
|
"Hello {name},\n\nYou have an account at {site} and a place in {group}. Set your password or passkey here; the link works until {expires}:\n\n{link}\n\nYour username is {username}.".to_string(),
|
||
|
|
),
|
||
|
|
(None, false, false) => (
|
||
|
|
"You now have a place in {group} at {site_name}".to_string(),
|
||
|
|
"Hello {name},\n\nYour account at {site} now has a place in {group}. Sign in as usual.".to_string(),
|
||
|
|
),
|
||
|
|
};
|
||
|
|
let now_ms = chrono::Utc::now().timestamp_millis();
|
||
|
|
let expires = invited
|
||
|
|
.link_expires_ms
|
||
|
|
.and_then(chrono::DateTime::<chrono::Utc>::from_timestamp_millis)
|
||
|
|
.map(|t| t.format("%Y-%m-%d %H:%M UTC").to_string())
|
||
|
|
.unwrap_or_default();
|
||
|
|
let mut vars = serde_json::Map::new();
|
||
|
|
vars.insert("name".into(), invited.display_name.clone().into());
|
||
|
|
vars.insert("username".into(), invited.username.clone().into());
|
||
|
|
vars.insert("email".into(), invited.email.clone().into());
|
||
|
|
vars.insert("group".into(), invited.group.clone().into());
|
||
|
|
vars.insert("link".into(), invited.reset_link.clone().unwrap_or_default().into());
|
||
|
|
vars.insert("site".into(), public_url().into());
|
||
|
|
vars.insert("site_name".into(), site_name(&site).into());
|
||
|
|
vars.insert("expires".into(), expires.into());
|
||
|
|
let text = render(&body, &vars);
|
||
|
|
let mail = Outgoing {
|
||
|
|
v: CONTRACT,
|
||
|
|
id: format!("invite:{}:{}:{now_ms}", invited.username, invited.group),
|
||
|
|
site: site_name(&site),
|
||
|
|
from: cfg.from.clone(),
|
||
|
|
reply_to: cfg.reply_to.clone(),
|
||
|
|
to: invited.email.clone(),
|
||
|
|
subject: render(&subject, &vars),
|
||
|
|
html: to_html(&text),
|
||
|
|
text,
|
||
|
|
};
|
||
|
|
if let Err(e) = publish(&state.jetstream, &mail).await {
|
||
|
|
tracing::error!(user = %invited.username, "publishing the invite mail failed: {e}");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#[cfg(test)]
|
||
|
|
mod tests {
|
||
|
|
use super::*;
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn templates_take_fields_and_leave_stray_braces_alone() {
|
||
|
|
let responses = serde_json::json!({ "name": "Kari", "email": " kari@x.no ", "count": 3, "hva": null });
|
||
|
|
let vars = vars_for("reports", "abc", "fixed", &responses);
|
||
|
|
assert_eq!(render("Hi {name}, {count} at {site}/decide?chain={chain}&email={email}", &vars), "Hi Kari, 3 at /decide?chain=abc&email=kari@x.no");
|
||
|
|
assert_eq!(render("{unknown} {hva} {not a key} {", &vars), " {not a key} {");
|
||
|
|
assert_eq!(vars["bucket"], "reports");
|
||
|
|
assert_eq!(vars["state"], "fixed");
|
||
|
|
}
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn the_html_alternative_keeps_markdown_and_drops_raw_html() {
|
||
|
|
let html = to_html("Hello **there**\n\n<script>x()</script>\n\n[link](https://x.no)");
|
||
|
|
assert!(html.contains("<strong>there</strong>"));
|
||
|
|
assert!(html.contains("<a href=\"https://x.no\">link</a>"));
|
||
|
|
assert!(!html.contains("<script>"));
|
||
|
|
assert!(html.starts_with("<!doctype html>"));
|
||
|
|
}
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn the_contract_round_trips() {
|
||
|
|
let mail = Outgoing {
|
||
|
|
v: CONTRACT,
|
||
|
|
id: "reports:abc:fixed".into(),
|
||
|
|
site: "Tomter Vel".into(),
|
||
|
|
from: "Tomter Vel <vel@example.no>".into(),
|
||
|
|
reply_to: None,
|
||
|
|
to: "kari@x.no".into(),
|
||
|
|
subject: "Fixed".into(),
|
||
|
|
text: "Done.".into(),
|
||
|
|
html: to_html("Done."),
|
||
|
|
};
|
||
|
|
let back: Outgoing = serde_json::from_slice(&serde_json::to_vec(&mail).unwrap()).unwrap();
|
||
|
|
assert_eq!(back, mail);
|
||
|
|
}
|
||
|
|
}
|