openlatch-client 0.1.18

//! Replay of `~/.openlatch/logs/fallback.jsonl` — hook-binary events
//! captured while the daemon was unreachable.
//!
//! Security note: hook entries are **pre-privacy-filter**. Every envelope
//! must go through `handlers::process_envelope` on replay — skipping the
//! filter would leak unredacted content to the cloud. Progress is
//! persisted to a sibling `.offset` file so mid-replay crashes resume
//! cleanly.

use std::io::{BufRead, BufReader, Seek, SeekFrom};
use std::path::Path;
use std::sync::Arc;

use crate::core::cloud::offset::{
    advance_past_oldest, count_entries_from, read_offset, write_offset, AdvanceStats,
};
use crate::core::envelope::EventEnvelope;
use crate::daemon::handlers;
use crate::daemon::AppState;

/// Name of the fallback log written by `openlatch-hook` when the daemon is
/// unreachable. Matches the path resolution in
/// `src/app/openlatch_hook/main.rs::openlatch_log_dir()`.
const FALLBACK_FILENAME: &str = "fallback.jsonl";
const OFFSET_FILENAME: &str = "fallback.jsonl.offset";

/// The replay loop. Returns immediately when `state.cloud_state` is `None`
/// (cloud forwarding disabled — nothing to replay to). Otherwise loops
/// forever, replaying on every `drain_notify` signal plus once at startup.
///
/// A plain future rather than a spawn, so the daemon can run it under its task
/// supervisor (`core::supervision::task`) and a panic mid-replay does not end
/// offline-event recovery for the rest of the process lifetime.
pub async fn run(state: Arc<AppState>) {
    let Some(cloud_state) = state.cloud_state.clone() else {
        return;
    };
    let notify = cloud_state.drain_notify.clone();
    // Startup replay: pick up entries left by a previous daemon run
    // (offline boot is the motivating case).
    replay_once(&state).await;
    loop {
        notify.notified().await;
        replay_once(&state).await;
    }
}

/// Single pass over `fallback.jsonl`. Reads from the persisted offset to
/// end of file, feeds each valid envelope through `process_envelope`, and
/// commits progress via the offset file. Stops on the first I/O error so a
/// transient filesystem issue doesn't retry-storm.
pub async fn replay_once(state: &Arc<AppState>) {
    // Emergency-mode early-return: when the cloud worker has flagged the
    // channel as saturated, every replay event would race live hooks for
    // capacity. Skip this pass entirely so the channel drains as fast as
    // the worker can post — the next `drain_notify` signal will retry.
    if let Some(cs) = state.cloud_state.as_ref() {
        if cs.is_emergency_mode() {
            tracing::debug!("fallback replay: skipping pass — cloud channel in emergency mode");
            return;
        }
    }

    let log_dir = crate::config::openlatch_dir().join("logs");
    let path = log_dir.join(FALLBACK_FILENAME);
    let offset_path = log_dir.join(OFFSET_FILENAME);

    if !path.exists() {
        // Remove any stale offset file if fallback.jsonl was cleared
        // manually. Keeps the directory tidy.
        let _ = std::fs::remove_file(&offset_path);
        return;
    }

    // Enforce the fallback size cap BEFORE the replay pass so a long
    // outage can't poison a fresh restart by handing the worker tens of
    // thousands of stale events all at once. Drop-oldest is implemented
    // by advancing the persisted offset, not by rewriting the file —
    // cheaper, atomic, and reuses the existing replay machinery.
    let max_bytes = state.config.cloud.fallback_max_bytes;
    if max_bytes > 0 {
        enforce_fallback_size_cap(&path, &offset_path, max_bytes);
    }

    let start_offset = read_offset(&offset_path);
    let file = match std::fs::File::open(&path) {
        Ok(f) => f,
        Err(e) => {
            tracing::warn!(
                code = "OL-1205",
                error = %e,
                path = %path.display(),
                "fallback replay: could not open fallback.jsonl"
            );
            return;
        }
    };

    let total_len = file.metadata().map(|m| m.len()).unwrap_or(0);
    if start_offset >= total_len {
        // Nothing new since last replay. Clean up both files so a long-
        // idle daemon doesn't carry empty artifacts around forever.
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&offset_path);
        return;
    }

    let mut reader = BufReader::new(file);
    if reader.seek(SeekFrom::Start(start_offset)).is_err() {
        tracing::warn!(
            code = "OL-1205",
            path = %path.display(),
            "fallback replay: seek failed — discarding offset and restarting from 0"
        );
        let _ = std::fs::remove_file(&offset_path);
        return;
    }

    let mut replayed: u64 = 0;
    let mut corrupt: u64 = 0;
    let mut consumed_offset = start_offset;
    let mut halted = false;

    let pending_hint = count_entries_from(&path, start_offset);
    tracing::info!(
        start_offset,
        total_bytes = total_len,
        pending = pending_hint,
        "fallback replay: starting"
    );
    crate::telemetry::capture_global(crate::telemetry::Event::fallback_replay_started(
        pending_hint,
    ));

    for line in reader.lines() {
        let Ok(raw) = line else {
            halted = true;
            break;
        };
        let line_len = raw.len() as u64 + 1; // +1 for newline
        if raw.trim().is_empty() {
            consumed_offset += line_len;
            continue;
        }

        let envelope: EventEnvelope = match serde_json::from_str(&raw) {
            Ok(e) => e,
            Err(e) => {
                tracing::debug!(
                    error = %e,
                    "fallback replay: discarding unparsable line"
                );
                corrupt += 1;
                consumed_offset += line_len;
                continue;
            }
        };

        // If the cloud worker has exited (e.g. daemon shutting down) the
        // replay path's `send().await` would return `Err` and we'd advance
        // the offset past an event that never reached the channel. Detect
        // that here so the offset stays anchored on the unsent line and
        // the next daemon start picks up exactly where we left off.
        if let Some(tx) = state.cloud_tx.as_ref() {
            if tx.is_closed() {
                tracing::info!(
                    "fallback replay: cloud channel closed — halting, will resume from offset {consumed_offset} on next start"
                );
                halted = true;
                break;
            }
        }

        // Push through the same pipeline as a live hook so privacy
        // filtering and enrichment both fire. We intentionally drop the
        // returned verdict — the agent that produced this envelope is
        // long gone; we only care that the event reaches the cloud and
        // the audit log.
        let started = std::time::Instant::now();
        let _ = handlers::process_envelope_for_replay(state.clone(), envelope, started).await;
        replayed += 1;
        consumed_offset += line_len;
    }

    if halted {
        tracing::warn!(
            code = "OL-1205",
            replayed,
            corrupt,
            "fallback replay: halted on read error — remainder will retry next cycle"
        );
        write_offset(&offset_path, consumed_offset);
        return;
    }

    if consumed_offset >= total_len {
        // Clean drain — delete both files.
        if let Err(e) = std::fs::remove_file(&path) {
            tracing::warn!(
                code = "OL-1205",
                error = %e,
                "fallback replay: could not remove drained fallback.jsonl"
            );
        }
        let _ = std::fs::remove_file(&offset_path);
    } else {
        write_offset(&offset_path, consumed_offset);
    }

    if replayed > 0 || corrupt > 0 {
        let remaining = count_entries_from(&path, consumed_offset);
        tracing::info!(
            replayed,
            corrupt,
            remaining,
            consumed_bytes = consumed_offset - start_offset,
            "fallback replay: completed"
        );
        crate::telemetry::capture_global(crate::telemetry::Event::fallback_replay_completed(
            replayed, corrupt, remaining,
        ));
    }
}

/// Cap the unread tail of `fallback.jsonl` to `max_bytes` by advancing
/// the persisted offset past oldest entries — same algorithm the outbox
/// uses, factored into `core::cloud::offset::advance_past_oldest`.
fn enforce_fallback_size_cap(path: &Path, offset_path: &Path, max_bytes: u64) {
    let Ok(metadata) = std::fs::metadata(path) else {
        return;
    };
    let total_len = metadata.len();
    let start_offset = read_offset(offset_path);
    if start_offset >= total_len {
        return;
    }
    let unread = total_len - start_offset;
    if unread <= max_bytes {
        return;
    }
    let excess = unread - max_bytes;
    let Some(AdvanceStats {
        new_offset,
        dropped,
    }) = advance_past_oldest(path, start_offset, excess)
    else {
        return;
    };
    tracing::warn!(
        code = "OL-1205",
        dropped,
        size_before = unread,
        max_bytes,
        new_offset,
        "fallback overflow: advancing offset past oldest entries to stay under size cap"
    );
    crate::telemetry::capture_global(crate::telemetry::Event::fallback_overflow(
        dropped, unread, max_bytes,
    ));
    write_offset(offset_path, new_offset);
}

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

    fn write_line(file: &mut std::fs::File, line: &str) -> u64 {
        use std::io::Write;
        let bytes = format!("{line}\n");
        file.write_all(bytes.as_bytes()).unwrap();
        bytes.len() as u64
    }

    #[test]
    fn enforce_size_cap_is_noop_when_unread_within_budget() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        let mut f = std::fs::File::create(&path).unwrap();
        for i in 0..10 {
            write_line(&mut f, &format!("{{\"id\":\"evt_{i}\"}}"));
        }
        drop(f);

        // Far above file size — nothing to evict.
        enforce_fallback_size_cap(&path, &offset_path, 10_000_000);

        // No offset file should be written.
        assert!(!offset_path.exists(), "no eviction → no offset file");
    }

    #[test]
    fn enforce_size_cap_advances_offset_past_oldest_when_over_budget() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        let mut f = std::fs::File::create(&path).unwrap();
        let mut sizes: Vec<u64> = Vec::new();
        for i in 0..20 {
            sizes.push(write_line(&mut f, &format!("{{\"id\":\"evt_{i:04}\"}}")));
        }
        drop(f);
        let total: u64 = sizes.iter().sum();
        // Cap at ~half the file — at least one line must be evicted.
        let cap = total / 2;

        enforce_fallback_size_cap(&path, &offset_path, cap);

        let new_offset = read_offset(&offset_path);
        assert!(new_offset > 0, "offset must advance past oldest entries");
        let unread = total - new_offset;
        assert!(
            unread <= cap,
            "unread tail ({unread}) must be ≤ cap ({cap}) after eviction"
        );
    }

    #[test]
    fn enforce_size_cap_handles_missing_file() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        // Must not panic when the file doesn't exist.
        enforce_fallback_size_cap(&path, &offset_path, 1000);
        assert!(!offset_path.exists());
    }

    #[test]
    fn enforce_size_cap_zero_disables_cap() {
        // The caller is responsible for the `max_bytes > 0` gate; this test
        // exists to assert that even if the function is called with 0, the
        // saturating arithmetic doesn't underflow. With max_bytes = 0,
        // `unread > 0` evicts everything — verify it does so without
        // panicking and lands on a sane offset.
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join(FALLBACK_FILENAME);
        let offset_path = tmp.path().join(OFFSET_FILENAME);
        let mut f = std::fs::File::create(&path).unwrap();
        write_line(&mut f, "{\"id\":\"evt_0\"}");
        drop(f);
        enforce_fallback_size_cap(&path, &offset_path, 0);
        // Either nothing evicted or the entire file evicted — both are
        // acceptable; what matters is no panic / underflow.
        let _ = read_offset(&offset_path);
    }
}