lodestone-proto 0.0.16

//! Bounded operator-to-gateway catalog intent wire.
//!
//! The codec is unreachable from runtime mutation until it is joined to the
//! catalog-owned management record and recovery migration before an
//! operator controller may submit it.

use std::error::Error;
use std::fmt::{self, Display, Formatter};

use lodestone_types::{
    BackendAuthorityPair, CatalogManagementGeneration, CollectionIndexRef, CrdGeneration, IndexAddress, IndexMode,
    IndexRevision, KubernetesObjectUidDigest, Mapping,
};
use serde::{Deserialize, Serialize};

pub const CATALOG_INTENT_FORMAT_VERSION: u16 = 3;
pub const CATALOG_INTENT_RESULT_FORMAT_VERSION: u16 = 1;
pub const CATALOG_RPC_FORMAT_VERSION: u16 = 2;
pub const CATALOG_OBSERVATION_FORMAT_VERSION: u16 = 3;
pub const CATALOG_OPERATION_RECEIPT_FORMAT_VERSION: u16 = 1;
pub const MAX_CATALOG_INTENT_BYTES: usize = 1024 * 1024;
pub const MAX_CATALOG_MAPPING_BYTES: usize = 512 * 1024;
pub const MAX_INITIAL_SHARDS: u16 = 4_096;

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CrdOwnerV1 {
    pub owner_uid_digest: KubernetesObjectUidDigest,
    pub observed_generation: CrdGeneration,
}

impl CrdOwnerV1 {
    fn validate(self) -> Result<(), CatalogIntentError> {
        if self.observed_generation.get() == 0 {
            return Err(CatalogIntentError::InvalidCrdGeneration);
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CatalogApplyPreconditionV1 {
    Absent,
    Owned {
        scope: CollectionIndexRef,
        expected_index_revision: IndexRevision,
        expected_management_generation: CatalogManagementGeneration,
    },
    ApiTransferPrepared {
        scope: CollectionIndexRef,
        // FENCING: adoption changes both index contents and catalog-owned
        // management mode, so neither CAS generation may be omitted.
        expected_index_revision: IndexRevision,
        expected_management_generation: CatalogManagementGeneration,
    },
}

impl CatalogApplyPreconditionV1 {
    fn validate(self) -> Result<(), CatalogIntentError> {
        match self {
            Self::Absent => Ok(()),
            Self::Owned {
                expected_index_revision,
                expected_management_generation,
                ..
            } => {
                if expected_index_revision.get() == 0 || expected_management_generation.get() == 0 {
                    return Err(CatalogIntentError::InvalidPrecondition);
                }
                Ok(())
            }
            Self::ApiTransferPrepared {
                expected_index_revision,
                expected_management_generation,
                ..
            } => {
                if expected_index_revision.get() == 0 || expected_management_generation.get() == 0 {
                    return Err(CatalogIntentError::InvalidPrecondition);
                }
                Ok(())
            }
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DesiredIndexV1 {
    pub address: IndexAddress,
    pub mode: IndexMode,
    #[serde(with = "crate::canonical_base64_bytes")]
    pub mapping_bytes: Vec<u8>,
    pub initial_shards: u16,
    pub authorities: BackendAuthorityPair,
}

impl DesiredIndexV1 {
    fn validate(&self) -> Result<(), CatalogIntentError> {
        if self.mapping_bytes.is_empty() || self.mapping_bytes.len() > MAX_CATALOG_MAPPING_BYTES {
            return Err(CatalogIntentError::MappingSize(self.mapping_bytes.len()));
        }
        let mapping: Mapping = serde_json::from_slice(&self.mapping_bytes)
            .map_err(|error| CatalogIntentError::Codec(error.to_string()))?;
        mapping
            .validate_current()
            .map_err(|error| CatalogIntentError::Mapping(error.to_string()))?;
        let canonical = serde_json::to_vec(&mapping).map_err(|error| CatalogIntentError::Codec(error.to_string()))?;
        if canonical != self.mapping_bytes {
            return Err(CatalogIntentError::NonCanonicalMapping);
        }
        if mapping.mode != self.mode {
            return Err(CatalogIntentError::MappingMode);
        }
        if self.initial_shards == 0 || self.initial_shards > MAX_INITIAL_SHARDS {
            return Err(CatalogIntentError::InitialShards(self.initial_shards));
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CatalogIntentKindV1 {
    Apply {
        owner: CrdOwnerV1,
        precondition: CatalogApplyPreconditionV1,
        desired: DesiredIndexV1,
    },
    Delete {
        owner: CrdOwnerV1,
        scope: CollectionIndexRef,
        expected_index_revision: IndexRevision,
        expected_management_generation: CatalogManagementGeneration,
    },
    ReleaseToApi {
        owner: CrdOwnerV1,
        scope: CollectionIndexRef,
        expected_index_revision: IndexRevision,
        expected_management_generation: CatalogManagementGeneration,
    },
}

impl CatalogIntentKindV1 {
    fn validate(&self) -> Result<(), CatalogIntentError> {
        match self {
            Self::Apply {
                owner,
                precondition,
                desired,
            } => {
                owner.validate()?;
                precondition.validate()?;
                desired.validate()
            }
            Self::Delete {
                owner,
                expected_index_revision,
                expected_management_generation,
                ..
            }
            | Self::ReleaseToApi {
                owner,
                expected_index_revision,
                expected_management_generation,
                ..
            } => {
                owner.validate()?;
                if expected_index_revision.get() == 0 || expected_management_generation.get() == 0 {
                    return Err(CatalogIntentError::InvalidPrecondition);
                }
                Ok(())
            }
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CatalogIntentV1 {
    pub format_version: u16,
    pub operation_id: [u8; 32],
    pub kind: CatalogIntentKindV1,
    pub semantic_digest: [u8; 32],
}

impl CatalogIntentV1 {
    pub fn validate(&self) -> Result<(), CatalogIntentError> {
        if self.format_version != CATALOG_INTENT_FORMAT_VERSION {
            return Err(CatalogIntentError::UnsupportedVersion(self.format_version));
        }
        if self.operation_id == [0; 32] || self.semantic_digest == [0; 32] {
            return Err(CatalogIntentError::InvalidOperationIdentity);
        }
        self.kind.validate()?;
        if self.semantic_digest != self.derived_semantic_digest()? {
            return Err(CatalogIntentError::SemanticDigestMismatch);
        }
        Ok(())
    }

    pub fn derive_semantic_digest(&mut self) -> Result<(), CatalogIntentError> {
        self.semantic_digest = self.derived_semantic_digest()?;
        Ok(())
    }

    fn derived_semantic_digest(&self) -> Result<[u8; 32], CatalogIntentError> {
        #[derive(Serialize)]
        #[serde(tag = "kind", rename_all = "snake_case")]
        enum StableSemantics<'a> {
            Apply {
                owner: &'a CrdOwnerV1,
                desired: &'a DesiredIndexV1,
            },
            Delete {
                owner: &'a CrdOwnerV1,
                scope: CollectionIndexRef,
            },
            ReleaseToApi {
                owner: &'a CrdOwnerV1,
                scope: CollectionIndexRef,
            },
        }
        let semantics = match &self.kind {
            CatalogIntentKindV1::Apply { owner, desired, .. } => StableSemantics::Apply { owner, desired },
            CatalogIntentKindV1::Delete { owner, scope, .. } => StableSemantics::Delete { owner, scope: *scope },
            CatalogIntentKindV1::ReleaseToApi { owner, scope, .. } => {
                StableSemantics::ReleaseToApi { owner, scope: *scope }
            }
        };
        // FENCING: mutable catalog CAS revisions select one submission
        // attempt; they are not operation semantics.  A response-loss retry
        // after the commit advances those revisions must retain this digest.
        let bytes = serde_json::to_vec(&(self.format_version, self.operation_id, semantics))
            .map_err(|error| CatalogIntentError::Codec(error.to_string()))?;
        let mut hasher = blake3::Hasher::new();
        hasher.update(b"lodestone/p7/catalog-intent/v3");
        hasher.update(&bytes);
        Ok(*hasher.finalize().as_bytes())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CatalogIntentResultKindV1 {
    Applied,
    DeleteAccepted,
    ReleasedToApi,
    Conflict,
    Refused,
    Unavailable,
}

/// Bounded projection of the catalog's one retained CRD-operation receipt.
///
/// This is not itself finalizer authority.  It lets a recovering controller
/// request the exact result by the durable identity instead of deriving a new
/// intent from post-commit catalog revisions.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CatalogOperationReceiptV1 {
    pub format_version: u16,
    pub operation_id: [u8; 32],
    pub semantic_digest: [u8; 32],
    pub kind: CatalogIntentResultKindV1,
    pub completed: bool,
}

impl CatalogOperationReceiptV1 {
    pub fn validate(self) -> Result<(), CatalogIntentError> {
        if self.format_version != CATALOG_OPERATION_RECEIPT_FORMAT_VERSION
            || self.operation_id == [0; 32]
            || self.semantic_digest == [0; 32]
            || !matches!(
                self.kind,
                CatalogIntentResultKindV1::Applied
                    | CatalogIntentResultKindV1::DeleteAccepted
                    | CatalogIntentResultKindV1::ReleasedToApi
            )
        {
            return Err(CatalogIntentError::InvalidObservation);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CatalogIntentResultV1 {
    pub format_version: u16,
    pub operation_id: [u8; 32],
    pub semantic_digest: [u8; 32],
    pub kind: CatalogIntentResultKindV1,
    pub scope: Option<CollectionIndexRef>,
    pub index_revision: Option<IndexRevision>,
    pub management_generation: Option<CatalogManagementGeneration>,
    pub result_digest: [u8; 32],
    pub retryable: bool,
}

impl CatalogIntentResultV1 {
    pub fn validate(&self) -> Result<(), CatalogIntentError> {
        if self.format_version != CATALOG_INTENT_RESULT_FORMAT_VERSION {
            return Err(CatalogIntentError::UnsupportedResultVersion(self.format_version));
        }
        if self.operation_id == [0; 32]
            || self.semantic_digest == [0; 32]
            || self.result_digest == [0; 32]
            || (self.retryable && self.kind != CatalogIntentResultKindV1::Unavailable)
            || self.index_revision.is_some_and(|revision| revision.get() == 0)
            || self
                .management_generation
                .is_some_and(|generation| generation.get() == 0)
        {
            return Err(CatalogIntentError::InvalidResult);
        }
        let succeeded = matches!(
            self.kind,
            CatalogIntentResultKindV1::Applied
                | CatalogIntentResultKindV1::DeleteAccepted
                | CatalogIntentResultKindV1::ReleasedToApi
        );
        if succeeded && (self.scope.is_none() || self.index_revision.is_none() || self.management_generation.is_none())
        {
            return Err(CatalogIntentError::InvalidResult);
        }
        Ok(())
    }
}

/// Read-only projection needed to derive the next fenced intent from catalog
/// authority. It deliberately omits storage internals and mutable Kubernetes
/// facts.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CatalogIndexObservationV1 {
    Absent,
    ApiManaged {
        scope: CollectionIndexRef,
        index_revision: IndexRevision,
        management_generation: CatalogManagementGeneration,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        prepared_owner_uid_digest: Option<KubernetesObjectUidDigest>,
    },
    CrdOwned {
        owner_uid_digest: KubernetesObjectUidDigest,
        observed_generation: CrdGeneration,
        scope: CollectionIndexRef,
        index_revision: IndexRevision,
        management_generation: CatalogManagementGeneration,
        desired_matches: bool,
    },
}

impl CatalogIndexObservationV1 {
    fn validate(self) -> Result<(), CatalogIntentError> {
        match self {
            Self::Absent => Ok(()),
            Self::ApiManaged {
                index_revision,
                management_generation,
                ..
            } => validate_observed_generations(index_revision, management_generation),
            Self::CrdOwned {
                observed_generation,
                index_revision,
                management_generation,
                ..
            } => {
                if observed_generation.get() == 0 {
                    return Err(CatalogIntentError::InvalidObservation);
                }
                validate_observed_generations(index_revision, management_generation)
            }
        }
    }
}

fn validate_observed_generations(
    index_revision: IndexRevision,
    management_generation: CatalogManagementGeneration,
) -> Result<(), CatalogIntentError> {
    if index_revision.get() == 0 || management_generation.get() == 0 {
        Err(CatalogIntentError::InvalidObservation)
    } else {
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CatalogObservationRequestV1 {
    pub format_version: u16,
    pub owner: CrdOwnerV1,
    pub desired: DesiredIndexV1,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operation_id: Option<[u8; 32]>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub semantic_digest: Option<[u8; 32]>,
}

impl CatalogObservationRequestV1 {
    pub fn validate(&self) -> Result<(), CatalogIntentError> {
        if self.format_version != CATALOG_OBSERVATION_FORMAT_VERSION {
            return Err(CatalogIntentError::UnsupportedObservationVersion(self.format_version));
        }
        self.owner.validate()?;
        self.desired.validate()?;
        match (self.operation_id, self.semantic_digest) {
            (None, None) => Ok(()),
            (Some(operation_id), Some(semantic_digest)) if operation_id != [0; 32] && semantic_digest != [0; 32] => {
                Ok(())
            }
            _ => Err(CatalogIntentError::InvalidOperationIdentity),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CatalogObservationResultV1 {
    pub format_version: u16,
    pub index: CatalogIndexObservationV1,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_operation: Option<CatalogOperationReceiptV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operation: Option<CatalogIntentResultV1>,
}

impl CatalogObservationResultV1 {
    pub fn validate_for(&self, request: &CatalogObservationRequestV1) -> Result<(), CatalogIntentError> {
        request.validate()?;
        if self.format_version != CATALOG_OBSERVATION_FORMAT_VERSION {
            return Err(CatalogIntentError::UnsupportedObservationVersion(self.format_version));
        }
        self.index.validate()?;
        if let Some(receipt) = self.last_operation {
            receipt.validate()?;
        }
        if let Some(operation) = &self.operation {
            operation.validate()?;
            if Some(operation.operation_id) != request.operation_id
                || Some(operation.semantic_digest) != request.semantic_digest
                || self.last_operation.is_none_or(|receipt| {
                    !receipt.completed
                        || receipt.operation_id != operation.operation_id
                        || receipt.semantic_digest != operation.semantic_digest
                        || receipt.kind != operation.kind
                })
            {
                return Err(CatalogIntentError::InvalidObservation);
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CatalogRpcRequestKindV1 {
    Apply { intent: CatalogIntentV1 },
    Observe { request: CatalogObservationRequestV1 },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CatalogRpcRequestV1 {
    pub format_version: u16,
    pub request: CatalogRpcRequestKindV1,
}

impl CatalogRpcRequestV1 {
    pub fn validate(&self) -> Result<(), CatalogIntentError> {
        if self.format_version != CATALOG_RPC_FORMAT_VERSION {
            return Err(CatalogIntentError::UnsupportedRpcVersion(self.format_version));
        }
        match &self.request {
            CatalogRpcRequestKindV1::Apply { intent } => intent.validate(),
            CatalogRpcRequestKindV1::Observe { request } => request.validate(),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CatalogRpcResponseKindV1 {
    Applied { result: CatalogIntentResultV1 },
    Observed { result: CatalogObservationResultV1 },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CatalogRpcResponseV1 {
    pub format_version: u16,
    pub response: CatalogRpcResponseKindV1,
}

impl CatalogRpcResponseV1 {
    pub fn validate_for(&self, request: &CatalogRpcRequestV1) -> Result<(), CatalogIntentError> {
        request.validate()?;
        if self.format_version != CATALOG_RPC_FORMAT_VERSION {
            return Err(CatalogIntentError::UnsupportedRpcVersion(self.format_version));
        }
        match (&request.request, &self.response) {
            (CatalogRpcRequestKindV1::Apply { intent }, CatalogRpcResponseKindV1::Applied { result }) => {
                result.validate()?;
                if result.operation_id != intent.operation_id || result.semantic_digest != intent.semantic_digest {
                    return Err(CatalogIntentError::InvalidResult);
                }
                Ok(())
            }
            (CatalogRpcRequestKindV1::Observe { request }, CatalogRpcResponseKindV1::Observed { result }) => {
                result.validate_for(request)
            }
            _ => Err(CatalogIntentError::InvalidResult),
        }
    }
}

pub fn encode_catalog_intent(intent: &CatalogIntentV1) -> Result<Vec<u8>, CatalogIntentError> {
    intent.validate()?;
    encode_bounded(intent)
}

pub fn decode_catalog_intent(bytes: &[u8]) -> Result<CatalogIntentV1, CatalogIntentError> {
    let intent: CatalogIntentV1 = decode_bounded(bytes)?;
    intent.validate()?;
    Ok(intent)
}

pub fn encode_catalog_intent_result(result: &CatalogIntentResultV1) -> Result<Vec<u8>, CatalogIntentError> {
    result.validate()?;
    encode_bounded(result)
}

pub fn decode_catalog_intent_result(bytes: &[u8]) -> Result<CatalogIntentResultV1, CatalogIntentError> {
    let result: CatalogIntentResultV1 = decode_bounded(bytes)?;
    result.validate()?;
    Ok(result)
}

pub fn encode_catalog_rpc_request(request: &CatalogRpcRequestV1) -> Result<Vec<u8>, CatalogIntentError> {
    request.validate()?;
    encode_bounded(request)
}

pub fn decode_catalog_rpc_request(bytes: &[u8]) -> Result<CatalogRpcRequestV1, CatalogIntentError> {
    let request: CatalogRpcRequestV1 = decode_bounded(bytes)?;
    request.validate()?;
    Ok(request)
}

pub fn encode_catalog_rpc_response(response: &CatalogRpcResponseV1) -> Result<Vec<u8>, CatalogIntentError> {
    if response.format_version != CATALOG_RPC_FORMAT_VERSION {
        return Err(CatalogIntentError::UnsupportedRpcVersion(response.format_version));
    }
    encode_bounded(response)
}

pub fn decode_catalog_rpc_response(
    bytes: &[u8],
    request: &CatalogRpcRequestV1,
) -> Result<CatalogRpcResponseV1, CatalogIntentError> {
    let response: CatalogRpcResponseV1 = decode_bounded(bytes)?;
    response.validate_for(request)?;
    Ok(response)
}

fn encode_bounded<T: Serialize>(value: &T) -> Result<Vec<u8>, CatalogIntentError> {
    let bytes = serde_json::to_vec(value).map_err(|error| CatalogIntentError::Codec(error.to_string()))?;
    if bytes.len() > MAX_CATALOG_INTENT_BYTES {
        return Err(CatalogIntentError::MessageTooLarge(bytes.len()));
    }
    Ok(bytes)
}

fn decode_bounded<T>(bytes: &[u8]) -> Result<T, CatalogIntentError>
where
    T: for<'de> Deserialize<'de> + Serialize,
{
    if bytes.len() > MAX_CATALOG_INTENT_BYTES {
        return Err(CatalogIntentError::MessageTooLarge(bytes.len()));
    }
    let value = serde_json::from_slice(bytes).map_err(|error| CatalogIntentError::Codec(error.to_string()))?;
    let canonical = serde_json::to_vec(&value).map_err(|error| CatalogIntentError::Codec(error.to_string()))?;
    if canonical != bytes {
        return Err(CatalogIntentError::NonCanonical);
    }
    Ok(value)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum CatalogIntentError {
    Codec(String),
    InitialShards(u16),
    InvalidCrdGeneration,
    InvalidOperationIdentity,
    InvalidObservation,
    InvalidPrecondition,
    InvalidResult,
    Mapping(String),
    MappingMode,
    MappingSize(usize),
    MessageTooLarge(usize),
    NonCanonical,
    NonCanonicalMapping,
    SemanticDigestMismatch,
    UnsupportedObservationVersion(u16),
    UnsupportedResultVersion(u16),
    UnsupportedRpcVersion(u16),
    UnsupportedVersion(u16),
}

impl Display for CatalogIntentError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "{self:?}")
    }
}

impl Error for CatalogIntentError {}

#[cfg(test)]
mod tests {
    use super::*;
    use lodestone_types::{
        BackendAuthorityId, CollectionId, FieldMapping, FieldType, IndexId, IndexName, SchemaGeneration,
    };

    fn desired(mode: IndexMode) -> Result<DesiredIndexV1, Box<dyn Error>> {
        let mapping = Mapping {
            format_version: lodestone_types::MAPPING_FORMAT_VERSION,
            schema_generation: SchemaGeneration::new(1),
            mode,
            fields: Default::default(),
            extensions: Default::default(),
        };
        Ok(DesiredIndexV1 {
            address: IndexAddress::new(CollectionId::new(1)?, IndexName::new("products")?),
            mode,
            mapping_bytes: serde_json::to_vec(&mapping)?,
            initial_shards: 2,
            authorities: BackendAuthorityPair::new(BackendAuthorityId::new("wal")?, BackendAuthorityId::new("data")?),
        })
    }

    fn owner(generation: u64) -> Result<CrdOwnerV1, Box<dyn Error>> {
        Ok(CrdOwnerV1 {
            owner_uid_digest: KubernetesObjectUidDigest::new([7; 32])?,
            observed_generation: CrdGeneration::new(generation),
        })
    }

    #[test]
    fn all_intent_kinds_round_trip_and_tampering_fails() -> Result<(), Box<dyn Error>> {
        let scope = CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(2)?);
        let kinds = vec![
            CatalogIntentKindV1::Apply {
                owner: owner(1)?,
                precondition: CatalogApplyPreconditionV1::Absent,
                desired: desired(IndexMode::Mutable)?,
            },
            CatalogIntentKindV1::Apply {
                owner: owner(2)?,
                precondition: CatalogApplyPreconditionV1::Owned {
                    scope,
                    expected_index_revision: IndexRevision::new(2),
                    expected_management_generation: CatalogManagementGeneration::new(1),
                },
                desired: desired(IndexMode::Mutable)?,
            },
            CatalogIntentKindV1::Apply {
                owner: owner(3)?,
                precondition: CatalogApplyPreconditionV1::ApiTransferPrepared {
                    scope,
                    expected_index_revision: IndexRevision::new(3),
                    expected_management_generation: CatalogManagementGeneration::new(2),
                },
                desired: desired(IndexMode::Mutable)?,
            },
            CatalogIntentKindV1::Delete {
                owner: owner(4)?,
                scope,
                expected_index_revision: IndexRevision::new(4),
                expected_management_generation: CatalogManagementGeneration::new(3),
            },
            CatalogIntentKindV1::ReleaseToApi {
                owner: owner(5)?,
                scope,
                expected_index_revision: IndexRevision::new(5),
                expected_management_generation: CatalogManagementGeneration::new(4),
            },
        ];
        for (ordinal, kind) in kinds.into_iter().enumerate() {
            let mut intent = CatalogIntentV1 {
                format_version: CATALOG_INTENT_FORMAT_VERSION,
                operation_id: [u8::try_from(ordinal.saturating_add(1))?; 32],
                kind,
                semantic_digest: [1; 32],
            };
            intent.derive_semantic_digest()?;
            let encoded = encode_catalog_intent(&intent)?;
            assert_eq!(decode_catalog_intent(&encoded)?, intent);
            let mut tampered = intent;
            tampered.semantic_digest = [9; 32];
            assert!(matches!(
                decode_catalog_intent(&serde_json::to_vec(&tampered)?),
                Err(CatalogIntentError::SemanticDigestMismatch)
            ));
        }
        Ok(())
    }

    #[test]
    fn large_valid_mapping_remains_reachable_inside_the_catalog_envelope() -> Result<(), Box<dyn Error>> {
        let mut mapping = Mapping::empty(IndexMode::Mutable);
        for ordinal in 0..850_u16 {
            mapping.fields.insert(
                format!("field_{ordinal:04}_{}", "x".repeat(250)),
                FieldMapping {
                    field_type: FieldType::Keyword,
                    stored: false,
                    null_value: None,
                },
            );
        }
        mapping.validate()?;
        let mapping_bytes = serde_json::to_vec(&mapping)?;
        assert!(mapping_bytes.len() > 400 * 1024);
        assert!(mapping_bytes.len() <= MAX_CATALOG_MAPPING_BYTES);
        let mut intent = CatalogIntentV1 {
            format_version: CATALOG_INTENT_FORMAT_VERSION,
            operation_id: [0x71; 32],
            kind: CatalogIntentKindV1::Apply {
                owner: owner(1)?,
                precondition: CatalogApplyPreconditionV1::Absent,
                desired: DesiredIndexV1 {
                    address: IndexAddress::new(CollectionId::new(1)?, IndexName::new("large-mapping")?),
                    mode: IndexMode::Mutable,
                    mapping_bytes,
                    initial_shards: 1,
                    authorities: BackendAuthorityPair::new(
                        BackendAuthorityId::new("wal")?,
                        BackendAuthorityId::new("data")?,
                    ),
                },
            },
            semantic_digest: [1; 32],
        };
        intent.derive_semantic_digest()?;
        let encoded = encode_catalog_intent(&intent)?;

        assert!(encoded.len() <= MAX_CATALOG_INTENT_BYTES);
        assert_eq!(decode_catalog_intent(&encoded)?, intent);
        let value: serde_json::Value = serde_json::from_slice(&encoded)?;
        assert!(value["kind"]["desired"]["mapping_bytes"].is_string());
        Ok(())
    }

    #[test]
    fn api_transfer_requires_both_catalog_cas_generations() -> Result<(), Box<dyn Error>> {
        let scope = CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(2)?);
        let mut intent = CatalogIntentV1 {
            format_version: CATALOG_INTENT_FORMAT_VERSION,
            operation_id: [1; 32],
            kind: CatalogIntentKindV1::Apply {
                owner: owner(1)?,
                precondition: CatalogApplyPreconditionV1::ApiTransferPrepared {
                    scope,
                    expected_index_revision: IndexRevision::new(1),
                    expected_management_generation: CatalogManagementGeneration::new(0),
                },
                desired: desired(IndexMode::Mutable)?,
            },
            semantic_digest: [1; 32],
        };
        intent.derive_semantic_digest()?;
        assert!(matches!(
            encode_catalog_intent(&intent),
            Err(CatalogIntentError::InvalidPrecondition)
        ));
        Ok(())
    }

    #[test]
    fn predecessor_newer_oversized_and_noncanonical_inputs_fail_closed() -> Result<(), Box<dyn Error>> {
        assert!(matches!(
            decode_catalog_intent(&vec![b'x'; MAX_CATALOG_INTENT_BYTES + 1]),
            Err(CatalogIntentError::MessageTooLarge(_))
        ));

        let current_desired = desired(IndexMode::Mutable)?;
        let mut historical_desired = current_desired.clone();
        let mut historical_mapping: serde_json::Value = serde_json::from_slice(&historical_desired.mapping_bytes)?;
        historical_mapping["format_version"] = serde_json::json!(1);
        historical_desired.mapping_bytes = serde_json::to_vec(&historical_mapping)?;
        assert!(matches!(
            historical_desired.validate(),
            Err(CatalogIntentError::Codec(message)) if message.contains("unsupported mapping format version")
        ));
        let mapping_bytes = current_desired.mapping_bytes.clone();
        let mut intent = CatalogIntentV1 {
            format_version: CATALOG_INTENT_FORMAT_VERSION,
            operation_id: [0x41; 32],
            kind: CatalogIntentKindV1::Apply {
                owner: owner(1)?,
                precondition: CatalogApplyPreconditionV1::Absent,
                desired: current_desired,
            },
            semantic_digest: [1; 32],
        };
        intent.derive_semantic_digest()?;

        let mut predecessor_intent = intent.clone();
        predecessor_intent.format_version = 2;
        assert!(matches!(
            decode_catalog_intent(&serde_json::to_vec(&predecessor_intent)?),
            Err(CatalogIntentError::UnsupportedVersion(2))
        ));

        let mut predecessor_byte_shape = serde_json::to_value(intent)?;
        predecessor_byte_shape["kind"]["desired"]["mapping_bytes"] = serde_json::Value::Array(
            mapping_bytes
                .into_iter()
                .map(|byte| serde_json::Value::from(u64::from(byte)))
                .collect(),
        );
        assert!(matches!(
            decode_catalog_intent(&serde_json::to_vec(&predecessor_byte_shape)?),
            Err(CatalogIntentError::Codec(_))
        ));

        let current_observation = CatalogObservationRequestV1 {
            format_version: CATALOG_OBSERVATION_FORMAT_VERSION,
            owner: owner(1)?,
            desired: desired(IndexMode::Mutable)?,
            operation_id: None,
            semantic_digest: None,
        };
        let predecessor_rpc = CatalogRpcRequestV1 {
            format_version: 1,
            request: CatalogRpcRequestKindV1::Observe {
                request: current_observation.clone(),
            },
        };
        assert!(matches!(
            decode_catalog_rpc_request(&serde_json::to_vec(&predecessor_rpc)?),
            Err(CatalogIntentError::UnsupportedRpcVersion(1))
        ));

        let mut predecessor_observation = current_observation;
        predecessor_observation.format_version = 2;
        let current_rpc = CatalogRpcRequestV1 {
            format_version: CATALOG_RPC_FORMAT_VERSION,
            request: CatalogRpcRequestKindV1::Observe {
                request: predecessor_observation,
            },
        };
        assert!(matches!(
            decode_catalog_rpc_request(&serde_json::to_vec(&current_rpc)?),
            Err(CatalogIntentError::UnsupportedObservationVersion(2))
        ));

        let unknown = br#"{"format_version":4,"operation_id":[1],"kind":{"kind":"future"},"semantic_digest":[1]}"#;
        assert!(decode_catalog_intent(unknown).is_err());
        let whitespace = br#" {"format_version":1}"#;
        assert!(decode_catalog_intent(whitespace).is_err());
        Ok(())
    }

    #[test]
    fn catalog_rpc_observation_round_trips_and_exact_checks_operation_identity() -> Result<(), Box<dyn Error>> {
        let request = CatalogObservationRequestV1 {
            format_version: CATALOG_OBSERVATION_FORMAT_VERSION,
            owner: owner(3)?,
            desired: desired(IndexMode::Mutable)?,
            operation_id: Some([0x31; 32]),
            semantic_digest: Some([0x32; 32]),
        };
        let rpc = CatalogRpcRequestV1 {
            format_version: CATALOG_RPC_FORMAT_VERSION,
            request: CatalogRpcRequestKindV1::Observe {
                request: request.clone(),
            },
        };
        let encoded = encode_catalog_rpc_request(&rpc)?;
        assert_eq!(decode_catalog_rpc_request(&encoded)?, rpc);

        let scope = CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(2)?);
        let response = CatalogRpcResponseV1 {
            format_version: CATALOG_RPC_FORMAT_VERSION,
            response: CatalogRpcResponseKindV1::Observed {
                result: CatalogObservationResultV1 {
                    format_version: CATALOG_OBSERVATION_FORMAT_VERSION,
                    index: CatalogIndexObservationV1::CrdOwned {
                        owner_uid_digest: request.owner.owner_uid_digest,
                        observed_generation: request.owner.observed_generation,
                        scope,
                        index_revision: IndexRevision::new(4),
                        management_generation: CatalogManagementGeneration::new(5),
                        desired_matches: true,
                    },
                    last_operation: Some(CatalogOperationReceiptV1 {
                        format_version: CATALOG_OPERATION_RECEIPT_FORMAT_VERSION,
                        operation_id: request.operation_id.ok_or("missing operation")?,
                        semantic_digest: request.semantic_digest.ok_or("missing semantic digest")?,
                        kind: CatalogIntentResultKindV1::Applied,
                        completed: true,
                    }),
                    operation: Some(CatalogIntentResultV1 {
                        format_version: CATALOG_INTENT_RESULT_FORMAT_VERSION,
                        operation_id: request.operation_id.ok_or("missing operation")?,
                        semantic_digest: request.semantic_digest.ok_or("missing semantic digest")?,
                        kind: CatalogIntentResultKindV1::Applied,
                        scope: Some(scope),
                        index_revision: Some(IndexRevision::new(4)),
                        management_generation: Some(CatalogManagementGeneration::new(5)),
                        result_digest: [0x33; 32],
                        retryable: false,
                    }),
                },
            },
        };
        let encoded = encode_catalog_rpc_response(&response)?;
        assert_eq!(decode_catalog_rpc_response(&encoded, &rpc)?, response);

        let mut substituted = response;
        let CatalogRpcResponseKindV1::Observed { result } = &mut substituted.response else {
            return Err("wrong response kind".into());
        };
        let operation = result.operation.as_mut().ok_or("missing operation result")?;
        operation.operation_id = [0x34; 32];
        assert!(matches!(
            substituted.validate_for(&rpc),
            Err(CatalogIntentError::InvalidObservation)
        ));
        Ok(())
    }

    #[test]
    fn observation_requires_paired_nonzero_operation_identity() -> Result<(), Box<dyn Error>> {
        let request = CatalogObservationRequestV1 {
            format_version: CATALOG_OBSERVATION_FORMAT_VERSION,
            owner: owner(1)?,
            desired: desired(IndexMode::Mutable)?,
            operation_id: Some([1; 32]),
            semantic_digest: None,
        };
        assert!(matches!(
            request.validate(),
            Err(CatalogIntentError::InvalidOperationIdentity)
        ));
        Ok(())
    }
}