kcode-kweb-db 0.1.0

A convergent signed-DAG store for Kweb nodes and objects
Documentation
use crate::{Error, Node, Result, TransactionId, ledger::sync_directory, projector::Projection};
use std::{
    fs::{self, OpenOptions},
    io::Write,
    path::Path,
};

const HEADER: &[u8] = b"KWNODE\0\x01\0\0";

pub(crate) fn rebuild(root: &Path, projection: &Projection) -> Result<()> {
    let path = root.join("nodes");
    match fs::symlink_metadata(&path) {
        Ok(metadata) => {
            if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
                return Err(Error::corrupt("nodes is not a real directory"));
            }
            fs::remove_dir_all(&path)?;
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(error.into()),
    }
    fs::create_dir(&path)?;
    sync_directory(root)?;
    for (id, projected) in &projection.nodes {
        write_node(&path, &projected.node, projected.visible_transaction)?;
        let text = id.to_string();
        sync_directory(&path.join(&text[..2]))?;
    }
    sync_directory(&path)?;
    Ok(())
}

pub(crate) fn write_projection(root: &Path, projection: &Projection) -> Result<()> {
    rebuild(root, projection)
}

fn write_node(base: &Path, node: &Node, visible_transaction: TransactionId) -> Result<()> {
    let id = node.id.to_string();
    let shard = base.join(&id[..2]);
    if !shard.exists() {
        fs::create_dir(&shard)?;
    }
    let final_path = shard.join(format!("{}.kwn", &id[2..]));
    let temporary_path = shard.join(format!(".tmp-{}", hex::encode(rand::random::<[u8; 8]>())));
    let bytes = encode_node(node, visible_transaction)?;
    let mut file = OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&temporary_path)?;
    file.write_all(&bytes)?;
    file.sync_all()?;
    drop(file);
    fs::rename(&temporary_path, &final_path)?;
    Ok(())
}

fn encode_node(node: &Node, visible_transaction: TransactionId) -> Result<Vec<u8>> {
    let mut bytes = Vec::new();
    bytes.extend_from_slice(HEADER);
    bytes.extend_from_slice(&node.id.0);
    bytes.extend_from_slice(&visible_transaction.0);
    write_string(&mut bytes, &node.data.short_name)?;
    write_string(&mut bytes, &node.data.short_description)?;
    write_string(&mut bytes, &node.data.long_description)?;
    write_string(&mut bytes, &node.last_author)?;
    bytes.extend_from_slice(&node.committed_at.timestamp().to_be_bytes());
    bytes.extend_from_slice(&node.committed_at.timestamp_subsec_nanos().to_be_bytes());
    bytes.extend_from_slice(&(node.connections.len() as u32).to_be_bytes());
    for connection in &node.connections {
        bytes.extend_from_slice(&connection.0);
    }
    Ok(bytes)
}

fn write_string(output: &mut Vec<u8>, value: &str) -> Result<()> {
    let length = u32::try_from(value.len())
        .map_err(|_| Error::corrupt("projected node string exceeds u32"))?;
    output.extend_from_slice(&length.to_be_bytes());
    output.extend_from_slice(value.as_bytes());
    Ok(())
}