From 19c8f541def7ffc72ffdcee7ce2861046da3154c Mon Sep 17 00:00:00 2001 From: Bendik Aagaard Lynghaug Date: Wed, 23 Sep 2026 19:45:48 +0200 Subject: [PATCH] Mail is content: a state's `mail:` and the invite mail, published for gdo `mail: { to, subject, body }` on a state in aggregates.yaml sends when a record enters that state, `to: submitter` or one address, templates over the record's fields plus {site}, {chain}, {email}. An invite mails its one-time link built in, reworded by `mail.invite` in site.yaml, which also carries the sender (`mail.from`, `mail.reply_to`); without it a site sends nothing and says so in the log. Portal renders the text and an HTML alternative and publishes one message per state change on portal.mail.send in the WORMHOLE stream, id : : so a retried transition sends once; uhhm/gdo holds the durable consumer and talks SMTP. The lint requires a required email field on every form recording into a bucket that mails the submitter. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01L4jrCgLiKKHAEFuZUJjckH --- CHANGELOG.md | 4 + src/aggregates/mod.rs | 80 +++++++++++ src/answers.rs | 2 + src/app.rs | 6 + src/content/mod.rs | 26 ++++ src/content/validate.rs | 23 ++++ src/lib.rs | 3 + src/mail.rs | 284 ++++++++++++++++++++++++++++++++++++++++ src/main.rs | 3 + 9 files changed, 431 insertions(+) create mode 100644 src/mail.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index c2bb6b6..dc2194c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ One line per change, grouped by the release that shipped it. Newest first. A release is a Cargo.toml bump plus a `v` tag; the content repos pin the tag they run (see the end of this file). +## 0.3.43 (2026-09-23) + +- Mail is content: `mail: { to, subject, body }` on a state in aggregates.yaml sends when a record enters that state (`to: submitter` or one address; `{field}`, `{site}`, `{chain}`, `{email}` in the templates), and an invite mails its one-time link built in; `site.yaml` gains `mail: { from, reply_to, invite }`. Portal renders and publishes on `portal.mail.send` in the `WORMHOLE` stream; uhhm/gdo delivers. The lint requires a required `email` field on every form recording into a bucket that mails the submitter. + ## 0.3.42 (2026-09-23) - people: the Kanidm person lookup reads the body, since Kanidm 1.11 answers a missing person with 200 and `null` rather than 404; before this every invite skipped account creation and failed on the group step diff --git a/src/aggregates/mod.rs b/src/aggregates/mod.rs index 0040807..6e9582c 100644 --- a/src/aggregates/mod.rs +++ b/src/aggregates/mod.rs @@ -36,6 +36,32 @@ pub struct AggregateSchema { /// rule has an explicit, auditable opt-out instead of a silent /// one, and so the next reader knows where the answers go. pub attended_by: Option, + /// The mail a record's arrival in a state sends (see `mail.rs`): + /// declared on the state in aggregates.yaml, next to the event + /// that enters it, so content decides when a person hears + /// something and the mailer never has to know a site. + pub mail_for_state: HashMap, +} + +/// A mail to send when a record enters a state. `to` is `submitter` +/// (the record's `email` field, which the lint then requires the +/// forms to collect) or one fixed address. Subject and body are +/// templates: `{field}` for any field of the record, plus `{site}` +/// (the instance's public URL), `{chain}` (the record id), `{email}`, +/// `{bucket}` and `{state}`. The body is markdown; the mailer sends it +/// as plain text with an HTML alternative. +#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MailSpec { + pub to: String, + pub subject: String, + pub body: String, +} + +impl MailSpec { + pub fn to_submitter(&self) -> bool { + self.to == "submitter" + } } impl AggregateSchema { @@ -43,6 +69,10 @@ impl AggregateSchema { self.event_for_state.contains_key(state) } + pub fn mail_for(&self, state: &str) -> Option<&MailSpec> { + self.mail_for_state.get(state) + } + pub fn allowed(&self, state: &str) -> &[String] { self.transitions .get(state) @@ -54,6 +84,8 @@ impl AggregateSchema { #[derive(Debug, serde::Deserialize)] struct RawState { event: String, + #[serde(default)] + mail: Option, } #[derive(Debug, serde::Deserialize)] @@ -123,6 +155,22 @@ pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result anyhow::Result Event { Event { id: uuid::Uuid::new_v4(), diff --git a/src/answers.rs b/src/answers.rs index 5b57fbe..384f39b 100644 --- a/src/answers.rs +++ b/src/answers.rs @@ -268,6 +268,7 @@ async fn apply_transition( .put(&item.item_id, data.into()) .await .map_err(|e| ServerFnError::new(e.to_string()))?; + crate::mail::on_state(state, bucket, &item.item_id, &item.to, &answer.responses).await; // The decision extends the DAG as a child of the answer's own // submission hash, published like any other answer. @@ -378,6 +379,7 @@ pub async fn self_transition_answer( .put(&item_id, data.into()) .await .map_err(|e| ServerFnError::new(e.to_string()))?; + crate::mail::on_state(&state, &self_transition.bucket, &item_id, &self_transition.to, &answer.responses).await; let parent_hashes = vec![item_id.clone()]; let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload_for_event, decided_ms); diff --git a/src/app.rs b/src/app.rs index 5f5d435..37d2c20 100644 --- a/src/app.rs +++ b/src/app.rs @@ -2245,6 +2245,7 @@ pub async fn submit_answer( if let Ok(bytes) = serde_json::to_vec(&invited) { let _ = state.nats.publish(crate::people::INVITED_SUBJECT, bytes.into()).await; } + crate::mail::on_invite(&state, &invited).await; } // `/shape/{curve.key}`: the next page may be chosen by the answer. let next = match alt.action.as_deref() { @@ -2298,6 +2299,11 @@ pub async fn submit_answer( { tracing::error!("failed to store answer in {bucket}: {e}"); } + // Arrival is a state too: what the schema says the first + // state sends, it sends now. + if let Some(initial) = state.aggregates.load().get(bucket).map(|s| s.initial.clone()) { + crate::mail::on_state(&state, bucket, &chain_hash, &initial, &responses).await; + } } Ok(SubmitResult { next, chain_hash }) diff --git a/src/content/mod.rs b/src/content/mod.rs index a3e0a59..9d4e869 100644 --- a/src/content/mod.rs +++ b/src/content/mod.rs @@ -543,6 +543,32 @@ pub struct SiteConfig { pub favicon: Option, #[serde(default)] pub hero: HeroConfig, + /// Who this site's mail comes from. Absent means the site sends + /// nothing: a state's `mail:` and the invite mail are logged and + /// skipped, never half-sent (see `mail.rs`). + #[serde(default)] + pub mail: Option, +} + +/// The sender identity, and the one built-in mail's wording. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct MailConfig { + /// `Name
` or a bare address, on a domain the host's + /// mail server may send for. + pub from: String, + #[serde(default)] + pub reply_to: Option, + /// Rewords the invite mail; the built-in text is in the site's + /// `lang`. `{name}`, `{username}`, `{email}`, `{group}`, `{link}`, + /// `{site}`, `{site_name}`, `{expires}`. + #[serde(default)] + pub invite: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct MailTemplate { + pub subject: String, + pub body: String, } /// What the landing page's hero is. `plain` is just the copy; `module` diff --git a/src/content/validate.rs b/src/content/validate.rs index 2d17c4c..b911525 100644 --- a/src/content/validate.rs +++ b/src/content/validate.rs @@ -92,6 +92,29 @@ pub fn validate_questions( questions: &std::collections::HashMap, aggregates: &std::collections::HashMap, ) -> anyhow::Result<()> { + // A mail to the submitter needs an address to go to: every form + // that records into that bucket has to collect a required email. + for (bucket, schema) in aggregates { + let submitter_states: Vec<&String> = schema.mail_for_state.iter().filter(|(_, m)| m.to_submitter()).map(|(s, _)| s).collect(); + if submitter_states.is_empty() { + continue; + } + for question in questions.values() { + for alternative in question.alternatives.iter().filter(|a| a.record_as.as_deref() == Some(bucket.as_str())) { + let collects_email = alternative + .features + .iter() + .flat_map(|f| &f.requirements) + .any(|r| r.name == "email" && r.kind == "email" && !r.optional); + if !collects_email { + anyhow::bail!( + "question {:?} alternative {:?}: records into {bucket:?}, whose state(s) {:?} mail the submitter, so it must collect a required `email` field of type email", + question.id, alternative.name, submitter_states + ); + } + } + } + } for question in questions.values() { if let Some(event) = &question.event { if question.is_dynamic() { diff --git a/src/lib.rs b/src/lib.rs index 0b0526b..40281ec 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,9 @@ pub mod access; #[cfg(feature = "ssr")] pub mod people; +#[cfg(feature = "ssr")] +pub mod mail; + #[cfg(feature = "ssr")] pub mod server; diff --git a/src/mail.rs b/src/mail.rs new file mode 100644 index 0000000..e467ea9 --- /dev/null +++ b/src/mail.rs @@ -0,0 +1,284 @@ +//! 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, + /// `::` or `invite:::` - + /// the dedup key. + pub id: String, + pub site: String, + pub from: String, + #[serde(default)] + pub reply_to: Option, + 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 { + 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 { + 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!( + "
{body}
" + ) +} + +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 { + 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::::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\n\n[link](https://x.no)"); + assert!(html.contains("there")); + assert!(html.contains("link")); + assert!(!html.contains("