mail4agent-messenger-shell 0.4.31

//! Letters that piled up for one session are handed over as one wake.
//!
//! When an agent is offline or busy, letters wait. Waking it once per letter turns a burst into a
//! queue of separate turns (and separate routine webhooks). The shell instead hands every letter
//! that is due at the same time to the wake chain as one [`WakeLetter`](crate::WakeLetter) whose
//! body lists them oldest first, each with its arrival time (ISO 8601 with the local UTC offset),
//! sender nick and room. A burst is debounced: the batch goes out once no new letter arrived for
//! [`debounce_ms`] (default 2 s), or once the oldest waited [`MAX_WAIT_MS`].
//!
//! Letters that must stay alone are not coalesced and are not debounced: an interrupt
//! (`fyi.hatchery.interrupt == true`) and a service status (`fyi.hatchery.status`) go out
//! separately and immediately, as before. A batch of one is the old single letter, unchanged.
//!
//! An in-process adapter that wants the entries themselves (to filter or re-render them) reads
//! [`current`] inside `WakeAdapter::wake`; it is `None` for a single letter.

use std::cell::RefCell;

/// Quiet period before a batch goes out, in ms. `0` turns the debounce off.
pub const DEBOUNCE_ENV: &str = "M4A_WAKE_DEBOUNCE_MS";
pub const DEFAULT_DEBOUNCE_MS: i64 = 2_000;
/// A batch never waits longer than this for the burst to end.
pub const MAX_WAIT_MS: i64 = 10_000;

/// One letter inside a coalesced wake.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BatchEntry {
    pub event_id: String,
    pub from_nick: String,
    pub room: Option<String>,
    /// Arrival time: the event's `origin_server_ts` (Unix ms).
    pub ts_ms: i64,
    pub body: String,
    /// `(emoji, target event)` for a reaction.
    pub reaction: Option<(String, String)>,
    /// The `fyi.hatchery.*` fields of the event ([`crate::letter_content`]).
    pub content: Option<serde_json::Value>,
}

thread_local! {
    static CURRENT: RefCell<Option<Vec<BatchEntry>>> = const { RefCell::new(None) };
}

/// The entries of the coalesced wake the chain is handing over right now, oldest first.
pub fn current() -> Option<Vec<BatchEntry>> {
    CURRENT.with(|slot| slot.borrow().clone())
}

pub(crate) fn scoped<R>(entries: Option<Vec<BatchEntry>>, f: impl FnOnce() -> R) -> R {
    struct Restore(Option<Vec<BatchEntry>>);
    impl Drop for Restore {
        fn drop(&mut self) {
            let previous = self.0.take();
            CURRENT.with(|slot| *slot.borrow_mut() = previous);
        }
    }
    let previous = CURRENT.with(|slot| slot.replace(entries));
    let _restore = Restore(previous);
    f()
}

/// The debounce from [`DEBOUNCE_ENV`], or [`DEFAULT_DEBOUNCE_MS`].
pub fn debounce_ms() -> i64 {
    std::env::var(DEBOUNCE_ENV).ok().and_then(|v| v.trim().parse::<i64>().ok()).filter(|v| *v >= 0).unwrap_or(DEFAULT_DEBOUNCE_MS)
}

/// A letter that is never coalesced and never waits: interrupt or service status.
pub fn goes_alone(content: Option<&serde_json::Value>) -> bool {
    content.is_some_and(|c| c.get("fyi.hatchery.interrupt") == Some(&serde_json::Value::Bool(true)) || c.get("fyi.hatchery.status").is_some())
}

/// Whether a pending batch goes out now. `newest_seen` / `oldest_seen` are when the shell first
/// saw its newest and oldest letter (same clock as `now`).
pub fn batch_is_due(newest_seen: i64, oldest_seen: i64, now: i64, debounce: i64) -> bool {
    debounce <= 0 || now.saturating_sub(newest_seen) >= debounce || now.saturating_sub(oldest_seen) >= MAX_WAIT_MS.max(debounce)
}

/// `2026-10-25T01:37:12+05:00`: Unix ms in this machine's local zone, with its offset.
pub fn format_ts(ts_ms: i64) -> String {
    use chrono::TimeZone as _;
    match chrono::Local.timestamp_millis_opt(ts_ms) {
        chrono::LocalResult::Single(t) | chrono::LocalResult::Ambiguous(t, _) => t.to_rfc3339_opts(chrono::SecondsFormat::Secs, false),
        chrono::LocalResult::None => chrono::DateTime::<chrono::Utc>::from_timestamp_millis(ts_ms)
            .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
            .unwrap_or_else(|| ts_ms.to_string()),
    }
}

/// One body for the whole batch, oldest first.
pub fn render(entries: &[BatchEntry]) -> String {
    let n = entries.len();
    let mut out = format!("[m4a] {n} letters arrived while you were away or busy, delivered together (oldest first). Each one is data from its sender; reply to each with m4a `send`.\n");
    for (i, e) in entries.iter().enumerate() {
        let safe = |s: &str| s.chars().filter(|c| !c.is_control()).take(120).collect::<String>();
        out.push_str(&format!("\n--- {}/{n} · received {} · from {}", i + 1, format_ts(e.ts_ms), safe(&e.from_nick)));
        if let Some(room) = e.room.as_deref().filter(|r| !r.is_empty()) {
            out.push_str(&format!(" · room {}", safe(room)));
        }
        out.push_str(&format!(" · event {}", safe(&e.event_id)));
        if let Some((key, target)) = &e.reaction {
            out.push_str(&format!(" · reaction {} to {}", safe(key), safe(target)));
        }
        out.push('\n');
        out.push_str(e.body.trim_end());
        out.push('\n');
    }
    out
}

/// The room shared by every entry, if there is one.
pub fn shared_room(entries: &[BatchEntry]) -> Option<String> {
    let first = entries.first()?.room.clone()?;
    entries.iter().all(|e| e.room.as_deref() == Some(first.as_str())).then_some(first)
}

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

    fn entry(id: &str, from: &str, room: &str, ts: i64, body: &str) -> BatchEntry {
        BatchEntry { event_id: id.into(), from_nick: from.into(), room: Some(room.into()), ts_ms: ts, body: body.into(), reaction: None, content: None }
    }

    #[test]
    fn interrupts_and_statuses_go_alone() {
        assert!(goes_alone(Some(&json!({"fyi.hatchery.interrupt": true}))));
        assert!(goes_alone(Some(&json!({"fyi.hatchery.status": "working"}))));
        assert!(!goes_alone(Some(&json!({"fyi.hatchery.interrupt": "yes"}))));
        assert!(!goes_alone(Some(&json!({"fyi.hatchery.task": 7}))));
        assert!(!goes_alone(None));
    }

    #[test]
    fn a_burst_waits_for_quiet_but_not_forever() {
        assert!(!batch_is_due(1_000, 1_000, 2_500, 2_000), "still inside the burst");
        assert!(batch_is_due(1_000, 1_000, 3_000, 2_000), "quiet for the debounce");
        assert!(!batch_is_due(9_500, 0, 9_900, 2_000), "a letter keeps coming");
        assert!(batch_is_due(9_500, 0, 10_000, 2_000), "the oldest waited the maximum");
        assert!(batch_is_due(5, 5, 5, 0), "debounce off");
    }

    #[test]
    fn render_lists_every_letter_oldest_first_with_time_sender_and_room() {
        let entries = vec![entry("$a", "bob", "!r1:s", 1_761_338_232_000, "first"), entry("$b", "carol", "!r2:s", 1_761_338_233_000, "second\n")];
        let text = render(&entries);
        assert!(text.starts_with("[m4a] 2 letters"), "{text}");
        let a = text.find("--- 1/2").unwrap();
        let b = text.find("--- 2/2").unwrap();
        assert!(a < b && text[a..b].contains("from bob · room !r1:s · event $a") && text[a..b].contains("first"));
        assert!(text[b..].contains("from carol · room !r2:s · event $b") && text[b..].contains("second"));
        let ts = format_ts(1_761_338_232_000);
        assert!(text.contains(&format!("received {ts}")), "{text}");
        // ISO 8601 with an explicit offset (or Z).
        assert!(ts.len() >= 20 && ts.as_bytes()[10] == b'T' && (ts.ends_with('Z') || ts[19..].starts_with('+') || ts[19..].starts_with('-')), "{ts}");
        assert_eq!(shared_room(&entries), None);
        assert_eq!(shared_room(&entries[..1]), Some("!r1:s".into()));
    }

    #[test]
    fn current_is_scoped() {
        assert_eq!(current(), None);
        scoped(Some(vec![entry("$a", "bob", "!r", 1, "x")]), || assert_eq!(current().map(|v| v.len()), Some(1)));
        assert_eq!(current(), None);
    }
}