polyc-state-connect 2026.8.3

//! Explicit durable-ingress contract encoding.

use std::time::Duration;

use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    command::{
        CommandEnvelope, CommandMetadata, CommandScope, FencingToken, Precondition, ResourceBounds,
    },
    consistency::Consistency,
    deadline::MonotonicInstant,
    digest::ContentDigest,
    error::StateError,
    id::{
        AggregateId, Audience, CommandId, NamespaceId, OperationFamily, OwnerId, PartitionId,
        ProtocolVersion, Purpose,
    },
    ingress::{
        AuthenticatedEnvelope, ClaimIngress, DispatchId, EdgeId, EnvelopeNonce, ExternalSubject,
        InboxDepth, IngressDecision, IngressItem, IngressPayload, IngressStatus, PayloadKind,
        Provenance, ReadInboxDepth, ReadIngressItem, ReceiveIngress, RecordIngressDecision,
        RejectionCode, SourceEventId, SourceIdentity,
    },
    journal::JournalRecordDraft,
    revision::{CommitRoot, JournalHead, JournalPosition, Revision},
};

use crate::wire::{Kernel, consistency, fixed_bytes, malformed, required};

pub(crate) fn scope_to_wire(scope: &CommandScope) -> pb::IngressScope {
    pb::IngressScope {
        aggregate: scope.aggregate().as_str().to_owned(),
        partition: scope.partition().as_str().to_owned(),
        namespace: scope.namespace().as_str().to_owned(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn scope_from_wire(value: pb::IngressScope) -> CommandScope {
    CommandScope::new(
        AggregateId::new(value.aggregate),
        PartitionId::new(value.partition),
        NamespaceId::new(value.namespace),
    )
}

fn metadata_to_wire(value: &CommandMetadata) -> pb::IngressCommandMetadata {
    pb::IngressCommandMetadata {
        command_id: value.command_id().as_str().to_owned(),
        digest: value.digest().as_bytes().to_vec(),
        scope: buffa::MessageField::some(scope_to_wire(value.scope())),
        purpose: value.envelope().purpose().as_str().to_owned(),
        command_audience: value.envelope().audience().as_str().to_owned(),
        max_payload_bytes: value.envelope().bounds().max_payload_bytes(),
        max_records: value.envelope().bounds().max_records(),
        precondition: buffa::MessageField::some(pb::Precondition::from(Kernel(
            value.precondition(),
        ))),
        fence: value.fence().map(FencingToken::get),
        protocol_version: value.protocol_version().get(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn metadata_from_wire(value: pb::IngressCommandMetadata) -> Result<CommandMetadata, StateError> {
    let mut metadata = CommandMetadata::new(
        CommandId::new(value.command_id),
        OperationFamily::new(polyc_state::ingress::FAMILY),
        ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>(
            "digest",
            &value.digest,
        )?),
        scope_from_wire(required(
            "scope",
            "an ingress command names its inbox",
            value.scope,
        )?),
        CommandEnvelope::new(
            Purpose::new(value.purpose),
            Audience::new(value.command_audience),
            ResourceBounds::new(value.max_payload_bytes, value.max_records),
        ),
    )
    .with_precondition(
        Kernel::<Precondition>::try_from(required(
            "precondition",
            "an ingress command declares its precondition",
            value.precondition,
        )?)?
        .into_inner(),
    )
    .with_protocol_version(ProtocolVersion::new(value.protocol_version));
    if let Some(fence) = value.fence {
        metadata = metadata.with_fence(FencingToken::new(fence));
    }
    Ok(metadata)
}

pub(crate) fn source_to_wire(value: &SourceIdentity) -> pb::IngressSourceIdentity {
    use pb::__buffa::oneof::ingress_source_identity::Event;
    let event = match value.event() {
        SourceEventId::Reported(id) => Event::Reported(id.clone()),
        SourceEventId::Derived(digest) => Event::Derived(digest.as_bytes().to_vec()),
    };
    pb::IngressSourceIdentity {
        edge: value.edge().as_str().to_owned(),
        event: Some(event),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn source_from_wire(
    value: pb::IngressSourceIdentity,
) -> Result<SourceIdentity, StateError> {
    use pb::__buffa::oneof::ingress_source_identity::Event;
    let event = match value.event {
        Some(Event::Reported(id)) => SourceEventId::reported(id),
        Some(Event::Derived(bytes)) => {
            SourceEventId::derived_from(ContentDigest::from_bytes(fixed_bytes::<
                { ContentDigest::LEN },
            >(
                "source.derived", &bytes
            )?))
        }
        None => {
            return Err(malformed(
                "source.event",
                "a source identity names its event",
            ));
        }
    };
    Ok(SourceIdentity::new(EdgeId::new(value.edge), event))
}

fn provenance_to_wire(value: &Provenance) -> pb::IngressProvenance {
    pb::IngressProvenance {
        external_subject: value.subject().as_str().to_owned(),
        envelope_nonce: value.envelope().nonce().as_str().to_owned(),
        remaining_validity_nanos: polyc_state::ingress::nanos_of(
            value.envelope().remaining_validity(),
        ),
        admitted_conversation_visibility: value.envelope().admitted_visibility().get(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn provenance_from_wire(value: pb::IngressProvenance) -> Provenance {
    Provenance::new(
        ExternalSubject::new(value.external_subject),
        AuthenticatedEnvelope::new(
            EnvelopeNonce::new(value.envelope_nonce),
            Duration::from_nanos(value.remaining_validity_nanos),
            polyc_state::ingress::AdmittedVisibility::new(value.admitted_conversation_visibility),
        ),
    )
}

fn payload_to_wire(value: &IngressPayload) -> pb::IngressPayload {
    pb::IngressPayload {
        kind: value.kind().as_str().to_owned(),
        payload: value.bytes().to_vec(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn payload_from_wire(value: pb::IngressPayload) -> IngressPayload {
    IngressPayload::new(PayloadKind::new(value.kind), value.payload)
}

fn changes_to_wire(changes: &[JournalRecordDraft]) -> Vec<pb::JournalRecordDraft> {
    changes
        .iter()
        .map(|change| pb::JournalRecordDraft::from(Kernel(change)))
        .collect()
}

fn changes_from_wire(
    changes: Vec<pb::JournalRecordDraft>,
) -> Result<Vec<JournalRecordDraft>, StateError> {
    changes
        .into_iter()
        .map(|change| Kernel::<JournalRecordDraft>::try_from(change).map(Kernel::into_inner))
        .collect()
}

pub(crate) fn receive_to_wire(command: &ReceiveIngress) -> pb::IngressReceiveCommand {
    let content_identity = (command.content_identity() != command.payload())
        .then(|| payload_to_wire(command.content_identity()));
    pb::IngressReceiveCommand {
        metadata: buffa::MessageField::some(metadata_to_wire(command.metadata())),
        source: buffa::MessageField::some(source_to_wire(command.source())),
        provenance: buffa::MessageField::some(provenance_to_wire(command.provenance())),
        payload: buffa::MessageField::some(payload_to_wire(command.payload())),
        lifetime_nanos: polyc_state::ingress::nanos_of(command.lifetime()),
        changes: changes_to_wire(command.changes()),
        content_identity: content_identity.into(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn receive_from_wire(
    value: pb::IngressReceiveCommand,
) -> Result<ReceiveIngress, StateError> {
    let content_identity = value.content_identity.into_option().map(payload_from_wire);
    let command = ReceiveIngress::new(
        metadata_from_wire(required(
            "metadata",
            "a receive carries metadata",
            value.metadata,
        )?)?,
        source_from_wire(required(
            "source",
            "a receive names its source",
            value.source,
        )?)?,
        provenance_from_wire(required(
            "provenance",
            "a receive carries authenticated provenance",
            value.provenance,
        )?),
        payload_from_wire(required(
            "payload",
            "a receive carries its payload",
            value.payload,
        )?),
        Duration::from_nanos(value.lifetime_nanos),
        changes_from_wire(value.changes)?,
    );
    Ok(match content_identity {
        Some(content_identity) => command.with_content_identity(content_identity),
        None => command,
    })
}

pub(crate) fn claim_to_wire(command: &ClaimIngress) -> pb::IngressClaimCommand {
    pb::IngressClaimCommand {
        metadata: buffa::MessageField::some(metadata_to_wire(command.metadata())),
        owner: command.owner().as_str().to_owned(),
        lease_nanos: polyc_state::ingress::nanos_of(command.lease()),
        changes: changes_to_wire(command.changes()),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn claim_from_wire(value: pb::IngressClaimCommand) -> Result<ClaimIngress, StateError> {
    Ok(ClaimIngress::new(
        metadata_from_wire(required(
            "metadata",
            "a claim carries metadata",
            value.metadata,
        )?)?,
        OwnerId::new(value.owner),
        Duration::from_nanos(value.lease_nanos),
        changes_from_wire(value.changes)?,
    ))
}

fn decision_to_wire(value: &IngressDecision) -> pb::IngressDecision {
    use pb::__buffa::oneof::ingress_decision::Decision;
    let decision = match value {
        IngressDecision::Rejected { code } => Decision::from(pb::IngressRejected {
            code: code.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        IngressDecision::Admitted { dispatch } => Decision::from(pb::IngressAdmitted {
            dispatch: dispatch.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
    };
    pb::IngressDecision {
        decision: Some(decision),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn decision_from_wire(value: pb::IngressDecision) -> Result<IngressDecision, StateError> {
    use pb::__buffa::oneof::ingress_decision::Decision;
    match value.decision {
        Some(Decision::Rejected(value)) => Ok(IngressDecision::Rejected {
            code: RejectionCode::new(value.code),
        }),
        Some(Decision::Admitted(value)) => Ok(IngressDecision::Admitted {
            dispatch: DispatchId::new(value.dispatch),
        }),
        None => Err(malformed("decision", "a decision declares its outcome")),
    }
}

pub(crate) fn decide_to_wire(command: &RecordIngressDecision) -> pb::IngressDecisionCommand {
    pb::IngressDecisionCommand {
        metadata: buffa::MessageField::some(metadata_to_wire(command.metadata())),
        source: buffa::MessageField::some(source_to_wire(command.source())),
        decision: buffa::MessageField::some(decision_to_wire(command.decision())),
        changes: changes_to_wire(command.changes()),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn decide_from_wire(
    value: pb::IngressDecisionCommand,
) -> Result<RecordIngressDecision, StateError> {
    Ok(RecordIngressDecision::new(
        metadata_from_wire(required(
            "metadata",
            "a decision carries metadata",
            value.metadata,
        )?)?,
        source_from_wire(required(
            "source",
            "a decision names its source",
            value.source,
        )?)?,
        decision_from_wire(required(
            "decision",
            "a decision declares its outcome",
            value.decision,
        )?)?,
        changes_from_wire(value.changes)?,
    ))
}

pub(crate) fn item_to_wire(value: &IngressItem) -> pb::IngressItem {
    use pb::__buffa::oneof::ingress_status::Status;
    let status = match value.status() {
        IngressStatus::Pending => Status::from(pb::IngressPending {
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        IngressStatus::Claimed { owner, until } => Status::from(pb::IngressClaimed {
            owner: owner.as_str().to_owned(),
            until_nanos: until.as_nanos(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        IngressStatus::Decided { decision } => Status::from(decision_to_wire(decision)),
        IngressStatus::Expired => Status::from(pb::IngressExpired {
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        IngressStatus::Exhausted => Status::from(pb::IngressExhausted {
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
    };
    pb::IngressItem {
        source: buffa::MessageField::some(source_to_wire(value.source())),
        digest: value.digest().as_bytes().to_vec(),
        provenance: buffa::MessageField::some(provenance_to_wire(value.provenance())),
        payload: buffa::MessageField::some(payload_to_wire(value.payload())),
        status: buffa::MessageField::some(pb::IngressStatus {
            status: Some(status),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        attempts: value.attempts(),
        fence: value.fence().get(),
        expires_at_nanos: value.expires_at().as_nanos(),
        revision: value.revision().get(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn item_from_wire(value: pb::IngressItem) -> Result<IngressItem, StateError> {
    use pb::__buffa::oneof::ingress_status::Status;
    let status = required("status", "an ingress item reports its status", value.status)?;
    let status = match status.status {
        Some(Status::Pending(_)) => IngressStatus::Pending,
        Some(Status::Claimed(value)) => IngressStatus::Claimed {
            owner: OwnerId::new(value.owner),
            until: MonotonicInstant::from_nanos(value.until_nanos),
        },
        Some(Status::Decided(value)) => IngressStatus::Decided {
            decision: decision_from_wire(*value)?,
        },
        Some(Status::Expired(_)) => IngressStatus::Expired,
        Some(Status::Exhausted(_)) => IngressStatus::Exhausted,
        None => return Err(malformed("status", "an ingress item reports its status")),
    };
    Ok(IngressItem::new(
        source_from_wire(required(
            "source",
            "an ingress item names its source",
            value.source,
        )?)?,
        ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>(
            "digest",
            &value.digest,
        )?),
        provenance_from_wire(required(
            "provenance",
            "an ingress item carries provenance",
            value.provenance,
        )?),
        payload_from_wire(required(
            "payload",
            "an ingress item carries a payload",
            value.payload,
        )?),
        MonotonicInstant::from_nanos(value.expires_at_nanos),
        Revision::new(value.revision),
    )
    .in_status(status)
    .claimed_times(value.attempts, FencingToken::new(value.fence)))
}

pub(crate) fn depth_to_wire(value: InboxDepth) -> pb::IngressDepth {
    pb::IngressDepth {
        edge_undecided: value.edge_undecided(),
        tenant_undecided: value.tenant_undecided(),
        edge_limit: value.edge_limit(),
        tenant_limit: value.tenant_limit(),
        claimed: value.claimed(),
        expired: value.expired(),
        exhausted: value.exhausted(),
        revision: value.revision().get(),
        feed_position: value.feed().position().get(),
        feed_root: value.feed().root().map(|root| root.as_bytes().to_vec()),
        consistency: pb::Consistency::from(Kernel(value.consistency())).into(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn depth_from_wire(value: pb::IngressDepth) -> Result<InboxDepth, StateError> {
    let root = value
        .feed_root
        .map(|root| {
            fixed_bytes::<{ CommitRoot::LEN }>("feed_root", &root).map(CommitRoot::from_bytes)
        })
        .transpose()?;
    let observed = consistency("consistency", value.consistency)?;
    if observed != Consistency::LinearizableCurrent {
        return Err(malformed(
            "consistency",
            "durable ingress reads are linearizable-current",
        ));
    }
    let depth = InboxDepth::new(
        value.edge_undecided,
        value.tenant_undecided,
        Revision::new(value.revision),
        JournalHead::new(JournalPosition::new(value.feed_position), root),
    )
    .with_outcomes(value.claimed, value.expired, value.exhausted);
    if depth.edge_limit() != value.edge_limit || depth.tenant_limit() != value.tenant_limit {
        return Err(malformed(
            "depth_limits",
            "the wire reports the bounds this build enforces",
        ));
    }
    Ok(depth)
}

pub(crate) fn depth_request_to_wire(request: &ReadInboxDepth) -> (pb::IngressScope, String) {
    (
        scope_to_wire(request.scope()),
        request.edge().as_str().to_owned(),
    )
}

pub(crate) fn item_request_to_wire(
    request: &ReadIngressItem,
) -> (pb::IngressScope, pb::IngressSourceIdentity) {
    (
        scope_to_wire(request.scope()),
        source_to_wire(request.source()),
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn receive_wire_preserves_distinct_durable_and_content_identity_payloads() {
        let command = polyc_state::ingress::cases::receive("receive-1", "event-1", b"durable");
        let kind = command.payload().kind().as_str().to_owned();
        let command = command.with_content_identity(IngressPayload::new(
            PayloadKind::new(kind),
            b"identity".to_vec(),
        ));

        let decoded = receive_from_wire(receive_to_wire(&command)).unwrap();

        assert_eq!(decoded, command);
        assert_eq!(decoded.payload().bytes(), b"durable");
        assert_eq!(decoded.content_identity().bytes(), b"identity");
    }

    /// The caller's audience decision crosses the transport unchanged.
    ///
    /// State never reads meaning from the number, so nothing downstream would
    /// notice it arriving as zero. The caller would: zero says the caller
    /// admitted nothing, and it withholds a persona's private material on
    /// that reading. A dropped field is therefore silent here and wrong
    /// there, which is why the assertion belongs at this boundary.
    #[test]
    fn receive_wire_preserves_the_callers_admitted_audience_decision() {
        let command = polyc_state::ingress::cases::receive("receive-1", "event-1", b"durable");

        let decoded = receive_from_wire(receive_to_wire(&command)).unwrap();

        assert_eq!(
            decoded.provenance().envelope().admitted_visibility(),
            polyc_state::ingress::cases::ADMITTED_VISIBILITY,
        );
    }

    #[test]
    fn receive_wire_defaults_an_omitted_content_identity_to_the_durable_payload() {
        let command = polyc_state::ingress::cases::receive("receive-1", "event-1", b"durable");
        let wire = receive_to_wire(&command);

        assert!(wire.content_identity.clone().into_option().is_none());

        let decoded = receive_from_wire(wire).unwrap();

        assert_eq!(decoded.content_identity(), decoded.payload());
    }
}