openlatch-client 0.1.18

//! Durable file-backed FIFO queue for cloud events that failed to forward.
//!
//! Events that survive the cloud worker's one retry are appended to
//! `~/.openlatch/outbox.jsonl`; the drain task replays them on daemon
//! startup and on every `CloudState::drain_notify` signal. Dedup happens
//! on the cloud side via the envelope's UUIDv7 `id`, so a crash between
//! successful POST and compaction results in at-worst one idempotent
//! replay. Drain progress is persisted in a sibling
//! `outbox.jsonl.offset` cursor so a halted drain reads only the unread
//! tail on the next pass.

use std::fs::{File, OpenOptions};
use std::io::{BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;

use crate::core::cloud::offset::{
    advance_past_oldest, count_entries_from, read_offset, write_offset, AdvanceStats,
};

/// Name of the on-disk outbox file, relative to the OpenLatch directory.
pub const OUTBOX_FILENAME: &str = "outbox.jsonl";

/// Statistics returned from a drain pass.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct DrainStats {
    pub drained: u64,
    pub failed: u64,
    pub corrupt: u64,
    /// Entries the drain closure asked to drop after repeatedly failing —
    /// cursor advances past them like a clean drain but they're tallied
    /// separately so telemetry can distinguish quarantine from delivery.
    pub quarantined: u64,
}

/// Grouping limits for one drain pass.
///
/// The drain replays entries through the same batched POST path the live
/// worker uses, so it must group them the same way: mirror the cloud's
/// `batch_max_events` and 256KB body caps here and the replay traffic has the
/// same shape as live traffic.
///
/// `Default` is deliberately one entry per group — the pre-batching
/// behaviour — so tests that assert per-entry semantics stay honest.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DrainLimits {
    /// Maximum number of entries handed to the callback in one group.
    pub max_entries: usize,
    /// Maximum raw JSONL payload bytes (sum of line lengths, excluding the
    /// newlines and the JSON array framing) in one group. Approximate by
    /// design: the caller re-splits under the exact wire cap before posting,
    /// so this only has to stop a group from growing wildly oversized.
    pub max_bytes: usize,
}

impl Default for DrainLimits {
    fn default() -> Self {
        Self {
            max_entries: 1,
            max_bytes: usize::MAX,
        }
    }
}

/// Outcome returned by the drain closure for each entry. `Forwarded` and
/// `Quarantined` both advance the cursor; `Err(())` halts the drain.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DrainOutcome {
    /// Cloud accepted the entry — count as drained.
    Forwarded,
    /// Closure has given up on this entry; advance past it without
    /// counting as delivered.
    Quarantined,
}

/// Errors surfaced by outbox operations. Kept intentionally small — callers
/// log `OL-1204` / `OL-1205` at a warn level and continue serving events;
/// outbox failures never crash the daemon.
#[derive(Debug, thiserror::Error)]
pub enum OutboxError {
    #[error("outbox I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("outbox serialization error: {0}")]
    Serde(#[from] serde_json::Error),
}

#[derive(Debug)]
pub struct Outbox {
    path: PathBuf,
    offset_path: PathBuf,
    tmp_path: PathBuf,
    max_bytes: u64,
    lock: Mutex<()>,
    // Reflect the **unread tail** (cursor → EOF). Maintained in lockstep
    // with file mutations so `/metrics` is O(1) — the endpoint is
    // unauthenticated and can be scraped while a drain holds the lock for
    // a large compaction.
    pending_entries: AtomicU64,
    pending_bytes: AtomicU64,
}

impl Outbox {
    /// Build an outbox rooted under `openlatch_dir`. `max_bytes == 0`
    /// disables the size cap. Scans the unread tail once to recover
    /// gauges after a daemon restart.
    pub fn new(openlatch_dir: &Path, max_bytes: u64) -> Self {
        let path = openlatch_dir.join(OUTBOX_FILENAME);
        let offset_path = openlatch_dir.join(format!("{OUTBOX_FILENAME}.offset"));
        let tmp_path = openlatch_dir.join(format!("{OUTBOX_FILENAME}.tmp"));
        let cursor = read_offset(&offset_path);
        let (entries, bytes) = scan_tail_gauges(&path, cursor);
        Self {
            path,
            offset_path,
            tmp_path,
            max_bytes,
            lock: Mutex::new(()),
            pending_entries: AtomicU64::new(entries),
            pending_bytes: AtomicU64::new(bytes),
        }
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Append a single envelope. Creates the parent directory and file on
    /// first use. Enforces the size cap via drop-oldest if this write
    /// would push the file past `max_bytes`.
    pub fn append(&self, envelope: &serde_json::Value) -> Result<(), OutboxError> {
        let _guard = self.lock.lock().unwrap_or_else(|e| e.into_inner());
        self.ensure_parent_dir()?;
        let line = serde_json::to_string(envelope)?;
        let mut f = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)?;
        writeln!(f, "{line}")?;
        f.sync_all()?;
        drop(f);

        let written = line.len() as u64 + 1;
        self.pending_entries.fetch_add(1, Ordering::Relaxed);
        self.pending_bytes.fetch_add(written, Ordering::Relaxed);

        if self.max_bytes > 0 {
            let size = file_byte_size(&self.path);
            if size > self.max_bytes {
                self.drop_oldest_until_under_cap_locked(size)?;
            }
        }
        Ok(())
    }

    /// Current pending byte count (unread tail). O(1) atomic load.
    pub fn byte_size(&self) -> u64 {
        self.pending_bytes.load(Ordering::Relaxed)
    }

    /// Current pending entry count (unread tail). O(1) atomic load.
    pub fn pending_count(&self) -> u64 {
        self.pending_entries.load(Ordering::Relaxed)
    }

    /// Drain the outbox: read entries from the persisted cursor, hand
    /// `post_fn` a **contiguous group** of up to `limits.max_entries` entries
    /// at a time, and advance the cursor past exactly the entries of a group
    /// that settled.
    ///
    /// Group semantics (the single-entry semantics, lifted to a group):
    ///
    /// - `Ok(outcomes)` — advance past **every** entry in the group. One
    ///   outcome per entry, positionally; a short vector counts the tail as
    ///   `Forwarded`.
    /// - `Err(())` — advance past **none** of them and halt the pass, exactly
    ///   as a single-entry failure does today. Advancing to the next group and
    ///   continuing would reorder delivery and burn attempt budget against a
    ///   connection that is already known to be down.
    ///
    /// Blank and corrupt lines never enter a group: they are counted and
    /// advanced past individually. Because the cursor is a single byte offset,
    /// an open group is always settled *before* a following corrupt line is
    /// skipped — otherwise advancing past the corrupt line would implicitly
    /// advance past the un-settled entries in front of it.
    ///
    /// A crash mid-group is safe: the cursor did not move, so the whole group
    /// is retried. Delivery stays at-least-once and the platform dedups on the
    /// envelope `id`.
    ///
    /// On halt the cursor is persisted at the first un-settled entry and the
    /// data file is left intact — the next drain resumes there without
    /// re-snapshotting the whole file. On clean completion the file is
    /// compacted (drained bytes dropped; concurrent appends preserved). When
    /// the file is fully consumed both files are removed.
    pub async fn drain<F, Fut>(
        &self,
        limits: DrainLimits,
        mut post_fn: F,
    ) -> Result<DrainStats, OutboxError>
    where
        F: FnMut(Vec<serde_json::Value>) -> Fut,
        Fut: std::future::Future<Output = Result<Vec<DrainOutcome>, ()>>,
    {
        // Snapshot only the unread tail under the lock, then release it
        // for the async POST loop so live appends aren't blocked by slow
        // network.
        let (start_offset, snapshot) = {
            let _guard = self.lock.lock().unwrap_or_else(|e| e.into_inner());
            let cursor = read_offset(&self.offset_path);
            let snapshot = self.read_tail_locked(cursor)?;
            (cursor, snapshot)
        };

        if snapshot.is_empty() {
            // Cursor at/past EOF (previous drain consumed everything but
            // compaction was skipped, e.g. crash). Reset both files.
            let _guard = self.lock.lock().unwrap_or_else(|e| e.into_inner());
            let total = file_byte_size(&self.path);
            if total == 0 || start_offset >= total {
                let _ = std::fs::remove_file(&self.path);
                let _ = std::fs::remove_file(&self.offset_path);
                self.pending_entries.store(0, Ordering::Relaxed);
                self.pending_bytes.store(0, Ordering::Relaxed);
            }
            return Ok(DrainStats::default());
        }

        let snapshot_total_bytes: u64 = snapshot.iter().map(|(_, b)| *b).sum();
        let snapshot_entries: u64 = snapshot.iter().filter(|(l, _)| !l.is_empty()).count() as u64;

        let mut stats = DrainStats::default();
        let mut bytes_advanced: u64 = 0;
        let mut halted = false;

        let max_entries = limits.max_entries.max(1);
        let mut i = 0usize;
        while i < snapshot.len() {
            let (raw, line_bytes) = &snapshot[i];
            if raw.is_empty() {
                bytes_advanced += line_bytes;
                i += 1;
                continue;
            }
            let first: serde_json::Value = match serde_json::from_str(raw) {
                Ok(v) => v,
                Err(_) => {
                    // Malformed JSON would fail every retry forever — drop it.
                    stats.corrupt += 1;
                    bytes_advanced += line_bytes;
                    i += 1;
                    continue;
                }
            };

            // Extend the group across following entries while they are
            // parseable and both caps still allow it. Anything that stops the
            // scan (blank line, corrupt line, cap reached) is left for the
            // next outer iteration, which only runs once this group settled.
            let mut group = vec![first];
            let mut group_bytes = *line_bytes;
            let mut payload = raw.len();
            let mut end = i + 1;
            while end < snapshot.len() && group.len() < max_entries {
                let (next_raw, next_line_bytes) = &snapshot[end];
                if next_raw.is_empty() || payload + next_raw.len() > limits.max_bytes {
                    break;
                }
                let Ok(value) = serde_json::from_str::<serde_json::Value>(next_raw) else {
                    break;
                };
                payload += next_raw.len();
                group_bytes += next_line_bytes;
                group.push(value);
                end += 1;
            }

            let group_len = group.len();
            match post_fn(group).await {
                Ok(outcomes) => {
                    for k in 0..group_len {
                        // A short outcome vector counts as delivered: `Ok`
                        // already means "advance past every entry", so the
                        // only question left is which counter moves.
                        match outcomes.get(k) {
                            Some(DrainOutcome::Quarantined) => stats.quarantined += 1,
                            _ => stats.drained += 1,
                        }
                    }
                    bytes_advanced += group_bytes;
                    i = end;
                }
                Err(()) => {
                    // Per-event counter, like every other field on DrainStats.
                    stats.failed += group_len as u64;
                    halted = true;
                    break;
                }
            }
        }

        let new_cursor = start_offset + bytes_advanced;
        let _guard = self.lock.lock().unwrap_or_else(|e| e.into_inner());
        if halted {
            write_offset(&self.offset_path, new_cursor);
            // Gauges reflect the snapshot's unread tail. Any appends that
            // landed during the async POST loop will be picked up on the
            // next clean drain — keeping gauges syscall-free here is the
            // whole point of the cursor.
            let unread_bytes = snapshot_total_bytes - bytes_advanced;
            let unread_entries = snapshot_entries
                .saturating_sub(stats.drained)
                .saturating_sub(stats.corrupt)
                .saturating_sub(stats.quarantined);
            self.pending_entries
                .store(unread_entries, Ordering::Relaxed);
            self.pending_bytes.store(unread_bytes, Ordering::Relaxed);
        } else {
            // Clean drain — keep any bytes appended past the snapshot,
            // discard the dead prefix.
            let tail = self.read_raw_tail_locked(new_cursor)?;
            self.rewrite_raw_locked(&tail)?;
        }
        Ok(stats)
    }

    /// Delete the outbox file entirely. Useful for tests and for the case
    /// where the user opts out of outbox via config.
    pub fn clear(&self) -> Result<(), OutboxError> {
        let _guard = self.lock.lock().unwrap_or_else(|e| e.into_inner());
        if let Err(e) = std::fs::remove_file(&self.path) {
            if e.kind() != std::io::ErrorKind::NotFound {
                return Err(e.into());
            }
        }
        let _ = std::fs::remove_file(&self.offset_path);
        self.pending_entries.store(0, Ordering::Relaxed);
        self.pending_bytes.store(0, Ordering::Relaxed);
        Ok(())
    }

    // -----------------------------------------------------------------
    // Internal helpers — must only be called with `self.lock` held
    // (the read-only ones also run from `Outbox::new` before the mutex
    // exists, safe because no other thread holds a reference yet).
    // -----------------------------------------------------------------

    fn ensure_parent_dir(&self) -> std::io::Result<()> {
        if let Some(parent) = self.path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        Ok(())
    }

    /// Read JSONL lines from `start_offset` to EOF. Returns
    /// `(line, byte_length_including_newline)` pairs so the drain loop
    /// can advance the cursor exactly.
    fn read_tail_locked(&self, start_offset: u64) -> Result<Vec<(String, u64)>, OutboxError> {
        let file = match File::open(&self.path) {
            Ok(f) => f,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => return Err(e.into()),
        };
        let total_len = file.metadata().map(|m| m.len()).unwrap_or(0);
        if start_offset >= total_len {
            return Ok(Vec::new());
        }
        let mut reader = BufReader::new(file);
        if reader.seek(SeekFrom::Start(start_offset)).is_err() {
            return Ok(Vec::new());
        }
        let mut out = Vec::new();
        use std::io::BufRead;
        for line in reader.lines() {
            let line = line?;
            // `writeln!` always emits `\n` on every platform, so +1 is exact.
            let bytes = line.len() as u64 + 1;
            out.push((line, bytes));
        }
        Ok(out)
    }

    /// Read raw bytes from `start_offset` to EOF. Used by the clean-drain
    /// compaction path so we don't lose byte fidelity when rewriting.
    fn read_raw_tail_locked(&self, start_offset: u64) -> Result<Vec<u8>, OutboxError> {
        let mut file = match File::open(&self.path) {
            Ok(f) => f,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => return Err(e.into()),
        };
        let total_len = file.metadata().map(|m| m.len()).unwrap_or(0);
        if start_offset >= total_len {
            return Ok(Vec::new());
        }
        file.seek(SeekFrom::Start(start_offset))?;
        let mut buf = Vec::with_capacity((total_len - start_offset) as usize);
        file.read_to_end(&mut buf)?;
        Ok(buf)
    }

    /// Rewrite the data file to `bytes` (or remove both data + offset if
    /// `bytes` is empty), then refresh the gauges and remove the offset.
    fn rewrite_raw_locked(&self, bytes: &[u8]) -> Result<(), OutboxError> {
        if bytes.is_empty() {
            if let Err(e) = std::fs::remove_file(&self.path) {
                if e.kind() != std::io::ErrorKind::NotFound {
                    return Err(e.into());
                }
            }
            let _ = std::fs::remove_file(&self.offset_path);
            self.pending_entries.store(0, Ordering::Relaxed);
            self.pending_bytes.store(0, Ordering::Relaxed);
            return Ok(());
        }
        self.ensure_parent_dir()?;
        {
            let mut tmp = File::create(&self.tmp_path)?;
            tmp.write_all(bytes)?;
            tmp.sync_all()?;
        }
        std::fs::rename(&self.tmp_path, &self.path)?;
        let _ = std::fs::remove_file(&self.offset_path);
        let entries = bytes
            .split(|&b| b == b'\n')
            .filter(|c| !c.is_empty())
            .count() as u64;
        self.pending_entries.store(entries, Ordering::Relaxed);
        self.pending_bytes
            .store(bytes.len() as u64, Ordering::Relaxed);
        Ok(())
    }

    /// Bring the file under `max_bytes` after an oversize append. First
    /// try to compact the dead prefix (bytes 0..cursor — already-drained
    /// entries); if that alone doesn't fix it, evict oldest unread
    /// entries via offset advance.
    fn drop_oldest_until_under_cap_locked(&self, current_size: u64) -> Result<(), OutboxError> {
        let cursor = read_offset(&self.offset_path);
        if cursor > 0 {
            let tail = self.read_raw_tail_locked(cursor)?;
            self.rewrite_raw_locked(&tail)?;
            let new_size = file_byte_size(&self.path);
            if new_size <= self.max_bytes {
                return Ok(());
            }
            return self.evict_oldest_unread_locked(new_size, 0);
        }
        self.evict_oldest_unread_locked(current_size, cursor)
    }

    fn evict_oldest_unread_locked(
        &self,
        current_size: u64,
        cursor: u64,
    ) -> Result<(), OutboxError> {
        let unread = current_size.saturating_sub(cursor);
        if unread <= self.max_bytes {
            return Ok(());
        }
        let excess = unread - self.max_bytes;
        let Some(AdvanceStats {
            new_offset,
            dropped,
        }) = advance_past_oldest(&self.path, cursor, excess)
        else {
            return Ok(());
        };
        tracing::warn!(
            code = crate::error::ERR_OUTBOX_WRITE_FAILED,
            dropped,
            size_before = current_size,
            max_bytes = self.max_bytes,
            new_offset,
            "outbox: advancing offset past oldest entries to stay under size cap"
        );
        crate::telemetry::capture_global(crate::telemetry::Event::cloud_outbox_overflow(
            dropped,
            current_size,
            self.max_bytes,
        ));
        write_offset(&self.offset_path, new_offset);
        let new_unread_bytes = current_size.saturating_sub(new_offset);
        let new_unread_entries = self
            .pending_entries
            .load(Ordering::Relaxed)
            .saturating_sub(dropped);
        self.pending_entries
            .store(new_unread_entries, Ordering::Relaxed);
        self.pending_bytes
            .store(new_unread_bytes, Ordering::Relaxed);
        Ok(())
    }
}

/// Filesystem-truth byte size. Used to enforce the size cap and to detect
/// concurrent appends — atomic gauges track our bookkeeping, this reads
/// the real metadata to avoid drift under concurrent file moves.
fn file_byte_size(path: &Path) -> u64 {
    std::fs::metadata(path).map(|m| m.len()).unwrap_or(0)
}

/// Count non-empty entries and unread byte length from `cursor` to EOF.
/// Used only at `Outbox::new` to recover gauges after a daemon restart.
fn scan_tail_gauges(path: &Path, cursor: u64) -> (u64, u64) {
    let total_len = file_byte_size(path);
    if cursor >= total_len {
        return (0, 0);
    }
    (count_entries_from(path, cursor), total_len - cursor)
}

// =====================================================================
// Tests
// =====================================================================

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

    fn make_outbox(tmp: &TempDir, max_bytes: u64) -> Outbox {
        Outbox::new(tmp.path(), max_bytes)
    }

    fn offset_path(tmp: &TempDir) -> PathBuf {
        tmp.path().join(format!("{OUTBOX_FILENAME}.offset"))
    }

    /// One outcome per entry in the group — the shape every "everything
    /// succeeded" drain callback returns.
    fn all_forwarded(n: usize) -> Result<Vec<DrainOutcome>, ()> {
        Ok(vec![DrainOutcome::Forwarded; n])
    }

    #[test]
    fn append_creates_file_and_writes_line() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        out.append(&json!({"id": "evt_1"})).unwrap();
        assert!(out.path().exists());
        assert_eq!(out.pending_count(), 1);
        assert!(out.byte_size() > 0);
    }

    #[test]
    fn append_is_newline_delimited_jsonl() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        out.append(&json!({"id": "a"})).unwrap();
        out.append(&json!({"id": "b"})).unwrap();
        let body = std::fs::read_to_string(out.path()).unwrap();
        let lines: Vec<&str> = body.lines().collect();
        assert_eq!(lines.len(), 2);
        assert!(lines[0].contains("\"a\""));
        assert!(lines[1].contains("\"b\""));
    }

    #[tokio::test]
    async fn drain_success_removes_file() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        out.append(&json!({"id": "a"})).unwrap();
        out.append(&json!({"id": "b"})).unwrap();
        let stats = out
            .drain(DrainLimits::default(), |batch| async move {
                all_forwarded(batch.len())
            })
            .await
            .unwrap();
        assert_eq!(stats.drained, 2);
        assert_eq!(stats.failed, 0);
        assert!(!out.path().exists());
        assert!(
            !offset_path(&tmp).exists(),
            "offset file must be cleaned up"
        );
        assert_eq!(out.pending_count(), 0);
        assert_eq!(out.byte_size(), 0);
    }

    #[tokio::test]
    async fn drain_persists_offset_on_halt() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        for i in 0..5 {
            out.append(&json!({"id": format!("evt_{i}")})).unwrap();
        }
        let counter = std::sync::atomic::AtomicUsize::new(0);
        let stats = out
            .drain(DrainLimits::default(), |batch| {
                let n = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let len = batch.len();
                async move {
                    if n < 2 {
                        all_forwarded(len)
                    } else {
                        Err(())
                    }
                }
            })
            .await
            .unwrap();
        assert_eq!(stats.drained, 2);
        assert_eq!(stats.failed, 1);
        assert!(
            offset_path(&tmp).exists(),
            "halted drain must persist cursor"
        );
        assert!(out.path().exists(), "data file must remain on halted drain");
        let cursor = read_offset(&offset_path(&tmp));
        assert!(cursor > 0, "cursor must advance past drained prefix");
        assert_eq!(out.pending_count(), 3);
        let body = std::fs::read_to_string(out.path()).unwrap();
        assert!(body.contains("evt_2"));
        assert!(body.contains("evt_3"));
        assert!(body.contains("evt_4"));
    }

    #[tokio::test]
    async fn drain_resumes_from_offset_across_restart() {
        let tmp = TempDir::new().unwrap();
        {
            let first = make_outbox(&tmp, 0);
            for i in 0..5 {
                first.append(&json!({"id": format!("evt_{i}")})).unwrap();
            }
            let counter = std::sync::atomic::AtomicUsize::new(0);
            let _ = first
                .drain(DrainLimits::default(), |batch| {
                    let n = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    let len = batch.len();
                    async move {
                        if n < 2 {
                            all_forwarded(len)
                        } else {
                            Err(())
                        }
                    }
                })
                .await
                .unwrap();
        }
        let recovered = make_outbox(&tmp, 0);
        assert_eq!(recovered.pending_count(), 3, "gauge from unread tail only");
        let drained_ids = std::sync::Mutex::new(Vec::<String>::new());
        let stats = recovered
            .drain(DrainLimits::default(), |batch| {
                let ids: Vec<String> = batch
                    .iter()
                    .map(|e| e["id"].as_str().unwrap_or("").to_string())
                    .collect();
                let drained_ids = &drained_ids;
                async move {
                    let n = ids.len();
                    drained_ids.lock().unwrap().extend(ids);
                    all_forwarded(n)
                }
            })
            .await
            .unwrap();
        assert_eq!(stats.drained, 3);
        let ids = drained_ids.into_inner().unwrap();
        assert_eq!(ids, vec!["evt_2", "evt_3", "evt_4"]);
        assert!(!recovered.path().exists(), "file removed after clean drain");
        assert!(
            !offset_path(&tmp).exists(),
            "offset removed after clean drain"
        );
    }

    #[tokio::test]
    async fn drain_clean_completion_truncates_and_clears_offset() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        out.append(&json!({"id": "a"})).unwrap();
        let stats = out
            .drain(DrainLimits::default(), |batch| async move {
                all_forwarded(batch.len())
            })
            .await
            .unwrap();
        assert_eq!(stats.drained, 1);
        assert!(!out.path().exists());
        assert!(!offset_path(&tmp).exists());
        assert_eq!(out.pending_count(), 0);
        assert_eq!(out.byte_size(), 0);
    }

    #[tokio::test]
    async fn drain_discards_corrupt_lines() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        std::fs::create_dir_all(tmp.path()).unwrap();
        std::fs::write(out.path(), b"{\"id\":\"a\"}\nnot json\n{\"id\":\"b\"}\n").unwrap();
        let stats = out
            .drain(DrainLimits::default(), |batch| async move {
                all_forwarded(batch.len())
            })
            .await
            .unwrap();
        assert_eq!(stats.drained, 2);
        assert_eq!(stats.corrupt, 1);
    }

    #[test]
    fn append_enforces_size_cap_by_dropping_oldest() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 200);
        for i in 0..20u32 {
            out.append(&json!({
                "id": format!("evt_{i:04}"),
                "padding": "x".repeat(20),
            }))
            .unwrap();
        }
        assert!(
            out.byte_size() <= 200,
            "outbox grew past cap: {} bytes",
            out.byte_size()
        );
        let body = std::fs::read_to_string(out.path()).unwrap();
        assert!(!body.contains("evt_0000"));
    }

    #[tokio::test]
    async fn drop_oldest_advances_offset_under_cap() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 200);
        for i in 0..20u32 {
            out.append(&json!({
                "id": format!("evt_{i:04}"),
                "padding": "x".repeat(20),
            }))
            .unwrap();
        }
        let cursor = read_offset(&offset_path(&tmp));
        assert!(
            cursor > 0,
            "evict_oldest_unread should have advanced offset rather than rewriting"
        );
        assert!(
            out.byte_size() <= 200,
            "unread tail must be ≤ cap: {} bytes",
            out.byte_size()
        );
    }

    #[tokio::test]
    async fn compaction_reclaims_dead_prefix_under_cap() {
        let tmp = TempDir::new().unwrap();
        let out = std::sync::Arc::new(Outbox::new(tmp.path(), 0));
        for i in 0..10u32 {
            out.append(&json!({"id": format!("evt_{i:04}"), "padding": "x".repeat(30)}))
                .unwrap();
        }
        let counter = std::sync::atomic::AtomicUsize::new(0);
        let _ = out
            .drain(DrainLimits::default(), |batch| {
                let n = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let len = batch.len();
                async move {
                    if n < 5 {
                        all_forwarded(len)
                    } else {
                        Err(())
                    }
                }
            })
            .await
            .unwrap();
        let cursor_after_halt = read_offset(&offset_path(&tmp));
        assert!(cursor_after_halt > 0, "halt should leave a dead prefix");
        let total_before = std::fs::metadata(out.path()).unwrap().len();
        assert!(total_before > cursor_after_halt);

        // Cap is exceeded by total_before + trigger append, but small
        // enough that compacting away the dead prefix alone fixes it.
        let unread_before = total_before - cursor_after_halt;
        let cap = unread_before + 128;
        assert!(
            total_before > cap,
            "test setup: file size before ({total_before}) must exceed cap ({cap})"
        );
        let capped = Outbox::new(tmp.path(), cap);
        capped.append(&json!({"id": "trigger"})).unwrap();
        let total_after = std::fs::metadata(capped.path()).unwrap().len();
        assert!(
            total_after < total_before,
            "compaction should have shrunk the file: before={total_before}, after={total_after}"
        );
        assert!(
            !offset_path(&tmp).exists(),
            "compaction-only path must clear the offset file"
        );
    }

    #[test]
    fn gauges_are_recovered_from_existing_file_on_new() {
        let tmp = TempDir::new().unwrap();
        {
            let first = make_outbox(&tmp, 0);
            first.append(&json!({"id": "a"})).unwrap();
            first.append(&json!({"id": "b"})).unwrap();
        }
        let recovered = make_outbox(&tmp, 0);
        assert_eq!(recovered.pending_count(), 2);
        assert!(recovered.byte_size() > 0);
    }

    #[test]
    fn gauges_reflect_unread_tail_only() {
        let tmp = TempDir::new().unwrap();
        {
            let out = make_outbox(&tmp, 0);
            for i in 0..4 {
                out.append(&json!({"id": format!("evt_{i}")})).unwrap();
            }
        }
        let body = std::fs::read_to_string(tmp.path().join(OUTBOX_FILENAME)).unwrap();
        let lines: Vec<&str> = body.split_inclusive('\n').collect();
        let prefix_bytes = (lines[0].len() + lines[1].len()) as u64;
        write_offset(&offset_path(&tmp), prefix_bytes);
        let recovered = make_outbox(&tmp, 0);
        assert_eq!(recovered.pending_count(), 2);
        assert_eq!(recovered.byte_size(), body.len() as u64 - prefix_bytes);
    }

    #[test]
    fn clear_removes_file_idempotently() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        out.append(&json!({"id": "a"})).unwrap();
        out.clear().unwrap();
        assert!(!out.path().exists());
        assert!(!offset_path(&tmp).exists());
        out.clear().unwrap();
    }

    #[tokio::test]
    async fn drain_quarantine_advances_past_entry() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        out.append(&json!({"id": "poison"})).unwrap();
        out.append(&json!({"id": "next"})).unwrap();
        let counter = std::sync::atomic::AtomicUsize::new(0);
        let stats = out
            .drain(DrainLimits::default(), |batch| {
                let n = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let len = batch.len();
                async move {
                    if n == 0 {
                        Ok::<Vec<DrainOutcome>, ()>(vec![DrainOutcome::Quarantined; len])
                    } else {
                        all_forwarded(len)
                    }
                }
            })
            .await
            .unwrap();
        assert_eq!(stats.drained, 1);
        assert_eq!(stats.quarantined, 1);
        assert_eq!(stats.failed, 0);
        assert_eq!(
            out.pending_count(),
            0,
            "quarantined entry must leave the unread tail"
        );
        assert!(
            !out.path().exists(),
            "clean drain still compacts when quarantined alongside forwarded"
        );
    }

    #[tokio::test]
    async fn concurrent_append_during_drain_still_visible() {
        let tmp = TempDir::new().unwrap();
        let out = std::sync::Arc::new(make_outbox(&tmp, 0));
        out.append(&json!({"id": "a"})).unwrap();
        out.append(&json!({"id": "b"})).unwrap();
        let writer = out.clone();
        let stats = out
            .drain(DrainLimits::default(), |batch| {
                let id = batch[0]["id"].as_str().unwrap_or("").to_string();
                let len = batch.len();
                let writer = writer.clone();
                async move {
                    if id == "a" {
                        writer.append(&json!({"id": "c"})).unwrap();
                    }
                    all_forwarded(len)
                }
            })
            .await
            .unwrap();
        assert_eq!(stats.drained, 2);
        assert_eq!(out.pending_count(), 1);
        let body = std::fs::read_to_string(out.path()).unwrap();
        assert!(body.contains("\"c\""));
    }

    /// The grouping contract: entries reach the callback in contiguous
    /// batches of at most `max_entries`, in file order, and the cursor
    /// advances past a whole group at a time.
    #[tokio::test]
    async fn drain_groups_contiguous_entries_up_to_max() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        for i in 0..12 {
            out.append(&json!({"id": format!("evt_{i:02}")})).unwrap();
        }
        let groups = std::sync::Mutex::new(Vec::<Vec<String>>::new());
        let stats = out
            .drain(
                DrainLimits {
                    max_entries: 5,
                    max_bytes: usize::MAX,
                },
                |batch| {
                    let ids: Vec<String> = batch
                        .iter()
                        .map(|e| e["id"].as_str().unwrap_or("").to_string())
                        .collect();
                    let groups = &groups;
                    async move {
                        let n = ids.len();
                        groups.lock().unwrap().push(ids);
                        all_forwarded(n)
                    }
                },
            )
            .await
            .unwrap();

        assert_eq!(stats.drained, 12);
        let groups = groups.into_inner().unwrap();
        assert_eq!(
            groups.iter().map(Vec::len).collect::<Vec<_>>(),
            vec![5, 5, 2],
            "12 entries at max_entries=5 must arrive as 5+5+2, not one per call"
        );
        assert_eq!(groups[0][0], "evt_00", "groups must preserve file order");
        assert_eq!(groups[2][1], "evt_11");
    }

    /// `Err(())` on a group advances the cursor past **none** of its entries
    /// and halts the pass — the whole group is retried next time.
    #[tokio::test]
    async fn drain_group_failure_advances_past_nothing() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        for i in 0..6 {
            out.append(&json!({"id": format!("evt_{i}")})).unwrap();
        }
        let stats = out
            .drain(
                DrainLimits {
                    max_entries: 3,
                    max_bytes: usize::MAX,
                },
                |_batch| async move { Err(()) },
            )
            .await
            .unwrap();

        assert_eq!(
            stats.failed, 3,
            "DrainStats.failed is per-event: a failed group of 3 is 3 failures"
        );
        assert_eq!(stats.drained, 0);
        assert_eq!(
            read_offset(&offset_path(&tmp)),
            0,
            "a failed first group must not move the cursor at all"
        );
        assert_eq!(
            out.pending_count(),
            6,
            "every entry stays pending after a failed group"
        );
    }

    /// The byte cap closes a group before the entry count does.
    #[tokio::test]
    async fn drain_group_respects_byte_cap() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        for i in 0..6 {
            out.append(&json!({"id": format!("evt_{i}"), "padding": "x".repeat(100)}))
                .unwrap();
        }
        let sizes = std::sync::Mutex::new(Vec::<usize>::new());
        let stats = out
            .drain(
                DrainLimits {
                    max_entries: 100,
                    // Room for two ~130-byte lines, not three.
                    max_bytes: 300,
                },
                |batch| {
                    let n = batch.len();
                    let sizes = &sizes;
                    async move {
                        sizes.lock().unwrap().push(n);
                        all_forwarded(n)
                    }
                },
            )
            .await
            .unwrap();

        assert_eq!(stats.drained, 6);
        let sizes = sizes.into_inner().unwrap();
        assert!(
            sizes.iter().all(|n| *n < 100),
            "byte cap must close groups well before the entry cap: {sizes:?}"
        );
        assert!(sizes.len() > 1, "6 entries must not fit in one 300B group");
    }

    /// Mixed-group cursor rule: a corrupt line never joins a group, and the
    /// group in front of it settles before the corrupt line is skipped —
    /// otherwise advancing past the corrupt line would implicitly advance
    /// past un-settled entries.
    #[tokio::test]
    async fn drain_settles_group_before_skipping_corrupt_line() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        std::fs::create_dir_all(tmp.path()).unwrap();
        std::fs::write(
            out.path(),
            b"{\"id\":\"a\"}\n{\"id\":\"b\"}\nnot json\n{\"id\":\"c\"}\n",
        )
        .unwrap();

        let groups = std::sync::Mutex::new(Vec::<Vec<String>>::new());
        let stats = out
            .drain(
                DrainLimits {
                    max_entries: 10,
                    max_bytes: usize::MAX,
                },
                |batch| {
                    let ids: Vec<String> = batch
                        .iter()
                        .map(|e| e["id"].as_str().unwrap_or("").to_string())
                        .collect();
                    let groups = &groups;
                    async move {
                        let n = ids.len();
                        groups.lock().unwrap().push(ids);
                        all_forwarded(n)
                    }
                },
            )
            .await
            .unwrap();

        assert_eq!(stats.drained, 3);
        assert_eq!(stats.corrupt, 1);
        let groups = groups.into_inner().unwrap();
        assert_eq!(
            groups,
            vec![
                vec!["a".to_string(), "b".to_string()],
                vec!["c".to_string()]
            ],
            "the corrupt line must split the group, not join it"
        );
    }

    /// A group may report per-entry outcomes: quarantined entries advance
    /// like forwarded ones but land in a different counter.
    #[tokio::test]
    async fn drain_group_reports_per_entry_outcomes() {
        let tmp = TempDir::new().unwrap();
        let out = make_outbox(&tmp, 0);
        for i in 0..3 {
            out.append(&json!({"id": format!("evt_{i}")})).unwrap();
        }
        let stats = out
            .drain(
                DrainLimits {
                    max_entries: 10,
                    max_bytes: usize::MAX,
                },
                |_batch| async move {
                    Ok::<Vec<DrainOutcome>, ()>(vec![
                        DrainOutcome::Forwarded,
                        DrainOutcome::Quarantined,
                        DrainOutcome::Forwarded,
                    ])
                },
            )
            .await
            .unwrap();

        assert_eq!(stats.drained, 2);
        assert_eq!(stats.quarantined, 1);
        assert_eq!(out.pending_count(), 0, "Ok advances past the whole group");
    }
}