kcode-kweb-db 0.1.0

A convergent signed-DAG store for Kweb nodes and objects
Documentation
use crate::{Error, Result, TransactionId, model::MAX_TRANSACTION_BYTES};
use sha2::{Digest, Sha256};
use std::{
    fs::{self, File, OpenOptions},
    io::{Read, Write},
    path::{Path, PathBuf},
};

const HEADER: &[u8] = b"KWLEDGER\0\x01\0\0";
const FRAME_PREFIX: usize = 8 + 32 + 32;

pub(crate) fn initialize(root: &Path) -> Result<()> {
    let path = ledger_path(root);
    match fs::symlink_metadata(&path) {
        Ok(metadata) => {
            if !metadata.file_type().is_file() || metadata.file_type().is_symlink() {
                return Err(Error::corrupt("ledger.kwl is not a regular file"));
            }
            let mut file = File::open(&path)?;
            let mut header = vec![0; HEADER.len()];
            file.read_exact(&mut header)
                .map_err(|_| Error::corrupt("ledger header is truncated"))?;
            if header != HEADER {
                return Err(Error::corrupt("unsupported ledger magic or version"));
            }
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            let mut file = OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(&path)?;
            file.write_all(HEADER)?;
            file.sync_all()?;
            sync_directory(root)?;
        }
        Err(error) => return Err(error.into()),
    }
    Ok(())
}

pub(crate) fn append(root: &Path, id: TransactionId, transaction: &[u8]) -> Result<()> {
    if transaction.len() > MAX_TRANSACTION_BYTES {
        return Err(Error::invalid_transaction(
            "ledger transaction exceeds the size limit",
        ));
    }
    let mut file = OpenOptions::new().append(true).open(ledger_path(root))?;
    file.write_all(&(transaction.len() as u64).to_be_bytes())?;
    file.write_all(&id.0)?;
    file.write_all(&Sha256::digest(transaction))?;
    file.write_all(transaction)?;
    file.sync_all()?;
    Ok(())
}

pub(crate) fn read_frames(root: &Path) -> Result<Vec<Vec<u8>>> {
    let path = ledger_path(root);
    let bytes = fs::read(&path)?;
    if bytes.len() < HEADER.len() || &bytes[..HEADER.len()] != HEADER {
        return Err(Error::corrupt("unsupported ledger magic or version"));
    }
    let mut offset = HEADER.len();
    let mut transactions = Vec::new();
    while offset < bytes.len() {
        let frame_start = offset;
        if bytes.len() - offset < 8 {
            truncate_tail(&path, frame_start)?;
            break;
        }
        let length = u64::from_be_bytes(bytes[offset..offset + 8].try_into().unwrap());
        offset += 8;
        let length = usize::try_from(length)
            .map_err(|_| Error::corrupt("ledger frame length does not fit memory"))?;
        if length > MAX_TRANSACTION_BYTES {
            return Err(Error::corrupt("ledger frame exceeds the size limit"));
        }
        let Some(frame_end) = offset
            .checked_add(FRAME_PREFIX - 8)
            .and_then(|value| value.checked_add(length))
        else {
            return Err(Error::corrupt("ledger frame length overflow"));
        };
        if frame_end > bytes.len() {
            truncate_tail(&path, frame_start)?;
            break;
        }
        let id = TransactionId(bytes[offset..offset + 32].try_into().unwrap());
        offset += 32;
        let checksum: [u8; 32] = bytes[offset..offset + 32].try_into().unwrap();
        offset += 32;
        let transaction = bytes[offset..offset + length].to_vec();
        offset += length;
        if TransactionId::for_signed_bytes(&transaction) != id {
            return Err(Error::corrupt("ledger transaction ID mismatch"));
        }
        if <[u8; 32]>::from(Sha256::digest(&transaction)) != checksum {
            return Err(Error::corrupt("ledger frame checksum mismatch"));
        }
        transactions.push(transaction);
    }
    Ok(transactions)
}

fn truncate_tail(path: &Path, length: usize) -> Result<()> {
    let file = OpenOptions::new().write(true).open(path)?;
    file.set_len(length as u64)?;
    file.sync_all()?;
    Ok(())
}

fn ledger_path(root: &Path) -> PathBuf {
    root.join("ledger.kwl")
}

pub(crate) fn sync_directory(path: &Path) -> Result<()> {
    File::open(path)?.sync_all()?;
    Ok(())
}

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

    #[test]
    fn partial_final_prefix_is_truncated() {
        let root = tempfile::tempdir().unwrap();
        initialize(root.path()).unwrap();
        let path = ledger_path(root.path());
        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
        file.write_all(&[0, 0, 0]).unwrap();
        file.sync_all().unwrap();
        assert!(read_frames(root.path()).unwrap().is_empty());
        assert_eq!(fs::metadata(path).unwrap().len(), HEADER.len() as u64);
    }
}