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 <bucket>:<record>
:<state> 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L4jrCgLiKKHAEFuZUJjckH

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<version>` 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

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<String>,
+    /// 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<String, MailSpec>,
+}
+
+/// 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<MailSpec>,
 }
 
 #[derive(Debug, serde::Deserialize)]
@@ -123,6 +155,22 @@ pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, Aggreg
             }
         }
 
+        let mut mail_for_state = HashMap::new();
+        for (state, raw_state) in raw_schema.states {
+            if let Some(mail) = raw_state.mail {
+                if mail.subject.trim().is_empty() || mail.body.trim().is_empty() {
+                    anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: mail needs a subject and a body");
+                }
+                if !mail.to_submitter() && !(mail.to.contains('@') && !mail.to.contains(char::is_whitespace)) {
+                    anyhow::bail!(
+                        "aggregates.yaml: bucket {bucket:?}: state {state:?}: mail.to must be `submitter` or one address, not {:?}",
+                        mail.to
+                    );
+                }
+                mail_for_state.insert(state, mail);
+            }
+        }
+
         out.insert(
             bucket.clone(),
             AggregateSchema {
@@ -132,6 +180,7 @@ pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, Aggreg
                 state_for_event,
                 transitions: raw_schema.transitions,
                 attended_by: raw_schema.attended_by,
+                mail_for_state,
             },
         );
     }
@@ -349,6 +398,37 @@ aggregates:
         .unwrap()
     }
 
+    #[test]
+    fn a_state_may_carry_the_mail_its_arrival_sends() {
+        let schemas = parse_aggregates_yaml(
+            r#"
+aggregates:
+  - bucket: reports
+    initial: open
+    states:
+      open: { event: received, mail: { to: submitter, subject: "Got it, {name}", body: "We read it. {site}/decide?chain={chain}" } }
+      fixed: { event: fixed, mail: { to: board@example.no, subject: "Fixed: {where}", body: "Done." } }
+      dropped: { event: dropped }
+    transitions:
+      open: [fixed, dropped]
+"#,
+        )
+        .unwrap();
+        let s = &schemas["reports"];
+        assert!(s.mail_for("open").unwrap().to_submitter());
+        assert_eq!(s.mail_for("fixed").unwrap().to, "board@example.no");
+        assert!(s.mail_for("dropped").is_none());
+
+        for bad in [
+            "mail: { to: submitter, subject: \"\", body: b }",
+            "mail: { to: \"the board\", subject: s, body: b }",
+            "mail: { to: submitter, subject: s, body: b, cc: x }",
+        ] {
+            let yaml = format!("aggregates:\n  - bucket: r\n    initial: open\n    states:\n      open: {{ event: received, {bad} }}\n");
+            assert!(parse_aggregates_yaml(&yaml).is_err(), "{bad}");
+        }
+    }
+
     fn event(seq: u64, event_type: &str) -> Event {
         Event {
             id: uuid::Uuid::new_v4(),

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);

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 })

src/content/mod.rs

@@ -543,6 +543,32 @@ pub struct SiteConfig {
     pub favicon: Option<String>,
     #[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<MailConfig>,
+}
+
+/// The sender identity, and the one built-in mail's wording.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+pub struct MailConfig {
+    /// `Name <address>` or a bare address, on a domain the host's
+    /// mail server may send for.
+    pub from: String,
+    #[serde(default)]
+    pub reply_to: Option<String>,
+    /// 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<MailTemplate>,
+}
+
+#[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`

src/content/validate.rs

@@ -92,6 +92,29 @@ pub fn validate_questions(
     questions: &std::collections::HashMap<String, Question>,
     aggregates: &std::collections::HashMap<String, crate::aggregates::AggregateSchema>,
 ) -> 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() {

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;
 

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,
+    /// `<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);
+    }
+}

src/main.rs

@@ -60,6 +60,9 @@ async fn main() -> anyhow::Result<()> {
     // into, and `answers::store_answer` creates it on first use.
     let jetstream = async_nats::jetstream::new(nats.clone());
     portal::events::store::ensure_stream(&jetstream).await?;
+    // Outgoing mail waits here for gdo; a mail is never lost to gdo
+    // being down, and never sent by portal itself.
+    portal::mail::ensure_stream(&jetstream).await?;
 
     let oidc_state = Arc::new(oidc::Oidc::from_env().await?);