Skip to main content

blotter/
store.rs

1use crate::error::{AppError, AppResult};
2use crate::{ListItem, LogEvent, Resolution, format_timestamp};
3use serde_json::{Value, json};
4use std::collections::{BTreeMap, HashMap};
5use std::fs::{self, File, OpenOptions, Permissions};
6use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
7#[cfg(unix)]
8use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
9use std::path::{Component, Path, PathBuf};
10use std::thread;
11use std::time::Duration;
12
13const LOCK_ATTEMPTS: usize = 50;
14const LOCK_DELAY: Duration = Duration::from_millis(100);
15
16#[derive(Debug, Clone)]
17pub struct ResolvedFile {
18    pub path: PathBuf,
19    pub explicit: bool,
20    pub repo: Option<PathBuf>,
21    pub warnings: Vec<String>,
22}
23
24impl ResolvedFile {
25    /// Repo root for cwd relativization. Only a log living inside the repo
26    /// stores repo-relative cwd; explicit and global logs are machine-local,
27    /// keep absolute cwd, and would otherwise lose all provenance now that
28    /// records carry no repo field.
29    pub fn cwd_repo(&self) -> Option<&Path> {
30        self.repo
31            .as_deref()
32            .filter(|root| self.path.starts_with(root))
33    }
34}
35
36#[derive(Debug, Default)]
37pub struct FoldResult {
38    pub items: Vec<ListItem>,
39    pub warnings: Vec<String>,
40    records: BTreeMap<String, LogEvent>,
41    orphan_amends: HashMap<String, LogEvent>,
42}
43
44pub struct LoadedFold {
45    pub items: Vec<ListItem>,
46    pub warnings: Vec<String>,
47}
48
49impl FoldResult {
50    pub fn record(&self, id: &str) -> Option<&LogEvent> {
51        self.records.get(id)
52    }
53
54    /// Materialize a just-appended resolve from the fold that made the append
55    /// decision. A new base resolve activates the latest earlier orphan amend,
56    /// exactly as a complete subsequent fold would; an appended amend itself
57    /// is necessarily the latest materialized amend.
58    pub(crate) fn materialized_appended_resolution(&self, event: &LogEvent) -> Resolution {
59        let LogEvent::Resolve { id, amend, .. } = event else {
60            unreachable!("only resolve events materialize resolutions")
61        };
62        let effective = if *amend {
63            event
64        } else {
65            self.orphan_amends.get(id).unwrap_or(event)
66        };
67        resolution_from_event(effective)
68    }
69}
70
71#[derive(Default)]
72struct WarningCounts {
73    torn: usize,
74    malformed: usize,
75    unknown: usize,
76    duplicate_cuts: usize,
77    duplicate_dogears: usize,
78    duplicate_resolves: usize,
79    orphans: usize,
80}
81
82pub(crate) struct ScannedLine<'a> {
83    pub line: usize,
84    pub raw: &'a [u8],
85    pub event: Result<LogEvent, ScanIssue>,
86}
87
88pub(crate) enum ScanIssue {
89    Malformed(String),
90    Unknown(Option<String>),
91    Torn,
92}
93
94pub fn discover(flag: Option<PathBuf>) -> AppResult<ResolvedFile> {
95    let cwd = std::env::current_dir().map_err(|error| AppError::from_io(error, Path::new(".")))?;
96    discover_from(&cwd, flag)
97}
98
99pub fn discover_from(cwd: &Path, flag: Option<PathBuf>) -> AppResult<ResolvedFile> {
100    let repo = find_repo_root(cwd);
101    if let Some(path) = flag {
102        return Ok(resolved_file(absolute(cwd, path), true, repo));
103    }
104    if let Some(path) = std::env::var_os("BLOTTER_FILE")
105        && !path.is_empty()
106    {
107        return Ok(resolved_file(
108            absolute(cwd, PathBuf::from(path)),
109            true,
110            repo,
111        ));
112    }
113    if let Some(root) = repo.clone() {
114        let path = default_log_path(&root);
115        return Ok(resolved_file(path, false, Some(root)));
116    }
117    let home = home_dir(cwd).ok_or_else(|| {
118        AppError::config(
119            "cannot resolve the home directory for the default blotter file",
120            "Set HOME or pass --file PATH.",
121        )
122    })?;
123    Ok(resolved_file(home.join(".blotter/log.jsonl"), false, None))
124}
125
126fn resolved_file(path: PathBuf, explicit: bool, repo: Option<PathBuf>) -> ResolvedFile {
127    ResolvedFile {
128        warnings: Vec::new(),
129        path,
130        explicit,
131        repo,
132    }
133}
134
135pub fn default_log_path(root: &Path) -> PathBuf {
136    root.join(".blotter.jsonl")
137}
138
139pub fn find_repo_root(start: &Path) -> Option<PathBuf> {
140    start
141        .ancestors()
142        .find(|candidate| candidate.join(".git").exists())
143        .map(Path::to_path_buf)
144}
145
146pub fn home_dir(cwd: &Path) -> Option<PathBuf> {
147    std::env::var_os("HOME")
148        .filter(|value| !value.is_empty())
149        .map(PathBuf::from)
150        .map(|home| absolute(cwd, home))
151}
152
153pub fn record_cwd(cwd: &Path, repo: Option<&Path>, home: Option<&Path>) -> String {
154    if let Some(relative) = repo.and_then(|root| cwd.strip_prefix(root).ok()) {
155        return match relative.as_os_str().is_empty() {
156            true => ".".into(),
157            false => relative.to_string_lossy().into_owned(),
158        };
159    }
160    match home.and_then(|root| cwd.strip_prefix(root).ok()) {
161        Some(relative) if relative.as_os_str().is_empty() => "~".into(),
162        Some(relative) => format!("~/{}", relative.to_string_lossy()),
163        None => cwd.to_string_lossy().into_owned(),
164    }
165}
166
167fn absolute(cwd: &Path, path: PathBuf) -> PathBuf {
168    let joined = if path.is_absolute() {
169        path
170    } else {
171        cwd.join(path)
172    };
173    let mut normalized = PathBuf::new();
174    for component in joined.components() {
175        match component {
176            Component::CurDir => {}
177            Component::ParentDir => {
178                normalized.pop();
179            }
180            other => normalized.push(other.as_os_str()),
181        }
182    }
183    normalized
184}
185
186pub fn with_shared<T>(path: &Path, action: impl FnOnce(&mut File) -> AppResult<T>) -> AppResult<T> {
187    let mut file = open_locked(path, false, || {
188        File::open(path).map_err(|error| AppError::from_log_open(error, path))
189    })?;
190    let result = action(&mut file);
191    let unlock = file
192        .unlock()
193        .map_err(|error| AppError::from_io(error, path));
194    match (result, unlock) {
195        (Err(error), _) | (Ok(_), Err(error)) => Err(error),
196        (Ok(value), Ok(())) => Ok(value),
197    }
198}
199
200pub fn read_or_empty<T>(
201    path: &Path,
202    explicit: bool,
203    warnings: &mut Vec<String>,
204    warning: &str,
205    suggested_fix: &str,
206    empty: impl FnOnce() -> T,
207    read: impl FnOnce(&mut File) -> AppResult<T>,
208) -> AppResult<(T, bool)> {
209    match with_shared(path, read) {
210        Ok(value) => Ok((value, true)),
211        Err(error) if error.code == "not_found" && error.exit_code == 66 && !explicit => {
212            warnings.push(warning.into());
213            Ok((empty(), false))
214        }
215        Err(error) if error.code == "not_found" && error.exit_code == 66 => {
216            Err(AppError::not_found(
217                format!("blotter file not found: {}", path.display()),
218                suggested_fix,
219            ))
220        }
221        Err(error) => Err(error),
222    }
223}
224
225pub fn load_folded(resolved: &ResolvedFile) -> AppResult<LoadedFold> {
226    let mut warnings = resolved.warnings.clone();
227    let (folded, _) = read_or_empty(
228        &resolved.path,
229        resolved.explicit,
230        &mut warnings,
231        "no blotter file yet; blotter add creates it",
232        "Pass an existing --file PATH or run `blotter add` to create a discovered default file.",
233        FoldResult::default,
234        |log| {
235            let bytes = read_bytes(log, &resolved.path)?;
236            Ok(fold_bytes(&bytes))
237        },
238    )?;
239    warnings.extend(folded.warnings);
240    Ok(LoadedFold {
241        items: folded.items,
242        warnings,
243    })
244}
245
246pub fn with_exclusive<T>(
247    path: &Path,
248    create: bool,
249    action: impl FnOnce(&mut File) -> AppResult<T>,
250) -> AppResult<T> {
251    if create && let Some(parent) = path.parent() {
252        std::fs::create_dir_all(parent).map_err(|error| AppError::from_io(error, parent))?;
253    }
254    let mut file = open_locked(path, true, || {
255        OpenOptions::new()
256            .read(true)
257            .append(true)
258            .create(create)
259            .open(path)
260            .map_err(|error| AppError::from_log_open(error, path))
261    })?;
262    let result = action(&mut file);
263    let unlock = file
264        .unlock()
265        .map_err(|error| AppError::from_io(error, path));
266    match (result, unlock) {
267        (Err(error), _) | (Ok(_), Err(error)) => Err(error),
268        (Ok(value), Ok(())) => Ok(value),
269    }
270}
271
272fn open_locked(
273    path: &Path,
274    exclusive: bool,
275    mut open: impl FnMut() -> AppResult<File>,
276) -> AppResult<File> {
277    let mut file = Some(open()?);
278    for attempt in 0..LOCK_ATTEMPTS {
279        if file.is_none() {
280            match open() {
281                Ok(opened) => file = Some(opened),
282                Err(error) if error.code == "not_found" => {
283                    if attempt + 1 < LOCK_ATTEMPTS {
284                        thread::sleep(LOCK_DELAY);
285                    }
286                    continue;
287                }
288                Err(error) => return Err(error),
289            }
290        }
291        let result = if exclusive {
292            file.as_ref().expect("file is open").try_lock()
293        } else {
294            file.as_ref().expect("file is open").try_lock_shared()
295        };
296        match result {
297            Ok(()) => {
298                if path_identity_matches(file.as_ref().expect("file is open"), path)? {
299                    return Ok(file.take().expect("file is open"));
300                }
301                let stale = file.take().expect("file is open");
302                let _ = stale.unlock();
303            }
304            Err(error) => {
305                let error: std::io::Error = error.into();
306                if error.kind() != std::io::ErrorKind::WouldBlock {
307                    return Err(AppError::from_io(error, path));
308                }
309                if attempt + 1 < LOCK_ATTEMPTS {
310                    thread::sleep(LOCK_DELAY);
311                }
312            }
313        }
314    }
315    Err(AppError::lock_timeout(path))
316}
317
318#[cfg(unix)]
319fn path_identity_matches(file: &File, path: &Path) -> AppResult<bool> {
320    // File::metadata uses fstat; fs::metadata obtains a fresh stat of the path.
321    let locked = file
322        .metadata()
323        .map_err(|error| AppError::from_io(error, path))?;
324    match std::fs::metadata(path) {
325        Ok(current) => Ok(locked.dev() == current.dev() && locked.ino() == current.ino()),
326        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
327        Err(error) => Err(AppError::from_io(error, path)),
328    }
329}
330
331#[cfg(not(unix))]
332fn path_identity_matches(_file: &File, _path: &Path) -> AppResult<bool> {
333    Ok(true)
334}
335
336pub fn read_bytes(file: &mut File, path: &Path) -> AppResult<Vec<u8>> {
337    file.seek(SeekFrom::Start(0))
338        .and_then(|_| {
339            let mut bytes = Vec::new();
340            file.read_to_end(&mut bytes).map(|_| bytes)
341        })
342        .map_err(|error| AppError::from_io(error, path))
343}
344
345pub fn write_new_file(path: &Path, bytes: &[u8], permissions: &Permissions) -> AppResult<PathBuf> {
346    let mut file = create_new_file(path, permissions, false)
347        .map_err(|error| AppError::from_io(error, path))?;
348    if let Err(error) = file.write_all(bytes) {
349        discard_new_file(file, path);
350        return Err(AppError::from_io(error, path));
351    }
352    if let Err(error) = file.sync_all() {
353        discard_new_file(file, path);
354        return Err(AppError::from_io(error, path));
355    }
356    Ok(path.to_path_buf())
357}
358
359pub fn append_file(path: &Path, bytes: &[u8], permissions: &Permissions) -> AppResult<PathBuf> {
360    let (mut file, created) = match create_new_file(path, permissions, false) {
361        Ok(file) => (file, true),
362        Err(error) if error.kind() == ErrorKind::AlreadyExists => (
363            OpenOptions::new()
364                .append(true)
365                .open(path)
366                .map_err(|error| AppError::from_io(error, path))?,
367            false,
368        ),
369        Err(error) => return Err(AppError::from_io(error, path)),
370    };
371    if let Err(error) = file.write_all(bytes) {
372        if created {
373            discard_new_file(file, path);
374        }
375        return Err(AppError::from_io(error, path));
376    }
377    if let Err(error) = file.sync_all() {
378        if created {
379            discard_new_file(file, path);
380        }
381        return Err(AppError::from_io(error, path));
382    }
383    Ok(path.to_path_buf())
384}
385
386pub fn replace_log(
387    path: &Path,
388    bytes: &[u8],
389    permissions: &Permissions,
390    temporary_suffix: &str,
391) -> AppResult<()> {
392    let temporary = suffixed_path(path, temporary_suffix);
393    let mut file = create_new_file(&temporary, permissions, true)
394        .map_err(|error| AppError::from_io(error, &temporary))?;
395    if let Err(error) = file.write_all(bytes) {
396        discard_new_file(file, &temporary);
397        return Err(AppError::from_io(error, &temporary));
398    }
399    if let Err(error) = file.sync_all() {
400        discard_new_file(file, &temporary);
401        return Err(AppError::from_io(error, &temporary));
402    }
403    drop(file);
404    if let Err(error) = fs::rename(&temporary, path) {
405        let _ = fs::remove_file(&temporary);
406        return Err(AppError::from_io(error, path));
407    }
408    if let Some(parent) = path.parent()
409        && let Ok(directory) = File::open(parent)
410    {
411        let _ = directory.sync_all();
412    }
413    Ok(())
414}
415
416/// Resolve a symlinked log path to its target before a copy-and-swap, so the
417/// backup, sidecar, and atomic replacement all act on the real file and the
418/// link survives. Only final-component links are chased; parent components
419/// keep their spelling so envelope paths stay stable for regular files.
420pub fn resolve_symlinked_log(path: &Path) -> AppResult<PathBuf> {
421    let mut current = path.to_path_buf();
422    for _ in 0..40 {
423        let metadata =
424            fs::symlink_metadata(&current).map_err(|error| AppError::from_io(error, &current))?;
425        if !metadata.file_type().is_symlink() {
426            return Ok(current);
427        }
428        let target = fs::read_link(&current).map_err(|error| AppError::from_io(error, &current))?;
429        current = if target.is_absolute() {
430            target
431        } else {
432            match current.parent() {
433                Some(parent) => parent.join(&target),
434                None => target,
435            }
436        };
437    }
438    Err(AppError::from_io(
439        std::io::Error::other("too many levels of symbolic links"),
440        path,
441    ))
442}
443
444pub fn suffixed_path(path: &Path, suffix: &str) -> PathBuf {
445    let mut value = path.as_os_str().to_os_string();
446    value.push(suffix);
447    PathBuf::from(value)
448}
449
450pub fn backup_timestamp(now: jiff::Timestamp) -> String {
451    format_timestamp(now)
452        .chars()
453        .filter(|character| !matches!(character, '-' | ':' | '.'))
454        .collect()
455}
456
457pub fn restore_hint(backup: &Path, path: &Path) -> String {
458    format!("cp {} {}", shell_quote(backup), shell_quote(path))
459}
460
461fn create_new_file(
462    path: &Path,
463    permissions: &Permissions,
464    set_permissions_on_non_unix: bool,
465) -> std::io::Result<File> {
466    let mut options = OpenOptions::new();
467    options.write(true).create_new(true);
468    #[cfg(unix)]
469    options.mode(permissions.mode());
470    let file = options.open(path)?;
471    #[cfg(unix)]
472    let permissions_result = {
473        let _ = set_permissions_on_non_unix;
474        file.set_permissions(permissions.clone())
475    };
476    #[cfg(not(unix))]
477    let permissions_result = set_permissions_on_non_unix
478        .then(|| file.set_permissions(permissions.clone()))
479        .transpose()
480        .map(|_| ());
481    if let Err(error) = permissions_result {
482        drop(file);
483        let _ = fs::remove_file(path);
484        return Err(error);
485    }
486    Ok(file)
487}
488
489fn discard_new_file(file: File, path: &Path) {
490    drop(file);
491    let _ = fs::remove_file(path);
492}
493
494fn shell_quote(path: &Path) -> String {
495    format!("'{}'", path.to_string_lossy().replace('\'', "'\\''"))
496}
497
498pub fn append_json<T: serde::Serialize>(
499    file: &mut File,
500    path: &Path,
501    prior: &[u8],
502    record: &T,
503) -> AppResult<()> {
504    let mut record_bytes = Vec::new();
505    serde_json::to_writer(&mut record_bytes, record)
506        .map_err(|error| AppError::internal(error.to_string()))?;
507    record_bytes.push(b'\n');
508    append_bytes(file, path, prior, &record_bytes)
509}
510
511pub fn append_unique(path: &Path, record: LogEvent, dry_run: bool) -> AppResult<(bool, LogEvent)> {
512    if dry_run {
513        return Ok((false, record));
514    }
515    let id = record.id().expect("new records have IDs").to_owned();
516    let kind = match &record {
517        LogEvent::Cut { .. } => "cut",
518        LogEvent::Dogear { .. } => "dogear",
519        _ => unreachable!("append_unique only receives cut or dogear records"),
520    };
521    with_exclusive(path, true, |log| {
522        let bytes = read_bytes(log, path)?;
523        let folded = fold_bytes(&bytes);
524        if let Some(existing) = folded.record(&id) {
525            return if std::mem::discriminant(&record) == std::mem::discriminant(existing) {
526                Ok((false, existing.clone()))
527            } else {
528                Err(AppError::internal(format!(
529                    "{kind} ID collides with an existing non-{kind} record"
530                )))
531            };
532        }
533        append_json(log, path, &bytes, &record)?;
534        Ok((true, record))
535    })
536}
537
538pub fn append_json_batch<T: serde::Serialize>(
539    file: &mut File,
540    path: &Path,
541    prior: &[u8],
542    records: &[T],
543) -> AppResult<()> {
544    let mut record_bytes = Vec::new();
545    for record in records {
546        serde_json::to_writer(&mut record_bytes, record)
547            .map_err(|error| AppError::internal(error.to_string()))?;
548        record_bytes.push(b'\n');
549    }
550    append_bytes(file, path, prior, &record_bytes)
551}
552
553fn append_bytes(file: &mut File, path: &Path, prior: &[u8], record_bytes: &[u8]) -> AppResult<()> {
554    append_bytes_with(file, path, prior, record_bytes, |file, bytes| {
555        file.write_all(bytes)
556    })
557}
558
559fn append_bytes_with(
560    file: &mut File,
561    path: &Path,
562    prior: &[u8],
563    record_bytes: &[u8],
564    write: impl FnOnce(&mut File, &[u8]) -> std::io::Result<()>,
565) -> AppResult<()> {
566    let original_len = file
567        .metadata()
568        .map_err(|error| AppError::from_io(error, path))?
569        .len();
570    let mut bytes = Vec::new();
571    if !prior.is_empty() && !prior.ends_with(b"\n") {
572        bytes.push(b'\n');
573    }
574    bytes.extend_from_slice(record_bytes);
575    // If the write fails, roll back to the pre-write length; if rollback also fails, surface both.
576    if let Err(error) = write(file, &bytes) {
577        if let Err(rollback) = file.set_len(original_len) {
578            return Err(AppError {
579                code: "io_error",
580                message: format!(
581                    "append failed: {error}; rollback to original length {original_len} failed: {rollback}"
582                ),
583                details: json!({}),
584                retryable: false,
585                suggested_fix: "Check the blotter file and filesystem, then retry.".into(),
586                exit_code: 74,
587            });
588        }
589        return Err(AppError::from_io(error, path));
590    }
591    Ok(())
592}
593
594/// Scan physical JSONL lines once. A final non-newline line is accepted only
595/// when its decoded JSON carries a recognized kind, so consumers cannot
596/// disagree on torn tails.
597pub(crate) fn scan(bytes: &[u8]) -> impl Iterator<Item = ScannedLine<'_>> + '_ {
598    // A sole empty segment is an empty file or a file holding only "\n", so
599    // it is not a physical line. An empty segment after a record is malformed.
600    let terminated = bytes.ends_with(b"\n");
601    let body = if terminated {
602        &bytes[..bytes.len() - 1]
603    } else {
604        bytes
605    };
606    let line_count = body.split(|byte| *byte == b'\n').count();
607    body.split(|byte| *byte == b'\n')
608        .enumerate()
609        .filter_map(move |(index, raw)| {
610            let final_line = index + 1 == line_count;
611            if raw.is_empty() && final_line && index == 0 {
612                return None;
613            }
614            let decoded = serde_json::from_slice::<Value>(raw);
615            let known = decoded.as_ref().ok().and_then(known_kind);
616            let event = if final_line && !terminated && known.is_none() {
617                Err(ScanIssue::Torn)
618            } else {
619                match decoded {
620                    Ok(value) => parse_event(value, known),
621                    Err(_) => Err(ScanIssue::Malformed("line is not valid JSON".into())),
622                }
623            };
624            Some(ScannedLine {
625                line: index + 1,
626                raw,
627                event,
628            })
629        })
630}
631
632fn known_kind(value: &Value) -> Option<&'static str> {
633    match value.get("kind").and_then(Value::as_str) {
634        Some("cut") => Some("cut"),
635        Some("dogear") => Some("dogear"),
636        Some("resolve") => Some("resolve"),
637        _ => None,
638    }
639}
640
641fn parse_event(value: Value, known: Option<&'static str>) -> Result<LogEvent, ScanIssue> {
642    let unknown = value.get("kind").and_then(Value::as_str).map(str::to_owned);
643    match serde_json::from_value::<LogEvent>(value) {
644        Ok(LogEvent::Unknown) => Err(ScanIssue::Unknown(unknown)),
645        Ok(event) => {
646            let ts = match &event {
647                LogEvent::Cut { ts, .. }
648                | LogEvent::Dogear { ts, .. }
649                | LogEvent::Resolve { ts, .. } => ts,
650                LogEvent::Unknown => unreachable!("unknown events are classified above"),
651            };
652            match ts.parse::<jiff::Timestamp>() {
653                Ok(_) => Ok(event),
654                Err(_) => Err(ScanIssue::Malformed(format!(
655                    "{} ts is not a full RFC3339 timestamp",
656                    known.expect("parsed events have a known kind")
657                ))),
658            }
659        }
660        Err(error) => match known {
661            Some(kind) => Err(ScanIssue::Malformed(format!(
662                "invalid {kind} record: {error}"
663            ))),
664            None => Err(ScanIssue::Unknown(unknown)),
665        },
666    }
667}
668
669fn resolution_from_event(event: &LogEvent) -> Resolution {
670    let LogEvent::Resolve {
671        ts,
672        agent,
673        note,
674        task,
675        pr,
676        commit,
677        url,
678        dropped,
679        amend,
680        ..
681    } = event
682    else {
683        unreachable!("only resolve events materialize resolutions")
684    };
685    Resolution {
686        ts: ts.clone(),
687        agent: agent.clone(),
688        note: note.clone(),
689        task: task.clone(),
690        pr: pr.clone(),
691        commit: commit.clone(),
692        url: url.clone(),
693        dropped: *dropped,
694        amended: *amend,
695    }
696}
697
698pub fn fold_bytes(bytes: &[u8]) -> FoldResult {
699    let mut records = BTreeMap::<String, LogEvent>::new();
700    let mut resolves = HashMap::<String, LogEvent>::new();
701    let mut amends = HashMap::<String, LogEvent>::new();
702    let mut counts = WarningCounts::default();
703    for scanned in scan(bytes) {
704        match scanned.event {
705            Err(ScanIssue::Malformed(_)) => counts.malformed += 1,
706            Err(ScanIssue::Unknown(_)) => counts.unknown += 1,
707            Err(ScanIssue::Torn) => counts.torn += 1,
708            Ok(mut event) => match &mut event {
709                LogEvent::Cut { tags, .. } => {
710                    // Fold normalizes legacy tag arrays for list output. Doctor
711                    // receives the scanner's unmodified parsed event instead.
712                    tags.sort();
713                    tags.dedup();
714                    let id = event.id().expect("parsed cuts have IDs").to_owned();
715                    if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id) {
716                        entry.insert(event);
717                    } else {
718                        counts.duplicate_cuts += 1;
719                    }
720                }
721                LogEvent::Dogear { tags, .. } => {
722                    tags.sort();
723                    tags.dedup();
724                    let id = event.id().expect("parsed dogears have IDs").to_owned();
725                    if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id) {
726                        entry.insert(event);
727                    } else {
728                        counts.duplicate_dogears += 1;
729                    }
730                }
731                LogEvent::Resolve { id, amend, .. } => {
732                    let id = id.clone();
733                    let amend = *amend;
734                    if amend {
735                        amends.insert(id, event);
736                    } else if let std::collections::hash_map::Entry::Vacant(entry) =
737                        resolves.entry(id)
738                    {
739                        entry.insert(event);
740                    } else {
741                        counts.duplicate_resolves += 1;
742                    }
743                }
744                LogEvent::Unknown => counts.unknown += 1,
745            },
746        }
747    }
748
749    // Base resolves remain first-wins. A latest amend only materializes when
750    // the full scan found a base resolve, so merge-reordered base resolves work.
751    let mut orphan_amends = HashMap::new();
752    for (id, amend) in amends {
753        match resolves.entry(id) {
754            std::collections::hash_map::Entry::Occupied(mut entry) => {
755                entry.insert(amend);
756            }
757            std::collections::hash_map::Entry::Vacant(entry) => {
758                counts.orphans += 1;
759                orphan_amends.insert(entry.into_key(), amend);
760            }
761        }
762    }
763
764    for id in resolves.keys() {
765        if !records.contains_key(id) {
766            counts.orphans += 1;
767        }
768    }
769    let mut items: Vec<_> = records
770        .values()
771        .cloned()
772        .map(|record| {
773            let resolution = record
774                .id()
775                .and_then(|id| resolves.get(id))
776                .map(resolution_from_event);
777            let item = ListItem::from_record(record, resolution);
778            let timestamp = item
779                .ts
780                .parse::<jiff::Timestamp>()
781                .expect("folded items have valid RFC3339 timestamps");
782            (item, timestamp)
783        })
784        .collect();
785    items.sort_by(|(left, left_timestamp), (right, right_timestamp)| {
786        match (left.kind.as_str(), right.kind.as_str()) {
787            ("cut", "cut") => right
788                .severity
789                .expect("cut has severity")
790                .rank()
791                .cmp(&left.severity.expect("cut has severity").rank())
792                .then_with(|| right_timestamp.cmp(left_timestamp))
793                .then_with(|| left.id.cmp(&right.id)),
794            ("dogear", "dogear") => right_timestamp
795                .cmp(left_timestamp)
796                .then_with(|| left.id.cmp(&right.id)),
797            ("cut", "dogear") => std::cmp::Ordering::Less,
798            ("dogear", "cut") => std::cmp::Ordering::Greater,
799            _ => left.kind.cmp(&right.kind),
800        }
801    });
802    let items = items.into_iter().map(|(item, _)| item).collect();
803
804    let mut warnings = Vec::new();
805    warning(&mut warnings, counts.torn, "torn final line");
806    warning(&mut warnings, counts.malformed, "malformed line");
807    warning(&mut warnings, counts.unknown, "unknown event");
808    warning(&mut warnings, counts.duplicate_cuts, "duplicate cut");
809    warning(&mut warnings, counts.duplicate_dogears, "duplicate dogear");
810    warning(
811        &mut warnings,
812        counts.duplicate_resolves,
813        "duplicate resolve",
814    );
815    warning(&mut warnings, counts.orphans, "orphan resolve");
816    FoldResult {
817        items,
818        warnings,
819        records,
820        orphan_amends,
821    }
822}
823
824fn warning(warnings: &mut Vec<String>, count: usize, label: &str) {
825    if count > 0 {
826        warnings.push(format!(
827            "skipped {count} {label}{}",
828            if count == 1 { "" } else { "s" }
829        ));
830    }
831}
832
833#[cfg(test)]
834mod tests {
835    use super::*;
836    use crate::{ItemStatus, Severity, compute_id};
837    use std::io::Write;
838    use tempfile::TempDir;
839
840    fn cut(id: &str) -> String {
841        cut_with_text(id, "x")
842    }
843
844    fn cut_with_text(id: &str, text: &str) -> String {
845        serde_json::json!({
846            "kind":"cut", "id":id, "ts":"2026-07-09T00:00:00.000Z",
847            "agent":"a", "text":text, "tags":[], "severity":"minor",
848            "cwd":"/tmp", "repo":null
849        })
850        .to_string()
851    }
852
853    fn resolve(id: &str) -> String {
854        serde_json::json!({
855            "kind":"resolve", "id":id, "ts":"2026-07-10T00:00:00.000Z",
856            "agent":"a", "note":null
857        })
858        .to_string()
859    }
860
861    #[cfg(unix)]
862    #[test]
863    fn exclusive_lock_reopens_a_replaced_path_before_appending() {
864        let temp = TempDir::new().unwrap();
865        let path = temp.path().join("cuts.jsonl");
866        std::fs::write(&path, b"old\n").unwrap();
867
868        let holder = OpenOptions::new()
869            .read(true)
870            .write(true)
871            .open(&path)
872            .unwrap();
873        holder.lock().unwrap();
874
875        let preopened = OpenOptions::new()
876            .read(true)
877            .append(true)
878            .open(&path)
879            .unwrap();
880        let (opened_tx, opened_rx) = std::sync::mpsc::channel();
881        let writer_path = path.clone();
882        let writer = std::thread::spawn(move || {
883            let mut first_open = Some(preopened);
884            let mut file = open_locked(&writer_path, true, || {
885                if let Some(file) = first_open.take() {
886                    // The writer now owns a descriptor for the old inode.
887                    opened_tx.send(()).unwrap();
888                    Ok(file)
889                } else {
890                    OpenOptions::new()
891                        .read(true)
892                        .append(true)
893                        .open(&writer_path)
894                        .map_err(|error| AppError::from_log_open(error, &writer_path))
895                }
896            })
897            .unwrap();
898            file.write_all(b"writer\n").unwrap();
899            file.unlock().unwrap();
900        });
901
902        opened_rx
903            .recv_timeout(std::time::Duration::from_secs(2))
904            .unwrap();
905        let replacement = temp.path().join("replacement.jsonl");
906        std::fs::write(&replacement, b"replacement\n").unwrap();
907        std::fs::rename(&replacement, &path).unwrap();
908        holder.unlock().unwrap();
909        writer.join().unwrap();
910
911        assert_eq!(std::fs::read(&path).unwrap(), b"replacement\nwriter\n");
912    }
913
914    #[test]
915    fn batch_append_rollback_restores_a_torn_tail_after_partial_write_failure() {
916        let temp = TempDir::new().unwrap();
917        let path = temp.path().join("cuts.jsonl");
918        let original = b"{\"kind\":\"cut\"}\n{\"kind\":";
919        std::fs::write(&path, original).unwrap();
920        let mut file = OpenOptions::new()
921            .read(true)
922            .append(true)
923            .open(&path)
924            .unwrap();
925
926        let error = append_bytes_with(
927            &mut file,
928            &path,
929            original,
930            b"{\"kind\":\"resolve\"}\n{\"kind\":\"resolve\"}\n",
931            |file, bytes| {
932                file.write_all(&bytes[..8])?;
933                Err(std::io::Error::other("injected partial write failure"))
934            },
935        )
936        .unwrap_err();
937
938        assert_eq!(error.code, "io_error");
939        assert_eq!(std::fs::read(&path).unwrap(), original);
940    }
941
942    #[test]
943    fn fold_matrix() {
944        let id = compute_id("2026-07-09T00:00:00.000Z", "a", "x", Severity::Minor, &[]);
945        let cases = [
946            ("cut", format!("{}\n", cut(&id)), 1, ItemStatus::Open, 0),
947            (
948                "resolve before cut",
949                format!("{}\n{}\n", resolve(&id), cut(&id)),
950                1,
951                ItemStatus::Resolved,
952                0,
953            ),
954            (
955                "duplicates",
956                format!(
957                    "{}\n{}\n{}\n{}\n",
958                    cut(&id),
959                    cut(&id),
960                    resolve(&id),
961                    resolve(&id)
962                ),
963                1,
964                ItemStatus::Resolved,
965                2,
966            ),
967            (
968                "unknown malformed orphan",
969                format!(
970                    "{{\"kind\":\"future\"}}\nnope\n{}\n{}\n",
971                    resolve("bl_deadbeef0000"),
972                    cut(&id)
973                ),
974                1,
975                ItemStatus::Open,
976                3,
977            ),
978            (
979                "torn tail",
980                format!("{}\n{{\"kind\":", cut(&id)),
981                1,
982                ItemStatus::Open,
983                1,
984            ),
985            (
986                "all adversarial orderings interleaved",
987                format!(
988                    "{}\n{{\"kind\":\"future\"}}\n{}\n{}\n{}\n{}\n{}\nnope\n{{\"kind\":",
989                    resolve(&id),
990                    cut(&id),
991                    cut(&id),
992                    cut_with_text(&id, "conflicting payload"),
993                    resolve(&id),
994                    resolve("bl_deadbeef0000"),
995                ),
996                1,
997                ItemStatus::Resolved,
998                6,
999            ),
1000        ];
1001        for (name, input, item_count, status, warning_count) in cases {
1002            let folded = fold_bytes(input.as_bytes());
1003            assert_eq!(folded.items.len(), item_count, "{name}");
1004            if !folded.items.is_empty() {
1005                assert_eq!(folded.items[0].status, status, "{name}");
1006                assert_eq!(folded.items[0].text, "x", "{name}");
1007            }
1008            assert_eq!(folded.warnings.len(), warning_count, "{name}");
1009        }
1010    }
1011}