kcode-session-control-journal 0.1.0

//! Durable mechanics for one checksummed `.session-control` record file.
//!
//! Domain interpretation, compaction selection, and coordination belong to
//! Session History or another caller.

use std::{
    ffi::OsString,
    fs::{File, OpenOptions},
    io::{Read, Write},
    path::{Path, PathBuf},
};

use anyhow::{Context as _, ensure};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use uuid::Uuid;

/// One uninterpreted session-control record.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Record {
    pub kind: String,
    pub recorded_at: String,
    pub value: Value,
}

/// An opened checksummed record file.
///
/// The caller owns all locking and domain policy.
pub struct Journal {
    path: PathBuf,
    records: Vec<Record>,
}

impl Journal {
    /// Creates and synchronizes a new empty journal.
    pub fn create(path: PathBuf) -> anyhow::Result<Self> {
        let file = OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(&path)
            .with_context(|| format!("creating {}", path.display()))?;
        file.sync_all()
            .with_context(|| format!("syncing {}", path.display()))?;
        sync_directory(parent(&path))?;
        Ok(Self {
            path,
            records: Vec::new(),
        })
    }

    /// Opens an existing journal, or returns `None` when it is absent.
    ///
    /// An incomplete final line is truncated and synchronized. Corruption in
    /// any complete line is rejected.
    pub fn open(path: PathBuf) -> anyhow::Result<Option<Self>> {
        let mut file = match OpenOptions::new().read(true).write(true).open(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => {
                return Err(error).with_context(|| format!("opening {}", path.display()));
            }
        };
        let mut bytes = Vec::new();
        file.read_to_end(&mut bytes)
            .with_context(|| format!("reading {}", path.display()))?;

        let mut records = Vec::new();
        let mut cursor = 0_usize;
        while cursor < bytes.len() {
            let Some(relative_end) = bytes[cursor..].iter().position(|byte| *byte == b'\n') else {
                let repaired_length =
                    u64::try_from(cursor).context("journal length does not fit in u64")?;
                file.set_len(repaired_length)
                    .with_context(|| format!("repairing {}", path.display()))?;
                file.sync_all()
                    .with_context(|| format!("syncing repaired {}", path.display()))?;
                break;
            };
            let end = cursor + relative_end;
            records.push(parse_record(&bytes[cursor..end])?);
            cursor = end + 1;
        }

        Ok(Some(Self { path, records }))
    }

    /// Returns validated records in file order.
    pub fn records(&self) -> &[Record] {
        &self.records
    }

    /// Appends and synchronizes one record before updating memory.
    pub fn append(
        &mut self,
        kind: impl Into<String>,
        recorded_at: impl Into<String>,
        value: Value,
    ) -> anyhow::Result<()> {
        let record = Record {
            kind: kind.into(),
            recorded_at: recorded_at.into(),
            value,
        };
        let encoded = encode_record(&record)?;
        let mut file = OpenOptions::new()
            .append(true)
            .open(&self.path)
            .with_context(|| format!("opening {} for append", self.path.display()))?;
        file.write_all(&encoded)
            .with_context(|| format!("appending {}", self.path.display()))?;
        file.sync_all()
            .with_context(|| format!("syncing {}", self.path.display()))?;
        self.records.push(record);
        Ok(())
    }

    /// Atomically replaces the file with caller-selected records.
    pub fn replace(&mut self, records: impl IntoIterator<Item = Record>) -> anyhow::Result<()> {
        let records = records.into_iter().collect::<Vec<_>>();
        let permissions = std::fs::metadata(&self.path)
            .with_context(|| format!("reading permissions for {}", self.path.display()))?
            .permissions();
        let temporary = temporary_path(&self.path)?;
        let mut temporary_created = false;
        let result = (|| -> anyhow::Result<()> {
            let mut file = OpenOptions::new()
                .create_new(true)
                .write(true)
                .open(&temporary)
                .with_context(|| format!("creating {}", temporary.display()))?;
            temporary_created = true;
            file.set_permissions(permissions)
                .with_context(|| format!("setting permissions on {}", temporary.display()))?;
            for record in &records {
                file.write_all(&encode_record(record)?)
                    .with_context(|| format!("writing {}", temporary.display()))?;
            }
            file.sync_all()
                .with_context(|| format!("syncing {}", temporary.display()))?;
            drop(file);
            std::fs::rename(&temporary, &self.path)
                .with_context(|| format!("replacing {}", self.path.display()))?;
            sync_directory(parent(&self.path))?;
            Ok(())
        })();
        if result.is_err() && temporary_created {
            let _ = std::fs::remove_file(&temporary);
        }
        result?;
        self.records = records;
        Ok(())
    }
}

fn parent(path: &Path) -> &Path {
    path.parent()
        .filter(|parent| !parent.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."))
}

fn temporary_path(path: &Path) -> anyhow::Result<PathBuf> {
    let file_name = path.file_name().context("journal path has no file name")?;
    let mut temporary_name = OsString::from(".");
    temporary_name.push(file_name);
    temporary_name.push(format!(".replace-{}.tmp", Uuid::new_v4()));
    Ok(parent(path).join(temporary_name))
}

fn sync_directory(path: &Path) -> anyhow::Result<()> {
    File::open(path)
        .with_context(|| format!("opening directory {} for sync", path.display()))?
        .sync_all()
        .with_context(|| format!("syncing directory {}", path.display()))
}

fn parse_record(line: &[u8]) -> anyhow::Result<Record> {
    let separator = line
        .iter()
        .position(|byte| *byte == b' ')
        .context("session-control record has no checksum separator")?;
    let expected =
        std::str::from_utf8(&line[..separator]).context("session-control checksum is not UTF-8")?;
    let payload = &line[separator + 1..];
    ensure!(
        hex_sha256(payload) == expected,
        "session-control record checksum mismatch"
    );
    let payload =
        std::str::from_utf8(payload).context("session-control JSON payload is not UTF-8")?;
    serde_json::from_str(payload).context("decoding session-control record")
}

fn encode_record(record: &Record) -> anyhow::Result<Vec<u8>> {
    let payload = serde_json::to_vec(record).context("encoding session-control record")?;
    let checksum = hex_sha256(&payload);
    let mut encoded = Vec::with_capacity(checksum.len() + payload.len() + 2);
    encoded.extend_from_slice(checksum.as_bytes());
    encoded.push(b' ');
    encoded.extend_from_slice(&payload);
    encoded.push(b'\n');
    Ok(encoded)
}

fn hex_sha256(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(64);
    for byte in Sha256::digest(bytes) {
        output.push(char::from(HEX[usize::from(byte >> 4)]));
        output.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    output
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::io::Write as _;

    use serde_json::json;

    use super::*;

    fn path(label: &str) -> PathBuf {
        let directory = std::env::temp_dir().join(format!(
            "kcode-session-control-journal-{label}-{}-{}",
            std::process::id(),
            Uuid::new_v4()
        ));
        fs::create_dir(&directory).unwrap();
        directory.join("test.session-control")
    }

    fn remove(path: &Path) {
        fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    fn record(kind: &str, number: u64) -> Record {
        Record {
            kind: kind.into(),
            recorded_at: format!("2026-08-01T00:00:0{number}Z"),
            value: json!({"number":number}),
        }
    }

    #[test]
    fn append_encoding_and_reopen_are_exact() {
        let path = path("append");
        let mut journal = Journal::create(path.clone()).unwrap();
        journal
            .append(
                "session_lifecycle",
                "2026-08-01T00:00:00Z",
                json!({"version":1}),
            )
            .unwrap();

        let payload = br#"{"kind":"session_lifecycle","recordedAt":"2026-08-01T00:00:00Z","value":{"version":1}}"#;
        let checksum = hex_sha256(payload);
        let mut expected = Vec::new();
        expected.extend_from_slice(checksum.as_bytes());
        expected.push(b' ');
        expected.extend_from_slice(payload);
        expected.push(b'\n');
        assert_eq!(fs::read(&path).unwrap(), expected);
        drop(journal);

        let reopened = Journal::open(path.clone()).unwrap().unwrap();
        assert_eq!(reopened.records().len(), 1);
        assert_eq!(reopened.records()[0].kind, "session_lifecycle");
        assert_eq!(reopened.records()[0].recorded_at, "2026-08-01T00:00:00Z");
        assert_eq!(reopened.records()[0].value, json!({"version":1}));
        remove(&path);
    }

    #[test]
    fn opening_repairs_only_an_incomplete_final_tail() {
        let path = path("tail");
        let mut journal = Journal::create(path.clone()).unwrap();
        journal.append("first", "time", json!(1)).unwrap();
        let complete = fs::read(&path).unwrap();
        drop(journal);

        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
        file.write_all(b"interrupted record").unwrap();
        file.sync_all().unwrap();
        drop(file);

        let reopened = Journal::open(path.clone()).unwrap().unwrap();
        assert_eq!(reopened.records().len(), 1);
        assert_eq!(reopened.records()[0].kind, "first");
        assert_eq!(fs::read(&path).unwrap(), complete);
        remove(&path);
    }

    #[test]
    fn complete_checksum_corruption_is_rejected() {
        let path = path("corruption");
        let mut journal = Journal::create(path.clone()).unwrap();
        journal.append("first", "time", json!(1)).unwrap();
        drop(journal);

        let mut bytes = fs::read(&path).unwrap();
        bytes[0] = if bytes[0] == b'0' { b'1' } else { b'0' };
        let mut file = OpenOptions::new()
            .write(true)
            .truncate(true)
            .open(&path)
            .unwrap();
        file.write_all(&bytes).unwrap();
        file.sync_all().unwrap();
        drop(file);

        let error = match Journal::open(path.clone()) {
            Err(error) => error,
            Ok(_) => panic!("corrupt complete record was accepted"),
        };
        assert!(error.to_string().contains("checksum mismatch"));
        remove(&path);
    }

    #[test]
    fn replace_preserves_order_updates_memory_and_reopens() {
        let path = path("replace");
        let mut journal = Journal::create(path.clone()).unwrap();
        journal.append("old", "time", json!(0)).unwrap();

        journal
            .replace([record("second", 2), record("first", 1)])
            .unwrap();
        assert_eq!(journal.records().len(), 2);
        assert_eq!(journal.records()[0].kind, "second");
        assert_eq!(journal.records()[1].kind, "first");
        drop(journal);

        let reopened = Journal::open(path.clone()).unwrap().unwrap();
        assert_eq!(reopened.records().len(), 2);
        assert_eq!(reopened.records()[0].kind, "second");
        assert_eq!(reopened.records()[0].value, json!({"number":2}));
        assert_eq!(reopened.records()[1].kind, "first");
        assert_eq!(reopened.records()[1].value, json!({"number":1}));
        remove(&path);
    }

    #[test]
    fn create_rejects_an_existing_path() {
        let path = path("collision");
        let journal = Journal::create(path.clone()).unwrap();
        let error = match Journal::create(path.clone()) {
            Err(error) => error,
            Ok(_) => panic!("create replaced an existing journal"),
        };
        assert_eq!(
            error
                .downcast_ref::<std::io::Error>()
                .map(std::io::Error::kind),
            Some(std::io::ErrorKind::AlreadyExists)
        );
        drop(journal);
        remove(&path);
    }

    #[test]
    fn create_and_replace_keep_the_target_in_its_parent() {
        let path = path("durability");
        assert!(Journal::open(path.clone()).unwrap().is_none());
        let mut journal = Journal::create(path.clone()).unwrap();

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt as _;
            fs::set_permissions(&path, fs::Permissions::from_mode(0o640)).unwrap();
        }

        journal.replace([record("retained", 1)]).unwrap();
        assert!(path.is_file());
        let entries = fs::read_dir(path.parent().unwrap())
            .unwrap()
            .map(|entry| entry.unwrap().path())
            .collect::<Vec<_>>();
        assert_eq!(entries, vec![path.clone()]);

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt as _;
            assert_eq!(
                fs::metadata(&path).unwrap().permissions().mode() & 0o777,
                0o640
            );
        }

        remove(&path);
    }
}