lodestone-types 0.0.16

use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::num::{NonZeroU64, NonZeroU128};
use std::str::FromStr;

use serde::de;
use serde::{Deserialize, Deserializer, Serialize, Serializer};

macro_rules! opaque_u128_id {
    ($name:ident, $description:literal) => {
        #[doc = $description]
        #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
        pub struct $name(NonZeroU128);

        impl $name {
            pub fn new(value: u128) -> Result<Self, IdentityError> {
                NonZeroU128::new(value).map(Self).ok_or(IdentityError::Zero {
                    kind: stringify!($name),
                })
            }

            pub fn from_be_bytes(bytes: [u8; 16]) -> Result<Self, IdentityError> {
                Self::new(u128::from_be_bytes(bytes))
            }

            #[must_use]
            pub const fn get(self) -> u128 {
                self.0.get()
            }

            #[must_use]
            pub const fn to_be_bytes(self) -> [u8; 16] {
                self.get().to_be_bytes()
            }

            #[must_use]
            pub fn checked_next(self) -> Option<Self> {
                self.get().checked_add(1).and_then(NonZeroU128::new).map(Self)
            }
        }

        impl Display for $name {
            fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                write!(formatter, "{:032x}", self.get())
            }
        }

        impl FromStr for $name {
            type Err = IdentityError;

            fn from_str(value: &str) -> Result<Self, Self::Err> {
                parse_hex_u128(stringify!($name), value).and_then(Self::new)
            }
        }

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

        impl<'de> Deserialize<'de> for $name {
            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
            where
                D: Deserializer<'de>,
            {
                let value = String::deserialize(deserializer)?;
                value.parse().map_err(de::Error::custom)
            }
        }
    };
}

opaque_u128_id!(
    CollectionId,
    "An immutable nonzero collection namespace encoded as 32 lower-case hexadecimal characters."
);
opaque_u128_id!(
    IndexId,
    "An immutable nonzero index-incarnation identity encoded as 32 lower-case hexadecimal characters."
);
opaque_u128_id!(
    ObjectIncarnation,
    "A never-reused nonzero physical-object identity within one backend authority."
);
opaque_u128_id!(
    ProtectionSnapshotId,
    "A never-reused semantic protection-snapshot identity derived from a burned control-authority allocator value."
);
opaque_u128_id!(
    RecoveryAttemptId,
    "A stable nonzero identity reused by every retry of one offline total-voter recovery attempt."
);
/// Stable object-namespace identity derived once from the bootstrap cluster
/// identity and persisted unchanged across recovery eras.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct ClusterStorageId([u8; 16]);

impl ClusterStorageId {
    const DERIVATION_DOMAIN: &'static [u8] = b"lodestone:cluster-storage-id:v1\0";

    pub fn derive_from_genesis(cluster_identity: ClusterIdentity) -> Result<Self, IdentityError> {
        if cluster_identity.recovery_generation() != RecoveryGeneration::genesis() {
            return Err(IdentityError::ClusterStorageIdNonGenesis(
                cluster_identity.recovery_generation().get(),
            ));
        }
        let mut hasher = blake3::Hasher::new();
        hasher.update(Self::DERIVATION_DOMAIN);
        hasher.update(&cluster_identity.recovery_era().to_be_bytes());
        let digest = hasher.finalize();
        let mut bytes = [0_u8; 16];
        bytes.copy_from_slice(&digest.as_bytes()[..16]);
        Ok(Self::from_be_bytes(bytes))
    }

    /// Decodes a persisted stable storage identity.
    #[must_use]
    pub const fn from_be_bytes(bytes: [u8; 16]) -> Self {
        Self(bytes)
    }

    #[must_use]
    pub const fn to_be_bytes(self) -> [u8; 16] {
        self.0
    }
}

impl Display for ClusterStorageId {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "{:032x}", u128::from_be_bytes(self.0))
    }
}

impl FromStr for ClusterStorageId {
    type Err = IdentityError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        let value = parse_hex_u128("ClusterStorageId", value)?;
        Ok(Self::from_be_bytes(value.to_be_bytes()))
    }
}

impl Serialize for ClusterStorageId {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.collect_str(self)
    }
}

impl<'de> Deserialize<'de> for ClusterStorageId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        String::deserialize(deserializer)?.parse().map_err(de::Error::custom)
    }
}

/// A checked, nonzero process incarnation local to one durable `NodeId`.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct NodeIncarnation(NonZeroU64);

impl NodeIncarnation {
    pub fn new(value: u64) -> Result<Self, IdentityError> {
        NonZeroU64::new(value).map(Self).ok_or(IdentityError::Zero {
            kind: "NodeIncarnation",
        })
    }

    #[must_use]
    pub const fn get(self) -> u64 {
        self.0.get()
    }

    #[must_use]
    pub fn checked_next(self) -> Option<Self> {
        self.get().checked_add(1).and_then(NonZeroU64::new).map(Self)
    }
}

impl Display for NodeIncarnation {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        Display::fmt(&self.get(), formatter)
    }
}

impl Serialize for NodeIncarnation {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_u64(self.get())
    }
}

impl<'de> Deserialize<'de> for NodeIncarnation {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        Self::new(u64::deserialize(deserializer)?).map_err(de::Error::custom)
    }
}

/// An opaque cluster fencing identity advanced only by offline total-voter
/// recovery.
///
/// Unlike allocator-backed identities, this type intentionally exposes no
/// numeric value or successor operation. Recovery-era order is meaningful
/// only inside the bounded offline lineage protocol.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct RecoveryEra(NonZeroU128);

impl RecoveryEra {
    pub fn new(value: u128) -> Result<Self, IdentityError> {
        NonZeroU128::new(value)
            .map(Self)
            .ok_or(IdentityError::Zero { kind: "RecoveryEra" })
    }

    pub fn from_be_bytes(bytes: [u8; 16]) -> Result<Self, IdentityError> {
        Self::new(u128::from_be_bytes(bytes))
    }

    #[must_use]
    pub const fn to_be_bytes(self) -> [u8; 16] {
        self.0.get().to_be_bytes()
    }
}

impl Display for RecoveryEra {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "{:032x}", self.0.get())
    }
}

impl FromStr for RecoveryEra {
    type Err = IdentityError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        parse_hex_u128("RecoveryEra", value).and_then(Self::new)
    }
}

impl Serialize for RecoveryEra {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.collect_str(self)
    }
}

impl<'de> Deserialize<'de> for RecoveryEra {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        String::deserialize(deserializer)?.parse().map_err(de::Error::custom)
    }
}

fn parse_hex_u128(kind: &'static str, value: &str) -> Result<u128, IdentityError> {
    if value.len() != 32 {
        return Err(IdentityError::HexLength {
            kind,
            actual: value.len(),
        });
    }
    if !value
        .bytes()
        .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(IdentityError::LowerHex { kind });
    }
    u128::from_str_radix(value, 16).map_err(|_| IdentityError::LowerHex { kind })
}

/// A bounded human-facing index name.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct IndexName(String);

impl IndexName {
    pub fn new(value: impl Into<String>) -> Result<Self, IdentityError> {
        let value = value.into();
        let length = value.len();
        if !(1..=255).contains(&length) {
            return Err(IdentityError::IndexNameLength(length));
        }
        if value.bytes().any(|byte| byte <= 0x1f || byte == 0x7f) {
            return Err(IdentityError::IndexNameControl);
        }
        Ok(Self(value))
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    #[must_use]
    pub fn into_inner(self) -> String {
        self.0
    }

    /// Encodes the bounded binary form as a big-endian u16 length followed by
    /// the exact UTF-8 bytes.
    #[must_use]
    pub fn to_binary(&self) -> Vec<u8> {
        let length = self.0.len() as u16;
        let mut bytes = Vec::with_capacity(2 + usize::from(length));
        bytes.extend_from_slice(&length.to_be_bytes());
        bytes.extend_from_slice(self.0.as_bytes());
        bytes
    }

    /// Decodes one exact bounded binary value, refusing trailing bytes.
    pub fn from_binary(bytes: &[u8]) -> Result<Self, IdentityError> {
        let Some(length_bytes) = bytes.get(..2) else {
            return Err(IdentityError::BinaryLength {
                kind: "IndexName",
                expected: 2,
                actual: bytes.len(),
            });
        };
        let length = usize::from(u16::from_be_bytes([length_bytes[0], length_bytes[1]]));
        if bytes.len() != 2 + length {
            return Err(IdentityError::BinaryLength {
                kind: "IndexName",
                expected: 2 + length,
                actual: bytes.len(),
            });
        }
        let value = std::str::from_utf8(&bytes[2..]).map_err(|_| IdentityError::IndexNameUtf8)?;
        Self::new(value)
    }
}

impl Display for IndexName {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

impl FromStr for IndexName {
    type Err = IdentityError;

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

impl<'de> Deserialize<'de> for IndexName {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        String::deserialize(deserializer)?.parse().map_err(de::Error::custom)
    }
}

/// Stable identity for one canonical object-storage namespace.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct BackendAuthorityId(String);

impl BackendAuthorityId {
    pub fn new(value: impl Into<String>) -> Result<Self, IdentityError> {
        let value = value.into();
        let bytes = value.as_bytes();
        if !(1..=64).contains(&bytes.len())
            || !bytes.first().is_some_and(u8::is_ascii_lowercase)
            || !bytes
                .iter()
                .skip(1)
                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-'))
        {
            return Err(IdentityError::BackendAuthority(value));
        }
        Ok(Self(value))
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Encodes the bounded binary form as a one-byte length followed by the
    /// exact ASCII identity.
    #[must_use]
    pub fn to_binary(&self) -> Vec<u8> {
        let length = self.0.len() as u8;
        let mut bytes = Vec::with_capacity(1 + usize::from(length));
        bytes.push(length);
        bytes.extend_from_slice(self.0.as_bytes());
        bytes
    }

    /// Decodes one exact bounded binary value, refusing trailing bytes.
    pub fn from_binary(bytes: &[u8]) -> Result<Self, IdentityError> {
        let Some(length) = bytes.first().copied() else {
            return Err(IdentityError::BinaryLength {
                kind: "BackendAuthorityId",
                expected: 1,
                actual: 0,
            });
        };
        let expected = 1 + usize::from(length);
        if bytes.len() != expected {
            return Err(IdentityError::BinaryLength {
                kind: "BackendAuthorityId",
                expected,
                actual: bytes.len(),
            });
        }
        let value = std::str::from_utf8(&bytes[1..]).map_err(|_| IdentityError::BackendAuthorityUtf8)?;
        Self::new(value)
    }
}

impl Display for BackendAuthorityId {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

impl FromStr for BackendAuthorityId {
    type Err = IdentityError;

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

impl<'de> Deserialize<'de> for BackendAuthorityId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        String::deserialize(deserializer)?.parse().map_err(de::Error::custom)
    }
}

/// The immutable catalog ownership identity of one index incarnation.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionIndexRef {
    collection_id: CollectionId,
    index_id: IndexId,
}

impl CollectionIndexRef {
    #[must_use]
    pub const fn new(collection_id: CollectionId, index_id: IndexId) -> Self {
        Self {
            collection_id,
            index_id,
        }
    }

    #[must_use]
    pub const fn collection_id(self) -> CollectionId {
        self.collection_id
    }

    #[must_use]
    pub const fn index_id(self) -> IndexId {
        self.index_id
    }
}

impl Display for CollectionIndexRef {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "{}/{}", self.collection_id, self.index_id)
    }
}

/// A human address that must be resolved once through the catalog.
#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct IndexAddress {
    collection_id: CollectionId,
    index_name: IndexName,
}

impl IndexAddress {
    #[must_use]
    pub const fn new(collection_id: CollectionId, index_name: IndexName) -> Self {
        Self {
            collection_id,
            index_name,
        }
    }

    #[must_use]
    pub const fn collection_id(&self) -> CollectionId {
        self.collection_id
    }

    #[must_use]
    pub fn index_name(&self) -> &IndexName {
        &self.index_name
    }
}

/// A shard identity that is meaningful at production boundaries.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ShardKey {
    collection_index_ref: CollectionIndexRef,
    shard_id: crate::ShardId,
}

impl ShardKey {
    #[must_use]
    pub const fn new(collection_index_ref: CollectionIndexRef, shard_id: crate::ShardId) -> Self {
        Self {
            collection_index_ref,
            shard_id,
        }
    }

    #[must_use]
    pub const fn collection_index_ref(self) -> CollectionIndexRef {
        self.collection_index_ref
    }

    #[must_use]
    pub const fn shard_id(self) -> crate::ShardId {
        self.shard_id
    }
}

/// The immutable WAL/data authority pair copied into an index catalog record.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct BackendAuthorityPair {
    wal: BackendAuthorityId,
    data: BackendAuthorityId,
}

impl BackendAuthorityPair {
    #[must_use]
    pub const fn new(wal: BackendAuthorityId, data: BackendAuthorityId) -> Self {
        Self { wal, data }
    }

    #[must_use]
    pub fn wal(&self) -> &BackendAuthorityId {
        &self.wal
    }

    #[must_use]
    pub fn data(&self) -> &BackendAuthorityId {
        &self.data
    }
}

/// Checked public-generation component advanced once per offline recovery.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct RecoveryGeneration(u32);

impl RecoveryGeneration {
    pub const MAX: u32 = 0x7fff_ffff;

    pub fn new(value: u32) -> Result<Self, IdentityError> {
        if value <= Self::MAX {
            Ok(Self(value))
        } else {
            Err(IdentityError::RecoveryGeneration(value))
        }
    }

    #[must_use]
    pub const fn genesis() -> Self {
        Self(0)
    }

    #[must_use]
    pub const fn get(self) -> u32 {
        self.0
    }

    #[must_use]
    pub fn checked_next(self) -> Option<Self> {
        self.0.checked_add(1).filter(|value| *value <= Self::MAX).map(Self)
    }
}

impl Serialize for RecoveryGeneration {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_u32(self.get())
    }
}

impl<'de> Deserialize<'de> for RecoveryGeneration {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        Self::new(u32::deserialize(deserializer)?).map_err(de::Error::custom)
    }
}

impl Display for RecoveryGeneration {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        Display::fmt(&self.get(), formatter)
    }
}

/// Recovery-stable positive public optimistic-concurrency term.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PrimaryTerm(u64);

impl PrimaryTerm {
    pub fn new(value: u64) -> Result<Self, IdentityError> {
        if value == 0 || value > i64::MAX as u64 || value as u32 == 0 {
            Err(IdentityError::PrimaryTerm(value))
        } else {
            Ok(Self(value))
        }
    }

    pub fn from_parts(recovery_generation: RecoveryGeneration, epoch: crate::Epoch) -> Result<Self, IdentityError> {
        if epoch.get() == 0 {
            return Err(IdentityError::ZeroEpoch);
        }
        let value = (u64::from(recovery_generation.get()) << 32) | u64::from(epoch.get());
        Self::new(value)
    }

    /// Constructs the term for a freshly bootstrapped writer before any total
    /// recovery generation change.
    #[must_use]
    pub const fn from_genesis_epoch(epoch: crate::Epoch) -> Self {
        Self(epoch.get() as u64)
    }

    #[must_use]
    pub const fn get(self) -> u64 {
        self.0
    }

    #[must_use]
    pub const fn epoch_component(self) -> crate::Epoch {
        crate::Epoch::new(self.0 as u32)
    }

    #[must_use]
    pub const fn recovery_generation_component(self) -> RecoveryGeneration {
        RecoveryGeneration((self.0 >> 32) as u32)
    }
}

impl Serialize for PrimaryTerm {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_u64(self.get())
    }
}

impl<'de> Deserialize<'de> for PrimaryTerm {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        Self::new(u64::deserialize(deserializer)?).map_err(de::Error::custom)
    }
}

impl Display for PrimaryTerm {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        Display::fmt(&self.get(), formatter)
    }
}

/// Complete internal writer/task fence. Public primary terms never replace it.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct FenceToken {
    recovery_era: RecoveryEra,
    epoch: crate::Epoch,
}

impl FenceToken {
    pub fn new(recovery_era: RecoveryEra, epoch: crate::Epoch) -> Result<Self, IdentityError> {
        if epoch.get() == 0 {
            return Err(IdentityError::ZeroEpoch);
        }
        Ok(Self { recovery_era, epoch })
    }

    #[must_use]
    pub const fn recovery_era(self) -> RecoveryEra {
        self.recovery_era
    }

    #[must_use]
    pub const fn epoch(self) -> crate::Epoch {
        self.epoch
    }

    /// Encodes the complete fence as 16-byte recovery era plus 4-byte
    /// big-endian epoch.
    #[must_use]
    pub fn to_binary(self) -> [u8; 20] {
        let mut bytes = [0_u8; 20];
        bytes[..16].copy_from_slice(&self.recovery_era.to_be_bytes());
        bytes[16..].copy_from_slice(&self.epoch.get().to_be_bytes());
        bytes
    }

    /// Decodes and validates one exact complete-fence binary value.
    pub fn from_binary(bytes: &[u8]) -> Result<Self, IdentityError> {
        if bytes.len() != 20 {
            return Err(IdentityError::BinaryLength {
                kind: "FenceToken",
                expected: 20,
                actual: bytes.len(),
            });
        }
        let mut era_bytes = [0_u8; 16];
        era_bytes.copy_from_slice(&bytes[..16]);
        let recovery_era = RecoveryEra::from_be_bytes(era_bytes)?;
        let epoch = crate::Epoch::new(u32::from_be_bytes([bytes[16], bytes[17], bytes[18], bytes[19]]));
        Self::new(recovery_era, epoch)
    }
}

impl<'de> Deserialize<'de> for FenceToken {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(deny_unknown_fields)]
        struct Wire {
            recovery_era: RecoveryEra,
            epoch: crate::Epoch,
        }

        let wire = Wire::deserialize(deserializer)?;
        Self::new(wire.recovery_era, wire.epoch).map_err(de::Error::custom)
    }
}

/// Exact opaque and numeric recovery identity pair stored in snapshots and markers.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterIdentity {
    recovery_era: RecoveryEra,
    recovery_generation: RecoveryGeneration,
}

impl ClusterIdentity {
    #[must_use]
    pub const fn new(recovery_era: RecoveryEra, recovery_generation: RecoveryGeneration) -> Self {
        Self {
            recovery_era,
            recovery_generation,
        }
    }

    #[must_use]
    pub const fn recovery_era(self) -> RecoveryEra {
        self.recovery_era
    }

    #[must_use]
    pub const fn recovery_generation(self) -> RecoveryGeneration {
        self.recovery_generation
    }
}

/// Validation failure for P5 durable identities.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum IdentityError {
    Zero {
        kind: &'static str,
    },
    HexLength {
        kind: &'static str,
        actual: usize,
    },
    LowerHex {
        kind: &'static str,
    },
    BinaryLength {
        kind: &'static str,
        expected: usize,
        actual: usize,
    },
    IndexNameLength(usize),
    IndexNameControl,
    IndexNameUtf8,
    BackendAuthority(String),
    BackendAuthorityUtf8,
    ClusterStorageIdNonGenesis(u32),
    RecoveryGeneration(u32),
    ZeroEpoch,
    PrimaryTerm(u64),
}

impl Display for IdentityError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Zero { kind } => write!(formatter, "{kind} must be nonzero"),
            Self::HexLength { kind, actual } => {
                write!(
                    formatter,
                    "{kind} must contain exactly 32 lower-case hexadecimal bytes; got {actual}"
                )
            }
            Self::LowerHex { kind } => {
                write!(
                    formatter,
                    "{kind} must contain exactly 32 lower-case hexadecimal characters"
                )
            }
            Self::BinaryLength { kind, expected, actual } => write!(
                formatter,
                "{kind} binary encoding requires exactly {expected} bytes; got {actual}"
            ),
            Self::IndexNameLength(length) => {
                write!(formatter, "index name is {length} bytes; expected 1..=255")
            }
            Self::IndexNameControl => formatter.write_str("index name contains an ASCII control character"),
            Self::IndexNameUtf8 => formatter.write_str("index name binary encoding is not valid UTF-8"),
            Self::BackendAuthority(value) => write!(
                formatter,
                "backend authority '{value}' must match [a-z][a-z0-9_-]{{0,63}}"
            ),
            Self::BackendAuthorityUtf8 => formatter.write_str("backend authority binary encoding is not valid UTF-8"),
            Self::ClusterStorageIdNonGenesis(generation) => write!(
                formatter,
                "cluster storage identity can be derived only from recovery generation zero; got {generation}"
            ),
            Self::RecoveryGeneration(value) => {
                write!(
                    formatter,
                    "recovery generation {value} exceeds the checked 31-bit domain"
                )
            }
            Self::ZeroEpoch => formatter.write_str("a fenced epoch or primary term epoch component must be nonzero"),
            Self::PrimaryTerm(value) => {
                write!(
                    formatter,
                    "primary term {value} must be in 1..=i64::MAX with a nonzero low epoch component"
                )
            }
        }
    }
}

impl Error for IdentityError {}

#[cfg(test)]
mod tests {
    use std::error::Error;

    use super::{
        BackendAuthorityId, ClusterIdentity, ClusterStorageId, CollectionId, CollectionIndexRef, FenceToken,
        IndexAddress, IndexId, IndexName, NodeIncarnation, ObjectIncarnation, PrimaryTerm, ProtectionSnapshotId,
        RecoveryAttemptId, RecoveryEra, RecoveryGeneration, ShardKey,
    };
    use crate::{Epoch, ShardId};

    #[test]
    fn opaque_ids_are_nonzero_fixed_lower_hex_and_big_endian() {
        let collection = CollectionId::new(0x102).ok();
        assert_eq!(
            collection.and_then(|value| serde_json::to_string(&value).ok()),
            Some("\"00000000000000000000000000000102\"".to_owned())
        );
        assert_eq!(
            collection.map(CollectionId::to_be_bytes),
            Some(0x102_u128.to_be_bytes())
        );
        assert!(CollectionId::new(0).is_err());
        assert!(serde_json::from_str::<CollectionId>("\"00000000000000000000000000000000\"").is_err());
        assert!(serde_json::from_str::<CollectionId>("\"000000000000000000000000000000AF\"").is_err());
        assert!(serde_json::from_str::<CollectionId>("1").is_err());
    }

    #[test]
    fn node_incarnation_is_nonzero_monotonic_and_numeric_on_the_wire() -> Result<(), Box<dyn Error>> {
        let incarnation = NodeIncarnation::new(7)?;
        assert_eq!(incarnation.get(), 7);
        assert_eq!(incarnation.checked_next().map(|value| value.get()), Some(8));
        assert_eq!(serde_json::to_string(&incarnation)?, "7");
        assert_eq!(serde_json::from_str::<NodeIncarnation>("7")?, incarnation);
        assert!(NodeIncarnation::new(0).is_err());
        assert!(serde_json::from_str::<NodeIncarnation>("0").is_err());
        Ok(())
    }

    #[test]
    fn recovery_attempt_identity_uses_the_opaque_hex_contract() -> Result<(), Box<dyn Error>> {
        let identity = RecoveryAttemptId::new(9)?;
        assert_eq!(identity.to_string(), "00000000000000000000000000000009");
        assert_eq!(identity.to_string().parse::<RecoveryAttemptId>()?, identity);
        let protection = ProtectionSnapshotId::new(11)?;
        assert_eq!(protection.to_string(), "0000000000000000000000000000000b");
        assert_eq!(protection.to_string().parse::<ProtectionSnapshotId>()?, protection);
        assert!(ProtectionSnapshotId::new(0).is_err());
        Ok(())
    }

    #[test]
    fn names_and_backend_authorities_enforce_catalog_bounds() {
        let name = IndexName::new("products");
        assert!(name.is_ok());
        if let Ok(value) = &name {
            assert_eq!(IndexName::from_binary(&value.to_binary()).as_ref(), Ok(value));
        }
        assert!(IndexName::new("").is_err());
        assert!(IndexName::new("bad\u{7f}name").is_err());
        assert!(IndexName::new("x".repeat(256)).is_err());
        let authority = BackendAuthorityId::new("standard_aps1");
        if let Ok(value) = &authority {
            assert_eq!(BackendAuthorityId::from_binary(&value.to_binary()).as_ref(), Ok(value));
        }
        assert!(BackendAuthorityId::new("Express").is_err());
        assert!(BackendAuthorityId::new("-leading").is_err());
        assert!(BackendAuthorityId::from_binary(&[3, b'a', b'b']).is_err());
    }

    #[test]
    fn scopes_and_addresses_round_trip_without_unknown_fields() -> Result<(), Box<dyn Error>> {
        let collection = CollectionId::new(1)?;
        let index = IndexId::new(2)?;
        let name = IndexName::new("events")?;
        let scope = CollectionIndexRef::new(collection, index);
        let address = IndexAddress::new(collection, name);
        let shard = ShardKey::new(scope, ShardId::new(7));
        let encoded_address = serde_json::to_vec(&address)?;
        assert_eq!(serde_json::from_slice::<IndexAddress>(&encoded_address)?, address);
        let encoded_shard = serde_json::to_vec(&shard)?;
        assert_eq!(serde_json::from_slice::<ShardKey>(&encoded_shard)?, shard);
        assert!(serde_json::from_str::<CollectionIndexRef>(
            r#"{"collection_id":"00000000000000000000000000000001","index_id":"00000000000000000000000000000002","extra":1}"#
        )
        .is_err());
        Ok(())
    }

    #[test]
    fn recovery_generation_projects_a_positive_signed_primary_term() -> Result<(), Box<dyn Error>> {
        let era = RecoveryEra::new(9)?;
        let generation = RecoveryGeneration::new(1)?;
        let term = PrimaryTerm::from_parts(generation, Epoch::new(7))?;
        assert_eq!(term.get(), (1_u64 << 32) | 7);
        assert_eq!(term.recovery_generation_component(), generation);
        assert_eq!(term.epoch_component(), Epoch::new(7));
        assert!(PrimaryTerm::from_parts(generation, Epoch::new(0)).is_err());
        assert!(PrimaryTerm::new(1_u64 << 32).is_err());
        assert!(RecoveryGeneration::new(RecoveryGeneration::MAX + 1).is_err());
        assert!(
            RecoveryGeneration::new(RecoveryGeneration::MAX)
                .ok()
                .and_then(RecoveryGeneration::checked_next)
                .is_none()
        );
        let fence = FenceToken::new(era, Epoch::new(7))?;
        assert_eq!(FenceToken::from_binary(&fence.to_binary())?, fence);
        assert!(FenceToken::from_binary(&[0; 20]).is_err());
        assert!(
            serde_json::from_str::<FenceToken>(r#"{"recovery_era":"00000000000000000000000000000009","epoch":0}"#)
                .is_err()
        );
        assert_eq!(ClusterIdentity::new(era, generation).recovery_generation(), generation);
        Ok(())
    }

    #[test]
    fn cluster_storage_identity_is_genesis_derived_and_stable() -> Result<(), Box<dyn Error>> {
        let genesis = ClusterIdentity::new(RecoveryEra::new(9)?, RecoveryGeneration::genesis());
        let storage_id = ClusterStorageId::derive_from_genesis(genesis)?;
        assert_eq!(storage_id.to_string().len(), 32);
        assert_eq!(storage_id.to_string().parse::<ClusterStorageId>()?, storage_id);
        assert_eq!(ClusterStorageId::from_be_bytes(storage_id.to_be_bytes()), storage_id);
        assert_eq!(
            serde_json::from_str::<ClusterStorageId>(&serde_json::to_string(&storage_id)?)?,
            storage_id
        );
        assert_eq!(
            ClusterStorageId::derive_from_genesis(genesis)?,
            ClusterStorageId::derive_from_genesis(genesis)?
        );
        let successor = ClusterIdentity::new(RecoveryEra::new(10)?, RecoveryGeneration::new(1)?);
        assert_eq!(
            ClusterStorageId::derive_from_genesis(successor),
            Err(super::IdentityError::ClusterStorageIdNonGenesis(1))
        );
        assert_eq!(ClusterStorageId::from_be_bytes([0; 16]).to_be_bytes(), [0; 16]);
        Ok(())
    }

    #[test]
    fn every_nonzero_physical_identity_uses_the_same_codec_contract() -> Result<(), Box<dyn Error>> {
        let index = IndexId::new(u128::MAX)?;
        let object = ObjectIncarnation::new(u128::MAX)?;
        assert_eq!(index.to_string(), object.to_string());
        assert!(ObjectIncarnation::from_be_bytes([0; 16]).is_err());
        Ok(())
    }
}