lodestone-proto 0.0.16

//! Bounded D1 security-watch and control-plane wire contracts.

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

use lodestone_types::{
    ApiKeyId, BackendAuthorityId, CollectionId, GatewayIncarnation, LogicalMillis, NodeRole, NodeToken,
    ObjectIncarnation, QuotaGrantId, SecurityPolicyRevision,
};
use serde::{Deserialize, Serialize};

pub const SECURITY_CONTROL_FORMAT_VERSION: u16 = 2;
pub const AUDIT_OBJECT_ALLOCATION_FORMAT_VERSION: u16 = 1;
pub const SECURITY_OWNER_CONTROL_FORMAT_VERSION: u16 = 1;
pub const SECURITY_ACTIVE_CONTROL_FORMAT_VERSION: u16 = 1;
pub const GATEWAY_READINESS_FORMAT_VERSION: u16 = 1;
pub const MAX_GATEWAY_READINESS_BYTES: usize = 128;
pub const SECURITY_ROOT_WATCH_FORMAT_VERSION: u16 = 2;
pub const MAX_SECURITY_ROOT_WATCH_FRAME_BYTES: usize = 1792 * 1024;
pub const MAX_SECURITY_DELTA_BYTES: usize = 256 * 1024;
pub const MAX_SECURITY_AUTHORIZATION_BUNDLE_BYTES: usize = 1200 * 1024;
pub const MAX_SECURITY_CONTROL_PAYLOAD_BYTES: usize = 1024 * 1024;
pub const MAX_SECURITY_CONTROL_BYTES: usize = 1_440 * 1024;
const _: () = assert!(MAX_SECURITY_CONTROL_BYTES < crate::RPC_MAX_CONTROL_PAYLOAD_BYTES);

/// Current root/delta/bundle request. The exact API-key and Collection
/// identities are parsed before any proof lookup, while random misses remain
/// subject to the Gateway's pre-lookup budget and root-bound negative cache.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityRootWatchRequestV1 {
    pub format_version: u16,
    pub gateway_node: NodeToken,
    pub gateway_incarnation: GatewayIncarnation,
    pub request: SecurityRootWatchRequestKindV1,
}

impl SecurityRootWatchRequestV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_ROOT_WATCH_FORMAT_VERSION {
            return Err(GatewaySecurityWireError::UnsupportedVersion(self.format_version));
        }
        self.request.validate()
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum SecurityRootWatchRequestKindV1 {
    Control {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        current_component_digest: Option<[u8; 32]>,
    },
    AuthorizationBundle {
        api_key_id: ApiKeyId,
        collection_id: CollectionId,
        expected_component_digest: [u8; 32],
    },
}

impl SecurityRootWatchRequestKindV1 {
    fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        match self {
            Self::Control {
                current_component_digest: Some(digest),
            } if *digest == [0; 32] => Err(GatewaySecurityWireError::InvalidWatchRequest),
            Self::Control { .. } => Ok(()),
            Self::AuthorizationBundle {
                api_key_id,
                expected_component_digest,
                ..
            } if api_key_id.get() != 0 && *expected_component_digest != [0; 32] => Ok(()),
            Self::AuthorizationBundle { .. } => Err(GatewaySecurityWireError::InvalidWatchRequest),
        }
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityRootHeartbeatV1 {
    pub format_version: u16,
    pub source: NodeToken,
    pub logical_time: LogicalMillis,
    pub bootstrap_closed: bool,
    pub ordering_revision: u64,
    pub component_digest: [u8; 32],
    pub delta_floor_revision: u64,
    pub delta_head_revision: u64,
}

impl SecurityRootHeartbeatV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_ROOT_WATCH_FORMAT_VERSION {
            return Err(GatewaySecurityWireError::UnsupportedVersion(self.format_version));
        }
        if self.ordering_revision == 0
            || self.component_digest == [0; 32]
            || self.delta_floor_revision == 0
            || self.delta_head_revision < self.delta_floor_revision
            || self.delta_head_revision > self.ordering_revision
        {
            return Err(GatewaySecurityWireError::InvalidHeartbeat);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityRootDeltaV1 {
    pub heartbeat: SecurityRootHeartbeatV1,
    pub predecessor_component_digest: [u8; 32],
    pub successor_component_digest: [u8; 32],
    pub successor_ordering_revision: u64,
    pub delta_digest: [u8; 32],
    #[serde(with = "crate::canonical_base64_bytes")]
    pub delta_bytes: Vec<u8>,
}

impl SecurityRootDeltaV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        self.heartbeat.validate()?;
        if self.predecessor_component_digest == [0; 32]
            || self.successor_component_digest == [0; 32]
            || self.predecessor_component_digest == self.successor_component_digest
            || self.successor_ordering_revision == 0
            || self.successor_ordering_revision > self.heartbeat.ordering_revision
            || self.delta_bytes.is_empty()
            || self.delta_bytes.len() > MAX_SECURITY_DELTA_BYTES
            || self.delta_digest != *blake3::hash(&self.delta_bytes).as_bytes()
        {
            return Err(GatewaySecurityWireError::InvalidSecurityDelta);
        }
        Ok(())
    }
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityAuthorizationBundleV1 {
    pub heartbeat: SecurityRootHeartbeatV1,
    pub api_key_id: ApiKeyId,
    pub collection_id: CollectionId,
    pub continuation_signing_root: [u8; 32],
    pub bundle_digest: [u8; 32],
    #[serde(with = "crate::canonical_base64_bytes")]
    pub bundle_bytes: Vec<u8>,
}

impl fmt::Debug for SecurityAuthorizationBundleV1 {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SecurityAuthorizationBundleV1")
            .field("heartbeat", &self.heartbeat)
            .field("api_key_id", &self.api_key_id)
            .field("collection_id", &self.collection_id)
            .field("continuation_signing_root", &"[REDACTED]")
            .field("bundle_digest", &self.bundle_digest)
            .field("bundle_bytes", &format_args!("{} bytes", self.bundle_bytes.len()))
            .finish()
    }
}

impl Drop for SecurityAuthorizationBundleV1 {
    fn drop(&mut self) {
        self.continuation_signing_root.fill(0);
        self.bundle_bytes.fill(0);
        std::hint::black_box(&mut self.continuation_signing_root);
        std::hint::black_box(&mut self.bundle_bytes);
    }
}

impl SecurityAuthorizationBundleV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        self.heartbeat.validate()?;
        if self.api_key_id.get() == 0
            || self.continuation_signing_root == [0; 32]
            || self.bundle_bytes.is_empty()
            || self.bundle_bytes.len() > MAX_SECURITY_AUTHORIZATION_BUNDLE_BYTES
            || self.bundle_digest != *blake3::hash(&self.bundle_bytes).as_bytes()
        {
            return Err(GatewaySecurityWireError::InvalidAuthorizationBundle);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum SecurityRootWatchEventV1 {
    Heartbeat(SecurityRootHeartbeatV1),
    Delta(SecurityRootDeltaV1),
    AuthorizationBundle(SecurityAuthorizationBundleV1),
    Gap { heartbeat: SecurityRootHeartbeatV1 },
    BundleUnavailable { heartbeat: SecurityRootHeartbeatV1 },
}

impl SecurityRootWatchEventV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        match self {
            Self::Heartbeat(heartbeat) | Self::Gap { heartbeat } | Self::BundleUnavailable { heartbeat } => {
                heartbeat.validate()
            }
            Self::Delta(delta) => delta.validate(),
            Self::AuthorizationBundle(bundle) => bundle.validate(),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityControlCommandV1 {
    pub format_version: u16,
    /// Exact authenticated workload token expected on the P6 mTLS channel.
    /// The serving voter separately constrains the admitted command phases by
    /// the caller's coordinated role; a token alone never grants a generic
    /// security mutation capability.
    pub caller: NodeToken,
    pub submitted_at: LogicalMillis,
    pub command_digest: [u8; 32],
    #[serde(with = "crate::canonical_base64_bytes")]
    pub command_bytes: Vec<u8>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityControlResponseV1 {
    pub format_version: u16,
    pub source: NodeToken,
    pub command_digest: [u8; 32],
    pub committed_revision: SecurityPolicyRevision,
    pub policy_snapshot_id: [u8; 32],
    pub bootstrap_closed: bool,
    #[serde(with = "crate::canonical_base64_bytes")]
    pub receipt_bytes: Vec<u8>,
    /// Exact pending action needed to construct activation after a successful
    /// prepare. Empty after activation or a completed prepare retry.
    #[serde(with = "crate::canonical_base64_bytes")]
    pub pending_action_bytes: Vec<u8>,
}

/// Narrow Gateway request for one catalog-owned control-backend incarnation.
/// The authenticated voter admits only the configured Standard control
/// authority and never exposes an allocator range to the Gateway.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct GatewayAuditObjectAllocationRequestV1 {
    pub format_version: u16,
    pub caller: NodeToken,
    pub request_id: [u8; 32],
    pub backend_authority: BackendAuthorityId,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct GatewayAuditObjectAllocationResponseV1 {
    pub format_version: u16,
    pub source: NodeToken,
    pub request_digest: [u8; 32],
    pub backend_authority: BackendAuthorityId,
    pub object_incarnation: ObjectIncarnation,
}

/// Bounded owner-local active-control query used only during Gateway startup.
/// The response contains no public-security record and no sibling-node state.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityOwnerControlRequestV1 {
    pub format_version: u16,
    pub caller: NodeToken,
    pub gateway_incarnation: GatewayIncarnation,
    pub request_id: [u8; 32],
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityQuotaPredecessorV1 {
    pub grant_id: QuotaGrantId,
    pub gateway_incarnation: GatewayIncarnation,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityOwnerControlResponseV1 {
    pub format_version: u16,
    pub source: NodeToken,
    pub request_digest: [u8; 32],
    pub logical_time: LogicalMillis,
    pub bootstrap_closed: bool,
    pub audit_predecessors: Vec<GatewayIncarnation>,
    pub quota_predecessors: Vec<SecurityQuotaPredecessorV1>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SecurityActiveControlOperationV1 {
    QuotaAllocate,
    QuotaReturn,
    QuotaReclaim,
    AuditHead,
    AuditReserve,
    AuditSettle,
    AuditReconcile,
    AuditBarrier,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SecurityActiveControlOutcomeV1 {
    Success,
    Capacity,
    Conflict,
    OwnerFenced,
    PolicyStale,
    NotQuiescent,
    Refused,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityActiveControlRequestV1 {
    pub format_version: u16,
    pub caller: NodeToken,
    pub gateway_incarnation: GatewayIncarnation,
    pub operation: SecurityActiveControlOperationV1,
    pub request_digest: [u8; 32],
    #[serde(with = "crate::canonical_base64_bytes")]
    pub request_bytes: Vec<u8>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SecurityActiveControlResponseV1 {
    pub format_version: u16,
    pub source: NodeToken,
    pub operation: SecurityActiveControlOperationV1,
    pub outcome: SecurityActiveControlOutcomeV1,
    pub request_digest: [u8; 32],
    pub response_digest: [u8; 32],
    #[serde(with = "crate::canonical_base64_bytes")]
    pub response_bytes: Vec<u8>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum GatewayReadinessStateV1 {
    BootstrapReady,
    ServingReady,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct GatewayReadinessResponseV1 {
    pub format_version: u16,
    pub state: GatewayReadinessStateV1,
}

impl GatewayReadinessResponseV1 {
    #[must_use]
    pub const fn new(state: GatewayReadinessStateV1) -> Self {
        Self {
            format_version: GATEWAY_READINESS_FORMAT_VERSION,
            state,
        }
    }

    fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != GATEWAY_READINESS_FORMAT_VERSION {
            return Err(GatewaySecurityWireError::UnsupportedVersion(self.format_version));
        }
        Ok(())
    }
}

impl SecurityControlResponseV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_CONTROL_FORMAT_VERSION {
            return Err(GatewaySecurityWireError::UnsupportedVersion(self.format_version));
        }
        if self.command_digest == [0; 32]
            || self.policy_snapshot_id == [0; 32]
            || self.receipt_bytes.is_empty()
            || self.receipt_bytes.len() > MAX_SECURITY_CONTROL_PAYLOAD_BYTES
            || self.pending_action_bytes.len() > MAX_SECURITY_CONTROL_PAYLOAD_BYTES
        {
            return Err(GatewaySecurityWireError::InvalidControlResponse);
        }
        Ok(())
    }
}

impl SecurityControlCommandV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_CONTROL_FORMAT_VERSION {
            return Err(GatewaySecurityWireError::UnsupportedVersion(self.format_version));
        }
        if self.command_bytes.is_empty()
            || self.command_bytes.len() > MAX_SECURITY_CONTROL_PAYLOAD_BYTES
            || self.command_digest != *blake3::hash(&self.command_bytes).as_bytes()
        {
            return Err(GatewaySecurityWireError::InvalidControlCommand);
        }
        Ok(())
    }
}

impl GatewayAuditObjectAllocationRequestV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != AUDIT_OBJECT_ALLOCATION_FORMAT_VERSION {
            return Err(GatewaySecurityWireError::UnsupportedVersion(self.format_version));
        }
        if self.request_id == [0; 32] {
            return Err(GatewaySecurityWireError::InvalidControlCommand);
        }
        Ok(())
    }
}

impl GatewayAuditObjectAllocationResponseV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != AUDIT_OBJECT_ALLOCATION_FORMAT_VERSION {
            return Err(GatewaySecurityWireError::UnsupportedVersion(self.format_version));
        }
        if self.request_digest == [0; 32] {
            return Err(GatewaySecurityWireError::InvalidControlResponse);
        }
        Ok(())
    }
}

impl SecurityOwnerControlRequestV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_OWNER_CONTROL_FORMAT_VERSION || self.request_id == [0; 32] {
            return Err(GatewaySecurityWireError::InvalidControlCommand);
        }
        Ok(())
    }
}

impl SecurityOwnerControlResponseV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_OWNER_CONTROL_FORMAT_VERSION
            || self.request_digest == [0; 32]
            || self.audit_predecessors.len() > 16
            || self.quota_predecessors.len() > 4_096
        {
            return Err(GatewaySecurityWireError::InvalidControlResponse);
        }
        let duplicate_audit = self
            .audit_predecessors
            .iter()
            .enumerate()
            .any(|(index, candidate)| self.audit_predecessors[..index].contains(candidate));
        let duplicate_quota = self.quota_predecessors.iter().enumerate().any(|(index, candidate)| {
            self.quota_predecessors[..index]
                .iter()
                .any(|prior| prior.grant_id == candidate.grant_id)
        });
        if duplicate_audit || duplicate_quota {
            return Err(GatewaySecurityWireError::InvalidControlResponse);
        }
        Ok(())
    }
}

impl SecurityActiveControlRequestV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_ACTIVE_CONTROL_FORMAT_VERSION
            || self.request_bytes.is_empty()
            || self.request_bytes.len() > MAX_SECURITY_CONTROL_PAYLOAD_BYTES
            || self.request_digest != *blake3::hash(&self.request_bytes).as_bytes()
        {
            return Err(GatewaySecurityWireError::InvalidControlCommand);
        }
        Ok(())
    }
}

impl SecurityActiveControlResponseV1 {
    pub fn validate(&self) -> Result<(), GatewaySecurityWireError> {
        if self.format_version != SECURITY_ACTIVE_CONTROL_FORMAT_VERSION
            || self.request_digest == [0; 32]
            || self.response_bytes.len() > MAX_SECURITY_CONTROL_PAYLOAD_BYTES
            || self.response_digest != *blake3::hash(&self.response_bytes).as_bytes()
            || (self.outcome == SecurityActiveControlOutcomeV1::Success) == self.response_bytes.is_empty()
        {
            return Err(GatewaySecurityWireError::InvalidControlResponse);
        }
        Ok(())
    }
}

pub fn encode_security_root_watch_request_v1(
    request: &SecurityRootWatchRequestV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    request.validate()?;
    encode_bounded_canonical(request, MAX_SECURITY_ROOT_WATCH_FRAME_BYTES)
}

pub fn decode_security_root_watch_request_v1(
    bytes: &[u8],
) -> Result<SecurityRootWatchRequestV1, GatewaySecurityWireError> {
    let request: SecurityRootWatchRequestV1 = decode_bounded_canonical(bytes, MAX_SECURITY_ROOT_WATCH_FRAME_BYTES)?;
    request.validate()?;
    Ok(request)
}

pub fn encode_security_root_watch_event_v1(
    event: &SecurityRootWatchEventV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    event.validate()?;
    encode_bounded_canonical(event, MAX_SECURITY_ROOT_WATCH_FRAME_BYTES)
}

pub fn decode_security_root_watch_event_v1(bytes: &[u8]) -> Result<SecurityRootWatchEventV1, GatewaySecurityWireError> {
    let event: SecurityRootWatchEventV1 = decode_bounded_canonical(bytes, MAX_SECURITY_ROOT_WATCH_FRAME_BYTES)?;
    event.validate()?;
    Ok(event)
}

pub fn encode_security_control_command_v1(
    command: &SecurityControlCommandV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    command.validate()?;
    encode_bounded_canonical(command, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_security_control_command_v1(bytes: &[u8]) -> Result<SecurityControlCommandV1, GatewaySecurityWireError> {
    let command: SecurityControlCommandV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    command.validate()?;
    Ok(command)
}

pub fn encode_security_control_response_v1(
    response: &SecurityControlResponseV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    response.validate()?;
    encode_bounded_canonical(response, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_security_control_response_v1(
    bytes: &[u8],
) -> Result<SecurityControlResponseV1, GatewaySecurityWireError> {
    let response: SecurityControlResponseV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    response.validate()?;
    Ok(response)
}

pub fn encode_gateway_audit_object_allocation_request_v1(
    request: &GatewayAuditObjectAllocationRequestV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    request.validate()?;
    encode_bounded_canonical(request, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_gateway_audit_object_allocation_request_v1(
    bytes: &[u8],
) -> Result<GatewayAuditObjectAllocationRequestV1, GatewaySecurityWireError> {
    let request: GatewayAuditObjectAllocationRequestV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    request.validate()?;
    Ok(request)
}

pub fn encode_gateway_audit_object_allocation_response_v1(
    response: &GatewayAuditObjectAllocationResponseV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    response.validate()?;
    encode_bounded_canonical(response, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_gateway_audit_object_allocation_response_v1(
    bytes: &[u8],
) -> Result<GatewayAuditObjectAllocationResponseV1, GatewaySecurityWireError> {
    let response: GatewayAuditObjectAllocationResponseV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    response.validate()?;
    Ok(response)
}

pub fn encode_security_owner_control_request_v1(
    request: &SecurityOwnerControlRequestV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    request.validate()?;
    encode_bounded_canonical(request, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_security_owner_control_request_v1(
    bytes: &[u8],
) -> Result<SecurityOwnerControlRequestV1, GatewaySecurityWireError> {
    let request: SecurityOwnerControlRequestV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    request.validate()?;
    Ok(request)
}

pub fn encode_security_owner_control_response_v1(
    response: &SecurityOwnerControlResponseV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    response.validate()?;
    encode_bounded_canonical(response, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_security_owner_control_response_v1(
    bytes: &[u8],
) -> Result<SecurityOwnerControlResponseV1, GatewaySecurityWireError> {
    let response: SecurityOwnerControlResponseV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    response.validate()?;
    Ok(response)
}

pub fn encode_security_active_control_request_v1(
    request: &SecurityActiveControlRequestV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    request.validate()?;
    encode_bounded_canonical(request, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_security_active_control_request_v1(
    bytes: &[u8],
) -> Result<SecurityActiveControlRequestV1, GatewaySecurityWireError> {
    let request: SecurityActiveControlRequestV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    request.validate()?;
    Ok(request)
}

pub fn encode_security_active_control_response_v1(
    response: &SecurityActiveControlResponseV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    response.validate()?;
    encode_bounded_canonical(response, MAX_SECURITY_CONTROL_BYTES)
}

pub fn decode_security_active_control_response_v1(
    bytes: &[u8],
) -> Result<SecurityActiveControlResponseV1, GatewaySecurityWireError> {
    let response: SecurityActiveControlResponseV1 = decode_bounded_canonical(bytes, MAX_SECURITY_CONTROL_BYTES)?;
    response.validate()?;
    Ok(response)
}

pub fn encode_gateway_readiness_response_v1(
    response: &GatewayReadinessResponseV1,
) -> Result<Vec<u8>, GatewaySecurityWireError> {
    response.validate()?;
    encode_bounded_canonical(response, MAX_GATEWAY_READINESS_BYTES)
}

pub fn decode_gateway_readiness_response_v1(
    bytes: &[u8],
) -> Result<GatewayReadinessResponseV1, GatewaySecurityWireError> {
    let response: GatewayReadinessResponseV1 = decode_bounded_canonical(bytes, MAX_GATEWAY_READINESS_BYTES)?;
    response.validate()?;
    Ok(response)
}

/// Applies the exact role-specific P7 health-payload contract. Existing P6
/// roles retain their empty success body; Packet J Gateways must explicitly
/// report `ServingReady`, never merely `BootstrapReady`.
#[must_use]
pub fn role_health_payload_is_ready(role: NodeRole, payload: &[u8]) -> bool {
    if role == NodeRole::Gateway {
        return decode_gateway_readiness_response_v1(payload)
            .is_ok_and(|readiness| readiness.state == GatewayReadinessStateV1::ServingReady);
    }
    payload.is_empty()
}

fn encode_bounded_canonical(value: &impl Serialize, maximum_bytes: usize) -> Result<Vec<u8>, GatewaySecurityWireError> {
    let bytes = serde_json::to_vec(value).map_err(|error| GatewaySecurityWireError::Codec(error.to_string()))?;
    if bytes.is_empty() || bytes.len() > maximum_bytes {
        return Err(GatewaySecurityWireError::FrameTooLarge(bytes.len()));
    }
    Ok(bytes)
}

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

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum GatewaySecurityWireError {
    UnsupportedVersion(u16),
    InvalidWatchRequest,
    InvalidProjectionPage,
    ProjectionDigestMismatch,
    InvalidHeartbeat,
    InvalidGap,
    InvalidSecurityDelta,
    InvalidAuthorizationBundle,
    InvalidControlCommand,
    InvalidControlResponse,
    FrameTooLarge(usize),
    NonCanonical,
    Codec(String),
}

impl Display for GatewaySecurityWireError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => write!(formatter, "unsupported security wire version {version}"),
            Self::InvalidWatchRequest => formatter.write_str("invalid security projection watch request"),
            Self::InvalidProjectionPage => formatter.write_str("invalid security projection page"),
            Self::ProjectionDigestMismatch => formatter.write_str("security projection page digest mismatch"),
            Self::InvalidHeartbeat => formatter.write_str("invalid security projection heartbeat"),
            Self::InvalidGap => formatter.write_str("invalid security projection gap"),
            Self::InvalidSecurityDelta => formatter.write_str("invalid security root delta"),
            Self::InvalidAuthorizationBundle => formatter.write_str("invalid security authorization bundle"),
            Self::InvalidControlCommand => formatter.write_str("invalid security control command"),
            Self::InvalidControlResponse => formatter.write_str("invalid security control response"),
            Self::FrameTooLarge(size) => write!(formatter, "security wire frame size {size} exceeds its bound"),
            Self::NonCanonical => formatter.write_str("security wire frame is not canonical"),
            Self::Codec(error) => write!(formatter, "security wire codec failed: {error}"),
        }
    }
}

impl Error for GatewaySecurityWireError {}

#[cfg(test)]
mod tests {
    use lodestone_types::{GatewayIncarnation, NodeId, NodeIncarnation, RecoveryEra};

    use super::*;

    fn source() -> Result<NodeToken, Box<dyn Error>> {
        Ok(NodeToken::new(
            RecoveryEra::new(4)?,
            NodeId::new(1),
            NodeIncarnation::new(2)?,
        )?)
    }

    #[test]
    fn security_control_allocation_and_readiness_frames_fail_closed() -> Result<(), Box<dyn Error>> {
        let control = SecurityControlCommandV1 {
            format_version: SECURITY_CONTROL_FORMAT_VERSION,
            caller: source()?,
            submitted_at: LogicalMillis::new(3),
            command_digest: *blake3::hash(b"command").as_bytes(),
            command_bytes: b"command".to_vec(),
        };
        let encoded_control = encode_security_control_command_v1(&control)?;
        assert_eq!(decode_security_control_command_v1(&encoded_control)?, control);
        let mut noncanonical = encoded_control;
        noncanonical.push(b' ');
        assert_eq!(
            decode_security_control_command_v1(&noncanonical),
            Err(GatewaySecurityWireError::NonCanonical)
        );

        let response = SecurityControlResponseV1 {
            format_version: SECURITY_CONTROL_FORMAT_VERSION,
            source: source()?,
            command_digest: control.command_digest,
            committed_revision: SecurityPolicyRevision::new(3),
            policy_snapshot_id: [7; 32],
            bootstrap_closed: true,
            receipt_bytes: b"receipt".to_vec(),
            pending_action_bytes: Vec::new(),
        };
        let encoded_response = encode_security_control_response_v1(&response)?;
        assert_eq!(decode_security_control_response_v1(&encoded_response)?, response);

        let allocation = GatewayAuditObjectAllocationRequestV1 {
            format_version: AUDIT_OBJECT_ALLOCATION_FORMAT_VERSION,
            caller: source()?,
            request_id: [11; 32],
            backend_authority: BackendAuthorityId::new("standard-control")?,
        };
        let encoded_allocation = encode_gateway_audit_object_allocation_request_v1(&allocation)?;
        assert_eq!(
            decode_gateway_audit_object_allocation_request_v1(&encoded_allocation)?,
            allocation
        );
        let allocation_response = GatewayAuditObjectAllocationResponseV1 {
            format_version: AUDIT_OBJECT_ALLOCATION_FORMAT_VERSION,
            source: source()?,
            request_digest: *blake3::hash(&encoded_allocation).as_bytes(),
            backend_authority: BackendAuthorityId::new("standard-control")?,
            object_incarnation: ObjectIncarnation::new(9)?,
        };
        let encoded_allocation_response = encode_gateway_audit_object_allocation_response_v1(&allocation_response)?;
        assert_eq!(
            decode_gateway_audit_object_allocation_response_v1(&encoded_allocation_response)?,
            allocation_response
        );

        let readiness = GatewayReadinessResponseV1::new(GatewayReadinessStateV1::BootstrapReady);
        let encoded_readiness = encode_gateway_readiness_response_v1(&readiness)?;
        assert_eq!(decode_gateway_readiness_response_v1(&encoded_readiness)?, readiness);
        assert!(!role_health_payload_is_ready(NodeRole::Gateway, &encoded_readiness));
        let serving = encode_gateway_readiness_response_v1(&GatewayReadinessResponseV1::new(
            GatewayReadinessStateV1::ServingReady,
        ))?;
        assert!(role_health_payload_is_ready(NodeRole::Gateway, &serving));
        assert!(!role_health_payload_is_ready(NodeRole::Gateway, &[]));
        assert!(role_health_payload_is_ready(NodeRole::Indexer, &[]));
        assert!(!role_health_payload_is_ready(NodeRole::Indexer, &serving));
        let mut newer_readiness = readiness;
        newer_readiness.format_version += 1;
        assert_eq!(
            encode_gateway_readiness_response_v1(&newer_readiness),
            Err(GatewaySecurityWireError::UnsupportedVersion(2))
        );
        Ok(())
    }

    #[test]
    fn security_root_watch_round_trips_bounded_control_delta_and_bundle() -> Result<(), Box<dyn Error>> {
        let heartbeat = SecurityRootHeartbeatV1 {
            format_version: SECURITY_ROOT_WATCH_FORMAT_VERSION,
            source: source()?,
            logical_time: LogicalMillis::new(10),
            bootstrap_closed: true,
            ordering_revision: 9,
            component_digest: [4; 32],
            delta_floor_revision: 3,
            delta_head_revision: 8,
        };
        let request = SecurityRootWatchRequestV1 {
            format_version: SECURITY_ROOT_WATCH_FORMAT_VERSION,
            gateway_node: source()?,
            gateway_incarnation: GatewayIncarnation::new([3; 32])?,
            request: SecurityRootWatchRequestKindV1::Control {
                current_component_digest: Some([2; 32]),
            },
        };
        let request_bytes = encode_security_root_watch_request_v1(&request)?;
        assert_eq!(decode_security_root_watch_request_v1(&request_bytes)?, request);

        let delta_bytes = br#"{"delta":1}"#.to_vec();
        let delta = SecurityRootWatchEventV1::Delta(SecurityRootDeltaV1 {
            heartbeat,
            predecessor_component_digest: [2; 32],
            successor_component_digest: [4; 32],
            successor_ordering_revision: 8,
            delta_digest: *blake3::hash(&delta_bytes).as_bytes(),
            delta_bytes,
        });
        let encoded = encode_security_root_watch_event_v1(&delta)?;
        assert_eq!(decode_security_root_watch_event_v1(&encoded)?, delta);

        let bundle_bytes = br#"{"bundle":1}"#.to_vec();
        let bundle = SecurityRootWatchEventV1::AuthorizationBundle(SecurityAuthorizationBundleV1 {
            heartbeat,
            api_key_id: ApiKeyId::new(7),
            collection_id: CollectionId::new(8)?,
            continuation_signing_root: [6; 32],
            bundle_digest: *blake3::hash(&bundle_bytes).as_bytes(),
            bundle_bytes,
        });
        let encoded = encode_security_root_watch_event_v1(&bundle)?;
        assert_eq!(decode_security_root_watch_event_v1(&encoded)?, bundle);
        Ok(())
    }

    #[test]
    fn security_active_and_owner_control_round_trip_exact_bounded_bytes() -> Result<(), Box<dyn Error>> {
        let caller = source()?;
        let owner_request = SecurityOwnerControlRequestV1 {
            format_version: SECURITY_OWNER_CONTROL_FORMAT_VERSION,
            caller,
            gateway_incarnation: GatewayIncarnation::new([7; 32])?,
            request_id: [8; 32],
        };
        let encoded = encode_security_owner_control_request_v1(&owner_request)?;
        assert_eq!(decode_security_owner_control_request_v1(&encoded)?, owner_request);

        let owner_response = SecurityOwnerControlResponseV1 {
            format_version: SECURITY_OWNER_CONTROL_FORMAT_VERSION,
            source: caller,
            request_digest: *blake3::hash(&encoded).as_bytes(),
            logical_time: LogicalMillis::new(9),
            bootstrap_closed: true,
            audit_predecessors: vec![GatewayIncarnation::new([9; 32])?],
            quota_predecessors: vec![SecurityQuotaPredecessorV1 {
                grant_id: QuotaGrantId::new(10),
                gateway_incarnation: GatewayIncarnation::new([10; 32])?,
            }],
        };
        let encoded = encode_security_owner_control_response_v1(&owner_response)?;
        assert_eq!(decode_security_owner_control_response_v1(&encoded)?, owner_response);

        let request_bytes = br#"{"operation_id":1}"#.to_vec();
        let active_request = SecurityActiveControlRequestV1 {
            format_version: SECURITY_ACTIVE_CONTROL_FORMAT_VERSION,
            caller,
            gateway_incarnation: GatewayIncarnation::new([11; 32])?,
            operation: SecurityActiveControlOperationV1::QuotaReturn,
            request_digest: *blake3::hash(&request_bytes).as_bytes(),
            request_bytes,
        };
        let encoded = encode_security_active_control_request_v1(&active_request)?;
        assert_eq!(decode_security_active_control_request_v1(&encoded)?, active_request);

        let response_bytes = br#"{"grant_id":10}"#.to_vec();
        let active_response = SecurityActiveControlResponseV1 {
            format_version: SECURITY_ACTIVE_CONTROL_FORMAT_VERSION,
            source: caller,
            operation: SecurityActiveControlOperationV1::QuotaReturn,
            outcome: SecurityActiveControlOutcomeV1::Success,
            request_digest: active_request.request_digest,
            response_digest: *blake3::hash(&response_bytes).as_bytes(),
            response_bytes,
        };
        let encoded = encode_security_active_control_response_v1(&active_response)?;
        assert_eq!(decode_security_active_control_response_v1(&encoded)?, active_response);

        let mut invalid = active_request;
        invalid.request_bytes.clear();
        invalid.request_digest = *blake3::hash(&invalid.request_bytes).as_bytes();
        assert_eq!(
            encode_security_active_control_request_v1(&invalid),
            Err(GatewaySecurityWireError::InvalidControlCommand)
        );
        Ok(())
    }

    #[test]
    fn security_control_maximum_binary_fields_fit_transport_bounds() -> Result<(), Box<dyn Error>> {
        let command_bytes = vec![u8::MAX; MAX_SECURITY_CONTROL_PAYLOAD_BYTES];
        let command = SecurityControlCommandV1 {
            format_version: SECURITY_CONTROL_FORMAT_VERSION,
            caller: source()?,
            submitted_at: LogicalMillis::new(3),
            command_digest: *blake3::hash(&command_bytes).as_bytes(),
            command_bytes,
        };
        let encoded_command = encode_security_control_command_v1(&command)?;
        assert!(encoded_command.len() <= MAX_SECURITY_CONTROL_BYTES);
        assert!(encoded_command.len() < crate::RPC_MAX_CONTROL_PAYLOAD_BYTES);
        assert_eq!(decode_security_control_command_v1(&encoded_command)?, command);

        Ok(())
    }
}