ljos 0.29.0

//! Mail from one seat to another.
//!
//! A message is a pack atom. Seats on one box share the pack and see it at
//! once. `ljos sync` exports the machine's own atoms, so the same atom
//! crosses machines in the sealed log. A receipt is its own atom: the pack
//! and the log are append-only, and an import tells two atoms apart by the
//! hash of their text, so the text of a message carries its id.

use std::collections::BTreeSet;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};

use anyhow::{bail, Context, Result};
use serde_json::{json, Value};

/// How many messages a prompt is shown. The rest stay unread.
const PROMPT_CAP: usize = 8;

/// The first packset that keeps kinds `message`, `receipt` and `group`.
///
/// 0.12.1 refuses them. A new kind has been a minor release: `outcome`
/// arrived in 0.12.0, and `prediction` and `rule` in 0.6.0.
pub(crate) const MAIL_PACKSET: &str = "0.13.0";

/// What a probe post of an empty `message` means.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Probe {
    /// The writer refused the kind.
    Refused,
    /// The kind passed and a later check refused the empty probe.
    Kept,
    /// The writer did not answer.
    Store,
}

/// Classify a probe error. Kind is checked before the text, so an empty
/// message is refused as unknown when the pack cannot keep mail, and as
/// missing text when it can.
fn probe_of(text: &str) -> Probe {
    if text.contains("unknown atom kind") {
        Probe::Refused
    } else if text.contains(": 400:") {
        Probe::Kept
    } else {
        Probe::Store
    }
}

fn unsupported_mail(version: Option<&str>) -> String {
    match version {
        Some(version) => format!(
            "this pack does not keep mail. packset {version} refused kind message. Mail needs packset {MAIL_PACKSET}."
        ),
        None => format!(
            "this pack does not keep mail. The writer refused kind message. Mail needs packset {MAIL_PACKSET}."
        ),
    }
}

/// One message as it is written.
#[derive(Debug, Clone)]
pub struct Outgoing {
    pub id: String,
    pub from: String,
    pub to: Vec<String>,
    pub body: String,
    pub interrupt: bool,
    pub issue: Option<String>,
    pub group: Option<String>,
    pub reply: Option<String>,
    pub scope: Option<String>,
}

/// What `ljos send` was asked to write.
pub struct Send<'a> {
    pub seat: Option<&'a str>,
    pub group: Option<&'a str>,
    pub text: &'a str,
    pub interrupt: bool,
    pub issue: Option<&'a str>,
    /// Write to a seat this pack has not heard from yet.
    pub new_seat: bool,
}

/// One message as a reader sees it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Letter {
    pub id: String,
    pub from: String,
    pub to: Vec<String>,
    pub body: String,
    pub interrupt: bool,
    pub issue: Option<String>,
    pub group: Option<String>,
    pub reply: Option<String>,
    pub scope: Option<String>,
    pub read: bool,
    ts: String,
    seq: usize,
}

/// A read receipt addressed to the seat that sent the message.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Receipt {
    pub msg: String,
    pub by: String,
    pub at: String,
    pub key: String,
}

fn mint_id(kind: &str, parts: &[&str]) -> String {
    static N: AtomicU64 = AtomicU64::new(1);
    let n = N.fetch_add(1, Ordering::Relaxed);
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let raw = format!(
        "{kind}\n{n}\n{nanos}\n{}\n{}",
        std::process::id(),
        parts.join("\n")
    );
    super::work_id(&raw)
}

fn check_name(kind: &str, name: &str) -> Result<String> {
    let name = name.trim();
    if name.is_empty()
        || name
            .chars()
            .any(|c| c.is_whitespace() || c == ',' || c == ':')
    {
        bail!("{kind}: {name:?} is not a name");
    }
    Ok(name.to_string())
}

fn tag(atom: &Value, prefix: &str) -> Option<String> {
    atom["entities"]
        .as_array()?
        .iter()
        .filter_map(Value::as_str)
        .find_map(|e| e.strip_prefix(prefix).map(str::to_string))
}

fn tags(atom: &Value, prefix: &str) -> Vec<String> {
    atom["entities"]
        .as_array()
        .into_iter()
        .flatten()
        .filter_map(Value::as_str)
        .filter_map(|e| e.strip_prefix(prefix).map(str::to_string))
        .collect()
}

fn has(atom: &Value, entity: &str) -> bool {
    atom["entities"]
        .as_array()
        .into_iter()
        .flatten()
        .any(|e| e.as_str() == Some(entity))
}

/// The body a person wrote. The lines above the blank line are the id and
/// the envelope, which keep two messages with the same words distinct.
#[must_use]
pub fn body_of(text: &str) -> String {
    text.split_once("\n\n")
        .map(|(_, body)| body.trim().to_string())
        .unwrap_or_else(|| text.trim().to_string())
}

/// The pack refuses an atom whose text is longer than this, in characters.
pub const ATOM_TEXT_CAP: usize = 500;

/// The longest body a message carries. Past it, a file and a deed
/// (`deedar create file`, then the accession in the message) say it better.
pub const MAIL_BODY_CAP: usize = 16_000;

/// A body past [`MAIL_BODY_CAP`] is refused before anything is written,
/// with the way to send it as a deed.
fn check_length(verb: &str, body: &str) -> Result<()> {
    let n = body.chars().count();
    if n > MAIL_BODY_CAP {
        bail!(
            "{verb}: the message is {n} characters, past {MAIL_BODY_CAP}; write it to a file, \
             `deedar create file --path FILE --agent SEAT`, and send the accession it prints. Nothing was sent"
        );
    }
    Ok(())
}

/// The text that travels. The id is in it, so an import that keys on the
/// text keeps two copies of the same words. A body that would push the
/// text past [`ATOM_TEXT_CAP`] is cut there with an ellipsis; the whole
/// body rides in the atom's `body` field, which [`message_atom`] sets.
#[must_use]
pub fn message_text(m: &Outgoing) -> String {
    let mut head = format!("msg {}\nfrom {}\nto {}", m.id, m.from, m.to.join(", "));
    if m.interrupt {
        head.push_str("\ninterrupt");
    }
    if let Some(issue) = &m.issue {
        head.push_str(&format!("\nissue {issue}"));
    }
    if let Some(group) = &m.group {
        head.push_str(&format!("\ngroup {group}"));
    }
    if let Some(reply) = &m.reply {
        head.push_str(&format!("\nreply {reply}"));
    }
    let body = m.body.trim();
    let room = ATOM_TEXT_CAP.saturating_sub(head.chars().count() + 2);
    if body.chars().count() <= room {
        return format!("{head}\n\n{body}");
    }
    let cut: String = body.chars().take(room.saturating_sub(1)).collect();
    format!("{head}\n\n{cut}\u{2026}")
}

/// The atom a message posts. The entities are what an inbox filters on.
#[must_use]
pub fn message_atom(m: &Outgoing, workspace: &str) -> Value {
    let mut entities = vec![
        format!("seat:{}", m.from),
        format!("from:{}", m.from),
        format!("msg:{}", m.id),
    ];
    for to in &m.to {
        entities.push(format!("to:{to}"));
    }
    if m.interrupt {
        entities.push("priority:interrupt".to_string());
    }
    if let Some(issue) = &m.issue {
        entities.push(format!("issue:{issue}"));
    }
    if let Some(group) = &m.group {
        entities.push(format!("group:{group}"));
    }
    if let Some(reply) = &m.reply {
        entities.push(format!("reply:{reply}"));
    }
    if let Some(scope) = &m.scope {
        entities.push(format!("scope:{scope}"));
    }
    let text = message_text(m);
    let mut atom = json!({
        "schema": "inside.atom/v1",
        "kind": "message",
        "level": "explicit",
        "text": text,
        "workspace": workspace,
        "entities": entities,
    });
    if body_of(&text) != m.body.trim() {
        atom["body"] = Value::String(m.body.trim().to_string());
    }
    atom
}

fn letter_of(atom: &Value, seq: usize) -> Option<Letter> {
    if atom["kind"].as_str() != Some("message") {
        return None;
    }
    let id = tag(atom, "msg:")?;
    let from = tag(atom, "from:").or_else(|| tag(atom, "seat:"))?;
    let to = tags(atom, "to:");
    if to.is_empty() {
        return None;
    }
    Some(Letter {
        id,
        from,
        to,
        // A long body rides whole in `body`; the text holds its start.
        body: atom["body"]
            .as_str()
            .map(|b| b.trim().to_string())
            .unwrap_or_else(|| body_of(atom["text"].as_str().unwrap_or(""))),
        interrupt: has(atom, "priority:interrupt"),
        issue: tag(atom, "issue:"),
        group: tag(atom, "group:"),
        reply: tag(atom, "reply:"),
        scope: tag(atom, "scope:"),
        read: false,
        ts: atom["ts"].as_str().unwrap_or("").to_string(),
        seq,
    })
}

fn readers_of(atoms: &[Value]) -> BTreeSet<(String, String)> {
    let mut out = BTreeSet::new();
    for atom in atoms {
        if atom["kind"].as_str() != Some("receipt") {
            continue;
        }
        let Some(msg) = tag(atom, "receipt:") else {
            continue;
        };
        let Some(by) = tag(atom, "from:").or_else(|| tag(atom, "seat:")) else {
            continue;
        };
        out.insert((msg, by));
    }
    out
}

/// Messages addressed to `seat`, interrupts first, then oldest first.
#[must_use]
pub fn inbox_letters(atoms: &[Value], seat: &str, all: bool) -> Vec<Letter> {
    let read = readers_of(atoms);
    let mut letters: Vec<Letter> = atoms
        .iter()
        .enumerate()
        .filter_map(|(seq, atom)| letter_of(atom, seq))
        .filter(|letter| letter.to.iter().any(|to| to == seat))
        .map(|mut letter| {
            letter.read = read.contains(&(letter.id.clone(), seat.to_string()));
            letter
        })
        .filter(|letter| all || !letter.read)
        .collect();
    letters.sort_by(|a, b| {
        b.interrupt
            .cmp(&a.interrupt)
            .then(a.ts.cmp(&b.ts))
            .then(a.seq.cmp(&b.seq))
    });
    letters
}

/// Receipts on messages this seat sent, newest stamp last.
#[must_use]
pub fn receipts_for(atoms: &[Value], seat: &str) -> Vec<Receipt> {
    let mut out = Vec::new();
    for atom in atoms {
        if atom["kind"].as_str() != Some("receipt") {
            continue;
        }
        if tag(atom, "to:").as_deref() != Some(seat) {
            continue;
        }
        let Some(msg) = tag(atom, "receipt:") else {
            continue;
        };
        let Some(by) = tag(atom, "from:") else {
            continue;
        };
        let text = atom["text"].as_str().unwrap_or("");
        let at = text
            .lines()
            .find_map(|l| l.strip_prefix("at "))
            .unwrap_or("")
            .to_string();
        out.push(Receipt {
            key: super::work_id(text),
            msg,
            by,
            at,
        });
    }
    out.sort_by(|a, b| a.at.cmp(&b.at).then(a.key.cmp(&b.key)));
    out
}

/// Seats in a group, in the order they were added. A later drop removes one.
#[must_use]
pub fn members(atoms: &[Value], group: &str) -> Vec<String> {
    let mut events: Vec<(String, usize, String, bool)> = Vec::new();
    for (seq, atom) in atoms.iter().enumerate() {
        if atom["kind"].as_str() != Some("group") {
            continue;
        }
        if tag(atom, "group:").as_deref() != Some(group) {
            continue;
        }
        let ts = atom["ts"].as_str().unwrap_or("").to_string();
        if let Some(seat) = tag(atom, "member:") {
            events.push((ts, seq, seat, false));
        } else if let Some(seat) = tag(atom, "drop:") {
            events.push((ts, seq, seat, true));
        }
    }
    events.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1)));
    let mut out: Vec<String> = Vec::new();
    for (_, _, seat, drop) in events {
        if drop {
            out.retain(|s| s != &seat);
        } else if !out.iter().any(|s| s == &seat) {
            out.push(seat);
        }
    }
    out
}

/// A body as the prompt shows it: one line of 160 characters, and when it
/// is cut, where the whole of it is. The prompt marks the letter read, so
/// a long body would otherwise be seen only by its start.
fn prompt_body(body: &str) -> String {
    let n = body.chars().count();
    let line = one_line(body, Some(160));
    if n > 160 {
        format!("{line} (all {n} characters: `ljos inbox --all`)")
    } else {
        line
    }
}

fn one_line(body: &str, limit: Option<usize>) -> String {
    let flat = body.split_whitespace().collect::<Vec<_>>().join(" ");
    match limit {
        Some(n) if flat.chars().count() > n => {
            let t: String = flat.chars().take(n.saturating_sub(3)).collect();
            format!("{t}...")
        }
        _ => flat,
    }
}

fn flags(letter: &Letter) -> String {
    let mut parts = Vec::new();
    if letter.interrupt {
        parts.push("interrupt".to_string());
    }
    if let Some(issue) = &letter.issue {
        parts.push(format!("issue:{issue}"));
    }
    if let Some(group) = &letter.group {
        parts.push(format!("group:{group}"));
    }
    if let Some(reply) = &letter.reply {
        parts.push(format!("reply:{reply}"));
    }
    if letter.read {
        parts.push("read".to_string());
    }
    parts.join(" ")
}

/// The inbox a seat prints. Receipts already shown are left out.
#[must_use]
pub fn render_inbox(
    atoms: &[Value],
    seat: &str,
    all: bool,
    seen_receipts: &BTreeSet<String>,
) -> (String, Vec<String>) {
    let letters = inbox_letters(atoms, seat, all);
    let receipts: Vec<Receipt> = receipts_for(atoms, seat)
        .into_iter()
        .filter(|r| !seen_receipts.contains(&r.key))
        .collect();
    if letters.is_empty() && receipts.is_empty() {
        let line = if all {
            "inbox: nothing\n"
        } else {
            "inbox: nothing unread\n"
        };
        return (line.to_string(), Vec::new());
    }
    let mut out = String::new();
    for letter in &letters {
        out.push_str(&format!(
            "{}\t{}\t{}\t{}\n",
            letter.id,
            letter.from,
            flags(letter),
            one_line(&letter.body, None)
        ));
    }
    let mut keys = Vec::new();
    for receipt in &receipts {
        out.push_str(&format!(
            "receipt\t{}\t{}\t{}\n",
            receipt.msg, receipt.by, receipt.at
        ));
        keys.push(receipt.key.clone());
    }
    (out, keys)
}

/// The note a prompt hook prints, and the ids to receipt when it is delivered.
///
/// Message ids are prefixed `mail:`. Receipt ids the sender is being told
/// about are prefixed `rcpt:` and are recorded locally, not written again.
#[must_use]
pub fn prompt_lines(letters: &[Letter], receipts: &[Receipt]) -> (String, Vec<String>) {
    if letters.is_empty() && receipts.is_empty() {
        return (String::new(), Vec::new());
    }
    let mut lines = vec!["mail:".to_string()];
    let mut ids = Vec::new();
    for letter in letters.iter().take(PROMPT_CAP) {
        let mark = if letter.interrupt { "! " } else { "" };
        let on = letter
            .issue
            .as_ref()
            .map(|issue| format!(" on {issue}"))
            .unwrap_or_default();
        lines.push(format!(
            "{mark}{} from {}{on}: {}",
            letter.id,
            letter.from,
            prompt_body(&letter.body)
        ));
        ids.push(format!("mail:{}", letter.id));
    }
    let rest = letters.len().saturating_sub(PROMPT_CAP);
    if rest > 0 {
        lines.push(format!("{rest} more in ljos inbox"));
    }
    for receipt in receipts.iter().take(PROMPT_CAP) {
        lines.push(format!("read {} by {}", receipt.msg, receipt.by));
        ids.push(format!("rcpt:{}", receipt.key));
    }
    (lines.join("\n"), ids)
}

fn find_letter(atoms: &[Value], id: &str) -> Result<Letter> {
    atoms
        .iter()
        .enumerate()
        .filter_map(|(seq, atom)| letter_of(atom, seq))
        .find(|letter| letter.id == id)
        .ok_or_else(|| anyhow::anyhow!("no message {id}"))
}

/// Every seat this pack has heard from: a sender, a reader, a group
/// member, and the seat any atom was written under.
#[must_use]
pub fn known_seats(atoms: &[Value]) -> BTreeSet<String> {
    let mut out = BTreeSet::new();
    for atom in atoms {
        for prefix in ["seat:", "from:", "member:"] {
            out.extend(tags(atom, prefix));
        }
        if let Some(seat) = atom["source"]["harness"].as_str() {
            out.insert(seat.to_string());
        }
    }
    out.retain(|s| !s.trim().is_empty());
    out
}

/// Refuse a seat the pack has never heard from. Mail waits under the name
/// it was sent to, so a message to `ljos` sat unread while the seat that
/// meant to read it was `ljos-bot` (Software-mwy5).
///
/// # Errors
///
/// `seat` is not in `known` and `new_seat` was not passed.
pub fn check_known(seat: &str, known: &BTreeSet<String>, new_seat: bool) -> Result<()> {
    if new_seat || known.contains(seat) {
        return Ok(());
    }
    let list = if known.is_empty() {
        "none yet".to_string()
    } else {
        known.iter().cloned().collect::<Vec<_>>().join(", ")
    };
    let guess = closest_seat(seat, known)
        .map(|s| format!("Did you mean {s}? "))
        .unwrap_or_default();
    bail!(
        "send: no seat named {seat} has written to this pack, so the message would wait unread. \
         {guess}Seats this pack knows: {list}. Pass --new-seat to write to a seat that has not started yet"
    )
}

/// The known seat a mistyped name most likely meant: one that starts with
/// it or holds it (`ljos` for `ljos-bot`), else the nearest by edits, at
/// most two away. `None` when nothing is that close.
#[must_use]
pub fn closest_seat<'a>(seat: &str, known: &'a BTreeSet<String>) -> Option<&'a str> {
    let lower = seat.to_lowercase();
    if lower.is_empty() {
        return None;
    }
    let holds = known
        .iter()
        .filter(|k| k.to_lowercase().contains(&lower) || lower.contains(&k.to_lowercase()))
        .min_by_key(|k| (k.len().abs_diff(seat.len()), k.as_str()));
    if let Some(k) = holds {
        return Some(k.as_str());
    }
    known
        .iter()
        .map(|k| (edits(&lower, &k.to_lowercase()), k))
        .filter(|(d, _)| *d <= 2)
        .min()
        .map(|(_, k)| k.as_str())
}

/// Levenshtein distance, in characters.
fn edits(a: &str, b: &str) -> usize {
    let b: Vec<char> = b.chars().collect();
    let mut row: Vec<usize> = (0..=b.len()).collect();
    for (i, ca) in a.chars().enumerate() {
        let mut prev = row[0];
        row[0] = i + 1;
        for (j, cb) in b.iter().enumerate() {
            let cur = row[j + 1];
            row[j + 1] = if ca == *cb {
                prev
            } else {
                1 + prev.min(cur).min(row[j])
            };
            prev = cur;
        }
    }
    row[b.len()]
}

/// Who a send addresses. A group send leaves the sender off the list.
pub fn recipients(
    from: &str,
    seat: Option<&str>,
    group: Option<&str>,
    members: &[String],
) -> Result<Vec<String>> {
    if let Some(group) = group {
        if seat.is_some() {
            bail!("send: name a seat or a group");
        }
        let to: Vec<String> = members
            .iter()
            .filter(|m| m.as_str() != from)
            .cloned()
            .collect();
        if to.is_empty() {
            bail!("send: {group} has no other member");
        }
        return Ok(to);
    }
    let Some(seat) = seat.map(str::trim).filter(|s| !s.is_empty()) else {
        bail!("send: name a seat or a group");
    };
    let seat = check_name("send", seat)?;
    if seat == from {
        bail!("send: {seat} is this seat");
    }
    Ok(vec![seat])
}

fn is_mail_atom(atom: &Value) -> bool {
    matches!(atom["kind"].as_str(), Some("message" | "receipt" | "group"))
}

fn writer_version(client: &packset_client::PacksetClient) -> Option<String> {
    client
        .status(Some(&client.workspace()))
        .ok()
        .and_then(|status| {
            status
                .get("version")
                .and_then(Value::as_str)
                .filter(|version| !version.is_empty())
                .map(str::to_string)
        })
}

/// Fail when this writer does not keep mail. An empty message is not
/// stored: the kind is refused, or the text is.
fn ensure_mail(client: &packset_client::PacksetClient, verb: &str) -> Result<()> {
    // The writer's version decides when it gives one. packset 0.12.1 checks
    // the text before the kind, so an empty probe reads the same from a
    // writer that keeps mail and from one that refuses it.
    if let Some(version) = writer_version(client) {
        match super::cmp_semver(&version, MAIL_PACKSET) {
            Some(std::cmp::Ordering::Less) => {
                bail!("{verb}: {}", unsupported_mail(Some(&version)));
            }
            Some(_) => return Ok(()),
            None => {}
        }
    }
    let probe = json!({
        "schema": "inside.atom/v1",
        "kind": "message",
        "level": "explicit",
        "text": "",
        "workspace": workspace_of(Some(&client.workspace())),
    });
    let posted = super::with_writer(|| client.post_atom(&probe).map_err(anyhow::Error::from));
    match posted {
        Ok(_) => Ok(()),
        Err(err) => match probe_of(&format!("{err:#}")) {
            Probe::Kept => Ok(()),
            Probe::Refused => {
                let version = writer_version(client);
                bail!("{verb}: {}", unsupported_mail(version.as_deref()));
            }
            Probe::Store => Err(err).context("mail: the pack did not answer"),
        },
    }
}

fn listed() -> Result<Vec<Value>> {
    let client = super::pack()?;
    let workspace = client.workspace();
    super::atoms_lean(&client, &workspace).context("mail: the pack did not answer")
}

fn load(verb: &str) -> Result<Vec<Value>> {
    let client = super::pack()?;
    let atoms = listed()?;
    if !atoms.iter().any(is_mail_atom) {
        ensure_mail(&client, verb)?;
    }
    Ok(atoms)
}

fn load_quiet() -> Option<Vec<Value>> {
    let client = super::pack().ok()?;
    let workspace = client.workspace();
    super::atoms_lean(&client, &workspace).ok()
}

fn post(atom: &Value, verb: &str) -> Result<Value> {
    let client = super::pack()?;
    super::with_writer(|| {
        client
            .post_atom(atom)
            .with_context(|| format!("{verb}: POST /v1/atoms failed"))
    })
}

fn workspace_of(client_workspace: Option<&str>) -> String {
    client_workspace.unwrap_or("seat").to_string()
}

fn stamp_source(atom: &mut Value) {
    atom["source"] = super::atom_source();
}

/// Write one message and print its id.
pub fn send(args: &Send<'_>) -> Result<String> {
    let text = args.text.trim();
    if text.is_empty() {
        bail!("send: empty text");
    }
    check_length("send", text)?;
    let from = super::seat_name();
    let group = args
        .group
        .map(|name| check_name("send", name))
        .transpose()?;
    let client = super::pack()?;
    // A named seat is checked before the pack, so "this seat" stays that
    // error. A group has no roster until the pack answers.
    let atoms = if group.is_some() {
        ensure_mail(&client, "send")?;
        listed()?
    } else {
        Vec::new()
    };
    let roster = group
        .as_deref()
        .map(|name| members(&atoms, name))
        .unwrap_or_default();
    let to = recipients(from.as_str(), args.seat, group.as_deref(), &roster)?;
    if group.is_none() {
        ensure_mail(&client, "send")?;
        if !args.new_seat {
            let known = known_seats(&listed()?);
            for seat in &to {
                check_known(seat, &known, false)?;
            }
        }
    }
    let issue = args
        .issue
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(|s| check_name("send", s))
        .transpose()?;
    let scope = issue.as_deref().and_then(super::sync::scope_for_issue);
    let outgoing = Outgoing {
        id: mint_id("msg", &[&from, text]),
        from,
        to: to.clone(),
        body: text.to_string(),
        interrupt: args.interrupt,
        issue,
        group,
        reply: None,
        scope,
    };
    let mut atom = message_atom(&outgoing, &workspace_of(Some(&client.workspace())));
    stamp_source(&mut atom);
    post(&atom, "send")?;
    let mut said = format!("sent {} to {}\n", outgoing.id, to.join(", "));
    said.push_str(&wake_after(&outgoing));
    Ok(said)
}

/// Start the wake of each recipient that set one, and say which. The
/// message is already written, so a wake that fails changes nothing here.
fn wake_after(m: &Outgoing) -> String {
    let events: Vec<super::wake::Event> =
        m.to.iter()
            .map(|seat| super::wake::Event {
                seat: seat.clone(),
                from: m.from.clone(),
                msg: m.id.clone(),
                issue: m.issue.clone(),
                interrupt: m.interrupt,
            })
            .collect();
    let woken = super::wake::spawn(&events);
    if woken.is_empty() {
        String::new()
    } else {
        format!("waking {}\n", woken.join(", "))
    }
}

/// Unread mail for this seat, then receipts on what it sent that it has not
/// been shown yet.
///
/// Listing does not write a receipt. `ljos hook --prompt` does, for each
/// message it shows. A pack that does not answer, or that refuses mail, is
/// an error. `ljos_inbox` calls this.
pub fn inbox(all: bool) -> Result<String> {
    let atoms = load("inbox")?;
    let seat = super::seat_name();
    let seen = receipt_seen();
    let (text, keys) = render_inbox(&atoms, &seat, all, &seen);
    remember_receipts(&keys);
    Ok(text)
}

fn already_read(atoms: &[Value], msg: &str, seat: &str) -> bool {
    readers_of(atoms).contains(&(msg.to_string(), seat.to_string()))
}

fn receipt_atom(msg: &Letter, reader: &str, at: &str, workspace: &str) -> Value {
    let id = mint_id("receipt", &[msg.id.as_str(), reader, at]);
    let text = format!("read {}\nby {reader}\nat {at}\nmsg {id}", msg.id);
    let mut entities = vec![
        format!("seat:{reader}"),
        format!("from:{reader}"),
        format!("to:{}", msg.from),
        format!("msg:{id}"),
        format!("receipt:{}", msg.id),
    ];
    if let Some(scope) = &msg.scope {
        entities.push(format!("scope:{scope}"));
    }
    json!({
        "schema": "inside.atom/v1",
        "kind": "receipt",
        "level": "explicit",
        "text": text,
        "workspace": workspace,
        "entities": entities,
    })
}

/// Write a receipt for one message. A second read of the same message
/// writes nothing.
pub fn read(id: &str) -> Result<String> {
    let id = id.trim();
    if id.is_empty() {
        bail!("read: name a message");
    }
    let atoms = load("read")?;
    let seat = super::seat_name();
    let letter = find_letter(&atoms, id).context("read")?;
    if !letter.to.iter().any(|to| to == &seat) {
        bail!("read: {id} is not for {seat}");
    }
    if already_read(&atoms, id, &seat) {
        return Ok(format!("read {id} already\n"));
    }
    let client = super::pack()?;
    let mut atom = receipt_atom(
        &letter,
        &seat,
        &super::now_utc(),
        &workspace_of(Some(&client.workspace())),
    );
    stamp_source(&mut atom);
    post(&atom, "read")?;
    Ok(format!("read {id}\n"))
}

fn read_ids(ids: &[String]) {
    let Some(atoms) = load_quiet() else {
        return;
    };
    let seat = super::seat_name();
    let Ok(client) = super::pack() else {
        return;
    };
    let workspace = client.workspace();
    let at = super::now_utc();
    for id in ids {
        let Ok(letter) = find_letter(&atoms, id) else {
            continue;
        };
        if !letter.to.iter().any(|to| to == &seat) || already_read(&atoms, id, &seat) {
            continue;
        }
        let mut atom = receipt_atom(&letter, &seat, &at, &workspace_of(Some(&workspace)));
        stamp_source(&mut atom);
        let _ = post(&atom, "read");
    }
}

/// Answer a message. The reply goes to its sender and keeps its issue.
pub fn reply(id: &str, text: &str, interrupt: bool) -> Result<String> {
    let id = id.trim();
    let text = text.trim();
    if id.is_empty() {
        bail!("reply: name a message");
    }
    if text.is_empty() {
        bail!("reply: empty text");
    }
    check_length("reply", text)?;
    let atoms = load("reply")?;
    let parent = find_letter(&atoms, id).context("reply")?;
    let from = super::seat_name();
    if parent.from == from {
        bail!("reply: {id} is from this seat");
    }
    let issue = parent.issue.clone();
    let scope = issue
        .as_deref()
        .and_then(super::sync::scope_for_issue)
        .or(parent.scope);
    let outgoing = Outgoing {
        id: mint_id("msg", &[&from, id, text]),
        from,
        to: vec![parent.from.clone()],
        body: text.to_string(),
        interrupt,
        issue,
        group: None,
        reply: Some(parent.id.clone()),
        scope,
    };
    let client = super::pack()?;
    let mut atom = message_atom(&outgoing, &workspace_of(Some(&client.workspace())));
    stamp_source(&mut atom);
    post(&atom, "reply")?;
    let mut line = format!(
        "sent {} to {}\nreply {}\n",
        outgoing.id, parent.from, parent.id
    );
    if let Some(issue) = &outgoing.issue {
        line.push_str(&format!("issue {issue}\n"));
    }
    line.push_str(&wake_after(&outgoing));
    Ok(line)
}

fn group_atom(name: &str, seat: &str, drop: bool, by: &str, workspace: &str) -> Value {
    let at = super::now_utc();
    let id = mint_id("group", &[name, seat, by, &at]);
    let verb = if drop { "drop" } else { "member" };
    let text = format!("group {id}\nname {name}\n{verb} {seat}\nby {by}\nat {at}");
    let mut entities = vec![
        format!("seat:{by}"),
        format!("group:{name}"),
        format!("msg:{id}"),
    ];
    if drop {
        entities.push(format!("drop:{seat}"));
    } else {
        entities.push(format!("member:{seat}"));
    }
    json!({
        "schema": "inside.atom/v1",
        "kind": "group",
        "level": "explicit",
        "text": text,
        "workspace": workspace,
        "entities": entities,
    })
}

/// List a group, or add and remove members. A drop is its own atom.
pub fn group(name: &str, add: &[String], remove: &[String]) -> Result<String> {
    let name = check_name("group", name)?;
    if add.is_empty() && remove.is_empty() {
        let atoms = load("group")?;
        let seats = members(&atoms, &name);
        if seats.is_empty() {
            return Ok(format!("{name}\t(no members)\n"));
        }
        let mut out = String::new();
        for seat in seats {
            out.push_str(&format!("{name}\t{seat}\n"));
        }
        return Ok(out);
    }
    let atoms = load("group")?;
    let by = super::seat_name();
    let client = super::pack()?;
    let workspace = client.workspace();
    let mut roster = members(&atoms, &name);
    let mut out = String::new();
    for seat in add {
        let seat = check_name("group", seat)?;
        if roster.iter().any(|s| s == &seat) {
            out.push_str(&format!("{seat} is already in {name}\n"));
            continue;
        }
        let mut atom = group_atom(&name, &seat, false, &by, &workspace_of(Some(&workspace)));
        stamp_source(&mut atom);
        post(&atom, "group")?;
        roster.push(seat.clone());
        out.push_str(&format!("added {seat} to {name}\n"));
    }
    for seat in remove {
        let seat = check_name("group", seat)?;
        if !roster.iter().any(|s| s == &seat) {
            out.push_str(&format!("{seat} is not in {name}\n"));
            continue;
        }
        let mut atom = group_atom(&name, &seat, true, &by, &workspace_of(Some(&workspace)));
        stamp_source(&mut atom);
        post(&atom, "group")?;
        roster.retain(|s| s != &seat);
        out.push_str(&format!("removed {seat} from {name}\n"));
    }
    Ok(out)
}

fn state_dir() -> PathBuf {
    std::env::var_os("XDG_STATE_HOME")
        .map(PathBuf::from)
        .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
        .unwrap_or_else(|| PathBuf::from(".local/state"))
        .join("ljos")
        .join("mail")
}

fn receipt_seen_path() -> PathBuf {
    state_dir().join("receipts-seen")
}

fn receipt_seen() -> BTreeSet<String> {
    std::fs::read_to_string(receipt_seen_path())
        .map(|text| {
            text.lines()
                .map(str::trim)
                .filter(|l| !l.is_empty())
                .map(str::to_string)
                .collect()
        })
        .unwrap_or_default()
}

fn remember_receipts(keys: &[String]) {
    if keys.is_empty() {
        return;
    }
    let path = receipt_seen_path();
    if let Some(dir) = path.parent() {
        let _ = std::fs::create_dir_all(dir);
    }
    let mut text = std::fs::read_to_string(&path).unwrap_or_default();
    for key in keys {
        if text.lines().any(|l| l == key) {
            continue;
        }
        text.push_str(key);
        text.push('\n');
    }
    let _ = std::fs::write(path, text);
}

/// What the prompt hook prints when the pack does not answer.
///
/// One line. The hook still exits 0, so the prompt is not blocked.
pub const MAIL_UNCHECKED: &str = "mail could not be checked";

/// Unread mail and unseen receipts for the prompt hook. A pack that does
/// not answer leaves one line, so unread mail is not silent.
#[must_use]
pub fn prompt_note() -> (String, Vec<String>) {
    let Some(atoms) = load_quiet() else {
        return (MAIL_UNCHECKED.to_string(), Vec::new());
    };
    let seat = super::seat_name();
    let letters = inbox_letters(&atoms, &seat, false);
    let seen = receipt_seen();
    let receipts: Vec<Receipt> = receipts_for(&atoms, &seat)
        .into_iter()
        .filter(|r| !seen.contains(&r.key))
        .collect();
    prompt_lines(&letters, &receipts)
}

/// The line a held stop opens with, before the mail it carries.
pub const STOP_MAIL_HEAD: &str =
    "Mail marked interrupt reached this seat while you worked. Read it before you stop:";

/// Unread interrupt mail for a stop hook, and the ids to receipt once the
/// runner has it. A running session takes it as one more round before it
/// goes idle, so a sender's interrupt does not wait for a prompt that may
/// never come. Plain mail waits for the next prompt or the seat's wake. A
/// pack that does not answer says nothing here: holding a stop on an
/// error would keep every turn open.
#[must_use]
pub fn stop_note() -> (String, Vec<String>) {
    let Some(atoms) = load_quiet() else {
        return (String::new(), Vec::new());
    };
    stop_lines(&atoms, &super::seat_name())
}

/// [`stop_note`] over atoms already read.
#[must_use]
pub fn stop_lines(atoms: &[Value], seat: &str) -> (String, Vec<String>) {
    let hot: Vec<Letter> = inbox_letters(atoms, seat, false)
        .into_iter()
        .filter(|l| l.interrupt)
        .collect();
    let (lines, ids) = prompt_lines(&hot, &[]);
    if lines.is_empty() {
        return (String::new(), Vec::new());
    }
    let reply = "Answer with `ljos reply ID TEXT`.";
    (format!("{STOP_MAIL_HEAD}\n{lines}\n{reply}"), ids)
}

/// Receipts for messages a runner was just shown, and a local mark for
/// receipts this seat was told about. A pack that does not answer leaves
/// the message unread, so the next prompt shows it again.
pub fn ack_delivered(ids: &[String]) {
    let mut reads = Vec::new();
    let mut seen = Vec::new();
    for id in ids {
        if let Some(msg) = id.strip_prefix("mail:") {
            reads.push(msg.to_string());
        } else if let Some(key) = id.strip_prefix("rcpt:") {
            seen.push(key.to_string());
        }
    }
    if !reads.is_empty() {
        read_ids(&reads);
    }
    remember_receipts(&seen);
}

#[cfg(test)]
mod tests {
    use super::*;

    fn outgoing(id: &str, from: &str, to: &[&str], body: &str) -> Outgoing {
        Outgoing {
            id: id.to_string(),
            from: from.to_string(),
            to: to.iter().map(|s| (*s).to_string()).collect(),
            body: body.to_string(),
            interrupt: false,
            issue: None,
            group: None,
            reply: None,
            scope: None,
        }
    }

    #[test]
    fn a_long_body_fits_the_pack_and_reads_back_whole() {
        let long = "the aarch64 log says the Highway test times out. ".repeat(40);
        let m = outgoing("msg-1", "saddle", &["inky"], &long);
        let atom = message_atom(&m, "seat");
        let text = atom["text"].as_str().unwrap();
        assert!(
            text.chars().count() <= ATOM_TEXT_CAP,
            "{}",
            text.chars().count()
        );
        assert!(text.ends_with('\u{2026}'), "{text}");
        assert!(
            text.starts_with("msg msg-1\nfrom saddle\nto inky\n\n"),
            "{text}"
        );
        assert_eq!(atom["body"].as_str(), Some(long.trim()));
        let letter = letter_of(&atom, 0).expect("a letter");
        assert_eq!(letter.body, long.trim());
        // A short body has no `body` field and reads back from the text.
        let short = message_atom(&outgoing("msg-2", "saddle", &["inky"], "ok"), "seat");
        assert!(short.get("body").is_none());
        assert_eq!(letter_of(&short, 0).unwrap().body, "ok");
        // Past the mail cap, a refusal that names the deed route.
        let err = check_length("send", &"x".repeat(MAIL_BODY_CAP + 1))
            .unwrap_err()
            .to_string();
        assert!(err.contains("deedar create file"), "{err}");
        assert!(check_length("send", &"x".repeat(MAIL_BODY_CAP)).is_ok());
        let shown = prompt_body(&long);
        assert!(shown.contains("`ljos inbox --all`"), "{shown}");
        assert_eq!(prompt_body("ok"), "ok");
    }

    fn atom(m: &Outgoing, ts: &str) -> Value {
        let mut v = message_atom(m, "seat");
        v["ts"] = json!(ts);
        v
    }

    #[test]
    fn the_same_words_stay_two_messages() {
        let a = outgoing("a", "alice", &["bob"], "ok");
        let b = outgoing("b", "carol", &["bob"], "ok");
        let left = message_text(&a);
        let right = message_text(&b);
        assert_ne!(left, right);
        assert_ne!(super::super::work_id(&left), super::super::work_id(&right));
        assert_eq!(body_of(&left), "ok");
        let atoms = [
            atom(&a, "2026-10-09T00:00:00Z"),
            atom(&b, "2026-10-09T00:00:01Z"),
        ];
        let plain = super::super::sync::log_plaintext(&atoms, "seat", "seat");
        assert_eq!(plain.lines().count(), 2, "{plain}");
        let got = super::super::sync::incoming(&plain, "other", &BTreeSet::new(), &BTreeSet::new());
        assert_eq!(got.post.len(), 2);
    }

    #[test]
    fn inbox_is_the_seat_named_and_an_interrupt_leads() {
        let mut late = outgoing("late", "alice", &["bob"], "when you can");
        late.issue = Some("acme-4".into());
        let mut hot = outgoing("hot", "carol", &["bob"], "the build is red");
        hot.interrupt = true;
        hot.issue = Some("acme-4".into());
        let other = outgoing("other", "alice", &["dave"], "not for bob");
        let atoms = [
            atom(&late, "2026-10-09T00:00:00Z"),
            atom(&hot, "2026-10-09T00:00:02Z"),
            atom(&other, "2026-10-09T00:00:01Z"),
        ];
        let letters = inbox_letters(&atoms, "bob", false);
        assert_eq!(
            letters.iter().map(|l| l.id.as_str()).collect::<Vec<_>>(),
            ["hot", "late"]
        );
        assert!(letters[0].interrupt);
        assert_eq!(letters[1].issue.as_deref(), Some("acme-4"));
        let (text, ids) = prompt_lines(&letters, &[]);
        assert!(
            text.starts_with("mail:\n! hot from carol on acme-4:"),
            "{text}"
        );
        assert!(ids.contains(&"mail:hot".to_string()), "{ids:?}");
        assert!(text.contains("late from alice on acme-4:"));
    }

    #[test]
    fn a_receipt_takes_the_message_out_of_the_unread_list() {
        let msg = outgoing("m1", "alice", &["bob"], "look at the log");
        let message = atom(&msg, "2026-10-09T00:00:00Z");
        let receipt = json!({
            "kind": "receipt",
            "text": "read m1\nby bob\nat 2026-10-09T00:01:00Z\nmsg r1",
            "ts": "2026-10-09T00:01:00Z",
            "entities": ["seat:bob", "from:bob", "to:alice", "receipt:m1", "msg:r1"],
        });
        let held = [message, receipt];
        let unread = inbox_letters(&held, "bob", false);
        assert!(unread.is_empty());
        let all = inbox_letters(&held, "bob", true);
        assert!(all[0].read);
        let (text, keys) = render_inbox(&held, "alice", false, &BTreeSet::new());
        assert!(
            text.contains("receipt\tm1\tbob\t2026-10-09T00:01:00Z"),
            "{text}"
        );
        assert_eq!(keys.len(), 1);
        let (again, _) = render_inbox(&held, "alice", false, &keys.into_iter().collect());
        assert_eq!(again, "inbox: nothing unread\n");
    }

    #[test]
    fn a_reply_keeps_the_issue_and_names_the_sender() {
        let mut parent = outgoing("p", "alice", &["bob"], "the build is red");
        parent.issue = Some("acme-4".into());
        parent.scope = Some("tools".into());
        let letter = letter_of(&atom(&parent, "2026-10-09T00:00:00Z"), 0).unwrap();
        let reply = Outgoing {
            id: "r".into(),
            from: "bob".into(),
            to: vec![letter.from.clone()],
            body: "looking".into(),
            interrupt: false,
            issue: letter.issue.clone(),
            group: None,
            reply: Some(letter.id.clone()),
            scope: letter.scope.clone(),
        };
        let atom = message_atom(&reply, "seat");
        let back = letter_of(&atom, 1).unwrap();
        assert_eq!(back.to, ["alice"]);
        assert_eq!(back.reply.as_deref(), Some("p"));
        assert_eq!(back.issue.as_deref(), Some("acme-4"));
        assert_eq!(back.scope.as_deref(), Some("tools"));
        assert!(super::super::sync::is_own(&atom));
    }

    #[test]
    fn a_group_send_reaches_the_other_members() {
        let atoms = [
            json!({
                "kind": "group",
                "ts": "2026-10-09T00:00:00Z",
                "entities": ["group:ops", "member:alice"],
            }),
            json!({
                "kind": "group",
                "ts": "2026-10-09T00:00:01Z",
                "entities": ["group:ops", "member:bob"],
            }),
            json!({
                "kind": "group",
                "ts": "2026-10-09T00:00:02Z",
                "entities": ["group:ops", "member:carol"],
            }),
            json!({
                "kind": "group",
                "ts": "2026-10-09T00:00:03Z",
                "entities": ["group:ops", "drop:carol"],
            }),
        ];
        assert_eq!(members(&atoms, "ops"), ["alice", "bob"]);
        let to = recipients("alice", None, Some("ops"), &members(&atoms, "ops")).unwrap();
        assert_eq!(to, ["bob"]);
        let err = recipients("alice", None, Some("ops"), &["alice".into()]).unwrap_err();
        assert!(err.to_string().contains("no other member"), "{err}");
        let err = recipients("alice", Some("alice"), None, &[]).unwrap_err();
        assert!(err.to_string().contains("this seat"), "{err}");
    }

    #[test]
    fn a_prompt_caps_the_note_and_keeps_the_rest_unread() {
        let letters: Vec<Letter> = (0..10)
            .map(|n| Letter {
                id: format!("m{n}"),
                from: "alice".into(),
                to: vec!["bob".into()],
                body: format!("line {n}"),
                interrupt: n == 9,
                issue: None,
                group: None,
                reply: None,
                scope: None,
                read: false,
                ts: format!("2026-10-09T00:00:{n:02}Z"),
                seq: n,
            })
            .collect();
        let mut ordered = letters.clone();
        ordered.sort_by(|a, b| b.interrupt.cmp(&a.interrupt).then(a.ts.cmp(&b.ts)));
        let (text, ids) = prompt_lines(&ordered, &[]);
        assert!(text.lines().next().unwrap() == "mail:");
        assert!(text.contains("! m9 from alice:"));
        assert!(text.contains("2 more in ljos inbox"), "{text}");
        assert_eq!(ids.iter().filter(|id| id.starts_with("mail:")).count(), 8);
        assert!(!ids.iter().any(|id| id == "mail:m7" || id == "mail:m8"));
    }

    #[test]
    fn a_misspelt_seat_is_answered_with_the_seat_it_meant() {
        let known: BTreeSet<String> = ["ljos-bot", "saddle", "inky"]
            .iter()
            .map(|s| (*s).to_string())
            .collect();
        assert_eq!(closest_seat("ljos", &known), Some("ljos-bot"));
        assert_eq!(closest_seat("sadle", &known), Some("saddle"));
        assert_eq!(closest_seat("eessi", &known), None);
        let err = check_known("ljos", &known, false).unwrap_err().to_string();
        assert!(err.contains("Did you mean ljos-bot?"), "{err}");
        let err = check_known("eessi", &known, false).unwrap_err().to_string();
        assert!(!err.contains("Did you mean"), "{err}");
        assert!(check_known("eessi", &known, true).is_ok());
    }

    #[test]
    fn a_stop_carries_unread_interrupt_mail_only() {
        let plain = outgoing("plain", "grok", &["claude"], "when you can");
        let mut hot = outgoing("hot", "grok", &["claude"], "take Software-d9n6 now");
        hot.interrupt = true;
        hot.issue = Some("Software-d9n6".into());
        let atoms = [
            atom(&plain, "2026-10-11T00:00:00Z"),
            atom(&hot, "2026-10-11T00:00:01Z"),
        ];
        let (text, ids) = stop_lines(&atoms, "claude");
        assert!(text.starts_with(STOP_MAIL_HEAD), "{text}");
        assert!(
            text.contains("! hot from grok on Software-d9n6: take Software-d9n6 now"),
            "{text}"
        );
        assert!(!text.contains("when you can"), "{text}");
        assert_eq!(ids, ["mail:hot"]);
        let read = json!({
            "kind": "receipt",
            "text": "read hot\nby claude\nat 2026-10-11T00:00:02Z\nmsg r1",
            "entities": ["seat:claude", "from:claude", "to:grok", "receipt:hot", "msg:r1"],
        });
        let after = [atoms[0].clone(), atoms[1].clone(), read];
        assert_eq!(stop_lines(&after, "claude"), (String::new(), Vec::new()));
        assert_eq!(stop_lines(&atoms, "grok"), (String::new(), Vec::new()));
    }

    #[test]
    fn an_unknown_kind_is_a_pack_without_mail() {
        let refused =
            "bad response: http://127.0.0.1:8761/v1/atoms: 400: unknown atom kind: message";
        assert_eq!(probe_of(refused), Probe::Refused);
        let kept = "bad response: http://127.0.0.1:8761/v1/atoms: 400: atom text is required";
        assert_eq!(probe_of(kept), Probe::Kept);
        let down = "http://127.0.0.1:8761/v1/atoms: http: connection refused";
        assert_eq!(probe_of(down), Probe::Store);
        let named = unsupported_mail(Some("0.12.1"));
        assert!(named.contains("packset 0.12.1"), "{named}");
        assert!(named.contains(MAIL_PACKSET), "{named}");
        assert!(named.contains("kind message"), "{named}");
        let quiet = unsupported_mail(None);
        assert!(quiet.contains(MAIL_PACKSET), "{quiet}");
        assert!(quiet.contains("refused kind message"), "{quiet}");
    }
}