smolboid/daemon/src/mailloop.rs

//! The mail loop: fetch from the daemon's mailbox, admit by ownership, apply
//! each mail to its feed, and send rate-limited error replies. The apply and
//! retention logic is pure and unit-tested; network glue stays thin.

use fumi_core::account::Account;
use fumi_core::address::Address;
use fumi_core::client;
use fumi_core::store::{now, Store};
use fumi_core::error::Error as FumiError;
use smolboid_core::feed::{Feed, MailLine, Record};
use smolboid_core::envelope;

use crate::config::Config;
use crate::dispatch::{self, Action, Command, Incoming};
use crate::feedstore::FeedStore;
use crate::state::{payload_key, reply_allowed, FeedState, PruneMode, State};

use std::sync::{Mutex, MutexGuard};

/// Everything needed to publish one record, before retention gets a say.
pub struct PendingRecord {
    pub timestamp: String,
    pub text: String,
    pub repost: bool,
    /// The Reply-To the mail carried, when the ordinary-send form was used.
    pub reply_to: Option<String>,
    pub mail: MailLine,
}

/// Failures that draw an error reply (PLAN.md, Management mail).
#[derive(Debug)]
pub enum ApplyError {
    OverQuota(bool),
    Collision(String),
    FeedFull,
    /// The author's Reply-To changed after the feed bound one: published
    /// records would stop reconstructing (PLAN.md, smol:reply_to).
    ReplyToChanged,
}

/// Applies a published record to the feed: quota, uniqueness, retention
/// (pruning in Fifo mode, rejecting in Off mode), then the append.
/// Returns the rendered size after appending, for the 90% warning.
pub fn apply_record(
    feed: &mut Feed,
    fs: &mut FeedState,
    cfg: &Config,
    record: PendingRecord,
    now_unix: i64,
) -> Result<(), ApplyError> {
    State::roll_counters(fs, now_unix);
    let (counter, limit) = if record.repost {
        (&mut fs.reposts, cfg.reposts_per_day)
    } else {
        (&mut fs.posts, cfg.posts_per_day)
    };
    if *counter >= limit {
        return Err(ApplyError::OverQuota(record.repost));
    }
    if feed.has_timestamp(&record.timestamp) {
        // The signature pins the time, so a bump is impossible; the author
        // resends a second later (post rule 2).
        return Err(ApplyError::Collision(record.timestamp));
    }

    // The Reply-To must be the author's own address (key = sender), and the
    // verifier derives it from `smol:author` plus the record's sender key. The
    // first publishing mail adopts the author's own spelling of the address;
    // a later change cannot be adopted without invalidating old records.
    if let Some(reply_to) = &record.reply_to {
        let Ok(addr) = Address::parse(reply_to) else {
            return Err(ApplyError::ReplyToChanged);
        };
        if addr.identity != Some(record.mail.sender) || addr.short() != feed.author().unwrap_or_default() {
            if feed.entries().is_empty() {
                feed.set_meta("smol:author", &addr.short());
            } else {
                return Err(ApplyError::ReplyToChanged);
            }
        }
    }

    let entry = if record.repost {
        Record::Repost {
            timestamp: record.timestamp.clone(),
            args: record.text.clone(),
        }
    } else {
        Record::Post {
            timestamp: record.timestamp.clone(),
            text: record.text.clone(),
        }
    };
    let rendered = entry.render();
    let addition = rendered.len() + 1 + record.mail.render().len() + 1;
    let mut size = feed.render().len() + addition;
    if size > cfg.cap_bytes {
        if fs.prune == PruneMode::Off {
            return Err(ApplyError::FeedFull);
        }
        // FIFO: drop oldest records until the new one fits. If even an empty
        // feed cannot hold it, the record simply exceeds the cap.
        while size > cfg.cap_bytes {
            let Some(oldest) = feed.entries().first().map(|e| e.record.timestamp().to_owned())
            else {
                return Err(ApplyError::FeedFull);
            };
            feed.delete(&oldest);
            size = feed.render().len() + addition;
        }
    }

    feed.append(entry, Some(record.mail));
    *counter += 1;

    // The 90% warning fires once per growth episode (PLAN.md, Retention).
    let size = feed.render().len();
    if size * 10 >= cfg.cap_bytes * 9 {
        if !fs.warned {
            fs.warned = true;
        }
    } else if fs.warned {
        fs.warned = false;
    }
    Ok(())
}

/// Applies command fields to feed and state. Deletion failure is reported so
/// the caller can reply.
pub fn apply_commands(
    feed: &mut Feed,
    fs: &mut FeedState,
    commands: &[Command],
) -> Result<usize, String> {
    for command in commands {
        match command {
            Command::Nick(nick) => {
                fs.nick = nick.clone();
                feed.set_meta("nick", nick);
            }
            Command::Delete(timestamp) => {
                if !feed.delete(timestamp) {
                    return Err(format!("no record at {timestamp}"));
                }
            }
            Command::Prune(mode) => fs.prune = *mode,
        }
    }
    Ok(commands.len())
}

pub struct Daemon {
    pub cfg: Config,
    pub store: Store,
    pub account: Account,
    pub state: Mutex<State>,
    pub feeds: FeedStore,
}

impl Daemon {
    /// One poll cycle. Returns a human summary line.
    pub fn poll(&self) -> Result<String, String> {
        let summary = client::fetch(&self.store, &self.account, false, false, self.cfg.timeout)
            .map_err(|e| format!("fetch: {e}"))?;
        if summary.stored == 0 {
            return Ok("no new mail".into());
        }
        let mut applied = 0;
        let mut dropped = 0;
        let mut replies: Vec<(String, String, String)> = Vec::new(); // (address, subject, body)

        let inbox = self
            .store
            .mail("inbox")
            .map_err(|e| format!("inbox: {e}"))?;
        for stored in &inbox {
            let outcome = self.process(stored.id, &stored.envelope, &mut replies);
            let _ = self.store.delete_local("inbox", &stored.id);
            match outcome {
                Ok(true) => applied += 1,
                Ok(false) => dropped += 1,
                Err(e) => eprintln!("mail: {e}"),
            }
        }
        self.send_replies(replies)?;
        Ok(format!(
            "{} applied, {} dropped",
            applied, dropped
        ))
    }

    /// Processes one stored envelope end to end. Ok(true) when the feed or
    /// state changed.
    fn process(
        &self,
        id: [u8; 32],
        envelope: &[u8],
        replies: &mut Vec<(String, String, String)>,
    ) -> Result<bool, String> {
        let opened = envelope::open(self.account.keys(), envelope, now())
            .map_err(|e| format!("open {}: {e}", &fumi_core::crypto::b32(&id)[..8]))?;
        let incoming = Incoming {
            sender: opened.sender,
            epk: opened.epk,
            sig: opened.sig,
            time: opened.time,
            body: opened.body,
        };

        let mut state = self.state.lock().unwrap();
        // Admission: a feed must exist and its owner must still be accepted.
        // Requests-tier or blocked mail has neither and is dropped without a
        // reply (PLAN.md, Admission and Error replies) — main-tier inbox rows
        // are all `mail("inbox")` ever returns.
        let index = match state.feed_of(&opened.sender) {
            Some(index) => index,
            None => match self.follow_rotation(&incoming, &mut state)? {
                Some(index) => index,
                None => return Ok(false),
            },
        };
        let feed_state = state.feeds[index].clone();
        if !self.accepted(&feed_state.address)? {
            return Ok(false);
        }

        self.apply(&incoming, index, &mut state, replies)
    }

    fn accepted(&self, address: &str) -> Result<bool, String> {
        match self.store.accepted_state(address) {
            Ok(Some(true)) => Ok(true),
            Ok(_) => Ok(false),
            Err(e) => Err(format!("accepted_state: {e}")),
        }
    }

    /// An unknown sender may be a rotated owner: a signed Reply-To names the
    /// address, and a verified chain from the key we hold to this one
    /// (SPEC.md §7) moves the feed to the new key, after which the very mail
    /// that announced the rotation applies as the rotated owner. Anything
    /// else is dropped.
    fn follow_rotation(
        &self,
        incoming: &Incoming,
        state: &mut MutexGuard<'_, State>,
    ) -> Result<Option<usize>, String> {
        let Ok(text) = std::str::from_utf8(&incoming.body) else {
            return Ok(None);
        };
        let (fields, _) = fumi_core::message::parse_frontmatter(text);
        let Some(reply_to) = fields.get("reply-to") else {
            return Ok(None);
        };
        let Ok(addr) = Address::parse(reply_to) else {
            return Ok(None);
        };
        let Some(user_key) = addr.identity else {
            return Ok(None);
        };
        if user_key != incoming.sender {
            return Ok(None);
        }
        let Some(index) = state.feeds.iter().position(|f| f.address == addr.short()) else {
            return Ok(None);
        };
        let Some(old) = state.feeds[index].latest_owner() else {
            return Ok(None);
        };
        let mut session = client::connect(&self.store, &addr, false, self.cfg.timeout)
            .map_err(|e| format!("rotation connect: {e}"))?;
        let resolved = client::resolve(session.transport(), &addr.user)
            .map_err(|e| format!("rotation resolve: {e}"))?;
        session.close();
        let (current, chain) = resolved;
        if !client::walk_chain(&addr.user, &old, &current, &chain) {
            return Ok(None);
        }
        if state.feeds[index].owner_keys.last().map(String::as_str)
            != Some(fumi_core::crypto::b32(&current).as_str())
        {
            state.feeds[index]
                .owner_keys
                .push(fumi_core::crypto::b32(&current));
        }
        Ok(Some(index))
    }

    fn apply(
        &self,
        incoming: &Incoming,
        index: usize,
        state: &mut MutexGuard<'_, State>,
        replies: &mut Vec<(String, String, String)>,
    ) -> Result<bool, String> {
        let username = state.feeds[index].username.clone();
        let address = state.feeds[index].address.clone();
        let mut feed = self.feeds.load(&username)?;
        let before = feed.render();

        let outcome = match dispatch::dispatch(incoming) {
            Ok(Action::Post { timestamp, text, reply_to, mail }) => {
                self.publish(
                    state, index, &mut feed, &address, incoming, replies,
                    PendingRecord { timestamp, text, reply_to, repost: false, mail },
                )?
            }
            Ok(Action::Repost { timestamp, args, reply_to, mail }) => {
                self.publish(
                    state, index, &mut feed, &address, incoming, replies,
                    PendingRecord { timestamp, text: args, reply_to, repost: true, mail },
                )?
            }
            Ok(Action::Commands(commands)) => {
                match apply_commands(&mut feed, &mut state.feeds[index], &commands) {
                    Ok(n) => n > 0,
                    Err(reason) => {
                        self.queue_reply(state, incoming, &address, &reason, replies);
                        false
                    }
                }
            }
            Ok(Action::Ignore) => false,
            Err(reason) => {
                self.queue_reply(state, incoming, &address, &reason, replies);
                false
            }
        };

        let after = feed.render();
        if after != before {
            self.feeds.save(&username, &feed)?;
        }
        // Counters, ledgers and prune modes change even when no record did.
        state
            .save(&self.cfg.data.join("state.json"))
            .map_err(|e| e.to_string())?;
        Ok(outcome)
    }

    #[allow(clippy::too_many_arguments)]
    fn publish(
        &self,
        state: &mut MutexGuard<'_, State>,
        index: usize,
        feed: &mut Feed,
        address: &str,
        incoming: &Incoming,
        replies: &mut Vec<(String, String, String)>,
        record: PendingRecord,
    ) -> Result<bool, String> {
        let was_warned = state.feeds[index].warned;
        match apply_record(feed, &mut state.feeds[index], &self.cfg, record, incoming.time) {
            Ok(()) => {
                // The transition false -> true sends the cap warning once per
                // growth episode (PLAN.md, Retention).
                if state.feeds[index].warned && !was_warned {
                    replies.push((
                        address.to_owned(),
                        "twtxt: feed at 90% of its size cap".into(),
                        format!(
                            "Your feed has reached 90% of the {}-byte cap.\n\n\
                             In hard-limit mode (the default) a full feed rejects new records; \
                             delete some with Twt-Delete, or choose FIFO pruning with \
                             Twt-Prune: fifo, which drops the oldest records to make room.",
                            self.cfg.cap_bytes
                        ),
                    ));
                }
                Ok(true)
            }
            Err(e) => {
                let reason = match &e {
                    ApplyError::OverQuota(repost) => format!(
                        "daily {} quota exceeded; try tomorrow",
                        if *repost { "repost" } else { "post" }
                    ),
                    ApplyError::Collision(ts) => {
                        format!("timestamp {ts} already exists; wait a second and resend")
                    }
                    ApplyError::FeedFull => "feed is at its size cap; delete records with Twt-Delete or choose Twt-Prune: fifo".to_owned(),
                    ApplyError::ReplyToChanged => format!(
                        "your Reply-To does not match the feed's bound author {}; \
                         send from that address or mail the operator",
                        feed.author().unwrap_or_default()
                    ),
                };
                self.queue_reply(state, incoming, address, &reason, replies);
                Ok(false)
            }
        }
    }

    /// One error reply per rejected payload, within the hourly budget.
    fn queue_reply(
        &self,
        state: &mut MutexGuard<'_, State>,
        incoming: &Incoming,
        address: &str,
        reason: &str,
        replies: &mut Vec<(String, String, String)>,
    ) {
        let key = payload_key(&incoming.sender, incoming.time, &incoming.body);
        if !reply_allowed(&mut **state, &incoming.sender, &key, now()) {
            return;
        }
        let text = std::str::from_utf8(&incoming.body)
            .map(|t| t.chars().take(200).collect::<String>())
            .unwrap_or_default();
        replies.push((
            address.to_owned(),
            format!("twtxt: {reason}"),
            format!(
                "Your mail could not be applied.\n\nReason: {reason}\n\nQuoted mail:\n{text}\n"
            ),
        ));
    }

    fn send_replies(&self, replies: Vec<(String, String, String)>) -> Result<(), String> {
        for (address, subject, body) in replies {
            let addr = Address::parse(&address).map_err(|e| format!("reply address: {e}"))?;
            client::send(
                &self.store,
                &self.account,
                &client::SendDraft {
                    address: &addr,
                    text: body,
                    subject: Some(&subject),
                    reply_to: None,
                    headers: &[],
                    anonymous: false,
                    no_pad: true,
                },
                self.cfg.timeout,
            )
            .map_err(|e: FumiError| format!("reply to {address}: {e}"))?;
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Config;
    use crate::state::State;
    use smolboid_core::verify;

    fn feed_state() -> FeedState {
        FeedState {
            username: "tobias".into(),
            address: "tobias@smol.place".into(),
            owner_keys: vec![fumi_core::crypto::b32(&[1; 32])],
            nick: "tobias".into(),
            prune: PruneMode::Off,
            warned: false,
            day: State::day(0),
            posts: 0,
            reposts: 0,
        }
    }

    /// A record renders at roughly 260 bytes: the post line plus a
    /// `# smol:mail` line of three base32 values.
    fn config(cap: usize) -> Config {
        Config {
            cap_bytes: cap,
            ..Config::default()
        }
    }

    fn record(ts: &str, text: &str) -> PendingRecord {
        PendingRecord {
            timestamp: ts.into(),
            text: text.into(),
            repost: false,
            reply_to: None,
            mail: MailLine {
                sender: [1; 32],
                epk: [2; 32],
                sig: [3; 64],
            },
        }
    }

    #[test]
    fn posts_publish_and_verify() {
        let mut feed = smolboid_core::Feed::new();
        let mut fs = feed_state();
        apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:47:00Z", "hello"), 0).unwrap();
        assert_eq!(fs.posts, 1);
        let report = verify(&feed);
        assert_eq!(report.states, vec![smolboid_core::State::Unsigned]);
        assert_eq!(feed.entries().len(), 1);
    }

    #[test]
    fn reposts_publish_as_repost_records() {
        let mut feed = smolboid_core::Feed::new();
        let mut fs = feed_state();
        let record = PendingRecord {
            timestamp: "2026-10-01T12:47:00Z".into(),
            text: "@<dana https://x/d.txt> #abc1234".into(),
            repost: true,
            reply_to: None,
            mail: MailLine { sender: [1; 32], epk: [2; 32], sig: [3; 64] },
        };
        apply_record(&mut feed, &mut fs, &config(1 << 20), record, 0).unwrap();
        let entry = &feed.entries()[0];
        assert!(matches!(&entry.record, Record::Repost { args, .. } if args == "@<dana https://x/d.txt> #abc1234"));
        assert_eq!(fs.reposts, 1);
        assert_eq!(fs.posts, 0);
        assert!(feed.render().starts_with("# smol:repost 2026-10-01T12:47:00Z @<dana https://x/d.txt> #abc1234\n"));
    }

    #[test]
    fn collisions_and_quotas_reject_without_appending() {
        let mut feed = smolboid_core::Feed::new();
        let mut fs = feed_state();
        let cfg = config(1 << 20);
        apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "one"), 0).unwrap();
        assert!(matches!(
            apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "two"), 0),
            Err(ApplyError::Collision(_))
        ));
        assert_eq!(feed.entries().len(), 1);

        let mut fs = feed_state();
        fs.posts = 50;
        assert!(matches!(
            apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:48:00Z", "x"), 0),
            Err(ApplyError::OverQuota(_))
        ));
        // A new UTC day rolls the counters.
        let mut fs = feed_state();
        fs.posts = 50;
        apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:48:00Z", "x"), 1_760_000_000).unwrap();
    }

    #[test]
    fn hard_limit_rejects_and_fifo_prunes_oldest() {
        let mut feed = smolboid_core::Feed::new();
        let mut fs = feed_state();
        let cfg = config(300);
        apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "first post here"), 0).unwrap();
        assert!(matches!(
            apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:48:00Z", "second post"), 0),
            Err(ApplyError::FeedFull)
        ));
        assert_eq!(feed.entries().len(), 1, "off mode removes nothing");

        fs.prune = PruneMode::Fifo;
        apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:48:00Z", "second post"), 0).unwrap();
        let entries = feed.entries();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].record.timestamp(), "2026-10-01T12:48:00Z");
        // The quota counts published posts; pruning does not refund it.
        assert_eq!(fs.posts, 2);
    }

    #[test]
    fn warning_flag_flips_at_ninety_percent() {
        let mut feed = smolboid_core::Feed::new();
        let mut fs = feed_state();
        let cfg = config(1300);
        apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "x"), 0).unwrap();
        assert!(!fs.warned);
        apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:01Z", &"y".repeat(700)), 0).unwrap();
        assert!(fs.warned, "crossing 90% sets the flag once");
        feed.delete("2026-10-01T12:47:01Z");
        apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:02Z", "z"), 0).unwrap();
        assert!(!fs.warned, "shrinking below the threshold resets it");
    }

    #[test]
    fn commands_touch_feed_and_state() {
        let mut feed = smolboid_core::Feed::new();
        let mut fs = feed_state();
        apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:47:00Z", "gone soon"), 0).unwrap();

        apply_commands(
            &mut feed,
            &mut fs,
            &[Command::Nick("toby".into()), Command::Prune(PruneMode::Fifo)],
        )
        .unwrap();
        assert_eq!(fs.nick, "toby");
        assert_eq!(feed.meta("nick"), Some("toby"));
        assert_eq!(fs.prune, PruneMode::Fifo);

        apply_commands(&mut feed, &mut fs, &[Command::Delete("2026-10-01T12:47:00Z".into())]).unwrap();
        assert_eq!(feed.entries().len(), 0);
        assert!(apply_commands(&mut feed, &mut fs, &[Command::Delete("2026-10-01T12:47:00Z".into())]).is_err());
    }
}