//! This phone's copy of the transcripts it has already been sent, so
//! reopening a session does not download it again. Ported from
//! `app/.../TranscriptCache.kt`; see `docs/TRANSCRIPT_CACHE.md`
//! for the design and `docs/CLIENT_CORE.md` for how this file corresponds to it.
//!
//! What is stored is the server's own JSON for one event per line, in
//! transcript order. Reading the cache means running the same [`seq_of`]
//! the network path runs, so a cached transcript and a fetched one cannot
//! draw differently, and an event type this build does not know keeps
//! every field it arrived with for the build that will. Rows are
//! deliberately *not* what is stored: a row is a rendering, and a cache of
//! rows would need throwing away on every update that touched the fold.
//!
//! Four rules run through all of it:
//! 1. what is on screen is what the server's transcript says, in order,
//!    with nothing missing -- the cache is a copy and is never inferred,
//!    folded or edited here;
//! 2. a cached line is never ahead of the live cursor, and the cursor never
//!    ahead of the cache;
//! 3. the cache is never load-bearing -- missing, evicted, damaged or
//!    unwritable all degrade to a cold open, never to a blank or a wrong
//!    screen;
//! 4. a line already on the phone is not fetched again.
//!
//! No JSON parser here: what it needs off a line is the sequence number and
//! whether the line is a streamed delta, both read with a regex-free scan
//! (see [`seq_of`] and [`is_delta`]). A line it cannot read that way is
//! treated as damage.

use std::collections::VecDeque;
use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

/// How much of this phone's cache directory all of one server's transcripts
/// may take. A dozen of the largest transcripts seen in the dev VM (21 MB
/// for 24,000 events) and a small fraction of a phone. A number to revisit
/// against real use rather than a measurement of anything.
pub const CACHE_BUDGET_BYTES: u64 = 256_000_000;

/// What the newest cached line says, which is what the probe checks against
/// the server.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CachedTail {
    pub seq: u64,
    pub line: String,
}

/// This phone's cache root for one server, holding one directory per session.
pub struct TranscriptCache {
    root: PathBuf,
    warn: std::sync::Arc<dyn Fn(&str) + Send + Sync>,
}

impl TranscriptCache {
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self::with_warn(root, |msg| eprintln!("ai-app: {msg}"))
    }

    pub fn with_warn(
        root: impl Into<PathBuf>,
        warn: impl Fn(&str) + Send + Sync + 'static,
    ) -> Self {
        Self {
            root: root.into(),
            warn: std::sync::Arc::new(warn),
        }
    }

    /// The cache for one session, whether or not anything has been stored
    /// for it yet.
    pub fn session(&self, id: &str) -> SessionCache {
        SessionCache::new(self.root.join(id), self.warn.clone())
    }

    /// Deletes every session directory not in `ids`, called after a
    /// successful list fetch. The path out for a session deleted on
    /// another device: nothing here would otherwise hear about it, and
    /// unlike a draft's few bytes what it leaves behind is megabytes.
    pub fn retain_only(&self, ids: &std::collections::HashSet<String>) {
        guard_io((), self.warn.as_ref(), || {
            for dir in session_dirs(&self.root)? {
                if let Some(name) = dir.file_name().and_then(|n| n.to_str())
                    && !ids.contains(name)
                {
                    fs::remove_dir_all(&dir)?;
                }
            }
            Ok(())
        });
    }

    /// Deletes least-recently-touched session directories, never `keep`,
    /// until the whole of this server's cache is under `budget`.
    /// Least-recently-touched rather than largest: what a reader is likely
    /// to open again is what they opened last, and evicting the big ones
    /// first would empty the cache for exactly the conversations it exists
    /// for.
    pub fn evict_to_budget(&self, keep: &str, budget: u64) {
        guard_io((), self.warn.as_ref(), || {
            let mut dirs: Vec<(PathBuf, std::time::SystemTime)> = session_dirs(&self.root)?
                .into_iter()
                .map(|d| {
                    let modified = fs::metadata(&d)
                        .and_then(|m| m.modified())
                        .unwrap_or(std::time::UNIX_EPOCH);
                    (d, modified)
                })
                .collect();
            dirs.sort_by_key(|(_, modified)| *modified);
            let mut total: u64 = dirs.iter().map(|(d, _)| dir_size(d)).sum();
            for (dir, _) in dirs {
                if total <= budget {
                    break;
                }
                if dir.file_name().and_then(|n| n.to_str()) == Some(keep) {
                    continue;
                }
                let was = dir_size(&dir);
                if fs::remove_dir_all(&dir).is_ok() {
                    total -= was;
                }
            }
            Ok(())
        });
    }

    pub fn purge_all(&self) {
        guard_io((), self.warn.as_ref(), || {
            if self.root.is_dir() {
                fs::remove_dir_all(&self.root)?;
            }
            Ok(())
        });
    }
}

fn session_dirs(root: &Path) -> io::Result<Vec<PathBuf>> {
    if !root.is_dir() {
        return Ok(Vec::new());
    }
    Ok(fs::read_dir(root)?
        .filter_map(|e| e.ok())
        .map(|e| e.path())
        .filter(|p| p.is_dir())
        .collect())
}

fn dir_size(path: &Path) -> u64 {
    let Ok(entries) = fs::read_dir(path) else {
        return 0;
    };
    entries
        .filter_map(|e| e.ok())
        .map(|e| {
            let p = e.path();
            if p.is_dir() {
                dir_size(&p)
            } else {
                fs::metadata(&p).map(|m| m.len()).unwrap_or(0)
            }
        })
        .sum()
}

/// One session's cached lines, as a directory of chunks.
///
/// A chunk is a set of lines *and a claim about what they cover*, and the
/// two are not the same thing: a coalesced page joins each run of streamed
/// deltas into one event carrying the seq of the run's oldest delta, so a
/// page whose newest event is seq 1,200 may cover everything up to the
/// 1,650 it was fetched with, and nothing in the lines says so. So coverage
/// is the half-open range in the file's name:
/// `<first>-<end>.rows.jsonl` (a coalesced page; `end` is the `before` it
/// was fetched with) or `<first>-<end>.raw.jsonl` (an uncoalesced page, or a
/// closed live run); `<first>-open.raw.jsonl` is the live run, whose end is
/// its last line's seq + 1.
///
/// Two chunks are adjacent when one's `end` is the other's `first`. Only
/// the contiguous run ending at the newest chunk -- the **suffix** -- is
/// ever served: chunks behind a gap are kept, because the gap is usually
/// closed by paging back through it, but nothing is served across one.
///
/// **The newest chunk is always raw**, which is what makes the stream
/// cursor and the probe well defined.
///
/// Nothing here is load-bearing. Every operation that touches the disk
/// answers as though the cache were empty when it cannot, and a write
/// failure disables writing for the rest of this instance's life so that a
/// full disk costs one log line rather than one per delta.
///
/// A `Mutex` around the writer state stands in for Kotlin's `@Synchronized`:
/// the stream appends live events from its own thread while a reader
/// scrolling back reads pages from another, and this is what keeps the open
/// chunk's name, its end and its writer from being read half-rotated.
pub struct SessionCache {
    dir: PathBuf,
    warn: std::sync::Arc<dyn Fn(&str) + Send + Sync>,
    state: Mutex<WriterState>,
}

#[derive(Default)]
struct WriterState {
    /// Set by the first write that fails: a second would fail the same way,
    /// once per delta.
    disabled: bool,
    writer: Option<fs::File>,
    open_file: Option<PathBuf>,
    open_end: u64,
}

impl SessionCache {
    fn new(dir: PathBuf, warn: std::sync::Arc<dyn Fn(&str) + Send + Sync>) -> Self {
        Self {
            dir,
            warn,
            state: Mutex::new(WriterState::default()),
        }
    }

    /// The newest line of the suffix, or `None` when there is none or the
    /// newest chunk is not raw.
    pub fn tail(&self) -> Option<CachedTail> {
        self.guard(None, |this, state| {
            let Some(newest) = this.suffix(state)?.last().cloned() else {
                return Ok(None);
            };
            let mut found = None;
            this.each_line(state, &newest, |line| {
                found = Some(CachedTail {
                    seq: seq_of(line).expect("chunk lines are checked in each_line"),
                    line: line.to_string(),
                });
                false
            });
            Ok(found)
        })
    }

    /// The newest `limit` lines of the suffix, oldest first -- the opening window.
    pub fn newest(&self, limit: usize) -> Vec<String> {
        self.guard(Vec::new(), |this, state| {
            let mut taken: VecDeque<String> = VecDeque::new();
            for chunk in this.suffix(state)?.iter().rev() {
                if taken.len() >= limit {
                    break;
                }
                this.each_line(state, chunk, |line| {
                    taken.push_front(line.to_string());
                    taken.len() < limit
                });
            }
            Ok(taken.into_iter().collect())
        })
    }

    /// The page of lines before `before`, oldest first, or `None` when the
    /// cache cannot answer.
    ///
    /// `None` is a miss -- the suffix does not cover the ground immediately
    /// below `before` -- and means the server has to be asked. Deliberately
    /// not an empty list: an empty page is how the screen is told it has
    /// reached the start of the conversation.
    ///
    /// With `rows` the count is rows rather than lines, mirroring the
    /// server's `parse_coalesced`. The deltas are not joined here -- the
    /// fold does that, and the joined row keeps the seq of its first delta
    /// either way.
    pub fn page(&self, before: u64, limit: usize, rows: bool) -> Option<Vec<String>> {
        self.guard(None, |this, state| {
            let suffix = this.suffix(state)?;
            let Some(newest) = suffix.last().cloned() else {
                return Ok(None);
            };
            // Above what is held, or at or below where it starts: either
            // way the run the caller is scrolling into is not continuous
            // with this one, and only the server has it.
            if before > newest.end || before <= suffix.first().unwrap().first {
                return Ok(None);
            }
            let mut taken: VecDeque<String> = VecDeque::new();
            let mut counted = 0usize;
            let mut in_run = false;
            let mut wanting = true;
            for chunk in suffix.iter().rev() {
                if !wanting {
                    break;
                }
                if chunk.first >= before {
                    continue;
                }
                this.each_line(state, chunk, |line| {
                    // The page is what is *before* the cursor; the rows at
                    // or above it are already on screen.
                    let seq = seq_of(line).expect("chunk lines are checked in each_line");
                    if seq >= before {
                        return true;
                    }
                    if rows {
                        let delta = is_delta(line);
                        // Stop only between rows: a delta continuing the
                        // run being gathered is part of a row already
                        // counted, and breaking on it would drop the half
                        // of that row already taken.
                        if counted >= limit && !(delta && in_run) {
                            wanting = false;
                        } else {
                            if !delta || !in_run {
                                counted += 1;
                            }
                            in_run = delta;
                        }
                    } else if taken.len() >= limit {
                        wanting = false;
                    }
                    if wanting {
                        taken.push_front(line.to_string());
                    }
                    wanting
                });
            }
            Ok(Some(taken.into_iter().collect()))
        })
    }

    /// The `end` of the nearest chunk at or below `before`, which is the
    /// floor a fetched page is asked with so that it stops where this
    /// phone's copy starts. `None` when there is no such chunk.
    pub fn covered_up_to(&self, before: u64) -> Option<u64> {
        self.guard(None, |this, state| {
            Ok(this
                .chunks(state)?
                .iter()
                .map(|c| c.end)
                .filter(|&end| end <= before)
                .max())
        })
    }

    /// Stores a fetched page covering `[first, end)`; `false` when it was
    /// not stored.
    ///
    /// Refused when it overlaps a chunk already here, because there is no
    /// clean cut: a coalesced event cannot be split at a seq inside its own
    /// delta run. The caller keeps that from arising by bounding what it
    /// fetches, and this is the guard for a page that arrives anyway.
    ///
    /// The newest chunk is never stored through here: the opening window
    /// and every live frame go through [`Self::append`], which is what
    /// keeps the newest chunk raw and open.
    pub fn store_page(&self, lines: &[String], first: u64, end: u64, rows: bool) -> bool {
        self.guard(false, |this, state| {
            if state.disabled || lines.is_empty() || end <= first {
                return Ok(false);
            }
            if this
                .chunks(state)?
                .iter()
                .any(|c| first < c.end && c.first < end)
            {
                return Ok(false);
            }
            debug_assert!(
                lines.iter().all(|l| !l.contains('\n')),
                "a stored page's lines must each be one line"
            );
            fs::create_dir_all(&this.dir)?;
            let kind = if rows { "rows" } else { "raw" };
            let mut content = lines.join("\n");
            content.push('\n');
            fs::write(
                this.dir.join(format!("{first}-{end}.{kind}.jsonl")),
                content,
            )?;
            Ok(true)
        })
    }

    /// Appends one live event, which is also how a freshly fetched opening
    /// window is stored.
    ///
    /// A seq equal to the open chunk's end extends it. A larger one is a
    /// gap -- which is what a `reset` looks like from here -- and closes
    /// the open chunk under the end it turned out to have. A smaller one is
    /// already covered and is ignored; the SSE contract is `seq > after`.
    pub fn append(&self, line: &str, seq: u64) {
        self.guard((), |this, state| {
            if state.disabled {
                return Ok(());
            }
            let Some(writer) = this.writer_for(state, seq)? else {
                return Ok(());
            };
            // Written as it arrived. A newline inside it would split one
            // event into two unreadable halves. No source here can produce
            // one -- an SSE `data:` field cannot hold a raw newline, and a
            // fetched line is one element of a compact JSON array -- but
            // that is a fact about the *server's* serializer rather than
            // anything this file controls, so it is checked rather than
            // trusted.
            debug_assert!(
                !line.contains('\n'),
                "a cached transcript line must be one line: {line}"
            );
            use std::io::Write;
            writer.write_all(line.as_bytes())?;
            writer.write_all(b"\n")?;
            state.open_end = seq + 1;
            Ok(())
        });
    }

    /// Flushes what [`Self::append`] has buffered.
    pub fn flush(&self) {
        self.guard((), |_this, state| {
            if let Some(writer) = state.writer.as_mut() {
                use std::io::Write;
                writer.flush()?;
            }
            Ok(())
        });
    }

    /// What [`Self::purge`] would discard, for the reload row in session settings.
    pub fn bytes(&self) -> u64 {
        self.guard(0, |this, _state| Ok(dir_size(&this.dir)))
    }

    /// Marks this session as visited, which is what eviction ranks by.
    pub fn touch(&self) {
        self.guard((), |this, _state| {
            if this.dir.is_dir() {
                let now = std::time::SystemTime::now();
                filetime_set_modified(&this.dir, now)?;
            }
            Ok(())
        });
    }

    pub fn purge(&self) {
        self.guard((), |this, state| {
            close_writer(state);
            if this.dir.is_dir() {
                fs::remove_dir_all(&this.dir)?;
            }
            Ok(())
        });
    }

    // -- chunks ------------------------------------------------------------------------------

    /// Every chunk on disk, oldest first. A name this does not recognise is
    /// not ours and is ignored. Recomputed per operation rather than kept:
    /// another operation may have changed the directory.
    fn chunks(&self, state: &mut WriterState) -> io::Result<Vec<Chunk>> {
        if let Some(writer) = state.writer.as_mut() {
            use std::io::Write;
            let _ = writer.flush();
        }
        if !self.dir.is_dir() {
            return Ok(Vec::new());
        }
        let mut out = Vec::new();
        for entry in fs::read_dir(&self.dir)? {
            let entry = entry?;
            let name = entry.file_name();
            let Some(name) = name.to_str() else { continue };
            let Some((first, end_str, kind)) = parse_chunk_name(name) else {
                continue;
            };
            let open = end_str == "open";
            let end = if open {
                self.open_end_of(state, &entry.path(), first)
            } else {
                end_str.parse::<u64>().ok()
            };
            // A chunk covering nothing is one that was created and never
            // written to -- an append whose very first write failed. It
            // says nothing, so it is not a chunk.
            if let Some(end) = end
                && end > first
            {
                out.push(Chunk {
                    file: entry.path(),
                    first,
                    end,
                    open,
                    rows: kind == "rows",
                });
            }
        }
        out.sort_by_key(|c| c.first);
        Ok(out)
    }

    /// The open chunk's end: its last line's seq plus one, or the in-memory
    /// end while this instance is the one writing it.
    ///
    /// An open chunk whose last line cannot be read is this app having died
    /// mid-write. That line is dropped and the file truncated to the last
    /// good one, which is the one place damage is repaired rather than
    /// discarded.
    fn open_end_of(&self, state: &WriterState, file: &Path, first: u64) -> Option<u64> {
        if state.open_file.as_deref() == Some(file) && state.open_end > 0 {
            return Some(state.open_end);
        }
        let _ = repair_tail(file);
        let mut end = first;
        each_line_backwards(file, |_offset, line| {
            if let Some(seq) = seq_of(line) {
                end = seq + 1;
            }
            false
        });
        Some(end)
    }

    /// The contiguous run of adjacent chunks ending at the newest one,
    /// oldest first.
    ///
    /// A newest chunk that is not raw cannot happen while this code is the
    /// only writer, and means the directory is not to be trusted -- so the
    /// session is discarded.
    fn suffix(&self, state: &mut WriterState) -> io::Result<Vec<Chunk>> {
        let all = self.chunks(state)?;
        let Some(newest) = all.last() else {
            return Ok(Vec::new());
        };
        if newest.rows {
            return Err(damaged(&newest.file));
        }
        let mut run: VecDeque<Chunk> = VecDeque::new();
        run.push_front(newest.clone());
        let mut index = all.len() - 1;
        while index > 0 && all[index - 1].end == run.front().unwrap().first {
            index -= 1;
            run.push_front(all[index].clone());
        }
        Ok(run.into_iter().collect())
    }

    /// Each line of `chunk`, newest first, until `take` says stop.
    ///
    /// Damage anywhere but at the tail of the open chunk was not written by
    /// this code, and there is no honest way to say what a chunk covers
    /// with a line of it unreadable -- so it is treated as damage rather
    /// than served partially. Propagated as an `Err` from the caller's
    /// `guard` closure via a panic-free sentinel: callers pass a `take`
    /// that never sees a non-seq line reach it, because `each_line_backwards`
    /// is stopped the moment one does and the damage is reported by the
    /// `Result` this returns.
    fn each_line(
        &self,
        _state: &mut WriterState,
        chunk: &Chunk,
        mut take: impl FnMut(&str) -> bool,
    ) {
        each_line_backwards(&chunk.file, |_offset, line| {
            if seq_of(line).is_none() {
                DAMAGED.with(|cell| *cell.borrow_mut() = Some(chunk.file.clone()));
                return false;
            }
            take(line)
        });
    }

    // -- writing -----------------------------------------------------------------------------

    /// The writer for the chunk `seq` belongs in, opening or rotating one
    /// as it has to.
    fn writer_for<'s>(
        &self,
        state: &'s mut WriterState,
        seq: u64,
    ) -> io::Result<Option<&'s mut fs::File>> {
        if state.writer.is_some() {
            if seq == state.open_end {
                return Ok(state.writer.as_mut());
            }
            if seq < state.open_end {
                return Ok(None);
            }
            // A gap: what this instance has written covers up to `open_end`,
            // and that is the name the chunk gets before a new one starts
            // at the arriving seq.
            let end = state.open_end;
            self.close_open_chunk(state, end);
        }
        fs::create_dir_all(&self.dir)?;
        // An open chunk left by an earlier instance, or by an earlier screen.
        let existing = self.chunks(state)?.into_iter().rfind(|c| c.open);
        if let Some(existing) = existing {
            if seq < existing.end {
                return Ok(None);
            }
            if seq == existing.end {
                state.open_file = Some(existing.file.clone());
                state.open_end = existing.end;
                let file = fs::OpenOptions::new().append(true).open(&existing.file)?;
                state.writer = Some(file);
                return Ok(state.writer.as_mut());
            }
            let first = existing.first;
            let end = existing.end;
            rename_chunk(&existing.file, &self.dir, first, end);
        }
        // A chunk that was created and never written to would otherwise be
        // left behind under a name a second one is about to want; it
        // covers nothing, so nothing is lost with it.
        if let Ok(entries) = fs::read_dir(&self.dir) {
            for entry in entries.flatten() {
                let name = entry.file_name();
                if let Some(name) = name.to_str()
                    && let Some((_, end, _)) = parse_chunk_name(name)
                    && end == "open"
                    && entry.metadata().map(|m| m.len()).unwrap_or(1) == 0
                {
                    let _ = fs::remove_file(entry.path());
                }
            }
        }
        let file_path = self.dir.join(format!("{seq}-open.raw.jsonl"));
        let file = fs::OpenOptions::new()
            .create(true)
            .write(true)
            .truncate(true)
            .open(&file_path)?;
        state.open_file = Some(file_path);
        state.open_end = seq;
        state.writer = Some(file);
        Ok(state.writer.as_mut())
    }

    /// Renames the open chunk to the range it turned out to cover, so it
    /// stops being open.
    fn close_open_chunk(&self, state: &mut WriterState, end: u64) {
        let file = state.open_file.clone();
        close_writer(state);
        let Some(file) = file else { return };
        if let Some((first, _, _)) = file
            .file_name()
            .and_then(|n| n.to_str())
            .and_then(parse_chunk_name)
        {
            rename_chunk(&file, &self.dir, first, end);
        }
    }

    // -- failure -----------------------------------------------------------------------------

    /// Runs `body`, answering `if_broken` when the directory cannot give a
    /// real answer. None of this is reported on screen: every read here has
    /// a network path beside it producing the same result, and the reader
    /// has nothing to do about it. Damage discards this session's cache,
    /// which makes the next open an ordinary cold one.
    fn guard<T>(
        &self,
        if_broken: T,
        body: impl FnOnce(&Self, &mut WriterState) -> io::Result<T>,
    ) -> T {
        let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner());
        // A disk that refused once will refuse again, once per delta, so
        // the first refusal is also the last.
        if state.disabled {
            return if_broken;
        }
        DAMAGED.with(|cell| *cell.borrow_mut() = None);
        let result = body(self, &mut state);
        // Damage takes priority over whatever `body` returned, `Ok` or
        // `Err`: `suffix` signals it by returning `Err(damaged(..))`
        // precisely so this check catches it before the branch below
        // mistakes it for a real I/O failure and disables the whole cache
        // over one corrupt session.
        if let Some(file) = DAMAGED.with(|cell| cell.borrow_mut().take()) {
            (self.warn)(&format!(
                "transcript cache damaged at {}; discarding {}",
                file.display(),
                self.dir.display()
            ));
            close_writer(&mut state);
            let _ = fs::remove_dir_all(&self.dir);
            return if_broken;
        }
        match result {
            Ok(value) => value,
            Err(e) => {
                (self.warn)(&format!("transcript cache unusable: {e}"));
                state.disabled = true;
                close_writer(&mut state);
                if_broken
            }
        }
    }
}

thread_local! {
    /// How [`SessionCache::each_line`] reports a line it cannot make sense
    /// of back up to [`SessionCache::guard`], since the callback it hands
    /// `each_line_backwards` cannot itself return a `Result`. Thread-local
    /// rather than a field: the guard that reads it always runs on the same
    /// call stack that could have set it, one `guard` call at a time.
    static DAMAGED: std::cell::RefCell<Option<PathBuf>> = const { std::cell::RefCell::new(None) };
}

fn damaged(file: &Path) -> io::Error {
    DAMAGED.with(|cell| *cell.borrow_mut() = Some(file.to_path_buf()));
    io::Error::other("damaged chunk")
}

#[derive(Debug, Clone)]
struct Chunk {
    file: PathBuf,
    first: u64,
    end: u64,
    open: bool,
    rows: bool,
}

fn close_writer(state: &mut WriterState) {
    state.writer = None;
    state.open_file = None;
    state.open_end = 0;
}

fn rename_chunk(file: &Path, dir: &Path, first: u64, end: u64) {
    let _ = fs::rename(file, dir.join(format!("{first}-{end}.raw.jsonl")));
}

/// `<first>-<end|open>.<rows|raw>.jsonl`; anything else in the directory is
/// not ours.
fn parse_chunk_name(name: &str) -> Option<(u64, &str, &str)> {
    let rest = name.strip_suffix(".jsonl")?;
    let (rest, kind) = rest.rsplit_once('.')?;
    if kind != "rows" && kind != "raw" {
        return None;
    }
    let (first, end) = rest.split_once('-')?;
    let first = first.parse::<u64>().ok()?;
    if end != "open" && end.parse::<u64>().is_err() {
        return None;
    }
    Some((first, end, kind))
}

/// One line's sequence number, or `None` when the line is not one of ours.
///
/// A hand-rolled scan rather than a JSON parse, so this module carries no
/// parser and stays testable with no server: the seq is the first field the
/// server writes, so the first match is the top-level one.
pub fn seq_of(line: &str) -> Option<u64> {
    find_number_field(line, "seq")
}

/// Whether a line is one streamed piece of a reply, which is what makes a
/// run of them one row.
pub fn is_delta(line: &str) -> bool {
    find_string_field(line, "type").as_deref() == Some("assistantText")
}

/// The value of `"key":N` (any amount of whitespace around the colon), or
/// `None`. Mirrors `Regex(""""seq"\s*:\s*(\d+)""")`'s first match.
fn find_number_field(line: &str, key: &str) -> Option<u64> {
    let pattern = format!("\"{key}\"");
    let at = line.find(&pattern)?;
    let after_key = &line[at + pattern.len()..];
    let after_colon = after_key.trim_start().strip_prefix(':')?;
    let digits: String = after_colon
        .trim_start()
        .chars()
        .take_while(|c| c.is_ascii_digit())
        .collect();
    if digits.is_empty() {
        None
    } else {
        digits.parse().ok()
    }
}

/// The value of `"key":"..."`, or `None`. Mirrors
/// `Regex(""""type"\s*:\s*"([^"]*)"""")`'s first match.
fn find_string_field(line: &str, key: &str) -> Option<String> {
    let pattern = format!("\"{key}\"");
    let at = line.find(&pattern)?;
    let after_key = &line[at + pattern.len()..];
    let after_colon = after_key.trim_start().strip_prefix(':')?;
    let after_quote = after_colon.trim_start().strip_prefix('"')?;
    let end = after_quote.find('"')?;
    Some(after_quote[..end].to_string())
}

/// How much of a file is read at a time when walking it backwards. One
/// block covers a page of a transcript comfortably, and the walk stops as
/// soon as the caller has what it asked for.
const READ_BLOCK: usize = 64 * 1024;

/// Calls `on_line` with each non-blank line of `file`, **newest first**,
/// along with the byte offset it starts at, until `on_line` answers false.
///
/// Every question the cache is asked is about the newest end of a chunk,
/// and a live run reaches the size of the conversation, so reading forwards
/// means reading a transcript to answer with the last eighty lines of it.
///
/// Splitting on bytes is safe because the separator is `\n`, which cannot
/// occur inside a multi-byte UTF-8 sequence; each line is decoded whole. A
/// missing file yields nothing.
fn each_line_backwards(file: &Path, mut on_line: impl FnMut(u64, &str) -> bool) {
    use std::io::{Read, Seek, SeekFrom};
    let Ok(mut handle) = fs::File::open(file) else {
        return;
    };
    let Ok(len) = handle.metadata().map(|m| m.len()) else {
        return;
    };
    let mut unread = len;
    let mut pending: Vec<u8> = Vec::new();
    while unread > 0 {
        let take = READ_BLOCK.min(unread as usize);
        let start = unread - take as u64;
        let mut block = vec![0u8; take];
        if handle.seek(SeekFrom::Start(start)).is_err() || handle.read_exact(&mut block).is_err() {
            return;
        }
        let mut buffer = block;
        buffer.extend_from_slice(&pending);
        // `buffer` is now `block` followed by `pending`; walk it backwards.
        let mut line_end = buffer.len();
        let mut at = buffer.len() as isize - 1;
        while at >= 0 {
            if buffer[at as usize] == b'\n' {
                let line_bytes = &buffer[at as usize + 1..line_end];
                let line = String::from_utf8_lossy(line_bytes);
                if !line.trim().is_empty() && !on_line(start + at as u64 + 1, &line) {
                    return;
                }
                line_end = at as usize;
            }
            at -= 1;
        }
        pending = buffer[..line_end].to_vec();
        unread = start;
    }
    // The first line of a file has no newline before it to be found.
    let first = String::from_utf8_lossy(&pending);
    if !first.trim().is_empty() {
        on_line(0, &first);
    }
}

/// Drops a final line that is not one of ours, by truncating the file to
/// where it starts.
///
/// This app having died mid-write is the one kind of damage that is
/// repaired rather than discarded: the tail of an append-only file is the
/// only place a partial line can be. A second bad line is not this, and is
/// left for the read path to notice.
fn repair_tail(file: &Path) -> io::Result<()> {
    let mut truncate_to: Option<u64> = None;
    each_line_backwards(file, |offset, line| {
        if seq_of(line).is_none() {
            truncate_to = Some(offset);
        }
        false
    });
    if let Some(truncate_to) = truncate_to {
        let handle = fs::OpenOptions::new().write(true).open(file)?;
        handle.set_len(truncate_to)?;
    }
    Ok(())
}

/// Runs `body`, translating an I/O or permission failure into `if_broken`
/// and a warning -- the disk half of [`SessionCache::guard`], shared with
/// [`TranscriptCache`]'s own maintenance.
fn guard_io<T>(
    if_broken: T,
    warn: &(impl Fn(&str) + ?Sized),
    body: impl FnOnce() -> io::Result<T>,
) -> T {
    match body() {
        Ok(value) => value,
        Err(e) => {
            warn(&format!("transcript cache unusable: {e}"));
            if_broken
        }
    }
}

/// Sets a path's modified time, without pulling in a crate for it: a single
/// `utimensat`-backed call would be one more platform-specific dependency
/// for one call site, so this touches the file instead, which every
/// filesystem this runs on updates the mtime for.
fn filetime_set_modified(path: &Path, _when: std::time::SystemTime) -> io::Result<()> {
    use std::io::Write;
    // Rewriting a marker file's contents (rather than the directory itself,
    // which `std` has no portable "touch" for) bumps the directory's own
    // mtime on every filesystem this cache runs on, because creating or
    // truncating an entry inside a directory always updates that
    // directory's mtime.
    let marker = path.join(".touch");
    let mut f = fs::OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(true)
        .open(&marker)?;
    f.write_all(b"")?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashSet;

    fn cache(temp: &Path) -> TranscriptCache {
        let said: std::sync::Arc<Mutex<Vec<String>>> = Default::default();
        let said2 = said.clone();
        TranscriptCache::with_warn(temp.join("v1/host_8443"), move |msg| {
            said2.lock().unwrap().push(msg.to_string());
        })
    }

    /// Like `cache`, but also hands back the messages it warned with.
    fn cache_with_log(temp: &Path) -> (TranscriptCache, std::sync::Arc<Mutex<Vec<String>>>) {
        let said: std::sync::Arc<Mutex<Vec<String>>> = Default::default();
        let said2 = said.clone();
        (
            TranscriptCache::with_warn(temp.join("v1/host_8443"), move |msg| {
                said2.lock().unwrap().push(msg.to_string());
            }),
            said,
        )
    }

    fn line(seq: u64, kind: &str) -> String {
        format!(r#"{{"seq":{seq},"ts":1.5,"type":"{kind}","id":"x"}}"#)
    }
    fn tool_line(seq: u64) -> String {
        line(seq, "toolStart")
    }
    fn delta(seq: u64) -> String {
        line(seq, "assistantText")
    }

    fn dir_of(temp: &Path, id: &str) -> PathBuf {
        temp.join("v1/host_8443").join(id)
    }

    fn names(temp: &Path, id: &str) -> Vec<String> {
        let dir = dir_of(temp, id);
        let Ok(entries) = fs::read_dir(&dir) else {
            return Vec::new();
        };
        let mut names: Vec<String> = entries
            .filter_map(|e| e.ok())
            .filter_map(|e| e.file_name().into_string().ok())
            .filter(|n| n != ".touch")
            .collect();
        names.sort();
        names
    }

    fn write_chunk(temp: &Path, id: &str, name: &str, lines: &[String]) {
        let dir = dir_of(temp, id);
        fs::create_dir_all(&dir).unwrap();
        let mut content = lines.join("\n");
        content.push('\n');
        fs::write(dir.join(name), content).unwrap();
    }

    fn seqs(lines: &Option<Vec<String>>) -> Option<Vec<u64>> {
        lines
            .as_ref()
            .map(|ls| ls.iter().map(|l| seq_of(l).unwrap()).collect())
    }
    fn seqs_vec(lines: &[String]) -> Vec<u64> {
        lines.iter().map(|l| seq_of(l).unwrap()).collect()
    }

    #[test]
    fn an_appended_run_is_one_open_chunk_and_its_newest_line_is_the_tail() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        for seq in 1..=3u64 {
            session.append(&tool_line(seq), seq);
        }
        session.flush();

        assert_eq!(names(temp.path(), "s"), vec!["1-open.raw.jsonl"]);
        assert_eq!(
            session.tail(),
            Some(CachedTail {
                seq: 3,
                line: tool_line(3)
            })
        );
        assert_eq!(session.newest(2), vec![tool_line(2), tool_line(3)]);
        // More than there is is what there is, which is a short opening
        // window and not a failure.
        assert_eq!(session.newest(80).len(), 3);
    }

    #[test]
    fn a_gap_in_the_stream_closes_the_open_chunk_under_the_end_it_turned_out_to_have() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        for seq in 1..=3u64 {
            session.append(&tool_line(seq), seq);
        }
        // What a `reset` looks like from here: the next event is not the
        // one after the last.
        session.append(&tool_line(90), 90);
        session.flush();

        assert_eq!(
            names(temp.path(), "s"),
            vec!["1-4.raw.jsonl", "90-open.raw.jsonl"]
        );
        assert_eq!(session.newest(80), vec![tool_line(90)]);
        assert_eq!(
            session.tail(),
            Some(CachedTail {
                seq: 90,
                line: tool_line(90)
            })
        );
    }

    #[test]
    fn an_event_already_covered_is_not_written_again() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        for seq in 1..=3u64 {
            session.append(&tool_line(seq), seq);
        }
        session.append(&tool_line(2), 2);
        session.flush();

        assert_eq!(seqs_vec(&session.newest(80)), vec![1, 2, 3]);
    }

    #[test]
    fn an_adjacent_page_extends_the_suffix_and_a_gap_stops_it() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        for seq in 100..=102u64 {
            session.append(&tool_line(seq), seq);
        }
        session.flush();

        // Adjacent: its end is the open chunk's first.
        let page: Vec<String> = (60..100u64).map(tool_line).collect();
        assert!(session.store_page(&page, 60, 100, true));
        assert_eq!(seqs(&session.page(100, 2, false)), Some(vec![98, 99]));
        assert_eq!(seqs_vec(&session.newest(80)).first(), Some(&60));

        // Behind a gap: kept on disk, because paging usually closes the
        // gap, but never served across it.
        let page2: Vec<String> = (1..10u64).map(tool_line).collect();
        assert!(session.store_page(&page2, 1, 10, true));
        assert_eq!(session.page(10, 5, false), None);
        assert_eq!(seqs_vec(&session.newest(200)).first(), Some(&60));
    }

    #[test]
    fn a_page_that_overlaps_what_is_here_is_not_stored() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        session.append(&tool_line(100), 100);
        session.flush();
        let page: Vec<String> = (60..100u64).map(tool_line).collect();
        assert!(session.store_page(&page, 60, 100, true));

        let overlap: Vec<String> = (50..80u64).map(tool_line).collect();
        assert!(!session.store_page(&overlap, 50, 80, true));
        assert!(!session.store_page(&[], 40, 60, true));
        assert_eq!(
            names(temp.path(), "s"),
            vec!["100-open.raw.jsonl", "60-100.rows.jsonl"]
        );
    }

    #[test]
    fn a_miss_is_null_and_never_an_empty_page() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        for seq in 100..=102u64 {
            session.append(&tool_line(seq), seq);
        }
        session.flush();

        // At or below where the run starts, so what the reader is
        // scrolling into is the server's.
        assert_eq!(session.page(100, 40, true), None);
        assert_eq!(session.page(40, 40, true), None);
        assert_eq!(cache.session("never-visited").page(100, 40, true), None);
    }

    #[test]
    fn a_page_starts_from_anywhere_inside_the_run_not_only_at_a_boundary() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        for seq in 1..=10u64 {
            session.append(&tool_line(seq), seq);
        }
        session.flush();

        assert_eq!(seqs(&session.page(8, 3, false)), Some(vec![5, 6, 7]));
        assert_eq!(seqs(&session.page(8, 99, false)), Some((1..=7).collect()));
    }

    #[test]
    fn a_page_counted_in_rows_folds_each_delta_run_into_one_and_cuts_only_between_rows() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        // Two replies of three deltas each, split by a tool call: the same
        // fixture as the server's `coalescing_counts_rows_and_joins_delta_runs`.
        let lines = vec![
            delta(1),
            delta(2),
            delta(3),
            tool_line(4),
            delta(5),
            delta(6),
            delta(7),
            tool_line(8),
        ];
        write_chunk(temp.path(), "s", "1-9.raw.jsonl", &lines);
        session.append(&tool_line(9), 9);
        session.flush();

        // Three rows: the tool call at 8, the run 5..7, and the tool call
        // at 4. The cut lands between rows, so the older run is not
        // started.
        assert_eq!(seqs(&session.page(9, 3, true)), Some(vec![4, 5, 6, 7, 8]));
        // One row is one whole run, however many deltas it is made of.
        assert_eq!(seqs(&session.page(9, 1, true)), Some(vec![8]));
        // A page of lines counts lines, which is what the anchor restore
        // asks for.
        assert_eq!(seqs(&session.page(9, 2, false)), Some(vec![7, 8]));
    }

    #[test]
    fn a_row_page_crosses_a_chunk_boundary_and_stops_short_at_the_oldest_chunk() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        write_chunk(
            temp.path(),
            "s",
            "5-9.raw.jsonl",
            &[delta(5), delta(6), tool_line(7), delta(8)],
        );
        session.append(&delta(9), 9);
        session.append(&tool_line(10), 10);
        session.flush();

        // A run straddling the boundary is one row, as it will be once folded.
        assert_eq!(seqs(&session.page(11, 2, true)), Some(vec![8, 9, 10]));
        // Asking for more rows than the suffix holds is a short page, not a
        // failure and not a claim that the conversation starts here.
        assert_eq!(seqs(&session.page(11, 40, true)), Some((5..=10).collect()));
    }

    #[test]
    fn the_floor_for_a_fetch_is_the_nearest_chunk_at_or_below_it() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        write_chunk(
            temp.path(),
            "s",
            "1-10.rows.jsonl",
            &(1..10u64).map(tool_line).collect::<Vec<_>>(),
        );
        write_chunk(
            temp.path(),
            "s",
            "10-40.rows.jsonl",
            &(10..40u64).map(tool_line).collect::<Vec<_>>(),
        );
        session.append(&tool_line(90), 90);
        session.flush();

        // The run behind the gap, which is what makes the fetched page
        // adjacent to it.
        assert_eq!(session.covered_up_to(90), Some(40));
        assert_eq!(session.covered_up_to(41), Some(40));
        assert_eq!(session.covered_up_to(10), Some(10));
        // Nothing at or below the oldest chunk's start, so the page is
        // bounded only by its limit.
        assert_eq!(session.covered_up_to(9), None);
    }

    #[test]
    fn a_newest_chunk_that_is_not_raw_discards_the_session() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        write_chunk(
            temp.path(),
            "s",
            "1-10.rows.jsonl",
            &(1..10u64).map(tool_line).collect::<Vec<_>>(),
        );

        // Only reachable by dying between closing one live run and opening
        // the next, and there is no cursor to be read off a coalesced line
        // -- so the open is a cold one.
        assert_eq!(session.tail(), None);
        assert!(!dir_of(temp.path(), "s").exists());
    }

    #[test]
    fn a_half_written_last_line_is_dropped_and_the_file_repaired() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        let dir = dir_of(temp.path(), "s");
        fs::create_dir_all(&dir).unwrap();
        fs::write(
            dir.join("1-open.raw.jsonl"),
            format!("{}\n{}\n{{\"se", tool_line(1), tool_line(2)),
        )
        .unwrap();

        assert_eq!(
            session.tail(),
            Some(CachedTail {
                seq: 2,
                line: tool_line(2)
            })
        );
        assert_eq!(
            fs::read_to_string(dir.join("1-open.raw.jsonl")).unwrap(),
            format!("{}\n{}\n", tool_line(1), tool_line(2))
        );
        // And the run continues from where the good tail left off.
        session.append(&tool_line(3), 3);
        session.flush();
        assert_eq!(seqs_vec(&session.newest(80)), vec![1, 2, 3]);
    }

    #[test]
    fn damage_anywhere_else_discards_the_session_when_a_read_reaches_it() {
        let temp = tempfile::tempdir().unwrap();
        let (cache, said) = cache_with_log(temp.path());
        let session = cache.session("s");
        write_chunk(
            temp.path(),
            "s",
            "1-open.raw.jsonl",
            &[tool_line(1), "not ours".to_string(), tool_line(3)],
        );

        // Not seen by the tail, which reads the newest line and stops.
        assert_eq!(
            session.tail(),
            Some(CachedTail {
                seq: 3,
                line: tool_line(3)
            })
        );
        // Reached by a read that walks past it: what is served is nothing,
        // and the session opens cold from here on.
        assert_eq!(session.newest(80), Vec::<String>::new());
        assert!(!dir_of(temp.path(), "s").exists());
        assert!(said.lock().unwrap().iter().any(|m| m.contains("damaged")));
    }

    #[test]
    fn a_name_this_does_not_recognise_is_ignored() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        write_chunk(temp.path(), "s", "notes.txt", &["hello".to_string()]);
        write_chunk(temp.path(), "s", "1-open.raw.jsonl", &[tool_line(1)]);

        assert_eq!(
            session.tail(),
            Some(CachedTail {
                seq: 1,
                line: tool_line(1)
            })
        );
    }

    #[test]
    fn a_chunk_larger_than_one_read_block_is_walked_across_the_boundaries() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        // Well past the 64 kB block the backwards reader takes at a time,
        // so a page has to be stitched across several of them -- including
        // a line that straddles a boundary.
        let padding = "x".repeat(300);
        let lines: Vec<String> = (1..=500u64)
            .map(|seq| format!(r#"{{"seq":{seq},"ts":1.5,"type":"toolStart","id":"{padding}"}}"#))
            .collect();
        write_chunk(temp.path(), "s", "1-open.raw.jsonl", &lines);

        assert_eq!(session.tail().unwrap().seq, 500);
        assert_eq!(session.newest(80), lines[420..].to_vec());
        assert_eq!(session.page(401, 999, false), Some(lines[0..400].to_vec()));
        // And a non-ASCII line, whose bytes a naive split could cut through
        // a character.
        let accented =
            r#"{"seq":501,"ts":1.5,"type":"assistantText","delta":"héllo — ok"}"#.to_string();
        session.append(&accented, 501);
        session.flush();
        assert_eq!(session.tail().unwrap().line, accented);
    }

    #[test]
    fn eviction_takes_the_least_recently_touched_and_never_the_one_on_screen() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        for (at, id) in ["old", "middle", "open"].iter().enumerate() {
            write_chunk(
                temp.path(),
                id,
                "1-open.raw.jsonl",
                &(1..=50u64).map(tool_line).collect::<Vec<_>>(),
            );
            let when =
                std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000 + at as u64);
            filetime_set_modified(&dir_of(temp.path(), id), when).unwrap();
            // The mtime touch above always sets "now", not `when` (see its
            // own doc) -- space the three writes out in real time instead,
            // since only relative order matters to eviction.
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        let each = dir_size(&dir_of(temp.path(), "old"));

        // Room for two of the three, so the oldest goes -- and the session
        // being read never does, however long ago it was last touched.
        cache.evict_to_budget("open", each * 2);
        let mut remaining = fs::read_dir(temp.path().join("v1/host_8443"))
            .unwrap()
            .filter_map(|e| e.ok())
            .filter_map(|e| e.file_name().into_string().ok())
            .collect::<Vec<_>>();
        remaining.sort();
        assert_eq!(remaining, vec!["middle", "open"]);

        cache.evict_to_budget("open", 0);
        let mut remaining = fs::read_dir(temp.path().join("v1/host_8443"))
            .unwrap()
            .filter_map(|e| e.ok())
            .filter_map(|e| e.file_name().into_string().ok())
            .collect::<Vec<_>>();
        remaining.sort();
        assert_eq!(remaining, vec!["open"]);
    }

    #[test]
    fn retaining_deletes_exactly_the_sessions_the_server_no_longer_lists() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        for id in ["a", "b", "c"] {
            write_chunk(temp.path(), id, "1-open.raw.jsonl", &[tool_line(1)]);
        }

        cache.retain_only(&HashSet::from(["a".to_string(), "c".to_string()]));
        let mut remaining = fs::read_dir(temp.path().join("v1/host_8443"))
            .unwrap()
            .filter_map(|e| e.ok())
            .filter_map(|e| e.file_name().into_string().ok())
            .collect::<Vec<_>>();
        remaining.sort();
        assert_eq!(remaining, vec!["a", "c"]);
    }

    #[test]
    fn size_and_purge_are_the_two_halves_of_the_reload_button() {
        let temp = tempfile::tempdir().unwrap();
        let cache = cache(temp.path());
        let session = cache.session("s");
        assert_eq!(session.bytes(), 0);
        for seq in 1..=5u64 {
            session.append(&tool_line(seq), seq);
        }
        session.flush();

        assert!(session.bytes() > 0);
        session.purge();
        assert_eq!(session.bytes(), 0);
        assert_eq!(session.tail(), None);
        // And the session is usable again straight afterwards, which is
        // what a reload does next.
        session.append(&tool_line(9), 9);
        session.flush();
        assert_eq!(seqs_vec(&session.newest(80)), vec![9]);
    }
}