portal/src/mail.rs

//! 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";
/// Where a reply to a case mail arrives, published by whatever reads
/// the mailbox (uhhm/gdo over JMAP). Portal turns it into a note on
/// the record it names and nothing else: no state moves on a reply.
pub const RECEIVED_SUBJECT: &str = "portal.mail.received";
/// 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,
}

/// A reply that came back, already matched to a record by whatever
/// read the mailbox. Portal trusts the match no further than it can
/// check it: the record must exist, the bucket must be one this site
/// has, and the sender must be the address the record itself holds.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Incoming {
    pub v: u8,
    /// The mail's own id, so the same reply read twice lands once.
    pub id: String,
    pub bucket: String,
    pub record: String,
    /// Who sent it, as the mail server saw it.
    pub from: String,
    #[serde(default)]
    pub subject: String,
    /// The reply itself, already stripped of quoted history and
    /// signatures by the reader; portal stores what it is given.
    pub text: String,
    pub received_ms: i64,
}

/// What a reply may add to a record, and what it may never do. Returns
/// the note to append, or why the reply is not one.
///
/// The address a reply comes back on is unguessable, so it is proof of
/// having been sent the mail - but a mailbox is a public door, and an
/// address can be forged. The sender must therefore also match the
/// address the record already holds, which is the one portal wrote to
/// in the first place.
pub fn note_for(incoming: &Incoming, answer: &crate::answers::Answer) -> Result<crate::answers::Note, String> {
    if incoming.v != CONTRACT {
        return Err(format!("reply contract v{} is not v{CONTRACT}", incoming.v));
    }
    let text = incoming.text.trim();
    if text.is_empty() {
        return Err("the reply has no text".to_string());
    }
    let held = answer
        .responses
        .get("email")
        .and_then(|v| v.as_str())
        .map(|e| e.trim().to_lowercase())
        .filter(|e| !e.is_empty())
        .ok_or_else(|| "the record holds no address to compare the sender with".to_string())?;
    if address_of(&incoming.from) != held {
        return Err(format!("{} is not the address this record was written to", incoming.from));
    }
    if answer.notes.iter().any(|n| n.id == incoming.id) {
        return Err("this reply is already on the record".to_string());
    }
    Ok(crate::answers::Note {
        id: incoming.id.clone(),
        from: address_of(&incoming.from),
        subject: incoming.subject.trim().to_string(),
        text: text.to_string(),
        received_ms: incoming.received_ms,
    })
}

/// The bare address out of `Name <a@b>` or `a@b`, folded for
/// comparison. Mail addresses are case-insensitive in the part that
/// matters here, and a display name is not identity.
pub fn address_of(from: &str) -> String {
    let inner = match (from.rfind('<'), from.rfind('>')) {
        (Some(open), Some(close)) if close > open => &from[open + 1..close],
        _ => from,
    };
    inner.trim().to_lowercase()
}

/// Listens for replies and writes each one onto the record it names.
/// A reply is a note and never a decision: nothing here moves a case,
/// so a mailbox - the one door anyone may knock on - cannot drive a
/// state machine even if someone forges their way past the checks.
pub async fn receive(state: crate::server::AppState) {
    use futures::StreamExt;
    let mut sub = match state.nats.subscribe(RECEIVED_SUBJECT).await {
        Ok(sub) => sub,
        Err(e) => {
            tracing::error!(error = %e, "cannot listen for replies");
            return;
        }
    };
    tracing::info!(subject = RECEIVED_SUBJECT, "listening for replies to case mail");
    while let Some(msg) = sub.next().await {
        let incoming: Incoming = match serde_json::from_slice(&msg.payload) {
            Ok(i) => i,
            Err(e) => {
                tracing::warn!("unreadable reply on {RECEIVED_SUBJECT}: {e}");
                continue;
            }
        };
        match note_onto(&state, &incoming).await {
            Ok(()) => tracing::info!(bucket = %incoming.bucket, record = %incoming.record, "a reply was added to the record"),
            Err(e) => tracing::warn!(bucket = %incoming.bucket, record = %incoming.record, "reply refused: {e}"),
        }
    }
}

/// The record's own bucket is the only one a reply may touch, and the
/// note is appended under the same read-modify-write the desks use.
async fn note_onto(state: &crate::server::AppState, incoming: &Incoming) -> anyhow::Result<()> {
    if !state.aggregates.load().contains_key(&incoming.bucket) && !crate::content::bucket_is_read(&state.questions.load(), &incoming.bucket) {
        anyhow::bail!("{} is not a bucket of this site", incoming.bucket);
    }
    let store = state.jetstream.get_key_value(&incoming.bucket).await?;
    let bytes = store
        .get(&incoming.record)
        .await?
        .ok_or_else(|| anyhow::anyhow!("no such record"))?;
    let mut answer: crate::answers::Answer = serde_json::from_slice(&bytes)?;
    let note = note_for(incoming, &answer).map_err(|e| anyhow::anyhow!(e))?;
    answer.notes.push(note);
    store.put(&incoming.record, serde_json::to_vec(&answer)?.into()).await?;
    Ok(())
}

/// 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);
        // A gdo that predates reply addresses sends none, and reads
        // one from a portal that does: the field is optional on the
        // wire in both directions.
        let older: Outgoing = serde_json::from_str(
            r#"{"v":1,"id":"a","site":"s","from":"f@x.no","to":"t@x.no","subject":"s","text":"t","html":"h"}"#,
        )
        .unwrap();
        assert_eq!(older.subject, "s");
        assert!(serde_json::to_string(&older).unwrap().contains("\"html\":\"h\""));
    }

    fn record(email: &str) -> crate::answers::Answer {
        crate::answers::Answer {
            id: "abc".into(),
            question_id: "/report".into(),
            alternative: "Report a hazard".into(),
            responses: serde_json::json!({ "email": email, "where": "The bend" }),
            state: "open".into(),
            submitted_ms: 1,
            decided_ms: None,
            decided_by: None,
            notes: Vec::new(),
        }
    }

    fn reply(from: &str, text: &str) -> Incoming {
        Incoming {
            v: CONTRACT,
            id: "mail-1".into(),
            bucket: "hazards".into(),
            record: "abc".into(),
            from: from.into(),
            subject: "Re: We have it".into(),
            text: text.into(),
            received_ms: 2,
        }
    }

    #[test]
    fn a_reply_from_the_address_on_the_record_becomes_a_note() {
        let note = note_for(&reply("Kari <KARI@x.no>", "  It is fixed now.  "), &record("kari@x.no")).unwrap();
        assert_eq!(note.from, "kari@x.no");
        assert_eq!(note.text, "It is fixed now.");
        assert_eq!(note.id, "mail-1");
        assert_eq!(note.received_ms, 2);
    }

    #[test]
    fn a_reply_from_anyone_else_is_refused() {
        // The address a reply arrives on is unguessable, but a From is
        // a claim, and a mailbox is a door anyone may knock on.
        let err = note_for(&reply("someone@else.no", "Let me in"), &record("kari@x.no")).unwrap_err();
        assert!(err.contains("not the address this record was written to"), "{err}");

        // Nothing to compare against is also a refusal, not a pass.
        let mut anonymous = record("kari@x.no");
        anonymous.responses = serde_json::json!({ "where": "The bend" });
        assert!(note_for(&reply("kari@x.no", "hi"), &anonymous).unwrap_err().contains("no address"));
    }

    #[test]
    fn an_empty_reply_and_a_repeated_one_add_nothing() {
        assert!(note_for(&reply("kari@x.no", "   \n "), &record("kari@x.no")).unwrap_err().contains("no text"));

        let mut already = record("kari@x.no");
        already.notes.push(note_for(&reply("kari@x.no", "It is fixed."), &already).unwrap());
        let err = note_for(&reply("kari@x.no", "It is fixed."), &already).unwrap_err();
        assert!(err.contains("already on the record"), "{err}");
    }

    #[test]
    fn a_reply_from_a_contract_this_portal_does_not_know_is_refused() {
        let mut future = reply("kari@x.no", "hello");
        future.v = CONTRACT + 1;
        assert!(note_for(&future, &record("kari@x.no")).unwrap_err().contains("contract"));
    }

    #[test]
    fn an_address_is_read_out_of_whatever_the_mail_server_said() {
        assert_eq!(address_of("Kari Nordmann <Kari@X.no>"), "kari@x.no");
        assert_eq!(address_of("  kari@x.no "), "kari@x.no");
        assert_eq!(address_of("<kari@x.no>"), "kari@x.no");
        // A display name that looks like an address is not the address.
        assert_eq!(address_of("\"admin@bank.no\" <kari@x.no>"), "kari@x.no");
    }
}