haematite 0.6.1

//! HBR1 record codec and ref-file naming for the branch ref store (§2.1,
//! §2.2). Split out of `refstore.rs` via `#[path]` so that file stays within
//! the branch module's 500-line cap; the record layout is documented on the
//! parent module.
//!
//! The pure decode + filename validation here are shared BY CONSTRUCTION with
//! the vacuum's read-only metadata reader (STORAGE-VACUUM.md §7 ⟨r4, M4⟩):
//! one parser, two callers, so the vacuum cannot accept or reject a record
//! the production store wouldn't. Creation and temp-debris cleanup stay in
//! the mutating [`super::BranchRefStore`] only.

use std::path::Path;

use super::{BranchRefError, BranchRefRecord, BranchShardRef};
use crate::branch::persist::{Reader, push_bytes, push_u64};
use crate::branch::policy::BranchKind;

const REF_MAGIC: &[u8; 4] = b"HBR1";
const KIND_EXTENSION_MAGIC: &[u8; 4] = b"HBK1";
const KIND_WORK: u64 = 0;
const KIND_NAMESPACE: u64 = 1;
pub const REF_EXTENSION: &str = "ref";

/// `<hex(blake3(name)[0..16])>.ref` (§2.1): the file name is derived from the
/// name hash so arbitrary byte-string names are filesystem-safe; the full name
/// lives inside the record.
pub fn ref_file_name(name: &str) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let digest = blake3::hash(name.as_bytes());
    let mut file_name = String::with_capacity(36);
    for byte in &digest.as_bytes()[..16] {
        file_name.push(char::from(HEX[usize::from(byte >> 4)]));
        file_name.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    file_name.push('.');
    file_name.push_str(REF_EXTENSION);
    file_name
}

/// Rewraps a decode failure so the error names the offending file (§2.2).
pub fn corrupt_in_file(path: &Path, error: &BranchRefError) -> BranchRefError {
    BranchRefError::Corrupt(format!("{}: {error}", path.display()))
}

pub fn encode_record(record: &BranchRefRecord) -> Vec<u8> {
    let mut bytes = Vec::new();
    bytes.extend_from_slice(REF_MAGIC);
    push_bytes(&mut bytes, record.name.as_bytes());
    push_u64(&mut bytes, record.created);
    push_u64(&mut bytes, record.seq);
    push_u64(&mut bytes, record.timestamp);
    push_u64(&mut bytes, record.shards.len() as u64);
    for shard in &record.shards {
        push_u64(&mut bytes, shard.shard_id as u64);
        bytes.extend_from_slice(shard.fork_anchor.as_bytes());
        bytes.extend_from_slice(shard.head.as_bytes());
    }
    push_u64(&mut bytes, record.parents.len() as u64);
    for parent in &record.parents {
        bytes.extend_from_slice(parent.as_bytes());
    }
    bytes.extend_from_slice(KIND_EXTENSION_MAGIC);
    push_u64(
        &mut bytes,
        match record.kind {
            BranchKind::Work => KIND_WORK,
            BranchKind::Namespace => KIND_NAMESPACE,
        },
    );
    match &record.namespace_lineage {
        Some(lineage) => {
            push_u64(&mut bytes, 1);
            push_bytes(&mut bytes, lineage.as_bytes());
        }
        None => push_u64(&mut bytes, 0),
    }
    bytes
}

pub fn decode_record(bytes: &[u8]) -> Result<BranchRefRecord, BranchRefError> {
    let mut reader = Reader::new(bytes);
    reader.expect_magic(*REF_MAGIC)?;
    let name = String::from_utf8(reader.read_bytes()?)
        .map_err(|_error| BranchRefError::Corrupt("branch name is not valid UTF-8".into()))?;
    let created = reader.read_u64()?;
    let seq = reader.read_u64()?;
    let timestamp = reader.read_u64()?;
    let shard_count = reader.read_usize()?;
    // Capacity hints are capped by the remaining input (PERSIST-003
    // discipline): a corrupt count fails as Corrupt when the bytes run
    // out, never aborts at the allocator. Each shard ref is 72 bytes.
    let mut shards: Vec<BranchShardRef> =
        Vec::with_capacity(shard_count.min(reader.remaining() / 72));
    for _ in 0..shard_count {
        let shard_id = reader.read_usize()?;
        if shards
            .last()
            .is_some_and(|previous| previous.shard_id >= shard_id)
        {
            return Err(BranchRefError::Corrupt(
                "shard ids are not strictly ascending".into(),
            ));
        }
        let fork_anchor = reader.read_hash()?;
        let head = reader.read_hash()?;
        shards.push(BranchShardRef {
            shard_id,
            fork_anchor,
            head,
        });
    }
    let parent_count = reader.read_usize()?;
    let mut parents = Vec::with_capacity(parent_count.min(reader.remaining() / 32));
    for _ in 0..parent_count {
        parents.push(reader.read_hash()?);
    }
    // The extension is deliberately tail-appended: a pre-marker HBR1 record
    // ends exactly here, so missing metadata has one unambiguous permissive
    // interpretation and all original field offsets remain unchanged.
    let (kind, namespace_lineage) = if reader.remaining() == 0 {
        (BranchKind::Work, None)
    } else {
        reader.expect_magic(*KIND_EXTENSION_MAGIC)?;
        let kind = match reader.read_u64()? {
            KIND_WORK => BranchKind::Work,
            KIND_NAMESPACE => BranchKind::Namespace,
            _ => return Err(BranchRefError::Corrupt("unknown branch kind".into())),
        };
        let lineage_tag = reader.read_u64()?;
        if kind == BranchKind::Namespace && lineage_tag == 1 {
            return Err(BranchRefError::Corrupt(
                "HBK1 record has kind=Namespace with lineage_tag=1".into(),
            ));
        }
        let namespace_lineage = match lineage_tag {
            0 => None,
            1 => Some(String::from_utf8(reader.read_bytes()?).map_err(|_error| {
                BranchRefError::Corrupt("namespace lineage is not valid UTF-8".into())
            })?),
            _ => {
                return Err(BranchRefError::Corrupt(
                    "unknown namespace lineage tag".into(),
                ));
            }
        };
        (kind, namespace_lineage)
    };
    reader.finish()?;
    Ok(BranchRefRecord {
        name,
        created,
        kind,
        namespace_lineage,
        seq,
        timestamp,
        shards,
        parents,
    })
}