loonfs-api 0.2.0

//! Every identifier newtype in the workspace, generated by the
//! `string_id!` and `numeric_id!` macros so all ids share one validated
//! surface.

use serde::{Deserialize, Serialize};
use std::fmt;
use thiserror::Error;
use uuid::Uuid;

const SERVER_GENERATED_ID_BODY_LEN: usize = 32;

// ---------------------------------------------------------------------------
// Validation errors
// ---------------------------------------------------------------------------

macro_rules! validation_error {
    ($name:ident, $message:literal) => {
        /// Describes why supplied text does not satisfy this identifier's validation contract.
        #[derive(Debug, Clone, PartialEq, Eq, Error)]
        #[error($message)]
        pub struct $name {
            value: String,
            reason: String,
        }

        impl $name {
            /// Returns the rejected input, or an empty string when echoing it would be unsafe.
            pub fn value(&self) -> &str {
                &self.value
            }

            /// Returns the specific grammar rule the rejected input violated.
            pub fn reason(&self) -> &str {
                &self.reason
            }
        }
    };
}

validation_error!(
    NamespaceIdValidationError,
    "invalid namespace_id {value:?}: {reason}"
);
validation_error!(
    CommitIdValidationError,
    "invalid commit_id {value:?}: {reason}"
);
validation_error!(
    GeneratedIdValidationError,
    "invalid generated id {value:?}: {reason}"
);
validation_error!(
    NameKeyValidationError,
    "invalid name_key {value:?}: {reason}"
);

// ---------------------------------------------------------------------------
// Id macros
// ---------------------------------------------------------------------------

/// Defines a validated string-id newtype.
///
/// Every string id gets the same surface: `parse` (the only fallible
/// constructor), `as_str`, `TryFrom<&str>`/`TryFrom<String>`/`FromStr`
/// (all delegating to `parse`), `AsRef<str>`, `Borrow<str>`, `Display`
/// (the plain inner string), and serde as a plain string with validation
/// on deserialize.
///
/// Two forms:
/// - `string_id!(Name, error = ErrType, validate = validator)` uses a custom
///   `fn(&str) -> Result<(), ErrType>` validator.
/// - `string_id!(Name, prefix = "xyz")` validates the project-standard
///   server-generated shape `xyz_<32 lowercase hex>` and adds a
///   `generate()` constructor.
///
/// Type-specific constructors that the macro cannot express (for example
/// `CommitId::generate` or `NameKey::for_display_name`) live in a separate
/// `impl` block next to the invocation.
macro_rules! string_id {
    (
        $(#[$meta:meta])*
        $name:ident,
        error = $error:ty,
        validate = $validate:expr
    ) => {
        $(#[$meta])*
        #[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
        #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
        #[cfg_attr(feature = "openapi", schema(value_type = String))]
        pub struct $name(String);

        impl $name {
            /// Parses and validates the id from its serialized form.
            pub fn parse(value: impl AsRef<str>) -> Result<Self, $error> {
                let value = value.as_ref();
                ($validate)(value)?;
                Ok(Self(value.to_owned()))
            }

            /// Returns the serialized id.
            pub fn as_str(&self) -> &str {
                &self.0
            }
        }

        impl TryFrom<&str> for $name {
            type Error = $error;

            fn try_from(value: &str) -> Result<Self, Self::Error> {
                Self::parse(value)
            }
        }

        impl TryFrom<String> for $name {
            type Error = $error;

            fn try_from(value: String) -> Result<Self, Self::Error> {
                Self::parse(value)
            }
        }

        impl std::str::FromStr for $name {
            type Err = $error;

            fn from_str(value: &str) -> Result<Self, Self::Err> {
                Self::parse(value)
            }
        }

        impl AsRef<str> for $name {
            fn as_ref(&self) -> &str {
                self.as_str()
            }
        }

        impl std::borrow::Borrow<str> for $name {
            fn borrow(&self) -> &str {
                self.as_str()
            }
        }

        impl std::fmt::Display for $name {
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                f.write_str(&self.0)
            }
        }

        impl serde::Serialize for $name {
            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
            where
                S: serde::Serializer,
            {
                serializer.serialize_str(&self.0)
            }
        }

        impl<'de> serde::Deserialize<'de> for $name {
            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
            where
                D: serde::Deserializer<'de>,
            {
                let value = <String as serde::Deserialize>::deserialize(deserializer)?;
                Self::parse(value).map_err(serde::de::Error::custom)
            }
        }
    };
    (
        $(#[$meta:meta])*
        $name:ident,
        prefix = $prefix:literal
    ) => {
        string_id! {
            $(#[$meta])*
            $name,
            error = GeneratedIdValidationError,
            validate = |value: &str| validate_generated_id($prefix, value)
        }

        impl $name {
            /// Generates a valid random id.
            pub fn generate() -> Self {
                Self(generated_id($prefix))
            }
        }
    };
}

/// Defines a numeric (`u64`) id newtype.
///
/// Every numeric id gets `Copy`, ordering and hashing derives, `From<u64>`,
/// `Display` as the plain inner number, and serde as a plain number. The
/// inner field stays public: numeric ids are constructed positionally
/// (`InodeId(7)`) and read via `.0`.
macro_rules! numeric_id {
    (
        $(#[$meta:meta])*
        $name:ident
    ) => {
        $(#[$meta])*
        #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
        #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
        #[cfg_attr(feature = "openapi", schema(value_type = u64))]
        pub struct $name(pub u64);

        impl From<u64> for $name {
            fn from(value: u64) -> Self {
                Self(value)
            }
        }

        impl fmt::Display for $name {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                write!(f, "{}", self.0)
            }
        }
    };
}

pub(crate) use string_id;

// ---------------------------------------------------------------------------
// Shared validators and generators
// ---------------------------------------------------------------------------

/// Generates a project-standard opaque durable identifier.
///
/// Generated server-side IDs use an underscore prefix plus a 32-character
/// lowercase UUID-simple body, such as `cs_<32hex>` or `chk_<32hex>`.
///
/// Ids in the id inventory generate through their newtype `generate()`
/// constructors; this helper stays public for free-form generated labels
/// (for example a server's per-request id) that have no validated id type.
pub fn generated_id(prefix: &'static str) -> String {
    format!("{prefix}_{}", Uuid::new_v4().simple())
}

fn generated_position_suffix() -> String {
    let suffix = Uuid::new_v4().simple().to_string();
    suffix[..16].to_owned()
}

/// Draws 128 fresh random bits.
///
/// [`generated_id`] spends six of its bits on UUID version and variant tags.
/// Content ids shard on their leading characters and must be uniform there,
/// so they draw from the system generator directly instead.
fn random_128() -> [u8; 16] {
    let mut bytes = [0_u8; 16];
    getrandom::fill(&mut bytes).expect("the system random generator must be available");
    bytes
}

fn validate_generated_id(
    prefix: &'static str,
    value: &str,
) -> Result<(), GeneratedIdValidationError> {
    let expected_prefix = format!("{prefix}_");
    let Some(body) = value.strip_prefix(&expected_prefix) else {
        return Err(generated_id_error(
            value,
            format!("must start with `{expected_prefix}`"),
        ));
    };
    if body.len() != SERVER_GENERATED_ID_BODY_LEN {
        return Err(generated_id_error(
            value,
            format!("body must be {SERVER_GENERATED_ID_BODY_LEN} lowercase hex characters"),
        ));
    }
    if !body.bytes().all(is_lower_hex_byte) {
        return Err(generated_id_error(
            value,
            "body must contain only lowercase hex characters".to_owned(),
        ));
    }
    Ok(())
}

fn validate_namespace_id(value: &str) -> Result<(), NamespaceIdValidationError> {
    validate_id_grammar(value).map_err(|reason| namespace_id_error(value, reason))?;
    // System tooling (for example the object-store doctor probes) writes
    // under namespace slots that must never collide with user namespaces.
    if value.starts_with("loonfs-") {
        return Err(namespace_id_error(
            value,
            "the `loonfs-` prefix is reserved for LoonFS system namespaces",
        ));
    }
    Ok(())
}

fn validate_commit_id(value: &str) -> Result<(), CommitIdValidationError> {
    validate_id_grammar(value).map_err(|reason| commit_id_error(value, reason))
}

/// Maximum name-key length in UTF-8 bytes. Keys are derived from display
/// names capped at [`crate::path::MAX_DISPLAY_NAME_BYTES`]; case folding
/// expands at most threefold in bytes, so 768 admits every key derivable
/// from a valid name while bounding row keys, filter keys, and cursors.
pub const MAX_NAME_KEY_BYTES: usize = 768;
/// Maximum validated namespace and commit id length in UTF-8 bytes.
pub const MAX_ID_BYTES: usize = 128;

fn validate_name_key(value: &str) -> Result<(), NameKeyValidationError> {
    if value.is_empty() {
        return Err(name_key_error(value, "must not be empty"));
    }
    if value.contains('/') {
        return Err(name_key_error(value, "must not contain `/`"));
    }
    if matches!(value, "." | "..") {
        return Err(name_key_error(value, "must not be `.` or `..`"));
    }
    if value.chars().any(|character| character.is_control()) {
        return Err(name_key_error(value, "must not contain control characters"));
    }
    if value.len() > MAX_NAME_KEY_BYTES {
        // An oversized or hostile name must not ride along in error payloads that serialize onto the wire.
        return Err(name_key_error(
            "",
            format!("exceeds the maximum name key length of {MAX_NAME_KEY_BYTES} bytes"),
        ));
    }
    Ok(())
}

fn validate_position_suffix_id(
    value: &str,
    position_label: (&str, &str),
) -> Result<(), GeneratedIdValidationError> {
    let Some((position, suffix)) = value.split_once('-') else {
        return Err(generated_id_error(
            value,
            format!(
                "must be `<20 digit {}>-<16 lowercase hex>`",
                position_label.0
            ),
        ));
    };
    if position.len() != 20 || !position.bytes().all(|byte| byte.is_ascii_digit()) {
        return Err(generated_id_error(
            value,
            format!("{} prefix must be 20 decimal digits", position_label.1),
        ));
    }
    if suffix.len() != 16 || !suffix.bytes().all(is_lower_hex_byte) {
        return Err(generated_id_error(
            value,
            "suffix must be 16 lowercase hex characters".to_owned(),
        ));
    }
    Ok(())
}

fn is_lower_hex_byte(byte: u8) -> bool {
    byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)
}

fn validate_id_grammar(value: &str) -> Result<(), String> {
    if value.is_empty() {
        return Err("must not be empty".to_owned());
    }
    if value.len() > MAX_ID_BYTES {
        return Err(format!("must be {MAX_ID_BYTES} bytes or fewer"));
    }
    if value.trim() != value {
        return Err("must not have leading or trailing whitespace".to_owned());
    }
    if matches!(value, "." | "..") {
        return Err("must not be `.` or `..`".to_owned());
    }

    let mut chars = value.chars();
    let first = chars
        .next()
        .expect("empty id returned before char validation");
    if !first.is_ascii_lowercase() && !first.is_ascii_digit() {
        return Err("must start with a lowercase ASCII letter or digit".to_owned());
    }
    if !chars.all(is_allowed_id_tail_char) {
        return Err(
            "must contain only lowercase ASCII letters, digits, `.`, `_`, or `-`".to_owned(),
        );
    }

    Ok(())
}

fn is_allowed_id_tail_char(ch: char) -> bool {
    ch.is_ascii_lowercase() || ch.is_ascii_digit() || matches!(ch, '.' | '_' | '-')
}

fn namespace_id_error(value: &str, reason: impl Into<String>) -> NamespaceIdValidationError {
    NamespaceIdValidationError {
        value: value.to_owned(),
        reason: reason.into(),
    }
}

fn commit_id_error(value: &str, reason: impl Into<String>) -> CommitIdValidationError {
    CommitIdValidationError {
        value: value.to_owned(),
        reason: reason.into(),
    }
}

fn generated_id_error(value: &str, reason: String) -> GeneratedIdValidationError {
    GeneratedIdValidationError {
        value: value.to_owned(),
        reason,
    }
}

fn name_key_error(value: &str, reason: impl Into<String>) -> NameKeyValidationError {
    NameKeyValidationError {
        value: value.to_owned(),
        reason: reason.into(),
    }
}

// ---------------------------------------------------------------------------
// String ids
// ---------------------------------------------------------------------------

string_id! {
    /// Durable id for one namespace.
    ///
    /// A namespace is one filesystem history. This id is not a display name and
    /// should not be reused after destruction.
    NamespaceId,
    error = NamespaceIdValidationError,
    validate = validate_namespace_id
}

string_id! {
    /// Durable id for an immutable content store.
    ///
    /// Content stores own file bytes. Namespaces point at content stores.
    ContentStoreId,
    prefix = "cs"
}

string_id! {
    /// Client-supplied idempotency key for one logical commit.
    ///
    /// Reuse the same `CommitId` when retrying the same request.
    CommitId,
    error = CommitIdValidationError,
    validate = validate_commit_id
}

impl CommitId {
    /// Generates a valid random commit id.
    pub fn generate() -> Self {
        Self(generated_id("c"))
    }
}

string_id! {
    /// Durable checkpoint identifier.
    ///
    /// A checkpoint is a durable bookmark to a namespace manifest version.
    CheckpointId,
    prefix = "chk"
}

string_id! {
    /// Durable id for one upload session.
    UploadId,
    prefix = "upl"
}

string_id! {
    /// Durable identity of one immutable content object.
    ///
    /// The body is 128 fully random bits, with no time component: content
    /// object keys shard on the id's leading characters, and a clock-derived
    /// prefix would put every upload in one window into one shard. The id
    /// names *which object*, never what it contains — integrity evidence
    /// rides [`crate::ContentRef`] beside it.
    ContentId,
    error = GeneratedIdValidationError,
    validate = |value: &str| validate_generated_id("con", value)
}

impl ContentId {
    /// Generates an id from 128 fresh random bits.
    pub fn generate() -> Self {
        Self(format!(
            "con_{}",
            crate::hex::hex_encode_bytes(&random_128())
        ))
    }

    /// Returns the two-character shard prefix content keys are grouped by.
    ///
    /// Every valid id has a 32-character lowercase hex body, so this never
    /// panics.
    pub fn shard_prefix(&self) -> &str {
        &self.0[CONTENT_ID_PREFIX_LEN..CONTENT_ID_PREFIX_LEN + CONTENT_ID_SHARD_LEN]
    }
}

/// Byte length of the `con_` marker that precedes a content id's hex body.
const CONTENT_ID_PREFIX_LEN: usize = "con_".len();
/// Number of leading body characters that select a content object's shard.
const CONTENT_ID_SHARD_LEN: usize = 2;

string_id! {
    /// Durable id for one metadata SST table file.
    MetadataTableId,
    prefix = "tbl"
}

string_id! {
    /// Durable id for one derived-index segment file.
    IndexSegmentId,
    prefix = "idx"
}

string_id! {
    /// Durable object id for one namespace manifest candidate.
    ManifestObjectId,
    error = GeneratedIdValidationError,
    validate = |value| {
        validate_position_suffix_id(value, ("manifest_id", "manifest id"))
    }
}

impl ManifestObjectId {
    /// Manifest object ids order by logical manifest position and stay unique
    /// under races.
    pub fn generate(manifest_id: ManifestId) -> Self {
        Self(format!(
            "{:020}-{}",
            manifest_id.0,
            generated_position_suffix()
        ))
    }
}

/// Logical manifest id encoded in a manifest object id's 20-digit prefix.
pub fn manifest_object_id_manifest_id(object_id: &str) -> Option<ManifestId> {
    validate_position_suffix_id(object_id, ("manifest_id", "manifest id")).ok()?;
    let (position, _) = object_id.split_once('-')?;
    position.parse().ok().map(ManifestId)
}

string_id! {
    /// Durable id for one WAL segment.
    WalSegmentId,
    error = GeneratedIdValidationError,
    validate = |value| {
        validate_position_suffix_id(value, ("start_seq", "position"))
    }
}

impl WalSegmentId {
    /// WAL segment ids order by history position and stay unique under races.
    ///
    /// The 20-digit prefix is the segment's `start_seq`, so listings sort by
    /// position and reclamation can range-scan below a boundary cursor. The
    /// 16-hex suffix keeps speculative writes unique: racing writers proposing
    /// different segments for the same position never collide, and the head
    /// compare-and-swap chooses among them. The name is an inspection and
    /// reclamation hint only — recovery authority is the head and chain.
    pub fn generate(start_seq: ChangeSeq) -> Self {
        Self(format!(
            "{:020}-{}",
            start_seq.0,
            generated_position_suffix()
        ))
    }
}

/// Start seq encoded in a WAL segment id's 20-digit position prefix.
///
/// Returns `None` when the value does not follow the generated id shape, so
/// listings can skip foreign objects instead of failing. Like the name
/// itself, the parsed position is an inspection and reclamation hint only —
/// recovery authority is the head and chain.
pub fn wal_segment_id_start_seq(segment_id: &str) -> Option<ChangeSeq> {
    validate_position_suffix_id(segment_id, ("start_seq", "position")).ok()?;
    let (position, _) = segment_id.split_once('-')?;
    position.parse().ok().map(ChangeSeq)
}

string_id! {
    /// Name-policy-derived directory entry key.
    ///
    /// Use this for exact name preconditions. Keep user-facing spelling in
    /// `DisplayName`.
    NameKey,
    error = NameKeyValidationError,
    validate = validate_name_key
}

impl NameKey {
    /// Computes the lookup key for a display name.
    pub fn for_display_name(display_name: &crate::DisplayName) -> Self {
        Self(crate::name_key_for_display_name(display_name.as_str()))
    }
}

// ---------------------------------------------------------------------------
// Numeric ids
// ---------------------------------------------------------------------------

numeric_id! {
    /// Numeric identity of a file or directory within a namespace.
    ///
    /// Inodes are stable across renames.
    InodeId
}

/// Inode 1 is always the root directory of a namespace.
pub const ROOT_INODE_ID: InodeId = InodeId(1);

numeric_id! {
    /// Monotonically increasing file revision counter within one file inode.
    RevisionNo
}

numeric_id! {
    /// Monotonically increasing namespace commit sequence number.
    ///
    /// This is the global visibility order for a namespace.
    ChangeSeq
}

numeric_id! {
    /// Monotonically increasing namespace manifest identity.
    ///
    /// This is the durable file-set version identity for namespace manifests.
    /// Initial/fork manifests may be seeded from the current head sequence, but
    /// later manifest ids can advance for checkpoint metadata, compaction, or
    /// fork/index metadata without a new namespace commit.
    ManifestId
}

numeric_id! {
    /// Monotonically increasing writer epoch for namespace write fencing.
    WriterEpoch
}

// ---------------------------------------------------------------------------
// Filesystem item kind
// ---------------------------------------------------------------------------

/// Filesystem item kind.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "snake_case")]
pub enum InodeKind {
    /// File with revision history.
    File,
    /// Directory with child bindings.
    ///
    /// The wire value is pinned to `"dir"`; only the Rust name spells the
    /// word out.
    #[serde(rename = "dir")]
    Directory,
}

impl fmt::Display for InodeKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::File => f.write_str("file"),
            Self::Directory => f.write_str("dir"),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        ChangeSeq, CheckpointId, CommitId, ContentId, ContentStoreId, ManifestId, ManifestObjectId,
        MetadataTableId, NameKey, NamespaceId, UploadId, WalSegmentId,
    };
    use std::collections::BTreeSet;

    #[test]
    fn namespace_id_parse_accepts_allowed_grammar() {
        let long_id = format!("a{}", "b".repeat(127));
        for value in ["demo", "demo-1", "demo_1", "demo.v1", &long_id] {
            let parsed = NamespaceId::parse(value).expect("valid namespace_id");
            assert_eq!(parsed.as_str(), value);
        }
    }

    #[test]
    fn namespace_id_parse_rejects_invalid_values() {
        let long_id = format!("a{}", "b".repeat(128));
        for value in [
            "", "/", "a/b", ".", "..", " demo", "demo ", "demo\n", "demo?", "demo#", "demo%",
            "Demo", &long_id,
        ] {
            assert!(
                NamespaceId::parse(value).is_err(),
                "expected invalid namespace_id {value:?}"
            );
        }
    }

    #[test]
    fn namespace_id_parse_rejects_reserved_system_prefix() {
        assert!(NamespaceId::parse("loonfs-doctor-abc").is_err());
        assert!(NamespaceId::parse("loonfs-").is_err());
        // The reservation is a prefix rule, not a substring rule.
        assert_eq!(
            NamespaceId::parse("my-loonfs-notes")
                .expect("non-prefixed use is allowed")
                .as_str(),
            "my-loonfs-notes"
        );
        // Commit ids share the base grammar but not the reservation.
        assert!(CommitId::parse("loonfs-retry-1").is_ok());
    }

    #[test]
    fn commit_id_parse_uses_same_allowed_grammar() {
        let parsed = CommitId::parse("c_demo-1").expect("valid commit_id");

        assert_eq!(parsed.as_str(), "c_demo-1");
        assert!(CommitId::parse("c/demo").is_err());
        assert!(CommitId::parse("C_demo").is_err());
    }

    #[test]
    fn identity_try_from_validates_values() {
        assert_eq!(
            NamespaceId::try_from("demo")
                .expect("valid namespace id")
                .as_str(),
            "demo"
        );
        assert_eq!(
            CommitId::try_from("commit-1")
                .expect("valid commit id")
                .as_str(),
            "commit-1"
        );
        assert_eq!(
            ContentStoreId::try_from("cs_00000000000000000000000000000001")
                .expect("valid content store id")
                .as_str(),
            "cs_00000000000000000000000000000001"
        );
        assert_eq!(
            CheckpointId::try_from("chk_00000000000000000000000000000001")
                .expect("valid checkpoint id")
                .as_str(),
            "chk_00000000000000000000000000000001"
        );
        assert_eq!(
            NameKey::try_from("report.txt".to_owned())
                .expect("valid name key")
                .as_str(),
            "report.txt"
        );
        assert_eq!(
            ManifestObjectId::try_from("00000000000000000042-0123456789abcdef")
                .expect("valid manifest object id")
                .as_str(),
            "00000000000000000042-0123456789abcdef"
        );

        assert!(NamespaceId::try_from("invalid/name").is_err());
        assert!(CommitId::try_from("invalid/name").is_err());
        assert!(ContentStoreId::try_from("cs_0000000000000000000000000000000g").is_err());
        assert!(CheckpointId::try_from("chk_0000000000000000000000000000000g").is_err());
        assert!(NameKey::try_from("a/b").is_err());
        assert!(ManifestObjectId::try_from("42-0123456789abcdef").is_err());
    }

    #[test]
    fn identity_from_str_delegates_to_parse() {
        let namespace_id: NamespaceId = "demo".parse().expect("valid namespace id");
        assert_eq!(namespace_id.as_str(), "demo");
        assert!("invalid/name".parse::<NamespaceId>().is_err());
    }

    #[test]
    fn identity_deserialize_validates_values() {
        let namespace_id: NamespaceId =
            serde_json::from_str(r#""demo""#).expect("valid namespace id json");
        assert_eq!(namespace_id.as_str(), "demo");
        let commit_id: CommitId =
            serde_json::from_str(r#""commit-1""#).expect("valid commit id json");
        assert_eq!(commit_id.as_str(), "commit-1");
        let content_store_id: ContentStoreId =
            serde_json::from_str(r#""cs_00000000000000000000000000000001""#)
                .expect("valid content store id json");
        assert_eq!(
            content_store_id.as_str(),
            "cs_00000000000000000000000000000001"
        );
        let checkpoint_id: CheckpointId =
            serde_json::from_str(r#""chk_00000000000000000000000000000001""#)
                .expect("valid checkpoint id json");
        assert_eq!(
            checkpoint_id.as_str(),
            "chk_00000000000000000000000000000001"
        );

        let namespace_error = serde_json::from_str::<NamespaceId>(r#""invalid/name""#)
            .expect_err("invalid namespace id json");
        assert!(namespace_error.to_string().contains("namespace_id"));
        let commit_error = serde_json::from_str::<CommitId>(r#""invalid/name""#)
            .expect_err("invalid commit id json");
        assert!(commit_error.to_string().contains("commit_id"));
        let content_store_error =
            serde_json::from_str::<ContentStoreId>(r#""cs_0000000000000000000000000000000g""#)
                .expect_err("invalid content store id json");
        assert!(content_store_error.to_string().contains("generated id"));
        let checkpoint_error =
            serde_json::from_str::<CheckpointId>(r#""chk_0000000000000000000000000000000g""#)
                .expect_err("invalid checkpoint id json");
        assert!(checkpoint_error.to_string().contains("generated id"));
    }

    #[test]
    fn generated_content_store_id_parse_requires_prefix_and_lower_hex_body() {
        let parsed = ContentStoreId::parse("cs_00000000000000000000000000000001")
            .expect("valid content store id");

        assert_eq!(parsed.as_str(), "cs_00000000000000000000000000000001");
        let hyphenated_content_store_id = ["cs", "1"].join("-");
        for value in [
            hyphenated_content_store_id.as_str(),
            "upl_00000000000000000000000000000001",
            "content-stores/foo",
            "cs_",
            "cs_abcdef",
            "cs_0000000000000000000000000000000",
            "cs_000000000000000000000000000000001",
            "cs_ABCDEF00000000000000000000000000",
            "cs_0000000000000000000000000000000g",
            " cs_00000000000000000000000000000001",
            "cs_00000000000000000000000000000001 ",
        ] {
            assert!(
                ContentStoreId::parse(value).is_err(),
                "expected invalid content store id {value:?}"
            );
        }
    }

    #[test]
    fn generated_upload_wal_segment_table_and_checkpoint_ids_reject_hyphenated_ids() {
        assert!(UploadId::parse("upl_00000000000000000000000000000001").is_ok());
        assert!(MetadataTableId::parse("tbl_00000000000000000000000000000001").is_ok());
        assert!(CheckpointId::parse("chk_00000000000000000000000000000001").is_ok());
        assert!(UploadId::parse(["upl", "123"].join("-")).is_err());
        assert!(WalSegmentId::parse("00000000000000000412-9f2a6c0e4b7d4a90").is_ok());
        assert!(ManifestObjectId::parse("00000000000000000412-9f2a6c0e4b7d4a90").is_ok());
        assert!(WalSegmentId::parse("412-9f2a6c0e4b7d4a90").is_err());
        assert!(WalSegmentId::parse("00000000000000000412-9F2A6C0E4B7D4A90").is_err());
        assert!(ManifestObjectId::parse("412-9f2a6c0e4b7d4a90").is_err());
        assert!(ManifestObjectId::parse("00000000000000000412-9F2A6C0E4B7D4A90").is_err());
        assert!(ManifestObjectId::parse("mf_9f2a6c0e4b7d4a90b13f0d8c5e6a2b41").is_err());
        assert!(WalSegmentId::parse("seg_9f2a6c0e4b7d4a90b13f0d8c5e6a2b41").is_err());
        assert!(MetadataTableId::parse(["tbl", "123"].join("-")).is_err());
        assert!(CheckpointId::parse(["chk", "123"].join("-")).is_err());
    }

    #[test]
    fn generated_runtime_ids_use_lower_hex_uuid_bodies() {
        let upload_id = UploadId::generate();
        let wal_segment_id = WalSegmentId::generate(ChangeSeq(412));
        let manifest_object_id = ManifestObjectId::generate(ManifestId(413));
        let metadata_table_id = MetadataTableId::generate();
        let checkpoint_id = CheckpointId::generate();

        assert_generated_id_shape(upload_id.as_str(), "upl");
        assert!(wal_segment_id.as_str().starts_with("00000000000000000412-"));
        assert!(manifest_object_id
            .as_str()
            .starts_with("00000000000000000413-"));
        assert_generated_id_shape(metadata_table_id.as_str(), "tbl");
        assert_generated_id_shape(checkpoint_id.as_str(), "chk");
        assert!(UploadId::parse(upload_id.as_str()).is_ok());
        assert!(WalSegmentId::parse(wal_segment_id.as_str()).is_ok());
        assert!(ManifestObjectId::parse(manifest_object_id.as_str()).is_ok());
        assert!(MetadataTableId::parse(metadata_table_id.as_str()).is_ok());
        assert!(CheckpointId::parse(checkpoint_id.as_str()).is_ok());
    }

    #[test]
    fn generated_wal_segment_ids_are_not_reused_across_samples() {
        // Same position, many proposers: the suffix keeps every proposal
        // distinct.
        let mut ids = BTreeSet::new();
        for _ in 0..128 {
            let id = WalSegmentId::generate(ChangeSeq(412));
            assert!(
                ids.insert(id.clone()),
                "generated duplicate WAL segment id {id}"
            );
        }
    }

    #[test]
    fn generated_manifest_object_ids_are_not_reused_across_samples() {
        let mut ids = BTreeSet::new();
        for _ in 0..128 {
            let id = ManifestObjectId::generate(ManifestId(412));
            assert!(
                ids.insert(id.clone()),
                "generated duplicate manifest object id {id}"
            );
        }
    }

    #[test]
    fn wal_segment_id_start_seq_reads_position_prefix() {
        assert_eq!(
            super::wal_segment_id_start_seq("00000000000000000412-9f2a6c0e4b7d4a90"),
            Some(ChangeSeq(412))
        );
        assert_eq!(super::wal_segment_id_start_seq("not-a-segment-id"), None);
    }

    #[test]
    fn manifest_object_id_manifest_id_reads_position_prefix() {
        assert_eq!(
            super::manifest_object_id_manifest_id("00000000000000000412-9f2a6c0e4b7d4a90"),
            Some(ManifestId(412))
        );
        assert_eq!(
            super::manifest_object_id_manifest_id("not-a-manifest-object-id"),
            None
        );
    }

    #[test]
    fn generated_content_ids_are_unique_and_shard_uniformly() {
        let mut ids = BTreeSet::new();
        let mut shards = BTreeSet::new();
        for _ in 0..512 {
            let id = ContentId::generate();
            assert_generated_id_shape(id.as_str(), "con");
            assert_eq!(
                id.shard_prefix(),
                &id.as_str()["con_".len().."con_".len() + 2]
            );
            shards.insert(id.shard_prefix().to_owned());
            assert!(
                ids.insert(id.clone()),
                "generated duplicate content id {id}"
            );
        }
        // 512 draws over 256 shards: a generator with a fixed or clock-derived
        // prefix would collapse into a handful of shards.
        assert!(
            shards.len() > 128,
            "content id shard prefixes are not spread: {} distinct",
            shards.len()
        );
    }

    #[test]
    fn content_id_parse_requires_the_generated_id_shape() {
        assert!(ContentId::parse("con_0123456789abcdef0123456789abcdef").is_ok());
        for value in [
            "con_",
            "con_abcdef",
            "con_0123456789ABCDEF0123456789abcdef",
            "con_0123456789abcdef0123456789abcde",
            "upl_0123456789abcdef0123456789abcdef",
            "0123456789abcdef0123456789abcdef",
        ] {
            assert!(
                ContentId::parse(value).is_err(),
                "expected invalid content id {value:?}"
            );
        }
    }

    fn assert_generated_id_shape(value: &str, prefix: &str) {
        let expected_prefix = format!("{prefix}_");
        let body = value
            .strip_prefix(&expected_prefix)
            .expect("generated id prefix");
        assert_eq!(body.len(), 32);
        assert!(
            body.bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)),
            "generated id body must be lowercase hex: {value}"
        );
    }

    #[test]
    fn name_key_parse_rejects_invalid_values() {
        assert_eq!(
            NameKey::parse("").expect_err("empty").reason(),
            "must not be empty"
        );
        assert_eq!(
            NameKey::parse("a/b").expect_err("slash").reason(),
            "must not contain `/`"
        );
        assert_eq!(
            NameKey::parse(".").expect_err("dot").reason(),
            "must not be `.` or `..`"
        );
        assert_eq!(
            NameKey::parse("a\u{0}b").expect_err("control").reason(),
            "must not contain control characters"
        );
        NameKey::parse("k".repeat(super::MAX_NAME_KEY_BYTES)).expect("cap is inclusive");
        assert_eq!(
            NameKey::parse("k".repeat(super::MAX_NAME_KEY_BYTES + 1))
                .expect_err("over cap")
                .reason(),
            "exceeds the maximum name key length of 768 bytes"
        );
    }

    #[test]
    fn name_key_serializes_as_string_and_validates_deserialize() {
        let name_key = NameKey::parse("report.txt").expect("valid name key");

        assert_eq!(
            serde_json::to_string(&name_key).expect("serialize name key"),
            "\"report.txt\""
        );
        assert_eq!(
            serde_json::from_str::<NameKey>("\"report.txt\"").expect("deserialize name key"),
            name_key
        );
        assert!(serde_json::from_str::<NameKey>("\"a/b\"").is_err());
    }
}