Release 0.3.47

iris learns stage_words; portal unchanged.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

CHANGELOG.md

@@ -4,6 +4,26 @@ 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.47 (2026-09-24)
+
+- iris: needs.yaml takes `stage_words`, each state in the site's own words, carried into the simulation model for the trainer's desk score. No change to portal itself.
+
+## 0.3.46 (2026-09-23)
+
+- An invite first looks the email up in Kanidm (`/v1/raw/search` on `mail`, when the service account may read people: `idm_people_pii_read`) and only adds an existing account, under whatever username, to the group; no second account, no reset link.
+
+## 0.3.45 (2026-09-23)
+
+- A mail stream that cannot be created at boot (subjects owned by an older stream) is a warning, not a failed start; 0.3.44 took the kasse sites down for two minutes on the rename.
+
+## 0.3.44 (2026-09-23)
+
+- The mail stream is named `mail` (was `WORMHOLE`); gdo 0.1.2 matches. Delete the old stream on each NATS after upgrading both.
+
+## 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

Cargo.lock

@@ -3265,7 +3265,7 @@ dependencies = [
 
 [[package]]
 name = "portal"
-version = "0.3.42"
+version = "0.3.47"
 dependencies = [
  "anyhow",
  "arc-swap",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "portal"
-version = "0.3.42"
+version = "0.3.47"
 edition = "2021"
 
 [lib]

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 = "mail";
+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,14 @@ 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.
+    // Best-effort: a stream that already owns these subjects under an
+    // older name (a rename mid-upgrade) must not keep a site from
+    // booting; mail waits until the old stream is removed.
+    if let Err(e) = portal::mail::ensure_stream(&jetstream).await {
+        tracing::warn!("the mail stream could not be created: {e} - outgoing mail is off until a stream owns portal.mail.>");
+    }
 
     let oidc_state = Arc::new(oidc::Oidc::from_env().await?);
 

src/people.rs

@@ -44,6 +44,13 @@ fn person_lookup_found(status: u16, body: &str) -> Option<bool> {
     }
 }
 
+/// The first entry's `name` in a `/v1/raw/search` answer, which is
+/// `{"entries": [{"attrs": {"name": ["…"], …}}]}`.
+fn name_from_search(body: &str) -> Option<String> {
+    let v: serde_json::Value = serde_json::from_str(body).ok()?;
+    v.get("entries")?.as_array()?.first()?.get("attrs")?.get("name")?.as_array()?.first()?.as_str().map(str::to_string)
+}
+
 pub struct Kanidm {
     base: String,
     token: String,
@@ -77,11 +84,34 @@ impl Kanidm {
         person_lookup_found(status, &text).ok_or_else(|| anyhow::anyhow!("kanidm: looking up {username}: http {status}"))
     }
 
+    /// The account that already carries this email, if the token may
+    /// see it (the service account in `idm_people_pii_read`). None
+    /// when nobody has it, or when the search is not permitted: then
+    /// the username derived from the email decides, as before.
+    async fn find_by_mail(&self, email: &str) -> anyhow::Result<Option<String>> {
+        let (status, text) = self
+            .call(
+                openidconnect::reqwest::Method::POST,
+                "/v1/raw/search",
+                Some(serde_json::json!({ "filter": { "eq": ["mail", email.trim().to_lowercase()] } })),
+            )
+            .await?;
+        if !(200..300).contains(&status) {
+            tracing::warn!(status, "kanidm: searching people by mail is not permitted for this token; matching by username instead");
+            return Ok(None);
+        }
+        Ok(name_from_search(&text))
+    }
+
     /// The whole onboarding, in the order that leaves the least behind
-    /// if a step fails: create, then email, then group, then the link.
+    /// if a step fails: find or create, then email, then group, then
+    /// the link. A person who already has an account, under whatever
+    /// username, is only added to the group; their credentials are
+    /// left alone and no link is made.
     pub async fn invite(&self, name: &str, email: &str, group: &str) -> anyhow::Result<Invited> {
-        let username = username_for(email);
-        let created = if self.person_exists(&username).await? {
+        let existing = self.find_by_mail(email).await?;
+        let username = existing.clone().unwrap_or_else(|| username_for(email));
+        let created = if existing.is_some() || self.person_exists(&username).await? {
             false
         } else {
             let (status, text) = self
@@ -171,6 +201,14 @@ mod tests {
         assert_eq!(username_for("Ærlig@x.no"), "rlig");
     }
 
+    #[test]
+    fn a_search_by_mail_yields_the_accounts_own_name() {
+        assert_eq!(name_from_search(r#"{"entries":[{"attrs":{"name":["ole.n"],"mail":["ole@x.no"]}}]}"#), Some("ole.n".into()));
+        assert_eq!(name_from_search(r#"{"entries":[]}"#), None);
+        assert_eq!(name_from_search("null"), None);
+        assert_eq!(name_from_search("not json"), None);
+    }
+
     #[test]
     fn a_null_person_answer_means_nobody_there() {
         assert_eq!(person_lookup_found(200, "null"), Some(false));