//! Daemon-side handlers for `AIMX/1` state-mutation verbs.
//!
//! `MARK-READ` and `MARK-UNREAD` rewrite the target email's TOML
//! frontmatter in place so the `read` field persists across restarts.
//! Files on disk are mode `0o600` owned by the mailbox's configured
//! owner (`<owner>:<owner>`), sitting inside `0o700` per-mailbox
//! directories. The MCP server runs as the invoking non-root user,
//! which is generally not the mailbox owner, so it cannot write the
//! file directly and routes through these verbs instead. The daemon
//! (running as root) performs the rewrite and re-applies
//! `<owner>:<owner> 0o600` via `chown_as_owner` on the temp file
//! before `rename(2)`.
//!
//! # Concurrency model
//!
//! MARK-*, inbound ingest, and MAILBOX-* all acquire the **same**
//! per-mailbox `tokio::sync::Mutex<()>` from the shared
//! [`crate::mailbox_locks::MailboxLocks`] map before touching any file
//! under that mailbox's tree.
//!
//! Lock hierarchy (see [`crate::mailbox_locks`] for the full rationale):
//!
//! ```text
//!   outer: per-mailbox mailbox_locks::MailboxLocks  (tokio::sync::Mutex)
//!   inner: process-wide mailbox_handler::CONFIG_WRITE_LOCK (std::sync::Mutex)
//! ```
//!
//! Always acquire outer → inner, never the reverse.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use crate::config::ConfigHandle;
use crate::frontmatter::InboundFrontmatter;
use crate::mailbox_locks::{MailboxLocks, MailboxState};
use crate::mcp::resolve_email_path_strict;
use crate::ownership::chown_as_owner;
use crate::send_protocol::{AckResponse, ErrCode, MarkRequest};
use crate::uds_authz::{Caller, enforce_mailbox_owner_or_root};

/// Per-connection shared state for the MARK verbs (and the MAILBOX-CRUD
/// verbs, which share the per-mailbox lock map so their config.toml
/// rewrite does not race with an in-flight MARK).
pub struct StateContext {
    /// Data directory root. `<data_dir>/inbox/<mailbox>/<id>.md` etc.
    ///
    /// Invariant: `data_dir == config_handle.load().data_dir` for the
    /// life of the daemon. `data_dir` is captured once at startup and
    /// never changes; the `Config` swap path (MAILBOX-CRUD in
    /// `mailbox_handler.rs`) deliberately never rewrites `data_dir`, so
    /// this snapshot and the live handle's `data_dir` cannot diverge in
    /// practice.
    pub data_dir: PathBuf,
    /// Live handle to the daemon's `Config`. MARK-* uses it to validate
    /// that the referenced mailbox exists at the time of the call (rather
    /// than at startup); MAILBOX-* uses it to append / remove stanzas and
    /// hot-swap the in-memory snapshot.
    pub config_handle: ConfigHandle,
    /// Shared per-mailbox write-lock map. Inbound ingest, MARK-*, and
    /// MAILBOX-* all serialize through these locks (see
    /// [`crate::mailbox_locks`]).
    pub locks: Arc<MailboxLocks>,
}

impl StateContext {
    /// Fresh lock map. Convenient for tests. Production callers share
    /// the daemon-wide map via [`StateContext::with_locks`].
    #[cfg(test)]
    pub fn new(data_dir: PathBuf, config_handle: ConfigHandle) -> Self {
        Self::with_locks(data_dir, config_handle, Arc::new(MailboxLocks::new()))
    }

    /// Construct a `StateContext` that shares an existing
    /// [`MailboxLocks`] map with other daemon-side contexts. Used by
    /// `run_serve` so the SMTP session (inbound ingest) and the UDS
    /// handlers all take the same lock per mailbox.
    pub fn with_locks(
        data_dir: PathBuf,
        config_handle: ConfigHandle,
        locks: Arc<MailboxLocks>,
    ) -> Self {
        Self {
            data_dir,
            config_handle,
            locks,
        }
    }

    /// Acquire (lazy-inserting if needed) the per-mailbox state handle.
    /// Returned `Arc` owns both the async write lock and the dedup
    /// cache; callers `.lock.lock().await` (async) or
    /// `.lock.blocking_lock()` (inside `spawn_blocking`).
    pub(crate) fn lock_for(&self, mailbox: &str) -> Arc<MailboxState> {
        self.locks.lock_for(mailbox)
    }
}

/// Validate that the email id is path-traversal-safe. Mirrors the checks
/// in `mcp::validate_email_id` but lives on the daemon side so a malicious
/// local client cannot bypass them.
fn validate_id(id: &str) -> Result<(), AckResponse> {
    if id.is_empty()
        || id.contains("..")
        || id.contains('/')
        || id.contains('\\')
        || id.contains('\0')
    {
        return Err(AckResponse::Err {
            code: ErrCode::NotFound,
            reason: format!("email id '{id}' contains invalid characters"),
        });
    }
    Ok(())
}

fn inbox_dir(data_dir: &Path, mailbox: &str) -> PathBuf {
    data_dir.join("inbox").join(mailbox)
}

/// Handle a `MARK-READ` / `MARK-UNREAD` request. Takes the shared
/// per-mailbox write lock (see [`crate::mailbox_locks`]) for the full
/// read → rewrite critical section. The lock is shared with the
/// inbound-ingest writer so MARK and ingest serialize against each
/// other, not just against other MARK calls.
pub async fn handle_mark(ctx: &StateContext, req: &MarkRequest, caller: &Caller) -> AckResponse {
    if let Err(e) = validate_id(&req.id) {
        return e;
    }

    // Mailbox existence is resolved live through the handle so a
    // freshly-created mailbox is immediately target-able from MARK.
    //
    // We return `ENOENT` (not `EACCES`) when the mailbox
    // is absent so the authz check itself cannot leak which mailboxes
    // exist. Once the mailbox resolves, the ownership check runs.
    let config_snapshot = ctx.config_handle.load();
    // `contains_key` + `get` is split to keep the existence check out
    // of the authz error path, but the `get` must succeed on the
    // snapshot we just checked. We match explicitly on `None` and
    // return `ENOENT` so authz can never be silently skipped if a
    // future refactor drops the `contains_key` pre-check.
    let mailbox_cfg = match config_snapshot.mailboxes.get(&req.mailbox) {
        Some(m) => m.clone(),
        None => {
            return AckResponse::Err {
                code: ErrCode::Enoent,
                reason: format!("mailbox '{}' does not exist", req.mailbox),
            };
        }
    };
    let verb = if req.read { "MARK-READ" } else { "MARK-UNREAD" };
    if let Err(reject) = enforce_mailbox_owner_or_root(verb, caller, &req.mailbox, &mailbox_cfg) {
        return AckResponse::Err {
            code: reject.code,
            reason: reject.reason,
        };
    }

    let state = ctx.lock_for(&req.mailbox);
    let _guard = state.lock.lock().await;

    let mailbox_dir = inbox_dir(&ctx.data_dir, &req.mailbox);
    let filepath = match resolve_email_path_strict(&mailbox_dir, &req.id) {
        Some(p) => p,
        None => {
            // A strict-resolver rejection can mean either "no such file"
            // or "symlink / canonical-containment violation". Both map
            // to `NotFound` on the wire (no existence leak per PRD
            // §6.1), but we log at `info` on `aimx::state` so operators
            // can spot probe attempts (PRD §7.3). The log payload is
            // intentionally coarse — we don't disambiguate symlink vs.
            // absent-file at the handler level to avoid leaking that
            // fact into anything that tails the log stream downstream.
            tracing::info!(
                target: "aimx::state",
                reason = "symlink_or_escape",
                mailbox = %req.mailbox,
                id = %req.id,
                "MARK rejected: email not resolvable under mailbox dir"
            );
            return AckResponse::Err {
                code: ErrCode::NotFound,
                reason: format!("email '{}' not found in inbox/{}", req.id, req.mailbox),
            };
        }
    };

    let content = match std::fs::read_to_string(&filepath) {
        Ok(c) => c,
        Err(e) => {
            return AckResponse::Err {
                code: ErrCode::Io,
                reason: format!("failed to read {}: {e}", filepath.display()),
            };
        }
    };

    let parts: Vec<&str> = content.splitn(3, "+++").collect();
    if parts.len() < 3 {
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("malformed frontmatter in {}", filepath.display()),
        };
    }

    let toml_str = parts[1].trim();
    let mut meta: InboundFrontmatter = match toml::from_str(toml_str) {
        Ok(m) => m,
        Err(e) => {
            return AckResponse::Err {
                code: ErrCode::Io,
                reason: format!("failed to parse frontmatter in {}: {e}", filepath.display()),
            };
        }
    };

    meta.read = req.read;
    // `read_at` is set when marking read (reflects the most
    // recent read, overwriting any prior timestamp on re-read) and
    // removed entirely when marking unread. Never serialized as null.
    meta.read_at = if req.read {
        Some(chrono::Utc::now())
    } else {
        None
    };

    let new_toml = match toml::to_string(&meta) {
        Ok(t) => t,
        Err(e) => {
            return AckResponse::Err {
                code: ErrCode::Io,
                reason: format!("failed to serialize frontmatter: {e}"),
            };
        }
    };
    let body = parts[2];

    let mut out = String::new();
    out.push_str("+++\n");
    out.push_str(&new_toml);
    out.push_str("+++");
    out.push_str(body);

    // Write-temp-then-rename with chown on the temp file BEFORE
    // `rename(2)`. Ordering matters (PRD §6.3): a post-rename chown
    // would briefly expose the file as `root:root` in a readable
    // directory. Doing the chown while the file is still under its
    // `.<stem>.tmp` name means the published inode already carries the
    // correct owner + mode the instant `rename(2)` lands.
    let parent = match filepath.parent() {
        Some(p) => p,
        None => {
            return AckResponse::Err {
                code: ErrCode::Io,
                reason: format!("no parent dir for {}", filepath.display()),
            };
        }
    };
    let tmp_name = format!(
        ".{}.tmp.{}",
        filepath
            .file_name()
            .and_then(|s| s.to_str())
            .unwrap_or("mark"),
        std::process::id()
    );
    let tmp_path = parent.join(&tmp_name);
    if let Err(e) = std::fs::write(&tmp_path, out) {
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to write {}: {e}", tmp_path.display()),
        };
    }
    // Chown + chmod BEFORE the rename publishes the new inode. A
    // failure here is not fatal for the MARK operation (the
    // containing dir is `0o700 <owner>:<owner>` so the file is
    // still safe from non-owners even if ownership drifts) but we
    // log so doctor can surface the drift.
    if let Err(e) = chown_as_owner(&tmp_path, &mailbox_cfg, 0o600) {
        tracing::warn!(
            target: "aimx::state",
            "chown temp file failed mailbox={mailbox} path={path} err={err}",
            mailbox = req.mailbox,
            path = tmp_path.display(),
            err = e,
        );
    }
    if let Err(e) = std::fs::rename(&tmp_path, &filepath) {
        let _ = std::fs::remove_file(&tmp_path);
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!(
                "failed to rename {} -> {}: {e}",
                tmp_path.display(),
                filepath.display()
            ),
        };
    }

    AckResponse::Ok
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::frontmatter::InboundFrontmatter;
    use tempfile::TempDir;

    fn sample_meta(id: &str, read: bool) -> InboundFrontmatter {
        InboundFrontmatter {
            id: id.to_string(),
            message_id: format!("<{id}@test.com>"),
            thread_id: "0123456789abcdef".to_string(),
            from: "sender@test.com".to_string(),
            to: "alice@test.com".to_string(),
            cc: None,
            reply_to: None,
            delivered_to: "alice@test.com".to_string(),
            subject: "Hello".to_string(),
            date: "2025-06-01T12:00:00Z".to_string(),
            received_at: "2025-06-01T12:00:01Z".to_string(),
            received_from_ip: None,
            size_bytes: 100,
            in_reply_to: None,
            references: None,
            attachments: vec![],
            list_id: None,
            auto_submitted: None,
            dkim: "none".to_string(),
            spf: "none".to_string(),
            dmarc: "none".to_string(),
            trusted: "none".to_string(),
            mailbox: "alice".to_string(),
            read,
            read_at: None,
            labels: vec![],
        }
    }

    fn write_email(dir: &Path, id: &str, meta: &InboundFrontmatter) {
        std::fs::create_dir_all(dir).unwrap();
        let toml = toml::to_string(meta).unwrap();
        let content = format!("+++\n{toml}+++\n\nbody of {id}\n");
        std::fs::write(dir.join(format!("{id}.md")), content).unwrap();
    }

    fn ctx(data_dir: &Path) -> StateContext {
        let mut mailboxes = std::collections::HashMap::new();
        mailboxes.insert(
            "alice".to_string(),
            crate::config::MailboxConfig {
                address: "alice@example.com".to_string(),
                owner: "root".to_string(),
                hooks: vec![],
                trust: None,
                trusted_senders: None,
                allow_root_catchall: false,
            },
        );
        let config = crate::config::Config {
            domain: "example.com".to_string(),
            data_dir: data_dir.to_path_buf(),
            dkim_selector: "aimx".to_string(),
            trust: "none".to_string(),
            trusted_senders: vec![],
            mailboxes,
            verify_host: None,
            enable_ipv6: false,
            signature: None,
            upgrade: None,
        };
        StateContext::new(data_dir.to_path_buf(), ConfigHandle::new(config))
    }

    #[tokio::test]
    async fn mark_read_toggles_frontmatter() {
        let tmp = TempDir::new().unwrap();
        let meta = sample_meta("2025-06-01-001", false);
        let inbox = tmp.path().join("inbox").join("alice");
        write_email(&inbox, "2025-06-01-001", &meta);

        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-001".to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Ok => {}
            other => panic!("expected Ok, got {other:?}"),
        }

        let content = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        let parts: Vec<&str> = content.splitn(3, "+++").collect();
        let fm: InboundFrontmatter = toml::from_str(parts[1].trim()).unwrap();
        assert!(fm.read);
        // Body preserved
        assert!(parts[2].contains("body of 2025-06-01-001"));
    }

    #[tokio::test]
    async fn mark_unread_after_read() {
        let tmp = TempDir::new().unwrap();
        let meta = sample_meta("2025-06-01-001", true);
        let inbox = tmp.path().join("inbox").join("alice");
        write_email(&inbox, "2025-06-01-001", &meta);

        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-001".to_string(),
            read: false,
        };
        assert!(matches!(
            handle_mark(&sctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        let content = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        let fm: InboundFrontmatter =
            toml::from_str(content.split("+++").nth(1).unwrap().trim()).unwrap();
        assert!(!fm.read);
    }

    #[tokio::test]
    async fn mark_unknown_mailbox_errors() {
        let tmp = TempDir::new().unwrap();
        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "ghost".to_string(),
            id: "2025-06-01-001".to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                // Unknown mailbox returns ENOENT so the
                // authz check cannot leak which mailboxes exist.
                assert_eq!(code, ErrCode::Enoent);
                assert!(reason.contains("ghost"), "{reason}");
            }
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn mark_missing_email_errors() {
        let tmp = TempDir::new().unwrap();
        std::fs::create_dir_all(tmp.path().join("inbox").join("alice")).unwrap();
        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2099-01-01-missing".to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, .. } => assert_eq!(code, ErrCode::NotFound),
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn mark_rejects_path_traversal_id() {
        let tmp = TempDir::new().unwrap();
        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "../../etc/passwd".to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::NotFound);
                assert!(reason.contains("invalid characters"), "{reason}");
            }
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn mark_read_in_bundle_layout() {
        // Bundle form: `<mailbox>/<id>/<id>.md`.
        let tmp = TempDir::new().unwrap();
        let mailbox = tmp.path().join("inbox").join("alice");
        let bundle = mailbox.join("2025-06-03-001");
        std::fs::create_dir_all(&bundle).unwrap();
        let meta = sample_meta("2025-06-03-001", false);
        let toml = toml::to_string(&meta).unwrap();
        let content = format!("+++\n{toml}+++\n\nbody\n");
        std::fs::write(bundle.join("2025-06-03-001.md"), content).unwrap();

        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-03-001".to_string(),
            read: true,
        };
        assert!(matches!(
            handle_mark(&sctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        let content = std::fs::read_to_string(bundle.join("2025-06-03-001.md")).unwrap();
        let fm: InboundFrontmatter =
            toml::from_str(content.split("+++").nth(1).unwrap().trim()).unwrap();
        assert!(fm.read);
    }

    #[tokio::test]
    async fn mark_rejects_truncated_frontmatter_as_io_error() {
        // Target file exists but has only one `+++` delimiter so the
        // splitn(3) produces fewer than 3 parts. Handler must surface
        // this as `ErrCode::Io` rather than panicking or silently
        // rewriting garbage.
        let tmp = TempDir::new().unwrap();
        let inbox = tmp.path().join("inbox").join("alice");
        std::fs::create_dir_all(&inbox).unwrap();
        std::fs::write(
            inbox.join("2025-06-01-002.md"),
            "+++\nid = \"2025-06-01-002\"\nno closing delimiter here\n",
        )
        .unwrap();

        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-002".to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Io);
                assert!(
                    reason.contains("malformed frontmatter"),
                    "expected 'malformed frontmatter' in reason, got: {reason}"
                );
            }
            other => panic!("expected Err(Io), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn mark_read_sets_read_at_timestamp() {
        let tmp = TempDir::new().unwrap();
        let meta = sample_meta("2025-06-01-001", false);
        let inbox = tmp.path().join("inbox").join("alice");
        write_email(&inbox, "2025-06-01-001", &meta);

        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-001".to_string(),
            read: true,
        };
        let before = chrono::Utc::now();
        assert!(matches!(
            handle_mark(&sctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));
        let after = chrono::Utc::now();

        let content = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        let fm: InboundFrontmatter =
            toml::from_str(content.split("+++").nth(1).unwrap().trim()).unwrap();
        assert!(fm.read);
        let ts = fm.read_at.expect("read_at must be set on MARK-READ");
        assert!(
            ts >= before && ts <= after,
            "read_at {ts} not between {before} and {after}"
        );
    }

    #[tokio::test]
    async fn mark_unread_removes_read_at_field() {
        let tmp = TempDir::new().unwrap();
        let mut meta = sample_meta("2025-06-01-001", true);
        meta.read_at = Some(chrono::Utc::now());
        let inbox = tmp.path().join("inbox").join("alice");
        write_email(&inbox, "2025-06-01-001", &meta);

        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-001".to_string(),
            read: false,
        };
        assert!(matches!(
            handle_mark(&sctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        let content = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        // Field must be removed entirely, not serialized as `null`.
        assert!(
            !content.contains("read_at"),
            "read_at must be removed on MARK-UNREAD; got:\n{content}"
        );
        let fm: InboundFrontmatter =
            toml::from_str(content.split("+++").nth(1).unwrap().trim()).unwrap();
        assert!(!fm.read);
        assert!(fm.read_at.is_none());
    }

    #[tokio::test]
    async fn mark_read_twice_updates_timestamp_to_most_recent() {
        // Re-MARK-READ overwrites the prior timestamp. The
        // field reflects "most recent read", not "first read".
        let tmp = TempDir::new().unwrap();
        let meta = sample_meta("2025-06-01-001", false);
        let inbox = tmp.path().join("inbox").join("alice");
        write_email(&inbox, "2025-06-01-001", &meta);

        let sctx = ctx(tmp.path());
        let req_read = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-001".to_string(),
            read: true,
        };
        let req_unread = MarkRequest {
            read: false,
            ..req_read.clone()
        };

        // First MARK-READ.
        assert!(matches!(
            handle_mark(&sctx, &req_read, &Caller::internal_root()).await,
            AckResponse::Ok
        ));
        let content = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        let fm: InboundFrontmatter =
            toml::from_str(content.split("+++").nth(1).unwrap().trim()).unwrap();
        let first = fm.read_at.expect("read_at set on first MARK-READ");

        // MARK-UNREAD clears it.
        assert!(matches!(
            handle_mark(&sctx, &req_unread, &Caller::internal_root()).await,
            AckResponse::Ok
        ));
        let content = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        let fm: InboundFrontmatter =
            toml::from_str(content.split("+++").nth(1).unwrap().trim()).unwrap();
        assert!(fm.read_at.is_none());

        // Give the monotonic wall clock room to advance between reads.
        // Millisecond resolution is plenty; chrono stores nanoseconds.
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        // Second MARK-READ writes a fresh, later timestamp.
        assert!(matches!(
            handle_mark(&sctx, &req_read, &Caller::internal_root()).await,
            AckResponse::Ok
        ));
        let content = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        let fm: InboundFrontmatter =
            toml::from_str(content.split("+++").nth(1).unwrap().trim()).unwrap();
        let second = fm.read_at.expect("read_at set on second MARK-READ");
        assert!(
            second > first,
            "second read_at ({second}) must be later than first ({first})"
        );
    }

    /// After a MARK-READ the rewritten file still lives
    /// at the same path with the same ownership + mode (0o600). The
    /// write-temp-then-rename dance chowns the temp file BEFORE the
    /// rename so the published inode lands with the correct owner
    /// instantly — no observable intermediate state.
    ///
    /// The test uses `owner = "testowner"` (not the reserved `root`)
    /// so the pluggable user resolver seam is active and a chown+chmod
    /// to the current process's uid/gid succeeds on any CI runner.
    #[tokio::test]
    async fn mark_read_preserves_mode_0600_on_rewrite() {
        use std::os::unix::fs::MetadataExt;

        let tmp = TempDir::new().unwrap();
        let meta = sample_meta("2025-06-01-001", false);
        let inbox = tmp.path().join("inbox").join("alice");
        write_email(&inbox, "2025-06-01-001", &meta);

        fn fake(name: &str) -> Option<crate::user_resolver::ResolvedUser> {
            if name == "testowner" {
                Some(crate::user_resolver::ResolvedUser {
                    name: "testowner".into(),
                    uid: unsafe { libc::geteuid() },
                    gid: unsafe { libc::getegid() },
                })
            } else {
                None
            }
        }
        let _r = crate::user_resolver::set_test_resolver(fake);

        // Build a dedicated context whose `alice` mailbox is owned by
        // `testowner` so the chown resolver seam is exercised.
        let mut mailboxes = std::collections::HashMap::new();
        mailboxes.insert(
            "alice".to_string(),
            crate::config::MailboxConfig {
                address: "alice@example.com".into(),
                owner: "testowner".into(),
                hooks: vec![],
                trust: None,
                trusted_senders: None,
                allow_root_catchall: false,
            },
        );
        let config = crate::config::Config {
            domain: "example.com".into(),
            data_dir: tmp.path().to_path_buf(),
            dkim_selector: "aimx".into(),
            trust: "none".into(),
            trusted_senders: vec![],
            mailboxes,
            verify_host: None,
            enable_ipv6: false,
            signature: None,
            upgrade: None,
        };
        let sctx = StateContext::new(tmp.path().to_path_buf(), ConfigHandle::new(config));

        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-001".to_string(),
            read: true,
        };
        assert!(matches!(
            handle_mark(&sctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        let target = inbox.join("2025-06-01-001.md");
        let md = std::fs::metadata(&target).unwrap();
        assert_eq!(
            md.mode() & 0o777,
            0o600,
            "MARK-READ rewrite must preserve mode 0o600 via chown-before-rename"
        );
        // No stray temp file left behind.
        let strays: Vec<_> = std::fs::read_dir(&inbox)
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
            .collect();
        assert!(strays.is_empty(), "no temp file must remain after rename");
    }

    /// When `chown_as_owner` fails (non-root process +
    /// unprivileged chown target), the handler logs a warning and
    /// **still succeeds** — the mail stays on disk with fresh content
    /// (`read = true`) and the published file contents match what the
    /// MARK verb produced. Non-fatal chown is the contract documented
    /// in PRD §6.3: the containing dir is already `0o700 owner:owner`
    /// on a properly-provisioned host so file-level chown drift doesn't
    /// break the isolation invariant.
    ///
    /// Skipped on root where chown-to-nobody succeeds (no failure to
    /// test). Uses a resolver that maps `nobody` to uid/gid 65534 so a
    /// non-root process cannot chown to it.
    #[tokio::test]
    async fn mark_nonfatal_on_chown_failure_preserves_rewrite() {
        use std::os::unix::fs::MetadataExt;

        let is_root = unsafe { libc::geteuid() == 0 };
        if is_root {
            // chown-to-nobody succeeds as root — the failure path we
            // want to test isn't reachable here. Skip.
            return;
        }

        let tmp = TempDir::new().unwrap();
        let meta = sample_meta("2025-06-01-001", false);
        let inbox = tmp.path().join("inbox").join("alice");
        write_email(&inbox, "2025-06-01-001", &meta);

        fn nobody(name: &str) -> Option<crate::user_resolver::ResolvedUser> {
            if name == "nobody" {
                Some(crate::user_resolver::ResolvedUser {
                    name: "nobody".into(),
                    uid: 65534,
                    gid: 65534,
                })
            } else {
                None
            }
        }
        let _r = crate::user_resolver::set_test_resolver(nobody);

        let mut mailboxes = std::collections::HashMap::new();
        mailboxes.insert(
            "alice".to_string(),
            crate::config::MailboxConfig {
                address: "alice@example.com".into(),
                owner: "nobody".into(),
                hooks: vec![],
                trust: None,
                trusted_senders: None,
                allow_root_catchall: false,
            },
        );
        let config = crate::config::Config {
            domain: "example.com".into(),
            data_dir: tmp.path().to_path_buf(),
            dkim_selector: "aimx".into(),
            trust: "none".into(),
            trusted_senders: vec![],
            mailboxes,
            verify_host: None,
            enable_ipv6: false,
            signature: None,
            upgrade: None,
        };
        let sctx = StateContext::new(tmp.path().to_path_buf(), ConfigHandle::new(config));

        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-001".to_string(),
            read: true,
        };
        // chown will fail (EPERM) but the handler must still return Ok.
        assert!(
            matches!(
                handle_mark(&sctx, &req, &Caller::internal_root()).await,
                AckResponse::Ok
            ),
            "chown failure must be non-fatal"
        );

        // The file must still be on disk after the rename — we just
        // verify the rewrite actually landed (content has read = true).
        let target = inbox.join("2025-06-01-001.md");
        assert!(
            target.exists(),
            "target file must survive chown-failure path"
        );
        let content = std::fs::read_to_string(&target).unwrap();
        assert!(
            content.contains("read = true"),
            "rewrite must have published the new content despite chown failure; got:\n{content}"
        );
        // And no stray tmp file remains.
        let strays: Vec<_> = std::fs::read_dir(&inbox)
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
            .collect();
        assert!(
            strays.is_empty(),
            "no tmp file must remain after a non-fatal chown failure"
        );
        // Guard against an unexpected ownership change despite the
        // failure log: the file should still be owned by the current
        // process's uid (since chown-to-65534 failed with EPERM).
        let md = std::fs::metadata(&target).unwrap();
        assert_eq!(
            md.uid(),
            unsafe { libc::geteuid() },
            "file ownership must be unchanged after chown failure"
        );
    }

    /// Cross-mailbox symlink attack. Alice plants a
    /// symlink in her own mailbox pointing at bob's real email. The
    /// MARK handler must reject (`NotFound`) and leave bob's file
    /// byte-for-byte unchanged; alice's symlink is also unchanged (no
    /// replacement-write lands on it).
    #[tokio::test]
    async fn mark_rejects_cross_mailbox_symlink_attack() {
        let tmp = TempDir::new().unwrap();

        // Configure two mailboxes owned by `root` so the authz check
        // short-circuits to the "internal root" bypass and we can
        // focus the assertion on the symlink rejection itself.
        let mut mailboxes = std::collections::HashMap::new();
        mailboxes.insert(
            "alice".to_string(),
            crate::config::MailboxConfig {
                address: "alice@example.com".into(),
                owner: "root".into(),
                hooks: vec![],
                trust: None,
                trusted_senders: None,
                allow_root_catchall: false,
            },
        );
        mailboxes.insert(
            "bob".to_string(),
            crate::config::MailboxConfig {
                address: "bob@example.com".into(),
                owner: "root".into(),
                hooks: vec![],
                trust: None,
                trusted_senders: None,
                allow_root_catchall: false,
            },
        );
        let config = crate::config::Config {
            domain: "example.com".into(),
            data_dir: tmp.path().to_path_buf(),
            dkim_selector: "aimx".into(),
            trust: "none".into(),
            trusted_senders: vec![],
            mailboxes,
            verify_host: None,
            enable_ipv6: false,
            signature: None,
            upgrade: None,
        };
        let sctx = StateContext::new(tmp.path().to_path_buf(), ConfigHandle::new(config));

        // Plant bob's real email.
        let bob_meta = sample_meta("2025-06-01-777", false);
        let bob_inbox = tmp.path().join("inbox").join("bob");
        write_email(&bob_inbox, "2025-06-01-777", &bob_meta);
        let bob_file = bob_inbox.join("2025-06-01-777.md");
        let bob_bytes_before = std::fs::read(&bob_file).unwrap();

        // Plant alice's symlink pointing at bob's real file.
        let alice_inbox = tmp.path().join("inbox").join("alice");
        std::fs::create_dir_all(&alice_inbox).unwrap();
        let alice_link = alice_inbox.join("2025-06-01-777.md");
        std::os::unix::fs::symlink(&bob_file, &alice_link).unwrap();

        // Sanity: alice's entry is a symlink on disk before the MARK.
        let pre = std::fs::symlink_metadata(&alice_link).unwrap();
        assert!(pre.file_type().is_symlink());

        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-777".to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, .. } => assert_eq!(code, ErrCode::NotFound),
            other => panic!("expected NotFound, got {other:?}"),
        }

        // Bob's real file is unchanged byte-for-byte.
        let bob_bytes_after = std::fs::read(&bob_file).unwrap();
        assert_eq!(
            bob_bytes_before, bob_bytes_after,
            "bob's file must be byte-for-byte unchanged after alice's probe"
        );

        // Alice's entry is still a symlink — no replacement-write
        // landed on it.
        let post = std::fs::symlink_metadata(&alice_link).unwrap();
        assert!(
            post.file_type().is_symlink(),
            "alice's planted symlink must not have been replaced by a regular file"
        );
    }

    /// Bundle-layout variant of the cross-mailbox
    /// symlink attack. Alice plants `alice/<id>/<id>.md` as a symlink
    /// to `bob/<id>/<id>.md`. Same expected outcome.
    #[tokio::test]
    async fn mark_rejects_cross_mailbox_bundle_symlink_attack() {
        let tmp = TempDir::new().unwrap();

        let mut mailboxes = std::collections::HashMap::new();
        for name in ["alice", "bob"] {
            mailboxes.insert(
                name.to_string(),
                crate::config::MailboxConfig {
                    address: format!("{name}@example.com"),
                    owner: "root".into(),
                    hooks: vec![],
                    trust: None,
                    trusted_senders: None,
                    allow_root_catchall: false,
                },
            );
        }
        let config = crate::config::Config {
            domain: "example.com".into(),
            data_dir: tmp.path().to_path_buf(),
            dkim_selector: "aimx".into(),
            trust: "none".into(),
            trusted_senders: vec![],
            mailboxes,
            verify_host: None,
            enable_ipv6: false,
            signature: None,
            upgrade: None,
        };
        let sctx = StateContext::new(tmp.path().to_path_buf(), ConfigHandle::new(config));

        // Bob has a real bundle.
        let id = "2025-06-01-888";
        let bob_bundle = tmp.path().join("inbox").join("bob").join(id);
        std::fs::create_dir_all(&bob_bundle).unwrap();
        let bob_meta = sample_meta(id, false);
        let toml_str = toml::to_string(&bob_meta).unwrap();
        let content = format!("+++\n{toml_str}+++\n\nbob body\n");
        let bob_inner = bob_bundle.join(format!("{id}.md"));
        std::fs::write(&bob_inner, content).unwrap();
        let bob_bytes_before = std::fs::read(&bob_inner).unwrap();

        // Alice has her own bundle directory but the inner `.md` is a
        // symlink to bob's file.
        let alice_bundle = tmp.path().join("inbox").join("alice").join(id);
        std::fs::create_dir_all(&alice_bundle).unwrap();
        let alice_inner = alice_bundle.join(format!("{id}.md"));
        std::os::unix::fs::symlink(&bob_inner, &alice_inner).unwrap();

        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: id.to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, .. } => assert_eq!(code, ErrCode::NotFound),
            other => panic!("expected NotFound, got {other:?}"),
        }

        let bob_bytes_after = std::fs::read(&bob_inner).unwrap();
        assert_eq!(bob_bytes_before, bob_bytes_after);
        let post = std::fs::symlink_metadata(&alice_inner).unwrap();
        assert!(post.file_type().is_symlink());
    }

    #[tokio::test]
    async fn mark_rejects_unparseable_frontmatter_toml_as_io_error() {
        // File has the `+++` delimiters but the TOML body is malformed.
        // Handler must return `ErrCode::Io` (read-failure-after-lock
        // contract).
        let tmp = TempDir::new().unwrap();
        let inbox = tmp.path().join("inbox").join("alice");
        std::fs::create_dir_all(&inbox).unwrap();
        std::fs::write(
            inbox.join("2025-06-01-003.md"),
            "+++\nthis is = not valid = toml\n+++\n\nbody\n",
        )
        .unwrap();

        let sctx = ctx(tmp.path());
        let req = MarkRequest {
            mailbox: "alice".to_string(),
            id: "2025-06-01-003".to_string(),
            read: true,
        };
        match handle_mark(&sctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, .. } => assert_eq!(code, ErrCode::Io),
            other => panic!("expected Err(Io), got {other:?}"),
        }
    }
}
