Skip to main content

blotter/
store.rs

1use crate::error::{AppError, AppResult};
2use crate::{ListItem, LogEvent, PromotionItem, Resolution, format_timestamp, normalized};
3use serde::Serialize;
4use serde_json::{Value, json};
5use std::collections::{BTreeMap, HashMap};
6use std::fs::{self, File, OpenOptions, Permissions};
7use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
8#[cfg(unix)]
9use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
10use std::path::{Component, Path, PathBuf};
11use std::thread;
12use std::time::Duration;
13
14const LOCK_ATTEMPTS: usize = 50;
15const LOCK_DELAY: Duration = Duration::from_millis(100);
16
17#[derive(Debug, Clone)]
18pub struct ResolvedFile {
19    pub path: PathBuf,
20    pub cwd: PathBuf,
21    pub explicit: bool,
22    pub repo: Option<PathBuf>,
23    pub warnings: Vec<String>,
24}
25
26impl ResolvedFile {
27    /// Repo root for cwd relativization. Only a log living inside the repo
28    /// stores repo-relative cwd; explicit and global logs are machine-local,
29    /// keep absolute cwd, and would otherwise lose all provenance now that
30    /// records carry no repo field.
31    pub fn cwd_repo(&self) -> Option<&Path> {
32        self.repo
33            .as_deref()
34            .filter(|root| self.path.starts_with(root))
35    }
36}
37
38#[derive(Debug, Default)]
39pub struct FoldResult {
40    pub items: Vec<ListItem>,
41    /// Folded promotions, `ts` descending then `id` ascending (r48). They are a
42    /// separate vector rather than a third arm of `items` because every analysis
43    /// command folds over cuts and dogears only; `list` is the one caller that
44    /// joins the two into its tagged union.
45    pub promotions: Vec<PromotionItem>,
46    pub warnings: Vec<String>,
47    records: BTreeMap<String, LogEvent>,
48    winning_amends: HashMap<String, LogEvent>,
49    lines: Vec<FoldedLine>,
50}
51
52/// One physical line that carried a parsed record, in file order. Archive needs
53/// the line numbers and per-ID groupings the fold already walks past; carrying
54/// them out of the fold is what keeps `plan_archive` to a single parse.
55#[derive(Debug, Clone)]
56pub struct FoldedLine {
57    pub line: usize,
58    pub id: String,
59    pub ts: jiff::Timestamp,
60}
61
62pub struct LoadedFold {
63    pub items: Vec<ListItem>,
64    pub promotions: Vec<PromotionItem>,
65    pub warnings: Vec<String>,
66}
67
68impl FoldResult {
69    pub fn record(&self, id: &str) -> Option<&LogEvent> {
70        self.records.get(id)
71    }
72
73    /// Physical lines carrying a parsed record, in file order. Empty unless the
74    /// fold was asked for them by `fold_bytes_with_lines`.
75    pub fn lines(&self) -> &[FoldedLine] {
76        &self.lines
77    }
78
79    /// Materialize a resolve against the fold that made the append decision,
80    /// reporting what a complete subsequent fold would show. A base resolve
81    /// activates an earlier orphan amend. An appended amend does *not* simply
82    /// win: the fold picks the amend with the latest timestamp, so a stored
83    /// amend carrying a later clock keeps the materialized fields, and only an
84    /// exact tie falls to the appended event as the last in file order.
85    /// Reached with a backdated `BLOTTER_NOW`, where the envelope would
86    /// otherwise report a note that no read command agrees with.
87    pub(crate) fn materialized_appended_resolution(&self, event: &LogEvent) -> Resolution {
88        let LogEvent::Resolve { id, amend, .. } = event else {
89            unreachable!("only resolve events materialize resolutions")
90        };
91        let effective = match self.winning_amends.get(id) {
92            Some(stored) if !*amend => stored,
93            Some(stored) if later_resolve(stored, event) => stored,
94            _ => event,
95        };
96        resolution_from_event(effective)
97    }
98}
99
100#[derive(Default)]
101struct WarningCounts {
102    torn: usize,
103    malformed: usize,
104    unknown: usize,
105    duplicate_cuts: usize,
106    duplicate_dogears: usize,
107    duplicate_promotions: usize,
108    duplicate_resolves: usize,
109    orphans: usize,
110    invalid_resolutions: usize,
111}
112
113pub(crate) struct ScannedLine<'a> {
114    pub line: usize,
115    pub raw: &'a [u8],
116    pub event: Result<LogEvent, ScanIssue>,
117}
118
119pub(crate) enum ScanIssue {
120    Malformed(String),
121    Unknown(Option<String>),
122    Torn,
123}
124
125/// The record version every v2 line carries.
126pub const RECORD_VERSION: u64 = 2;
127
128/// The probe's own kind list (r50). It stays at four names even while the fold
129/// knows three: through Phase 3 a `promotion` line is a known raw kind and must
130/// carry `"v":2` even though nothing reads it yet. `known_kind` below is the
131/// scan's list and is deliberately not this one.
132const PROBE_KINDS: [&str; 4] = ["cut", "dogear", "resolve", "promotion"];
133
134/// The first line whose raw JSON names a known kind and does not carry `"v":2`.
135#[derive(Debug, Clone)]
136pub struct VersionProbe {
137    /// 1-based physical line number.
138    pub line: usize,
139    /// The offending `v` verbatim, or `None` when the key was absent. Absent and
140    /// wrong are told apart by this being `None`, never by null-ness: a stored
141    /// `"v":null` reports `Some(Value::Null)`.
142    pub found_version: Option<Value>,
143}
144
145/// Inspect each scanned line's **raw** JSON, before and independently of the
146/// scan's classification (r50): a v1 cut is a record missing required fields,
147/// which the scan calls malformed, so keying on classification would exempt the
148/// exact file this probe exists to catch. A log holding no line with a known raw
149/// kind passes.
150pub fn probe_version(bytes: &[u8]) -> Option<VersionProbe> {
151    // Walk the same physical-line segmenter `scan` uses, so `line` is the
152    // number `scan` would report, but parse each line once and read only
153    // `kind` and `v`: the probe runs before the fold on every read, and going
154    // through `scan` here made every read command parse each line twice more.
155    for (line, raw) in physical_lines(bytes).1 {
156        let Ok(value) = serde_json::from_slice::<Value>(raw) else {
157            continue;
158        };
159        let known = value
160            .get("kind")
161            .and_then(Value::as_str)
162            .is_some_and(|kind| PROBE_KINDS.contains(&kind));
163        if !known {
164            continue;
165        }
166        match value.get("v") {
167            // Only a JSON integer literal whose value is 2: `2.0` and `2e0`
168            // decode as floats and `as_u64` declines them.
169            Some(found) if found.as_u64() == Some(RECORD_VERSION) => {}
170            found => {
171                return Some(VersionProbe {
172                    line,
173                    found_version: found.cloned(),
174                });
175            }
176        }
177    }
178    None
179}
180
181/// The one choke point every read path calls immediately after `read_bytes`,
182/// under the lock it already holds and before the fold, any tear-heal byte, any
183/// append, and any copy-and-swap.
184pub fn check_version(bytes: &[u8], path: &Path) -> AppResult<()> {
185    match probe_version(bytes) {
186        None => Ok(()),
187        Some(probe) => Err(AppError::unsupported_log_version(
188            path,
189            probe.line,
190            probe.found_version.as_ref(),
191        )),
192    }
193}
194
195/// A stored line: `v` first, then the event's own members. `LogEvent` is
196/// internally tagged on `kind`, so serde emits `kind` first for it; `v` belongs
197/// to this write-path-only wrapper rather than to `LogEvent`, which is what
198/// keeps `v` out of every envelope (r50).
199#[derive(Serialize)]
200struct Stored<'a> {
201    v: u64,
202    #[serde(flatten)]
203    event: &'a LogEvent,
204}
205
206impl<'a> Stored<'a> {
207    fn new(event: &'a LogEvent) -> Self {
208        Self {
209            v: RECORD_VERSION,
210            event,
211        }
212    }
213}
214
215pub fn discover(flag: Option<PathBuf>) -> AppResult<ResolvedFile> {
216    let cwd = std::env::current_dir().map_err(|error| AppError::from_io(error, Path::new(".")))?;
217    discover_from(&cwd, flag)
218}
219
220pub fn discover_from(cwd: &Path, flag: Option<PathBuf>) -> AppResult<ResolvedFile> {
221    let repo = find_repo_root(cwd);
222    if let Some(path) = flag {
223        return Ok(resolved_file(cwd, absolute(cwd, path), true, repo));
224    }
225    if let Some(path) = std::env::var_os("BLOTTER_FILE")
226        && !path.is_empty()
227    {
228        return Ok(resolved_file(
229            cwd,
230            absolute(cwd, PathBuf::from(path)),
231            true,
232            repo,
233        ));
234    }
235    if let Some(root) = repo.clone() {
236        let path = default_log_path(&root);
237        return Ok(resolved_file(cwd, path, false, Some(root)));
238    }
239    let home = home_dir(cwd).ok_or_else(|| {
240        AppError::config(
241            "cannot resolve the home directory for the default blotter file",
242            "Set HOME or pass --file PATH.",
243        )
244    })?;
245    Ok(resolved_file(
246        cwd,
247        home.join(".blotter/log.jsonl"),
248        false,
249        None,
250    ))
251}
252
253fn resolved_file(cwd: &Path, path: PathBuf, explicit: bool, repo: Option<PathBuf>) -> ResolvedFile {
254    ResolvedFile {
255        warnings: Vec::new(),
256        path,
257        cwd: cwd.to_path_buf(),
258        explicit,
259        repo,
260    }
261}
262
263pub fn default_log_path(root: &Path) -> PathBuf {
264    root.join(".blotter.jsonl")
265}
266
267pub fn find_repo_root(start: &Path) -> Option<PathBuf> {
268    start
269        .ancestors()
270        .find(|candidate| candidate.join(".git").exists())
271        .map(Path::to_path_buf)
272}
273
274pub fn home_dir(cwd: &Path) -> Option<PathBuf> {
275    std::env::var_os("HOME")
276        .filter(|value| !value.is_empty())
277        .map(PathBuf::from)
278        .map(|home| absolute(cwd, home))
279}
280
281pub fn record_cwd(cwd: &Path, repo: Option<&Path>, home: Option<&Path>) -> String {
282    if let Some(relative) = repo.and_then(|root| cwd.strip_prefix(root).ok()) {
283        return match relative.as_os_str().is_empty() {
284            true => ".".into(),
285            false => relative.to_string_lossy().into_owned(),
286        };
287    }
288    // The whole-string scanner, not a prefix-anchored match: a dash-encoded home
289    // (`/private/tmp/<session>/-Users-<user>-<repo>`) appears mid-path, and only
290    // this scanner applies the generic `/Users/` and `/home/` rules that
291    // `doctor --leaks` gates on. Its exact-home branch subsumes strip_prefix.
292    crate::redact::rewrite_home_paths(&cwd.to_string_lossy(), home)
293}
294
295/// Absolutize a log path. `.` folds away textually, but `..` cannot: when a
296/// component is a symlink to a directory elsewhere, the OS resolves `..`
297/// against the link's target, and a lexical `pop()` would name a different file
298/// than the one every later open, lock, backup, and `meta.file` acts on. A path
299/// carrying `..` therefore resolves through the OS — the longest existing
300/// ancestor is canonicalized and only the components that do not exist yet fold
301/// lexically. The final component is never canonicalized: a final-component
302/// symlink is `resolve_symlinked_log`'s policy, not this function's. A path with
303/// no `..` keeps its spelling, because the lexical join already names what the
304/// OS opens.
305fn absolute(cwd: &Path, path: PathBuf) -> PathBuf {
306    let joined = if path.is_absolute() {
307        path
308    } else {
309        cwd.join(path)
310    };
311    let components: Vec<Component> = joined.components().collect();
312    if !components
313        .iter()
314        .any(|component| matches!(component, Component::ParentDir))
315    {
316        return fold_lexically(PathBuf::new(), &components);
317    }
318    let trailing = match components.last() {
319        Some(Component::Normal(_)) => components.len() - 1,
320        _ => components.len(),
321    };
322    let mut resolved = resolve_existing_prefix(&components[..trailing]);
323    if let Some(Component::Normal(name)) = components.get(trailing) {
324        resolved.push(name);
325    }
326    resolved
327}
328
329/// Canonicalize the longest prefix of `components` that exists, then fold the
330/// remainder lexically. A path that exists resolves exactly as the OS resolves
331/// it; one that does not yet exist still resolves, with the lexical fold applied
332/// only to the components no directory backs.
333fn resolve_existing_prefix(components: &[Component]) -> PathBuf {
334    for split in (1..=components.len()).rev() {
335        // Verbatim, never folded: canonicalize must see `..` itself, or the
336        // fold would answer for the link instead of for its target.
337        let mut candidate = PathBuf::new();
338        for component in &components[..split] {
339            candidate.push(component.as_os_str());
340        }
341        if let Ok(canonical) = fs::canonicalize(&candidate) {
342            return fold_lexically(canonical, &components[split..]);
343        }
344    }
345    fold_lexically(PathBuf::new(), components)
346}
347
348fn fold_lexically(mut base: PathBuf, components: &[Component]) -> PathBuf {
349    for component in components {
350        match component {
351            Component::CurDir => {}
352            Component::ParentDir => {
353                base.pop();
354            }
355            other => base.push(other.as_os_str()),
356        }
357    }
358    base
359}
360
361pub fn with_shared<T>(path: &Path, action: impl FnOnce(&mut File) -> AppResult<T>) -> AppResult<T> {
362    let mut file = open_locked(path, false, || {
363        // O_NONBLOCK does not make a FIFO fail; it makes the open return
364        // immediately instead of blocking for a writer, and the regular-file
365        // check in open_locked is what rejects it. The flag has no effect on
366        // regular-file reads or writes on Linux or macOS.
367        #[cfg(unix)]
368        let opened = OpenOptions::new()
369            .read(true)
370            .custom_flags(libc::O_NONBLOCK)
371            .open(path);
372        #[cfg(not(unix))]
373        let opened = File::open(path);
374        opened.map_err(|error| AppError::from_log_open(error, path))
375    })?;
376    let result = action(&mut file);
377    let unlock = file
378        .unlock()
379        .map_err(|error| AppError::from_io(error, path));
380    match (result, unlock) {
381        (Err(error), _) | (Ok(_), Err(error)) => Err(error),
382        (Ok(value), Ok(())) => Ok(value),
383    }
384}
385
386pub fn read_or_empty<T>(
387    path: &Path,
388    explicit: bool,
389    warnings: &mut Vec<String>,
390    warning: &str,
391    suggested_fix: &str,
392    empty: impl FnOnce() -> T,
393    read: impl FnOnce(&mut File) -> AppResult<T>,
394) -> AppResult<(T, bool)> {
395    match with_shared(path, read) {
396        Ok(value) => Ok((value, true)),
397        Err(error) if error.code == "not_found" && error.exit_code == 66 && !explicit => {
398            warnings.push(warning.into());
399            Ok((empty(), false))
400        }
401        Err(error) if error.code == "not_found" && error.exit_code == 66 => {
402            Err(AppError::not_found(
403                format!("blotter file not found: {}", path.display()),
404                suggested_fix,
405            ))
406        }
407        Err(error) => Err(error),
408    }
409}
410
411pub fn load_folded(resolved: &ResolvedFile) -> AppResult<LoadedFold> {
412    let mut warnings = resolved.warnings.clone();
413    let (folded, _) = read_or_empty(
414        &resolved.path,
415        resolved.explicit,
416        &mut warnings,
417        "no blotter file yet; blotter add creates it",
418        "Pass an existing --file PATH or run `blotter add` to create a discovered default file.",
419        FoldResult::default,
420        |log| {
421            let bytes = read_bytes(log, &resolved.path)?;
422            check_version(&bytes, &resolved.path)?;
423            Ok(fold_bytes(&bytes))
424        },
425    )?;
426    warnings.extend(folded.warnings);
427    Ok(LoadedFold {
428        items: folded.items,
429        promotions: folded.promotions,
430        warnings,
431    })
432}
433
434pub fn with_exclusive<T>(
435    path: &Path,
436    create: bool,
437    action: impl FnOnce(&mut File) -> AppResult<T>,
438) -> AppResult<T> {
439    if create && let Some(parent) = path.parent() {
440        std::fs::create_dir_all(parent).map_err(|error| AppError::from_io(error, parent))?;
441    }
442    let mut file = open_locked(path, true, || {
443        let mut options = OpenOptions::new();
444        options.read(true).append(true).create(create);
445        // See with_shared: O_NONBLOCK only keeps the open from blocking on a
446        // FIFO; open_locked's regular-file check is what rejects one.
447        #[cfg(unix)]
448        options.custom_flags(libc::O_NONBLOCK);
449        options
450            .open(path)
451            .map_err(|error| AppError::from_log_open(error, path))
452    })?;
453    let result = action(&mut file);
454    let unlock = file
455        .unlock()
456        .map_err(|error| AppError::from_io(error, path));
457    match (result, unlock) {
458        (Err(error), _) | (Ok(_), Err(error)) => Err(error),
459        (Ok(value), Ok(())) => Ok(value),
460    }
461}
462
463fn open_locked(
464    path: &Path,
465    exclusive: bool,
466    mut open: impl FnMut() -> AppResult<File>,
467) -> AppResult<File> {
468    let mut file = Some(regular_file(open()?, path)?);
469    // The last reopen that found nothing. Kept so an exhausted budget whose
470    // final failure was a vanished log answers not_found (66) instead of
471    // blaming contention that never happened; any later failure clears it.
472    let mut missing: Option<AppError> = None;
473    for attempt in 0..LOCK_ATTEMPTS {
474        if file.is_none() {
475            match open() {
476                // A reopen that succeeds needs no clear: the lock attempt below
477                // ends this iteration in a branch that returns or clears.
478                Ok(opened) => file = Some(regular_file(opened, path)?),
479                Err(error) if error.code == "not_found" => {
480                    missing = Some(error);
481                    delay_before_retry(attempt);
482                    continue;
483                }
484                Err(error) => return Err(error),
485            }
486        }
487        let result = if exclusive {
488            file.as_ref().expect("file is open").try_lock()
489        } else {
490            file.as_ref().expect("file is open").try_lock_shared()
491        };
492        match result {
493            Ok(()) => {
494                if path_identity_matches(file.as_ref().expect("file is open"), path)? {
495                    return Ok(file.take().expect("file is open"));
496                }
497                let stale = file.take().expect("file is open");
498                let _ = stale.unlock();
499                // The path names another inode now. Reopening costs the same
500                // delay every other retry pays, so the attempt budget cannot
501                // burn through in microseconds and report a timeout nobody
502                // waited for.
503                missing = None;
504                delay_before_retry(attempt);
505            }
506            Err(error) => {
507                let error: std::io::Error = error.into();
508                if error.kind() != std::io::ErrorKind::WouldBlock {
509                    return Err(AppError::from_io(error, path));
510                }
511                missing = None;
512                delay_before_retry(attempt);
513            }
514        }
515    }
516    Err(missing.unwrap_or_else(|| AppError::lock_timeout(path)))
517}
518
519/// Pay the retry delay unless this was the last attempt, where the caller
520/// returns instead of retrying.
521fn delay_before_retry(attempt: usize) {
522    if attempt + 1 < LOCK_ATTEMPTS {
523        thread::sleep(LOCK_DELAY);
524    }
525}
526
527/// Reject a log path that is not a regular file, before the lock and before any
528/// read. flock reports ENOTSUP on a macOS FIFO, so a post-lock check would
529/// surface io_error instead of invalid_input, and a path that can never be valid
530/// would first burn the whole retry budget. `File::metadata` is fstat on the open
531/// handle, so this cannot race a swap the way a path stat can.
532fn regular_file(file: File, path: &Path) -> AppResult<File> {
533    let metadata = file
534        .metadata()
535        .map_err(|error| AppError::from_io(error, path))?;
536    if !metadata.is_file() {
537        return Err(AppError::invalid_input(
538            format!("blotter file is not a regular file: {}", path.display()),
539            "Point --file PATH or BLOTTER_FILE at a regular JSONL file; FIFOs and devices are not accepted.",
540        ));
541    }
542    Ok(file)
543}
544
545#[cfg(unix)]
546fn path_identity_matches(file: &File, path: &Path) -> AppResult<bool> {
547    // File::metadata uses fstat; fs::metadata obtains a fresh stat of the path.
548    let locked = file
549        .metadata()
550        .map_err(|error| AppError::from_io(error, path))?;
551    match std::fs::metadata(path) {
552        Ok(current) => Ok(locked.dev() == current.dev() && locked.ino() == current.ino()),
553        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
554        Err(error) => Err(AppError::from_io(error, path)),
555    }
556}
557
558#[cfg(not(unix))]
559fn path_identity_matches(_file: &File, _path: &Path) -> AppResult<bool> {
560    Ok(true)
561}
562
563pub fn read_bytes(file: &mut File, path: &Path) -> AppResult<Vec<u8>> {
564    file.seek(SeekFrom::Start(0))
565        .and_then(|_| {
566            let mut bytes = Vec::new();
567            file.read_to_end(&mut bytes).map(|_| bytes)
568        })
569        .map_err(|error| AppError::from_io(error, path))
570}
571
572pub fn write_new_file(path: &Path, bytes: &[u8], permissions: &Permissions) -> AppResult<PathBuf> {
573    let mut file = create_new_file(path, permissions, false)
574        .map_err(|error| AppError::from_io(error, path))?;
575    if let Err(error) = file.write_all(bytes) {
576        discard_new_file(file, path);
577        return Err(AppError::from_io(error, path));
578    }
579    if let Err(error) = file.sync_all() {
580        discard_new_file(file, path);
581        return Err(AppError::from_io(error, path));
582    }
583    Ok(path.to_path_buf())
584}
585
586pub fn append_file(path: &Path, bytes: &[u8], permissions: &Permissions) -> AppResult<PathBuf> {
587    let (mut file, created) = match create_new_file(path, permissions, false) {
588        Ok(file) => (file, true),
589        Err(error) if error.kind() == ErrorKind::AlreadyExists => (
590            OpenOptions::new()
591                .append(true)
592                .open(path)
593                .map_err(|error| AppError::from_io(error, path))?,
594            false,
595        ),
596        Err(error) => return Err(AppError::from_io(error, path)),
597    };
598    if let Err(error) = file.write_all(bytes) {
599        if created {
600            discard_new_file(file, path);
601        }
602        return Err(AppError::from_io(error, path));
603    }
604    if let Err(error) = file.sync_all() {
605        if created {
606            discard_new_file(file, path);
607        }
608        return Err(AppError::from_io(error, path));
609    }
610    Ok(path.to_path_buf())
611}
612
613pub fn replace_log(
614    path: &Path,
615    bytes: &[u8],
616    permissions: &Permissions,
617    temporary_suffix: &str,
618) -> AppResult<()> {
619    let temporary = suffixed_path(path, temporary_suffix);
620    let mut file = create_new_file(&temporary, permissions, true)
621        .map_err(|error| AppError::from_io(error, &temporary))?;
622    if let Err(error) = file.write_all(bytes) {
623        discard_new_file(file, &temporary);
624        return Err(AppError::from_io(error, &temporary));
625    }
626    if let Err(error) = file.sync_all() {
627        discard_new_file(file, &temporary);
628        return Err(AppError::from_io(error, &temporary));
629    }
630    drop(file);
631    if let Err(error) = fs::rename(&temporary, path) {
632        let _ = fs::remove_file(&temporary);
633        return Err(AppError::from_io(error, path));
634    }
635    if let Some(parent) = path.parent()
636        && let Ok(directory) = File::open(parent)
637    {
638        let _ = directory.sync_all();
639    }
640    Ok(())
641}
642
643/// Resolve a symlinked log path to its target before a copy-and-swap, so the
644/// backup, sidecar, and atomic replacement all act on the real file and the
645/// link survives. Only final-component links are chased; parent components
646/// keep their spelling so envelope paths stay stable for regular files.
647pub fn resolve_symlinked_log(path: &Path) -> AppResult<PathBuf> {
648    let mut current = path.to_path_buf();
649    for _ in 0..40 {
650        let metadata =
651            fs::symlink_metadata(&current).map_err(|error| AppError::from_io(error, &current))?;
652        if !metadata.file_type().is_symlink() {
653            return Ok(current);
654        }
655        let target = fs::read_link(&current).map_err(|error| AppError::from_io(error, &current))?;
656        current = if target.is_absolute() {
657            target
658        } else {
659            match current.parent() {
660                Some(parent) => parent.join(&target),
661                None => target,
662            }
663        };
664    }
665    Err(AppError::from_io(
666        std::io::Error::other("too many levels of symbolic links"),
667        path,
668    ))
669}
670
671pub fn suffixed_path(path: &Path, suffix: &str) -> PathBuf {
672    let mut value = path.as_os_str().to_os_string();
673    value.push(suffix);
674    PathBuf::from(value)
675}
676
677pub fn backup_timestamp(now: jiff::Timestamp) -> String {
678    format_timestamp(now)
679        .chars()
680        .filter(|character| !matches!(character, '-' | ':' | '.'))
681        .collect()
682}
683
684pub fn restore_hint(backup: &Path, path: &Path) -> String {
685    format!("cp {} {}", shell_quote(backup), shell_quote(path))
686}
687
688fn create_new_file(
689    path: &Path,
690    permissions: &Permissions,
691    set_permissions_on_non_unix: bool,
692) -> std::io::Result<File> {
693    let mut options = OpenOptions::new();
694    options.write(true).create_new(true);
695    #[cfg(unix)]
696    options.mode(permissions.mode());
697    let file = options.open(path)?;
698    #[cfg(unix)]
699    let permissions_result = {
700        let _ = set_permissions_on_non_unix;
701        file.set_permissions(permissions.clone())
702    };
703    #[cfg(not(unix))]
704    let permissions_result = set_permissions_on_non_unix
705        .then(|| file.set_permissions(permissions.clone()))
706        .transpose()
707        .map(|_| ());
708    if let Err(error) = permissions_result {
709        drop(file);
710        let _ = fs::remove_file(path);
711        return Err(error);
712    }
713    Ok(file)
714}
715
716fn discard_new_file(file: File, path: &Path) {
717    drop(file);
718    let _ = fs::remove_file(path);
719}
720
721fn shell_quote(path: &Path) -> String {
722    format!("'{}'", path.to_string_lossy().replace('\'', "'\\''"))
723}
724
725pub fn append_json(file: &mut File, path: &Path, prior: &[u8], record: &LogEvent) -> AppResult<()> {
726    let mut record_bytes = Vec::new();
727    serde_json::to_writer(&mut record_bytes, &Stored::new(record))
728        .map_err(|error| AppError::internal(error.to_string()))?;
729    record_bytes.push(b'\n');
730    append_bytes(file, path, prior, &record_bytes)
731}
732
733pub fn append_unique(path: &Path, record: LogEvent, dry_run: bool) -> AppResult<(bool, LogEvent)> {
734    if dry_run {
735        return Ok((false, record));
736    }
737    let id = record.id().expect("new records have IDs").to_owned();
738    let kind = match &record {
739        LogEvent::Cut { .. } => "cut",
740        LogEvent::Dogear { .. } => "dogear",
741        _ => unreachable!("append_unique only receives cut or dogear records"),
742    };
743    with_exclusive(path, true, |log| {
744        let bytes = read_bytes(log, path)?;
745        // Before the fold and before the tear-healing byte `append_bytes` would
746        // add: a refusal writes zero bytes.
747        check_version(&bytes, path)?;
748        let records = fold_records(&bytes);
749        if let Some(existing) = records.get(&id) {
750            return if std::mem::discriminant(&record) == std::mem::discriminant(existing) {
751                Ok((false, existing.clone()))
752            } else {
753                Err(AppError::internal(format!(
754                    "{kind} ID collides with an existing non-{kind} record"
755                )))
756            };
757        }
758        append_json(log, path, &bytes, &record)?;
759        Ok((true, record))
760    })
761}
762
763pub fn append_json_batch(
764    file: &mut File,
765    path: &Path,
766    prior: &[u8],
767    records: &[LogEvent],
768) -> AppResult<()> {
769    let mut record_bytes = Vec::new();
770    for record in records {
771        serde_json::to_writer(&mut record_bytes, &Stored::new(record))
772            .map_err(|error| AppError::internal(error.to_string()))?;
773        record_bytes.push(b'\n');
774    }
775    append_bytes(file, path, prior, &record_bytes)
776}
777
778fn append_bytes(file: &mut File, path: &Path, prior: &[u8], record_bytes: &[u8]) -> AppResult<()> {
779    append_bytes_with(file, path, prior, record_bytes, |file, bytes| {
780        file.write_all(bytes)
781    })
782}
783
784fn append_bytes_with(
785    file: &mut File,
786    path: &Path,
787    prior: &[u8],
788    record_bytes: &[u8],
789    write: impl FnOnce(&mut File, &[u8]) -> std::io::Result<()>,
790) -> AppResult<()> {
791    let original_len = file
792        .metadata()
793        .map_err(|error| AppError::from_io(error, path))?
794        .len();
795    let mut bytes = Vec::new();
796    if !is_empty_log(prior) && !prior.ends_with(b"\n") {
797        bytes.push(b'\n');
798    }
799    bytes.extend_from_slice(record_bytes);
800    // If the write fails, roll back to the pre-write length; if rollback also fails, surface both.
801    if let Err(error) = write(file, &bytes) {
802        if let Err(rollback) = file.set_len(original_len) {
803            return Err(AppError {
804                code: "io_error",
805                message: format!(
806                    "append failed: {error}; rollback to original length {original_len} failed: {rollback}"
807                ),
808                details: json!({}),
809                retryable: false,
810                suggested_fix: "Check the blotter file and filesystem, then retry.".into(),
811                exit_code: 74,
812            });
813        }
814        return Err(AppError::from_io(error, path));
815    }
816    Ok(())
817}
818
819/// A log holding no physical line: an empty file, or the single newline that
820/// `scan` reads as a terminator rather than a line (r26). The appender uses
821/// this so it adds no tear-healing separator to such a log; `scan` encodes
822/// the same rule structurally by skipping an empty first segment, so a log
823/// the appender calls empty stays empty for the reader.
824pub(crate) fn is_empty_log(bytes: &[u8]) -> bool {
825    bytes.is_empty() || bytes == b"\n"
826}
827
828/// Scan physical JSONL lines once. A final non-newline line is accepted only
829/// when its decoded JSON carries a recognized kind, so consumers cannot
830/// disagree on torn tails.
831/// Splits a log into its physical lines, numbered from 1, and reports whether
832/// the final line is newline-terminated. Shared by `scan` and `probe_version`
833/// so both report the same line number for the same bytes.
834///
835/// A leading empty segment is the terminator of an empty log, not a physical
836/// line: the log was empty or held only "\n" when the record was appended,
837/// and an append-only writer cannot remove the byte that precedes it. An
838/// empty segment after a record is still a line, and `scan` reports it as
839/// malformed.
840fn physical_lines(bytes: &[u8]) -> (bool, impl Iterator<Item = (usize, &[u8])> + '_) {
841    let terminated = bytes.ends_with(b"\n");
842    let body = if terminated {
843        &bytes[..bytes.len() - 1]
844    } else {
845        bytes
846    };
847    let lines = body
848        .split(|byte| *byte == b'\n')
849        .enumerate()
850        .filter(|(index, raw)| !(raw.is_empty() && *index == 0))
851        .map(|(index, raw)| (index + 1, raw));
852    (terminated, lines)
853}
854
855pub(crate) fn scan(bytes: &[u8]) -> impl Iterator<Item = ScannedLine<'_>> + '_ {
856    let (terminated, lines) = physical_lines(bytes);
857    let last_line = physical_lines(bytes).1.map(|(line, _)| line).last();
858    lines.map(move |(line, raw)| {
859        let final_line = Some(line) == last_line;
860        let decoded = serde_json::from_slice::<Value>(raw);
861        let known = decoded.as_ref().ok().and_then(known_kind);
862        let event = if final_line && !terminated && known.is_none() {
863            Err(ScanIssue::Torn)
864        } else {
865            match decoded {
866                Ok(value) => parse_event(value, known),
867                Err(_) => Err(ScanIssue::Malformed("line is not valid JSON".into())),
868            }
869        };
870        ScannedLine { line, raw, event }
871    })
872}
873
874fn known_kind(value: &Value) -> Option<&'static str> {
875    match value.get("kind").and_then(Value::as_str) {
876        Some("cut") => Some("cut"),
877        Some("dogear") => Some("dogear"),
878        Some("resolve") => Some("resolve"),
879        Some("promotion") => Some("promotion"),
880        _ => None,
881    }
882}
883
884fn parse_event(value: Value, known: Option<&'static str>) -> Result<LogEvent, ScanIssue> {
885    let unknown = value.get("kind").and_then(Value::as_str).map(str::to_owned);
886    match serde_json::from_value::<LogEvent>(value) {
887        Ok(LogEvent::Unknown) => Err(ScanIssue::Unknown(unknown)),
888        Ok(event) => {
889            let ts = match &event {
890                LogEvent::Cut { ts, .. }
891                | LogEvent::Dogear { ts, .. }
892                | LogEvent::Resolve { ts, .. }
893                | LogEvent::Promotion { ts, .. } => ts,
894                LogEvent::Unknown => unreachable!("unknown events are classified above"),
895            };
896            match ts.parse::<jiff::Timestamp>() {
897                Ok(_) => Ok(event),
898                Err(_) => Err(ScanIssue::Malformed(format!(
899                    "{} ts is not a full RFC3339 timestamp",
900                    known.expect("parsed events have a known kind")
901                ))),
902            }
903        }
904        Err(error) => match known {
905            Some(kind) => Err(ScanIssue::Malformed(format!(
906                "invalid {kind} record: {error}"
907            ))),
908            None => Err(ScanIssue::Unknown(unknown)),
909        },
910    }
911}
912
913/// Is `stored` strictly later than `candidate`? Only strictly: the fold breaks
914/// an exact tie toward the last event in file order, and `candidate` is the one
915/// being appended. An unparseable timestamp never wins, so this cannot panic on
916/// a hand-edited log the way the fold's validated parse would.
917fn later_resolve(stored: &LogEvent, candidate: &LogEvent) -> bool {
918    let timestamp = |event: &LogEvent| match event {
919        LogEvent::Resolve { ts, .. } => ts.parse::<jiff::Timestamp>().ok(),
920        _ => None,
921    };
922    match (timestamp(stored), timestamp(candidate)) {
923        (Some(stored), Some(candidate)) => stored > candidate,
924        _ => false,
925    }
926}
927
928fn resolution_from_event(event: &LogEvent) -> Resolution {
929    let LogEvent::Resolve {
930        ts,
931        agent,
932        note,
933        task,
934        pr,
935        commit,
936        url,
937        dropped,
938        amend,
939        disposition,
940        disposition_ts,
941        promotion,
942        ..
943    } = event
944    else {
945        unreachable!("only resolve events materialize resolutions")
946    };
947    Resolution {
948        ts: ts.clone(),
949        agent: agent.clone(),
950        note: note.clone(),
951        task: task.clone(),
952        pr: pr.clone(),
953        commit: commit.clone(),
954        url: url.clone(),
955        dropped: *dropped,
956        amended: *amend,
957        disposition: *disposition,
958        disposition_ts: disposition_ts.clone(),
959        promotion: promotion.clone(),
960    }
961}
962
963/// The `sources[]` of every promotion in a folded log, keyed by promotion ID.
964/// Rules (5) and (6) below join against it, and `doctor` builds the same map
965/// while it scans so both answer a hand-edited log identically.
966pub type PromotionSources = HashMap<String, Vec<String>>;
967
968/// The six r48 invalid-resolution rules, in their permanent numbering, evaluated
969/// only for an event the fold has already joined to its record. An orphan — a
970/// resolve joining to no record — is never evaluated.
971pub(crate) fn broken_resolution_rules(
972    event: &LogEvent,
973    record_kind: &str,
974    promotions: &PromotionSources,
975) -> Vec<&'static str> {
976    let LogEvent::Resolve {
977        id,
978        disposition,
979        disposition_ts,
980        promotion,
981        ..
982    } = event
983    else {
984        unreachable!("only resolve events are validated")
985    };
986    let mut broken = Vec::new();
987    if record_kind == "cut" && disposition.is_none() {
988        broken.push("resolve targets a cut without a disposition");
989    }
990    if record_kind == "dogear" && disposition.is_some() {
991        broken.push("resolve targets a dogear with a disposition");
992    }
993    if disposition.is_some() != disposition_ts.is_some() {
994        broken.push("disposition and disposition_ts must be present together");
995    }
996    if let Some(promotion) = promotion {
997        if *disposition != Some(crate::Disposition::Promoted) {
998            broken.push("a promotion link requires disposition promoted");
999        }
1000        match promotions.get(promotion) {
1001            None => broken.push("promotion link names no promotion in this log"),
1002            // The mutual-link rule the CLI enforces on write, enforced here on
1003            // read, so a hand-written one-way link never materializes.
1004            Some(sources) if !sources.contains(id) => {
1005                broken.push("promotion does not name this record as a source");
1006            }
1007            Some(_) => {}
1008        }
1009    }
1010    broken
1011}
1012
1013/// Records-only fold for the append path. `append_unique` needs one fact — does
1014/// this ID already exist — so it skips the resolution join, the ListItem clones,
1015/// the timestamp parses, and the sort that `fold_bytes` would discard, inside
1016/// the exclusive lock. Tag normalization must match `fold_bytes`: the duplicate
1017/// branch returns this record straight into the add/dogear response envelope.
1018fn fold_records(bytes: &[u8]) -> BTreeMap<String, LogEvent> {
1019    let mut records = BTreeMap::<String, LogEvent>::new();
1020    for scanned in scan(bytes) {
1021        let Ok(mut event) = scanned.event else {
1022            continue;
1023        };
1024        match &mut event {
1025            LogEvent::Cut { tags, .. } | LogEvent::Dogear { tags, .. } => {
1026                tags.sort();
1027                tags.dedup();
1028            }
1029            LogEvent::Promotion { sources, .. } => *sources = normalized(sources),
1030            LogEvent::Resolve { .. } | LogEvent::Unknown => continue,
1031        }
1032        let id = event.id().expect("parsed records have IDs").to_owned();
1033        records.entry(id).or_insert(event);
1034    }
1035    records
1036}
1037
1038pub fn fold_bytes(bytes: &[u8]) -> FoldResult {
1039    fold_bytes_inner(bytes, false)
1040}
1041
1042/// The same fold, additionally carrying the `(line, id, ts)` tuple of every
1043/// physical line that parsed into a record. Only `archive` needs them, and the
1044/// tuples cost one owned ID per physical line, so every other caller keeps the
1045/// cheaper `fold_bytes`. Collecting them changes no fold verdict: the tuples are
1046/// written from the scanner's own output and nothing reads them back.
1047pub fn fold_bytes_with_lines(bytes: &[u8]) -> FoldResult {
1048    fold_bytes_inner(bytes, true)
1049}
1050
1051fn fold_bytes_inner(bytes: &[u8], collect_lines: bool) -> FoldResult {
1052    let mut lines = Vec::new();
1053    let mut records = BTreeMap::<String, LogEvent>::new();
1054    let mut resolves = HashMap::<String, LogEvent>::new();
1055    // Amends carry their parsed timestamp so the winner is chosen by clock, not
1056    // by byte position, without reparsing the incumbent for every candidate.
1057    let mut amends = HashMap::<String, (jiff::Timestamp, LogEvent)>::new();
1058    let mut resolve_events = Vec::<LogEvent>::new();
1059    let mut counts = WarningCounts::default();
1060    for scanned in scan(bytes) {
1061        let line = scanned.line;
1062        match scanned.event {
1063            Err(ScanIssue::Malformed(_)) => counts.malformed += 1,
1064            Err(ScanIssue::Unknown(_)) => counts.unknown += 1,
1065            Err(ScanIssue::Torn) => counts.torn += 1,
1066            Ok(mut event) => {
1067                if collect_lines
1068                    && let Some(id) = event.id()
1069                    && let Some(ts) = event_timestamp(&event)
1070                {
1071                    lines.push(FoldedLine {
1072                        line,
1073                        id: id.to_owned(),
1074                        ts,
1075                    });
1076                }
1077                match &mut event {
1078                    LogEvent::Cut { tags, .. } => {
1079                        // Fold normalizes legacy tag arrays for list output. Doctor
1080                        // receives the scanner's unmodified parsed event instead.
1081                        tags.sort();
1082                        tags.dedup();
1083                        let id = event.id().expect("parsed cuts have IDs").to_owned();
1084                        if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id)
1085                        {
1086                            entry.insert(event);
1087                        } else {
1088                            counts.duplicate_cuts += 1;
1089                        }
1090                    }
1091                    LogEvent::Dogear { tags, .. } => {
1092                        tags.sort();
1093                        tags.dedup();
1094                        let id = event.id().expect("parsed dogears have IDs").to_owned();
1095                        if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id)
1096                        {
1097                            entry.insert(event);
1098                        } else {
1099                            counts.duplicate_dogears += 1;
1100                        }
1101                    }
1102                    LogEvent::Promotion { sources, .. } => {
1103                        // Sorted-unique on read as tags are, so the fold and the
1104                        // hash agree about what the source set is.
1105                        *sources = normalized(sources);
1106                        let id = event.id().expect("parsed promotions have IDs").to_owned();
1107                        if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id)
1108                        {
1109                            entry.insert(event);
1110                        } else {
1111                            counts.duplicate_promotions += 1;
1112                        }
1113                    }
1114                    // Resolve events are held back: validity is knowable only
1115                    // after the join to their record, and r50 requires the
1116                    // invalid ones to be discarded *before* winners are chosen,
1117                    // so an invalid event cannot occupy the base slot.
1118                    LogEvent::Resolve { .. } => resolve_events.push(event),
1119                    LogEvent::Unknown => counts.unknown += 1,
1120                }
1121            }
1122        }
1123    }
1124
1125    let promotion_sources = promotion_sources(&records);
1126    for event in resolve_events {
1127        let LogEvent::Resolve { id, ts, amend, .. } = &event else {
1128            unreachable!("only resolve events are held back")
1129        };
1130        let id = id.clone();
1131        let amend = *amend;
1132        if let Some(kind) = records.get(&id).and_then(record_kind)
1133            && !broken_resolution_rules(&event, kind, &promotion_sources).is_empty()
1134        {
1135            // Discarded entirely: it materializes nothing and is counted only
1136            // in `skipped N invalid resolutions`, never as a duplicate or an
1137            // orphan.
1138            counts.invalid_resolutions += 1;
1139            continue;
1140        }
1141        if amend {
1142            let timestamp = ts
1143                .parse::<jiff::Timestamp>()
1144                .expect("parsed resolves have valid RFC3339 timestamps");
1145            match amends.entry(id) {
1146                std::collections::hash_map::Entry::Occupied(mut entry) => {
1147                    // `>=`, not `>`: equal timestamps are reachable under a
1148                    // frozen BLOTTER_NOW, and there the last amend in file
1149                    // order keeps winning.
1150                    if timestamp >= entry.get().0 {
1151                        entry.insert((timestamp, event));
1152                    }
1153                }
1154                std::collections::hash_map::Entry::Vacant(entry) => {
1155                    entry.insert((timestamp, event));
1156                }
1157            }
1158        } else if let std::collections::hash_map::Entry::Vacant(entry) = resolves.entry(id) {
1159            entry.insert(event);
1160        } else {
1161            counts.duplicate_resolves += 1;
1162        }
1163    }
1164
1165    // Base resolves remain first-wins. The winning amend is the one with the
1166    // latest timestamp, with the last in file order breaking an exact tie; file
1167    // position never decides, because a `merge=union` log concatenates branches
1168    // in branch order. A latest amend only materializes when the full scan found
1169    // a base resolve, so merge-reordered base resolves work. Every winner is
1170    // also kept in `winning_amends`, whether or not a base resolve claimed it,
1171    // so `materialized_appended_resolution` can apply the same rule to an amend
1172    // that has not been folded yet.
1173    let mut winning_amends = HashMap::new();
1174    for (id, (_, amend)) in amends {
1175        winning_amends.insert(id.clone(), amend.clone());
1176        match resolves.entry(id) {
1177            std::collections::hash_map::Entry::Occupied(mut entry) => {
1178                entry.insert(amend);
1179            }
1180            // An amend with no base resolve stays out of `resolves`, so the
1181            // record remains open, exactly as before.
1182            std::collections::hash_map::Entry::Vacant(_) => counts.orphans += 1,
1183        }
1184    }
1185
1186    for id in resolves.keys() {
1187        if !records.contains_key(id) {
1188            counts.orphans += 1;
1189        }
1190    }
1191    let mut items: Vec<_> = records
1192        .values()
1193        .filter(|record| !matches!(record, LogEvent::Promotion { .. }))
1194        .cloned()
1195        .map(|record| {
1196            let resolution = record
1197                .id()
1198                .and_then(|id| resolves.get(id))
1199                .map(resolution_from_event);
1200            let item = ListItem::from_record(record, resolution);
1201            let timestamp = item
1202                .ts
1203                .parse::<jiff::Timestamp>()
1204                .expect("folded items have valid RFC3339 timestamps");
1205            (item, timestamp)
1206        })
1207        .collect();
1208    items.sort_by(|(left, left_timestamp), (right, right_timestamp)| {
1209        match (left.kind.as_str(), right.kind.as_str()) {
1210            ("cut", "cut") => right
1211                .impact
1212                .expect("cut has impact")
1213                .rank()
1214                .cmp(&left.impact.expect("cut has impact").rank())
1215                .then_with(|| right_timestamp.cmp(left_timestamp))
1216                .then_with(|| left.id.cmp(&right.id)),
1217            ("dogear", "dogear") => right_timestamp
1218                .cmp(left_timestamp)
1219                .then_with(|| left.id.cmp(&right.id)),
1220            ("cut", "dogear") => std::cmp::Ordering::Less,
1221            ("dogear", "cut") => std::cmp::Ordering::Greater,
1222            _ => left.kind.cmp(&right.kind),
1223        }
1224    });
1225    let items = items.into_iter().map(|(item, _)| item).collect();
1226
1227    // Promotions order by `ts` descending then `id` ascending (r48), the same
1228    // rule dogears follow; they are never interleaved with the two kinds above.
1229    let mut promotions: Vec<_> = records
1230        .values()
1231        .filter(|record| matches!(record, LogEvent::Promotion { .. }))
1232        .cloned()
1233        .map(|record| {
1234            let item = PromotionItem::from_record(record);
1235            let timestamp = item
1236                .ts
1237                .parse::<jiff::Timestamp>()
1238                .expect("folded promotions have valid RFC3339 timestamps");
1239            (item, timestamp)
1240        })
1241        .collect();
1242    promotions.sort_by(|(left, left_ts), (right, right_ts)| {
1243        right_ts.cmp(left_ts).then_with(|| left.id.cmp(&right.id))
1244    });
1245    let promotions = promotions.into_iter().map(|(item, _)| item).collect();
1246
1247    let mut warnings = Vec::new();
1248    warning(&mut warnings, counts.torn, "torn final line");
1249    warning(&mut warnings, counts.malformed, "malformed line");
1250    warning(&mut warnings, counts.unknown, "unknown event");
1251    warning(&mut warnings, counts.duplicate_cuts, "duplicate cut");
1252    warning(&mut warnings, counts.duplicate_dogears, "duplicate dogear");
1253    warning(
1254        &mut warnings,
1255        counts.duplicate_promotions,
1256        "duplicate promotion",
1257    );
1258    warning(
1259        &mut warnings,
1260        counts.duplicate_resolves,
1261        "duplicate resolve",
1262    );
1263    warning(&mut warnings, counts.orphans, "orphan resolve");
1264    warning(
1265        &mut warnings,
1266        counts.invalid_resolutions,
1267        "invalid resolution",
1268    );
1269    FoldResult {
1270        items,
1271        promotions,
1272        warnings,
1273        records,
1274        winning_amends,
1275        lines,
1276    }
1277}
1278
1279/// The record kind a resolve event joins to, or `None` for an event that is not
1280/// an identity-bearing record.
1281fn record_kind(event: &LogEvent) -> Option<&'static str> {
1282    match event {
1283        LogEvent::Cut { .. } => Some("cut"),
1284        LogEvent::Dogear { .. } => Some("dogear"),
1285        LogEvent::Promotion { .. } => Some("promotion"),
1286        LogEvent::Resolve { .. } | LogEvent::Unknown => None,
1287    }
1288}
1289
1290/// The `sources[]` of every folded promotion, for rules (5) and (6).
1291fn promotion_sources(records: &BTreeMap<String, LogEvent>) -> PromotionSources {
1292    records
1293        .iter()
1294        .filter_map(|(id, event)| match event {
1295            LogEvent::Promotion { sources, .. } => Some((id.clone(), sources.clone())),
1296            _ => None,
1297        })
1298        .collect()
1299}
1300
1301/// The parsed timestamp of a record-carrying event. `parse_event` already
1302/// rejected an unparseable one, so `None` only covers `Unknown`.
1303fn event_timestamp(event: &LogEvent) -> Option<jiff::Timestamp> {
1304    match event {
1305        LogEvent::Cut { ts, .. }
1306        | LogEvent::Dogear { ts, .. }
1307        | LogEvent::Resolve { ts, .. }
1308        | LogEvent::Promotion { ts, .. } => ts.parse().ok(),
1309        LogEvent::Unknown => None,
1310    }
1311}
1312
1313fn warning(warnings: &mut Vec<String>, count: usize, label: &str) {
1314    if count > 0 {
1315        warnings.push(format!(
1316            "skipped {count} {label}{}",
1317            if count == 1 { "" } else { "s" }
1318        ));
1319    }
1320}
1321
1322#[cfg(test)]
1323mod tests {
1324    use super::*;
1325    use crate::{Impact, ItemStatus, compute_id};
1326    use std::io::Write;
1327    use tempfile::TempDir;
1328
1329    fn cut(id: &str) -> String {
1330        cut_with_text(id, "x")
1331    }
1332
1333    fn cut_with_text(id: &str, text: &str) -> String {
1334        serde_json::json!({
1335            "v":2, "kind":"cut", "id":id, "ts":"2026-07-09T00:00:00.000Z",
1336            "agent":"a", "text":text, "tags":[], "impact":"low",
1337            "cwd":"/tmp", "repo":null
1338        })
1339        .to_string()
1340    }
1341
1342    fn resolve(id: &str) -> String {
1343        serde_json::json!({
1344            "v":2, "kind":"resolve", "id":id, "ts":"2026-07-10T00:00:00.000Z",
1345            "agent":"a", "note":null,
1346            "disposition":"fixed", "disposition_ts":"2026-07-10T00:00:00.000Z"
1347        })
1348        .to_string()
1349    }
1350
1351    #[cfg(unix)]
1352    #[test]
1353    fn exclusive_lock_reopens_a_replaced_path_before_appending() {
1354        let temp = TempDir::new().unwrap();
1355        let path = temp.path().join("cuts.jsonl");
1356        std::fs::write(&path, b"old\n").unwrap();
1357
1358        let holder = OpenOptions::new()
1359            .read(true)
1360            .write(true)
1361            .open(&path)
1362            .unwrap();
1363        holder.lock().unwrap();
1364
1365        let preopened = OpenOptions::new()
1366            .read(true)
1367            .append(true)
1368            .open(&path)
1369            .unwrap();
1370        let (opened_tx, opened_rx) = std::sync::mpsc::channel();
1371        let writer_path = path.clone();
1372        let writer = std::thread::spawn(move || {
1373            let mut first_open = Some(preopened);
1374            let mut file = open_locked(&writer_path, true, || {
1375                if let Some(file) = first_open.take() {
1376                    // The writer now owns a descriptor for the old inode.
1377                    opened_tx.send(()).unwrap();
1378                    Ok(file)
1379                } else {
1380                    OpenOptions::new()
1381                        .read(true)
1382                        .append(true)
1383                        .open(&writer_path)
1384                        .map_err(|error| AppError::from_log_open(error, &writer_path))
1385                }
1386            })
1387            .unwrap();
1388            file.write_all(b"writer\n").unwrap();
1389            file.unlock().unwrap();
1390        });
1391
1392        opened_rx
1393            .recv_timeout(std::time::Duration::from_secs(2))
1394            .unwrap();
1395        let replacement = temp.path().join("replacement.jsonl");
1396        std::fs::write(&replacement, b"replacement\n").unwrap();
1397        std::fs::rename(&replacement, &path).unwrap();
1398        holder.unlock().unwrap();
1399        writer.join().unwrap();
1400
1401        assert_eq!(std::fs::read(&path).unwrap(), b"replacement\nwriter\n");
1402    }
1403
1404    #[cfg(unix)]
1405    #[test]
1406    fn a_permanent_path_identity_mismatch_still_pays_the_retry_delay() {
1407        // The locked descriptor never names the requested path, so every
1408        // attempt mismatches. The budget must still span the published bound
1409        // rather than burning through in microseconds.
1410        let temp = TempDir::new().unwrap();
1411        let path = temp.path().join("cuts.jsonl");
1412        let other = temp.path().join("other.jsonl");
1413        std::fs::write(&path, b"").unwrap();
1414        std::fs::write(&other, b"").unwrap();
1415
1416        let started = std::time::Instant::now();
1417        let error = open_locked(&path, true, || {
1418            OpenOptions::new()
1419                .read(true)
1420                .append(true)
1421                .open(&other)
1422                .map_err(|error| AppError::from_log_open(error, &other))
1423        })
1424        .expect_err("a permanent identity mismatch never locks the path");
1425        let elapsed = started.elapsed();
1426
1427        assert_eq!(error.code, "lock_timeout");
1428        assert_eq!(error.exit_code, 75);
1429        assert!(
1430            elapsed >= LOCK_DELAY * (LOCK_ATTEMPTS as u32 - 1),
1431            "gave up after {elapsed:?}"
1432        );
1433    }
1434
1435    #[cfg(unix)]
1436    #[test]
1437    fn a_log_that_vanishes_during_the_retry_budget_reports_not_found() {
1438        // First open lands on another inode, so the identity check rejects it;
1439        // every reopen then finds nothing. Exhaustion must name the missing
1440        // log, not contention that never happened.
1441        let temp = TempDir::new().unwrap();
1442        let path = temp.path().join("cuts.jsonl");
1443        let other = temp.path().join("other.jsonl");
1444        std::fs::write(&other, b"").unwrap();
1445
1446        let mut first = true;
1447        let error = open_locked(&path, true, || {
1448            let target = if std::mem::take(&mut first) {
1449                other.as_path()
1450            } else {
1451                path.as_path()
1452            };
1453            OpenOptions::new()
1454                .read(true)
1455                .append(true)
1456                .open(target)
1457                .map_err(|error| AppError::from_log_open(error, target))
1458        })
1459        .expect_err("a log that never appears cannot be locked");
1460
1461        assert_eq!(error.code, "not_found");
1462        assert_eq!(error.exit_code, 66);
1463    }
1464
1465    #[test]
1466    fn batch_append_rollback_restores_a_torn_tail_after_partial_write_failure() {
1467        let temp = TempDir::new().unwrap();
1468        let path = temp.path().join("cuts.jsonl");
1469        let original = b"{\"kind\":\"cut\"}\n{\"kind\":";
1470        std::fs::write(&path, original).unwrap();
1471        let mut file = OpenOptions::new()
1472            .read(true)
1473            .append(true)
1474            .open(&path)
1475            .unwrap();
1476
1477        let error = append_bytes_with(
1478            &mut file,
1479            &path,
1480            original,
1481            b"{\"kind\":\"resolve\"}\n{\"kind\":\"resolve\"}\n",
1482            |file, bytes| {
1483                file.write_all(&bytes[..8])?;
1484                Err(std::io::Error::other("injected partial write failure"))
1485            },
1486        )
1487        .unwrap_err();
1488
1489        assert_eq!(error.code, "io_error");
1490        assert_eq!(std::fs::read(&path).unwrap(), original);
1491    }
1492
1493    #[test]
1494    fn fold_matrix() {
1495        let id = compute_id("2026-07-09T00:00:00.000Z", "a", "x", Impact::Low, &[]);
1496        let cases = [
1497            ("cut", format!("{}\n", cut(&id)), 1, ItemStatus::Open, 0),
1498            (
1499                "resolve before cut",
1500                format!("{}\n{}\n", resolve(&id), cut(&id)),
1501                1,
1502                ItemStatus::Resolved,
1503                0,
1504            ),
1505            (
1506                "duplicates",
1507                format!(
1508                    "{}\n{}\n{}\n{}\n",
1509                    cut(&id),
1510                    cut(&id),
1511                    resolve(&id),
1512                    resolve(&id)
1513                ),
1514                1,
1515                ItemStatus::Resolved,
1516                2,
1517            ),
1518            (
1519                "unknown malformed orphan",
1520                format!(
1521                    "{{\"v\":2,\"kind\":\"future\"}}\nnope\n{}\n{}\n",
1522                    resolve("bl_deadbeef000000000000"),
1523                    cut(&id)
1524                ),
1525                1,
1526                ItemStatus::Open,
1527                3,
1528            ),
1529            (
1530                "torn tail",
1531                format!("{}\n{{\"kind\":", cut(&id)),
1532                1,
1533                ItemStatus::Open,
1534                1,
1535            ),
1536            (
1537                "all adversarial orderings interleaved",
1538                format!(
1539                    "{}\n{{\"v\":2,\"kind\":\"future\"}}\n{}\n{}\n{}\n{}\n{}\nnope\n{{\"kind\":",
1540                    resolve(&id),
1541                    cut(&id),
1542                    cut(&id),
1543                    cut_with_text(&id, "conflicting payload"),
1544                    resolve(&id),
1545                    resolve("bl_deadbeef000000000000"),
1546                ),
1547                1,
1548                ItemStatus::Resolved,
1549                6,
1550            ),
1551        ];
1552        for (name, input, item_count, status, warning_count) in cases {
1553            let folded = fold_bytes(input.as_bytes());
1554            assert_eq!(folded.items.len(), item_count, "{name}");
1555            if !folded.items.is_empty() {
1556                assert_eq!(folded.items[0].status, status, "{name}");
1557                assert_eq!(folded.items[0].text, "x", "{name}");
1558            }
1559            assert_eq!(folded.warnings.len(), warning_count, "{name}");
1560        }
1561    }
1562}