openlatch-client 0.6.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 capture-time cutoff of the last organization switch (PLT-7 C-8), as
/// Unix milliseconds. An entry captured before it belongs to the old
/// organization and is never replayed, even one the hook appended after apply
/// moved the cursor (it built the envelope, failed to reach the daemon, then
/// wrote it).
const SWITCH_CUTOFF_FILENAME: &str = "fallback.switch-cutoff";

/// The replay loop. Returns immediately when `state.cloud_state` is `None`
/// (cloud forwarding disabled — nothing to replay to). Otherwise loops
/// forever, replaying on every `fallback_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.fallback_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 `fallback_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;
        }
    }

    // PLT-7 D-11 — a pass holds the switch's I/O gate from its first read to its
    // cursor commit, so pausing for an organization switch waits for it, and no
    // pass starts while uploads are paused.
    let switch = state.cloud_state.as_ref().map(|cs| cs.switch.clone());
    let _pass = match switch.as_ref() {
        Some(switch) => match switch.begin_io_pass().await {
            Some(pass) => Some(pass),
            None => {
                tracing::debug!(
                    "fallback replay: skipping pass — uploads paused for an organization switch"
                );
                return;
            }
        },
        None => None,
    };
    let pass_generation = switch.as_ref().map(|s| s.generation());
    // A cursor this pass computed is void once a switch applied after it
    // started: apply moved the cursor to the end, and this one would move it
    // back over the old organization's events.
    let switched_since_start =
        || matches!((switch.as_ref(), pass_generation), (Some(s), Some(g)) if s.is_stale(g));

    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 switch_cutoff = read_switch_cutoff(&log_dir);
    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;
            }
        };

        // Captured before the last organization switch: the old organization's
        // event, never replayed under the new key.
        if switch_cutoff.is_some_and(|cutoff| envelope.time.timestamp_millis() < cutoff) {
            tracing::debug!(
                "fallback replay: dropping an entry captured before the organization switch"
            );
            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 switched_since_start() {
        tracing::debug!("fallback replay: an organization switch applied mid-pass; discarding this pass's cursor");
        return;
    }

    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,
        ));
    }
}

/// Drop the unread tail of `logs/fallback.jsonl` for an organization switch
/// (PLT-7 C-8 apply step 1): the replay cursor moves to the end of the file, so
/// no event the hook captured for the old organization is ever replayed under
/// the new key. Returns how many entries were dropped.
///
/// Called by apply with uploads paused and the replay quiesced, so no pass is
/// reading the file. It first records the capture-time cutoff, so an entry the
/// hook appends later for the old organization is dropped too. When the cursor
/// cannot be persisted the file itself is removed — every byte before the
/// cursor was already replayed. `log_dir` is `<openlatch_dir>/logs`.
///
/// # Errors
///
/// When neither the skip nor the removal can be verified, or the cutoff cannot
/// be written: apply must then stay paused, since the tail would replay under
/// the new key.
pub(crate) fn skip_unread_tail_in(log_dir: &Path) -> Result<u64, String> {
    write_switch_cutoff(log_dir, chrono::Utc::now().timestamp_millis())?;
    let path = log_dir.join(FALLBACK_FILENAME);
    let offset_path = log_dir.join(OFFSET_FILENAME);
    let end = match std::fs::metadata(&path) {
        Ok(metadata) => metadata.len(),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(e) => return Err(format!("cannot read {}: {e}", path.display())),
    };
    let start = read_offset(&offset_path);
    if start >= end {
        return Ok(0);
    }
    let dropped = count_entries_from(&path, start);
    write_offset(&offset_path, end);
    if read_offset(&offset_path) != end {
        match std::fs::remove_file(&path) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                return Err(format!(
                    "cannot drop the unread tail of {}: {e}",
                    path.display()
                ))
            }
        }
    }
    Ok(dropped)
}

fn write_switch_cutoff(log_dir: &Path, cutoff_ms: i64) -> Result<(), String> {
    let path = log_dir.join(SWITCH_CUTOFF_FILENAME);
    std::fs::create_dir_all(log_dir)
        .and_then(|()| crate::core::fs_secure::write_owner_only(&path, &cutoff_ms.to_string()))
        .map_err(|e| format!("cannot record the switch cutoff at {}: {e}", path.display()))
}

fn read_switch_cutoff(log_dir: &Path) -> Option<i64> {
    std::fs::read_to_string(log_dir.join(SWITCH_CUTOFF_FILENAME))
        .ok()
        .and_then(|s| s.trim().parse().ok())
}

/// How many entries of `logs/fallback.jsonl` are not replayed yet.
pub(crate) fn unread_entries_in(log_dir: &Path) -> u64 {
    let path = log_dir.join(FALLBACK_FILENAME);
    let start = read_offset(&log_dir.join(OFFSET_FILENAME));
    match std::fs::metadata(&path) {
        Ok(m) if m.len() > start => count_entries_from(&path, start),
        _ => 0,
    }
}

/// 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 fallback_replay_skip_unread_tail_moves_the_cursor_to_the_end() {
        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 first = write_line(&mut f, "{\"id\":\"evt_read\"}");
        let mut total = first;
        for i in 0..3 {
            total += write_line(&mut f, &format!("{{\"id\":\"evt_unread_{i}\"}}"));
        }
        drop(f);
        write_offset(&offset_path, first);

        assert_eq!(unread_entries_in(tmp.path()), 3);
        assert_eq!(skip_unread_tail_in(tmp.path()), Ok(3));
        assert_eq!(unread_entries_in(tmp.path()), 0);
        assert_eq!(read_offset(&offset_path), total);
        // Nothing left to drop the second time.
        assert_eq!(skip_unread_tail_in(tmp.path()), Ok(0));
        // The cutoff is recorded, so a late append from before the switch is caught.
        assert!(read_switch_cutoff(tmp.path()).is_some());
    }

    #[test]
    fn fallback_replay_skip_unread_tail_without_a_file_drops_nothing() {
        let tmp = TempDir::new().unwrap();
        assert_eq!(skip_unread_tail_in(tmp.path()), Ok(0));
        assert!(!tmp.path().join(OFFSET_FILENAME).exists());
    }

    #[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);
    }
}