hayson-engine 0.27.0

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

use crate::errors::HaysonCorruptLogError;

use super::entry::{parse_buffer_entry, BufferEntry};

#[derive(Debug)]
pub enum BufferParseError {
    Io(io::Error),
    Corrupt(HaysonCorruptLogError),
}

impl From<io::Error> for BufferParseError {
    fn from(value: io::Error) -> Self {
        BufferParseError::Io(value)
    }
}

impl From<HaysonCorruptLogError> for BufferParseError {
    fn from(value: HaysonCorruptLogError) -> Self {
        BufferParseError::Corrupt(value)
    }
}

/// Reads every `*.jsonl` file directly inside `folder` (not recursive), in filename order, and
/// every line inside each file in file order — this is the enumeration order later fed to `merge`
/// before `buffered_at` sorting takes over.
pub fn parse_buffer_folder(folder: &Path) -> Result<Vec<BufferEntry>, BufferParseError> {
    let mut files: Vec<_> = fs::read_dir(folder)?
        .filter_map(|entry| entry.ok())
        .map(|entry| entry.path())
        .filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("jsonl"))
        .collect();
    files.sort();

    let mut entries = Vec::new();
    for file in files {
        let contents = fs::read_to_string(&file)?;
        let path_label = file.to_string_lossy().into_owned();
        for (line_no, line) in contents.lines().enumerate() {
            if line.is_empty() {
                continue;
            }
            entries.push(parse_buffer_entry(line, &path_label, line_no)?);
        }
    }
    Ok(entries)
}

/// Reads every folder in `folders` via `parse_buffer_folder` and concatenates the results in the
/// order the folders were given — `merge` is the one that imposes the real `buffered_at` order.
pub fn parse_buffer_folders(folders: &[impl AsRef<Path>]) -> Result<Vec<BufferEntry>, BufferParseError> {
    let mut entries = Vec::new();
    for folder in folders {
        entries.extend(parse_buffer_folder(folder.as_ref())?);
    }
    Ok(entries)
}

#[cfg(test)]
mod tests {
    use serde_json::{Map, Value};
    use tempfile::tempdir;

    use super::*;
    use crate::buffer::entry::{serialize_buffer_entry, BufferEntry, BufferOp};

    fn write_folder(dir: &Path, file_name: &str, entries: &[BufferEntry]) {
        let lines: Vec<String> = entries.iter().map(serialize_buffer_entry).collect();
        fs::write(dir.join(file_name), lines.join("\n") + "\n").unwrap();
    }

    fn put(table: &str, key: &str, source: &str, buffered_at: i64, name: &str) -> BufferEntry {
        let mut data = Map::new();
        data.insert("name".into(), Value::String(name.into()));
        BufferEntry {
            table: table.into(),
            key: key.into(),
            op: BufferOp::Put,
            data: Some(data),
            buffered_at,
            source: source.into(),
        }
    }

    #[test]
    fn reads_two_folders_and_keeps_each_entrys_own_source() {
        let dir_a = tempdir().unwrap();
        let dir_b = tempdir().unwrap();
        write_folder(
            dir_a.path(),
            "buf.jsonl",
            &[put("users", "u1", "branch-a", 100, "Ania")],
        );
        write_folder(
            dir_b.path(),
            "buf.jsonl",
            &[put("users", "u2", "branch-b", 200, "Piotr")],
        );

        let entries = parse_buffer_folders(&[dir_a.path(), dir_b.path()]).unwrap();

        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].source, "branch-a");
        assert_eq!(entries[0].key, "u1");
        assert_eq!(entries[1].source, "branch-b");
        assert_eq!(entries[1].key, "u2");
    }

    #[test]
    fn ignores_files_without_the_jsonl_extension() {
        let dir = tempdir().unwrap();
        write_folder(dir.path(), "buf.jsonl", &[put("users", "u1", "branch-a", 100, "Ania")]);
        fs::write(dir.path().join("readme.txt"), "not a buffer file").unwrap();

        let entries = parse_buffer_folder(dir.path()).unwrap();
        assert_eq!(entries.len(), 1);
    }
}