hayson-engine 0.27.0

use std::fs;
use std::io;
use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::atomic::write_atomic;
use crate::crypto::Sealer;

pub const JOURNAL_FILE: &str = ".hayson.txn";
pub const JOURNAL_VERSION: u32 = 1;

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct JournalHeader {
    // NOTE: "hayson" is the pre-5.0.0 key — read it, or an interrupted unit is missed at recovery.
    #[serde(alias = "hayson")]
    pub journal_version: u32,
    pub kind: String,
    pub id: String,
    // NOTE: `started_at` is what 4.0.0 wrote before this alias — read it, or its unit is dropped.
    #[serde(alias = "started_at")]
    pub started_at: i64,
    pub tables: Vec<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JournalEntry {
    pub t: String,
    pub s: u64,
    pub l: String,
}

#[derive(Debug, Clone)]
pub struct Journal {
    pub header: JournalHeader,
    pub entries: Vec<JournalEntry>,
}

pub fn write_journal(path: &Path, journal: &Journal) -> io::Result<()> {
    write_journal_sealed(path, journal, None)
}

/// [`write_journal`] whose file is sealed with `sealer` when one is given: a journal holds the very
/// lines about to be appended to the logs, so in an encrypted database it must not be plain.
pub fn write_journal_sealed(path: &Path, journal: &Journal, sealer: Option<&Sealer>) -> io::Result<()> {
    let mut lines = Vec::with_capacity(journal.entries.len() + 1);
    lines.push(serde_json::to_string(&journal.header)?);
    for entry in &journal.entries {
        lines.push(serde_json::to_string(entry)?);
    }
    let mut contents = lines.join("\n");
    contents.push('\n');
    match sealer {
        Some(sealer) => write_atomic(path, &sealer.seal_blob("journal", contents.as_bytes()), None),
        None => write_atomic(path, contents.as_bytes(), None),
    }
}

/// Reads back a journal written by [`write_journal`]. Any failure to parse it — missing file,
/// truncated write, a header from a future/foreign format — is reported as "nothing to recover"
/// rather than an error: an unreadable journal cannot be finished, only left alone.
pub fn read_journal(path: &Path) -> Option<Journal> {
    read_journal_sealed(path, None)
}

/// [`read_journal`] for a sealed journal. One that does not open is unreadable like any other.
pub fn read_journal_sealed(path: &Path, sealer: Option<&Sealer>) -> Option<Journal> {
    let bytes = fs::read(path).ok()?;
    let bytes = match sealer {
        Some(sealer) => sealer.open_blob("journal", &bytes).ok()?,
        None => bytes,
    };
    let raw = String::from_utf8(bytes).ok()?;
    let mut lines = raw.lines();
    let header: JournalHeader = serde_json::from_str(lines.next()?).ok()?;
    if header.journal_version != JOURNAL_VERSION || header.kind != "unit" {
        return None;
    }
    let mut entries = Vec::new();
    for line in lines {
        if line.is_empty() {
            continue;
        }
        entries.push(serde_json::from_str(line).ok()?);
    }
    Some(Journal { header, entries })
}

pub fn remove_journal(path: &Path) {
    let _ = fs::remove_file(path);
    let _ = fs::remove_file(format!("{}.tmp", path.display()));
}

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

    #[test]
    fn round_trips_header_and_entries() {
        let dir = tempdir().unwrap();
        let path = dir.path().join(JOURNAL_FILE);
        let journal = Journal {
            header: JournalHeader {
                journal_version: JOURNAL_VERSION,
                kind: "unit".into(),
                id: "abc".into(),
                started_at: 42,
                tables: vec!["users".into(), "orders".into()],
            },
            entries: vec![
                JournalEntry {
                    t: "users".into(),
                    s: 1,
                    l: "{\"_k\":\"u1\"}".into(),
                },
                JournalEntry {
                    t: "orders".into(),
                    s: 1,
                    l: "{\"_k\":\"o1\"}".into(),
                },
            ],
        };

        write_journal(&path, &journal).unwrap();
        let read = read_journal(&path).unwrap();
        assert_eq!(read.header.id, "abc");
        assert_eq!(read.header.tables, vec!["users".to_string(), "orders".to_string()]);
        assert_eq!(read.entries.len(), 2);

        remove_journal(&path);
        assert!(read_journal(&path).is_none());
    }

    #[test]
    fn a_missing_journal_reads_as_nothing_to_recover() {
        let dir = tempdir().unwrap();
        let path = dir.path().join(JOURNAL_FILE);
        assert!(read_journal(&path).is_none());
    }

    #[test]
    fn writes_the_header_key_the_format_documents() {
        let dir = tempdir().unwrap();
        let path = dir.path().join(JOURNAL_FILE);
        let journal = Journal {
            header: JournalHeader {
                journal_version: JOURNAL_VERSION,
                kind: "unit".into(),
                id: "abc".into(),
                started_at: 1_786_995_100_337,
                tables: vec!["users".into()],
            },
            entries: Vec::new(),
        };

        write_journal(&path, &journal).unwrap();

        let header = fs::read_to_string(&path).unwrap().lines().next().unwrap().to_string();
        assert!(header.contains("\"startedAt\":1786995100337"), "{header}");
        assert!(!header.contains("started_at"), "{header}");
        assert!(header.contains("\"journalVersion\":1"), "{header}");
        assert!(
            !header.contains("\"hayson\""),
            "the old marker key is never written again: {header}"
        );
    }

    /// A unit left unfinished by a release that wrote the marker under `"hayson"` still has to be
    /// finished — nothing migrates a journal, and a missed one loses a half-applied write.
    #[test]
    fn reads_the_pre_rename_marker_key_an_older_release_left_behind() {
        let dir = tempdir().unwrap();
        let path = dir.path().join(JOURNAL_FILE);
        fs::write(
            &path,
            "{\"hayson\":1,\"kind\":\"unit\",\"id\":\"abc\",\"startedAt\":42,\"tables\":[\"users\"]}\n{\"t\":\"users\",\"s\":7,\"l\":\"{}\"}\n",
        )
        .unwrap();

        let read = read_journal(&path).expect("a journal under the old marker key is still recoverable");
        assert_eq!(read.header.journal_version, JOURNAL_VERSION);
        assert_eq!(read.header.id, "abc");
        assert_eq!(read.entries.len(), 1);
    }

    /// A unit left unfinished by a release that wrote the snake-case key still has to be finished,
    /// not silently dropped — the whole point of the journal is that nothing gets lost.
    #[test]
    fn reads_the_snake_case_header_an_older_release_left_behind() {
        let dir = tempdir().unwrap();
        let path = dir.path().join(JOURNAL_FILE);
        fs::write(
            &path,
            "{\"hayson\":1,\"kind\":\"unit\",\"id\":\"abc\",\"started_at\":42,\"tables\":[\"users\"]}\n{\"t\":\"users\",\"s\":7,\"l\":\"{}\"}\n",
        )
        .unwrap();

        let read = read_journal(&path).expect("an older journal is still recoverable");
        assert_eq!(read.header.started_at, 42);
        assert_eq!(read.entries.len(), 1);
    }
}