openlatch-client 0.1.18

//! Atomic offset-file helpers shared by the durable outbox and the
//! fallback-replay queue.

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

/// Read the persisted byte offset. Returns `0` when the file is missing,
/// unreadable, or contains a non-numeric value.
pub(crate) fn read_offset(path: &Path) -> u64 {
    std::fs::read_to_string(path)
        .ok()
        .and_then(|s| s.trim().parse::<u64>().ok())
        .unwrap_or(0)
}

/// Persist a byte offset atomically via tmp + fsync + rename.
pub(crate) fn write_offset(path: &Path, value: u64) {
    if let Some(parent) = path.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    let tmp = path.with_extension("offset.tmp");
    match std::fs::File::create(&tmp) {
        Ok(mut f) => {
            if writeln!(f, "{value}").is_ok() && f.sync_all().is_ok() {
                let _ = std::fs::rename(&tmp, path);
            }
        }
        Err(e) => {
            tracing::debug!(
                error = %e,
                "offset: could not persist value — will re-scan on next pass"
            );
        }
    }
}

/// Stats returned by [`advance_past_oldest`].
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) struct AdvanceStats {
    /// Byte offset after the walk (≥ `cursor`).
    pub new_offset: u64,
    /// Non-empty JSONL entries the cursor advanced past.
    pub dropped: u64,
}

/// Walk JSONL lines forward from `cursor` until `excess` bytes have been
/// covered, returning the new offset on a clean line boundary. Returns
/// `None` (with no offset write) if the file can't be opened, the seek
/// fails, or no progress was made.
///
/// The caller is responsible for emitting tracing + telemetry and for
/// persisting the new offset via [`write_offset`] — this function does
/// the line walk only.
pub(crate) fn advance_past_oldest(path: &Path, cursor: u64, excess: u64) -> Option<AdvanceStats> {
    if excess == 0 {
        return None;
    }
    let file = std::fs::File::open(path).ok()?;
    let mut reader = BufReader::new(file);
    if reader.seek(SeekFrom::Start(cursor)).is_err() {
        return None;
    }
    let mut dropped: u64 = 0;
    let mut new_offset = cursor;
    for line in reader.lines() {
        let raw = line.ok()?;
        let line_len = raw.len() as u64 + 1;
        new_offset += line_len;
        if !raw.trim().is_empty() {
            dropped += 1;
        }
        if new_offset - cursor >= excess {
            break;
        }
    }
    if new_offset == cursor {
        return None;
    }
    Some(AdvanceStats {
        new_offset,
        dropped,
    })
}

/// Count non-empty JSONL entries from `cursor` to EOF. Returns 0 on any
/// read error.
pub(crate) fn count_entries_from(path: &Path, cursor: u64) -> u64 {
    let Ok(file) = std::fs::File::open(path) else {
        return 0;
    };
    let mut reader = BufReader::new(file);
    if reader.seek(SeekFrom::Start(cursor)).is_err() {
        return 0;
    }
    reader
        .lines()
        .map_while(Result::ok)
        .filter(|l| !l.trim().is_empty())
        .count() as u64
}

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

    #[test]
    fn read_offset_returns_zero_for_missing_file() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("missing.offset");
        assert_eq!(read_offset(&path), 0);
    }

    #[test]
    fn write_then_read_round_trips_value() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("v.offset");
        write_offset(&path, 4_242);
        assert_eq!(read_offset(&path), 4_242);
    }

    #[test]
    fn read_offset_returns_zero_for_garbage() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("garbage.offset");
        std::fs::write(&path, b"not a number\n").unwrap();
        assert_eq!(read_offset(&path), 0);
    }

    fn write_lines(tmp: &TempDir, name: &str, lines: &[&str]) -> std::path::PathBuf {
        let path = tmp.path().join(name);
        let mut f = std::fs::File::create(&path).unwrap();
        for l in lines {
            writeln!(f, "{l}").unwrap();
        }
        path
    }

    #[test]
    fn advance_past_oldest_walks_to_line_boundary() {
        let tmp = TempDir::new().unwrap();
        let path = write_lines(
            &tmp,
            "data.jsonl",
            &[r#"{"id":"a"}"#, r#"{"id":"b"}"#, r#"{"id":"c"}"#],
        );
        // First two entries: 11 bytes each (10 + newline). excess=12 → advance past two.
        let stats = advance_past_oldest(&path, 0, 12).unwrap();
        assert_eq!(stats.dropped, 2);
        assert_eq!(stats.new_offset, 22);
    }

    #[test]
    fn advance_past_oldest_returns_none_when_no_progress() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("empty.jsonl");
        std::fs::write(&path, b"").unwrap();
        assert!(advance_past_oldest(&path, 0, 100).is_none());
    }

    #[test]
    fn count_entries_from_skips_blank_lines() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("data.jsonl");
        std::fs::write(&path, b"{\"id\":\"a\"}\n\n{\"id\":\"b\"}\n").unwrap();
        assert_eq!(count_entries_from(&path, 0), 2);
    }
}