polyc-state-connect 2026.9.0

//! The mapping between the commit feed's vocabulary and its encoding.
//!
//! Same rules as [`crate::wire`] and [`crate::journal::wire`], and the same
//! helpers: every conversion is an explicit `From` or `TryFrom` naming each
//! field, [`Kernel`] carries the local type the orphan rule needs, kernel to
//! wire is infallible, and wire to kernel returns [`StateError::Malformed`]
//! naming the field it could not interpret.
//!
//! Two things are deliberately not on the wire. A feed command's resource
//! bounds, for the same reason a journal command's are not — they belong to the
//! operation family, and a bound a caller could send is a bound a caller could
//! widen. And the stream contract on a subscription request: the contract is the
//! server's declaration, read through `DescribeFeedStream`, not something a
//! caller proposes.

use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    command::{
        CommandEnvelope, CommandMetadata, CommandScope, FencingToken, Precondition, ResourceBounds,
    },
    digest::ContentDigest,
    error::StateError,
    feed::{
        self, AcknowledgeProjectorCursor, CommitEnvelope, CompactFeedPrefix, ConsumerPolicy,
        CreateSnapshot, FeedAnchor, FeedChunk, FeedCompaction, FeedCursor, FeedReadStart,
        FeedRecord, FeedRetention, FeedSnapshot, ProjectorRegistration, ProjectorStatus,
        RegisterProjector, SourceCheckpoint, SubscribeCommits,
    },
    id::{AggregateId, Audience, CommandId, ConsumerId, NamespaceId, Purpose, SnapshotId},
    journal::JournalAttestation,
    journal::JournalRecord,
    page::Positioned as _,
    receipt::Receipt,
    revision::{JournalPosition, JournalSource},
};

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

// ---------------------------------------------------------------------------
// Consumer policy
// ---------------------------------------------------------------------------

impl From<Kernel<ConsumerPolicy>> for pb::ConsumerPolicy {
    fn from(value: Kernel<ConsumerPolicy>) -> Self {
        match value.0 {
            ConsumerPolicy::Required => Self::CONSUMER_POLICY_REQUIRED,
            ConsumerPolicy::Optional => Self::CONSUMER_POLICY_OPTIONAL,
        }
    }
}

/// Reads a consumer's retention policy off the wire.
///
/// An unset policy is refused rather than defaulted. Defaulting either way is
/// wrong: to required, and an unregistered peer could pin a partition's storage
/// forever; to optional, and a projector that meant to hold its prefix silently
/// stops holding it.
fn consumer_policy(
    field: &str,
    value: buffa::EnumValue<pb::ConsumerPolicy>,
) -> Result<ConsumerPolicy, StateError> {
    match known(field, value)? {
        pb::ConsumerPolicy::CONSUMER_POLICY_REQUIRED => Ok(ConsumerPolicy::Required),
        pb::ConsumerPolicy::CONSUMER_POLICY_OPTIONAL => Ok(ConsumerPolicy::Optional),
        pb::ConsumerPolicy::CONSUMER_POLICY_UNSPECIFIED => Err(malformed(
            field,
            "a projector declares whether its lag holds the feed prefix",
        )),
    }
}

// ---------------------------------------------------------------------------
// Feed commands
// ---------------------------------------------------------------------------

/// Returns the bounds every feed command declares.
///
/// The family's own, not the caller's. A feed command carries no payload of its
/// own — a snapshot, a registration, an acknowledgement, and a compaction are
/// all a handful of fields — so the byte bound is the one the transport already
/// enforces, and the record bound is one command.
const fn feed_bounds() -> ResourceBounds {
    ResourceBounds::new(crate::MAX_WIRE_MESSAGE_BYTES as u64, 1)
}

impl From<Kernel<&CommandMetadata>> for pb::FeedCommand {
    /// Encodes the identity, addressing, and preconditions a feed command
    /// carries.
    fn from(value: Kernel<&CommandMetadata>) -> Self {
        let metadata = value.0;
        Self {
            command_id: metadata.command_id().as_str().to_owned(),
            source: buffa::MessageField::default(),
            aggregate: metadata.scope().aggregate().as_str().to_owned(),
            namespace: metadata.scope().namespace().as_str().to_owned(),
            purpose: metadata.envelope().purpose().as_str().to_owned(),
            audience: metadata.envelope().audience().as_str().to_owned(),
            digest: metadata.digest().as_bytes().to_vec(),
            precondition: buffa::MessageField::some(pb::Precondition::from(Kernel(
                metadata.precondition(),
            ))),
            fence: metadata.fence().map(FencingToken::get),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedCommand> for Kernel<CommandMetadata> {
    type Error = StateError;

    /// Reads the metadata a feed command carries.
    ///
    /// The protocol version is deliberately left at the current one rather than
    /// read off the command: it rides on the call's `CallContext` and is checked
    /// at admission, before a handler runs. A second copy inside the command
    /// would be a second place a peer could disagree with itself.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `digest` when the digest is not
    /// [`ContentDigest::LEN`] bytes, and `precondition` when the command names
    /// none.
    fn try_from(value: pb::FeedCommand) -> Result<Self, Self::Error> {
        let pb::FeedCommand {
            command_id,
            source,
            aggregate,
            namespace,
            purpose,
            audience,
            digest,
            precondition,
            fence,
            __buffa_unknown_fields: _,
        } = value;
        let digest =
            ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>("digest", &digest)?);
        let precondition = Kernel::<Precondition>::try_from(required(
            "precondition",
            "a precondition names what durable state the command requires",
            precondition,
        )?)?
        .into_inner();

        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "a feed command names an exact physical source",
            source,
        )?)?
        .into_inner();
        let mut metadata = CommandMetadata::new(
            CommandId::new(command_id),
            feed::family(),
            digest,
            CommandScope::new(
                AggregateId::new(aggregate),
                source.partition().clone(),
                NamespaceId::new(namespace),
            ),
            CommandEnvelope::new(
                Purpose::new(purpose),
                Audience::new(audience),
                feed_bounds(),
            ),
        )
        .with_precondition(precondition);
        if let Some(fence) = fence {
            metadata = metadata.with_fence(FencingToken::new(fence));
        }
        Ok(Self(metadata))
    }
}

impl From<Kernel<&CreateSnapshot>> for pb::FeedCommand {
    fn from(value: Kernel<&CreateSnapshot>) -> Self {
        let mut command = Self::from(Kernel(value.0.metadata()));
        command.source =
            buffa::MessageField::some(pb::JournalSource::from(Kernel(value.0.source())));
        command
    }
}

impl From<Kernel<&RegisterProjector>> for pb::FeedCommand {
    fn from(value: Kernel<&RegisterProjector>) -> Self {
        let mut command = Self::from(Kernel(value.0.metadata()));
        command.source = buffa::MessageField::some(pb::JournalSource::from(Kernel(
            value.0.registration().source(),
        )));
        command
    }
}

impl From<Kernel<&AcknowledgeProjectorCursor>> for pb::FeedCommand {
    fn from(value: Kernel<&AcknowledgeProjectorCursor>) -> Self {
        let mut command = Self::from(Kernel(value.0.metadata()));
        command.source =
            buffa::MessageField::some(pb::JournalSource::from(Kernel(value.0.cursor().source())));
        command
    }
}

impl From<Kernel<&CompactFeedPrefix>> for pb::FeedCommand {
    fn from(value: Kernel<&CompactFeedPrefix>) -> Self {
        let mut command = Self::from(Kernel(value.0.metadata()));
        command.source =
            buffa::MessageField::some(pb::JournalSource::from(Kernel(value.0.source())));
        command
    }
}

// ---------------------------------------------------------------------------
// Exact source checkpoints and cursors
// ---------------------------------------------------------------------------

impl From<Kernel<&SourceCheckpoint>> for pb::SourceCheckpoint {
    fn from(value: Kernel<&SourceCheckpoint>) -> Self {
        let checkpoint = value.0;
        Self {
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(checkpoint.source()))),
            feed_position: checkpoint.feed_position().get(),
            journal_position: checkpoint.journal_position().get(),
            evidence_leaf: checkpoint.evidence_leaf(),
            covering_attestation: buffa::MessageField::some(pb::JournalAttestation::from(Kernel(
                checkpoint.covering_attestation(),
            ))),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::SourceCheckpoint> for Kernel<SourceCheckpoint> {
    type Error = StateError;

    fn try_from(value: pb::SourceCheckpoint) -> Result<Self, Self::Error> {
        let pb::SourceCheckpoint {
            source,
            feed_position,
            journal_position,
            evidence_leaf,
            covering_attestation,
            __buffa_unknown_fields: _,
        } = value;
        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "a checkpoint names an exact physical source",
            source,
        )?)?
        .into_inner();
        let attestation = Kernel::<JournalAttestation>::try_from(required(
            "covering_attestation",
            "a checkpoint carries the later signed root that covers its journal prefix",
            covering_attestation,
        )?)?
        .into_inner();
        Ok(Self(SourceCheckpoint::try_new(
            source,
            JournalPosition::new(feed_position),
            JournalPosition::new(journal_position),
            evidence_leaf,
            attestation,
        )?))
    }
}

impl From<Kernel<&FeedCursor>> for pb::FeedCursor {
    fn from(value: Kernel<&FeedCursor>) -> Self {
        let cursor = value.0;
        Self {
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(cursor.source()))),
            checkpoint: cursor.checkpoint().map_or_else(
                buffa::MessageField::default,
                |checkpoint| {
                    buffa::MessageField::some(pb::SourceCheckpoint::from(Kernel(checkpoint)))
                },
            ),
            snapshot: cursor
                .snapshot()
                .map(|snapshot| snapshot.as_str().to_owned()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedCursor> for Kernel<FeedCursor> {
    type Error = StateError;

    fn try_from(value: pb::FeedCursor) -> Result<Self, Self::Error> {
        let pb::FeedCursor {
            source,
            checkpoint,
            snapshot,
            __buffa_unknown_fields: _,
        } = value;
        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "a feed cursor names an exact physical source",
            source,
        )?)?
        .into_inner();
        let checkpoint = checkpoint
            .into_option()
            .map(|checkpoint| {
                Kernel::<SourceCheckpoint>::try_from(checkpoint).map(Kernel::into_inner)
            })
            .transpose()?;
        if checkpoint
            .as_ref()
            .is_some_and(|checkpoint| checkpoint.source() != &source)
        {
            return Err(malformed(
                "checkpoint",
                "a cursor and its checkpoint name the same physical source",
            ));
        }
        let cursor = match (snapshot, checkpoint) {
            (Some(snapshot), Some(checkpoint)) if !snapshot.is_empty() => {
                FeedCursor::in_snapshot(SnapshotId::new(snapshot), checkpoint)
            }
            (Some(_), Some(_)) => {
                return Err(malformed("snapshot", "snapshot provenance is non-empty"));
            }
            (Some(_), None) => {
                return Err(malformed(
                    "checkpoint",
                    "snapshot provenance accompanies an issued checkpoint",
                ));
            }
            (None, Some(checkpoint)) => FeedCursor::at(checkpoint),
            (None, None) => FeedCursor::origin(source),
        };
        Ok(Self(cursor))
    }
}

impl From<Kernel<&FeedReadStart>> for pb::FeedReadStart {
    fn from(value: Kernel<&FeedReadStart>) -> Self {
        use pb::__buffa::oneof::feed_read_start::Start;
        let start = match value.0 {
            FeedReadStart::Snapshot(snapshot) => Start::Snapshot(snapshot.as_str().to_owned()),
            FeedReadStart::Resume(cursor) => {
                Start::from(pb::FeedCursor::from(Kernel(cursor.as_ref())))
            }
        };
        Self {
            start: Some(start),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedReadStart> for Kernel<FeedReadStart> {
    type Error = StateError;

    fn try_from(value: pb::FeedReadStart) -> Result<Self, Self::Error> {
        use pb::__buffa::oneof::feed_read_start::Start;
        let pb::FeedReadStart {
            start,
            __buffa_unknown_fields: _,
        } = value;
        let start = match start {
            Some(Start::Snapshot(snapshot)) if !snapshot.is_empty() => {
                FeedReadStart::Snapshot(SnapshotId::new(snapshot))
            }
            Some(Start::Snapshot(_)) => {
                return Err(malformed(
                    "start",
                    "a snapshot start names a non-empty identity",
                ));
            }
            Some(Start::Resume(cursor)) => FeedReadStart::Resume(Box::new(
                Kernel::<FeedCursor>::try_from(*cursor)?.into_inner(),
            )),
            None => return Err(malformed("start", "a feed read names where it begins")),
        };
        Ok(Self(start))
    }
}

// ---------------------------------------------------------------------------
// Feed entries
// ---------------------------------------------------------------------------

impl From<Kernel<&CommitEnvelope>> for pb::CommitEnvelope {
    fn from(value: Kernel<&CommitEnvelope>) -> Self {
        let envelope = value.0;
        Self {
            checkpoint: buffa::MessageField::some(pb::SourceCheckpoint::from(Kernel(
                envelope.checkpoint(),
            ))),
            command_id: envelope.command_id().as_str().to_owned(),
            digest: envelope.digest().as_bytes().to_vec(),
            fence: envelope.fence().map(FencingToken::get),
            head_before: envelope.head_before().get(),
            head_after: envelope.head_after().get(),
            record_count: envelope.record_count(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::CommitEnvelope> for Kernel<CommitEnvelope> {
    type Error = StateError;

    /// Reads what one commit recorded about itself.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `digest` when the digest is not
    /// [`ContentDigest::LEN`] bytes — a commit whose digest cannot be read is
    /// one a projector could not compare against anything.
    fn try_from(value: pb::CommitEnvelope) -> Result<Self, Self::Error> {
        let pb::CommitEnvelope {
            checkpoint,
            command_id,
            digest,
            fence,
            head_before,
            head_after,
            record_count,
            __buffa_unknown_fields: _,
        } = value;
        let checkpoint = Kernel::<SourceCheckpoint>::try_from(required(
            "checkpoint",
            "a commit envelope carries its exact source evidence point",
            checkpoint,
        )?)?
        .into_inner();
        let digest =
            ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>("digest", &digest)?);
        if command_id.is_empty() {
            return Err(malformed(
                "command_id",
                "a commit envelope names a non-empty command identity",
            ));
        }
        let envelope = CommitEnvelope::new(
            checkpoint,
            CommandId::new(command_id),
            digest,
            JournalPosition::new(head_before),
            JournalPosition::new(head_after),
            record_count,
        );
        Ok(Self(match fence {
            Some(fence) => envelope.with_fence(FencingToken::new(fence)),
            None => envelope,
        }))
    }
}

impl From<Kernel<&FeedRecord>> for pb::FeedRecord {
    fn from(value: Kernel<&FeedRecord>) -> Self {
        let entry = value.0;
        Self {
            position: entry.position().get(),
            envelope: buffa::MessageField::some(pb::CommitEnvelope::from(Kernel(entry.envelope()))),
            records: entry
                .records()
                .iter()
                .map(|record| pb::JournalRecord::from(Kernel(record)))
                .collect(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedRecord> for Kernel<FeedRecord> {
    type Error = StateError;

    /// Reads one entry of the durable feed.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `envelope` when the entry
    /// carries none — an entry with no envelope is a batch of records with
    /// nothing saying which commit produced them — and whatever the envelope and
    /// record conversions themselves report.
    fn try_from(value: pb::FeedRecord) -> Result<Self, Self::Error> {
        let pb::FeedRecord {
            position,
            envelope,
            records,
            __buffa_unknown_fields: _,
        } = value;
        let envelope = Kernel::<CommitEnvelope>::try_from(required(
            "envelope",
            "a feed entry carries the commit it describes",
            envelope,
        )?)?
        .into_inner();
        let records = records
            .into_iter()
            .map(|record| Kernel::<JournalRecord>::try_from(record).map(Kernel::into_inner))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Self(FeedRecord::new(
            JournalPosition::new(position),
            envelope,
            records,
        )))
    }
}

impl From<Kernel<&FeedChunk>> for pb::FeedChunk {
    fn from(value: Kernel<&FeedChunk>) -> Self {
        let chunk = value.0;
        Self {
            records: chunk
                .records()
                .iter()
                .map(|record| pb::FeedRecord::from(Kernel(record)))
                .collect(),
            next: buffa::MessageField::some(pb::FeedCursor::from(Kernel(chunk.next_cursor()))),
            end: pb::StreamEnd::from(Kernel(chunk.end())).into(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedChunk> for Kernel<FeedChunk> {
    type Error = StateError;

    /// Reads one bounded chunk of a commit feed.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `end` when the chunk does not
    /// say why it stopped — which is what distinguishes a drain from exhaustion
    /// — and whatever the entry conversions report.
    fn try_from(value: pb::FeedChunk) -> Result<Self, Self::Error> {
        let pb::FeedChunk {
            records,
            next,
            end,
            __buffa_unknown_fields: _,
        } = value;
        let end = crate::wire::stream_end("end", end)?;
        let records = records
            .into_iter()
            .map(|record| Kernel::<FeedRecord>::try_from(record).map(Kernel::into_inner))
            .collect::<Result<Vec<_>, _>>()?;
        let next = Kernel::<FeedCursor>::try_from(required(
            "next",
            "every feed chunk carries the exact cursor to persist",
            next,
        )?)?
        .into_inner();
        Ok(Self(FeedChunk::new(records, next, end)))
    }
}

// ---------------------------------------------------------------------------
// Snapshots
// ---------------------------------------------------------------------------

impl From<Kernel<&FeedSnapshot>> for pb::FeedSnapshot {
    fn from(value: Kernel<&FeedSnapshot>) -> Self {
        let snapshot = value.0;
        Self {
            snapshot: snapshot.id().as_str().to_owned(),
            checkpoint: buffa::MessageField::some(pb::SourceCheckpoint::from(Kernel(
                snapshot.anchor().checkpoint(),
            ))),
            receipt: buffa::MessageField::some(pb::Receipt::from(Kernel(snapshot.receipt()))),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedSnapshot> for Kernel<FeedSnapshot> {
    type Error = StateError;

    /// Reads one immutable snapshot of a partition's feed.
    ///
    /// The anchor is rebuilt from the identity rather than from the loose
    /// partition and position fields, and then checked against them. An identity
    /// that disagreed with the fields beside it would let a consumer bootstrap
    /// from one prefix while believing it bootstrapped from another.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `snapshot` when the identity is
    /// not one the contract could have issued or does not match the partition
    /// and position beside it, and `receipt` when a snapshot arrives without the
    /// durable receipt that recorded it.
    fn try_from(value: pb::FeedSnapshot) -> Result<Self, Self::Error> {
        let pb::FeedSnapshot {
            snapshot,
            checkpoint,
            receipt,
            __buffa_unknown_fields: _,
        } = value;
        let anchor = FeedAnchor::parse(&SnapshotId::new(snapshot))?;
        let checkpoint = Kernel::<SourceCheckpoint>::try_from(required(
            "checkpoint",
            "a feed snapshot carries its exact source evidence point",
            checkpoint,
        )?)?
        .into_inner();
        if anchor.checkpoint() != &checkpoint {
            return Err(malformed(
                "snapshot",
                "a snapshot identity names the exact checkpoint beside it",
            ));
        }
        let receipt = Kernel::<Receipt>::try_from(required(
            "receipt",
            "a recorded snapshot carries the receipt that recorded it",
            receipt,
        )?)?
        .into_inner();
        Ok(Self(FeedSnapshot::new(anchor, receipt)))
    }
}

// ---------------------------------------------------------------------------
// Subscriptions
// ---------------------------------------------------------------------------

impl From<Kernel<(&DeclaredCall, &SubscribeCommits)>> for pb::SubscribeCommitsRequest {
    fn from(value: Kernel<(&DeclaredCall, &SubscribeCommits)>) -> Self {
        let (declared, request) = value.0;
        Self {
            context: buffa::MessageField::some(pb::CallContext::from(Kernel(declared))),
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(request.source()))),
            start: buffa::MessageField::some(pb::FeedReadStart::from(Kernel(request.start()))),
            max_chunk_commits: request.max_chunk_commits(),
            consumer: request
                .consumer()
                .map(|consumer| consumer.as_str().to_owned()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

/// One decoded subscription request, including its transport admission context.
#[derive(Debug)]
pub struct SubscriptionRequest {
    /// The caller's declared admission and budget context.
    pub context: buffa::MessageField<pb::CallContext, buffa::Inline<pb::CallContext>>,
    /// The exact source, start, bound, and optional registered consumer.
    pub subscription: SubscribeCommits,
}

impl TryFrom<pb::SubscribeCommitsRequest> for Kernel<SubscriptionRequest> {
    type Error = StateError;

    /// Reads one complete subscription request off the wire.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `start` when the request omits
    /// where it begins. The error names `partition` when the request names none.
    /// A subscription without a partition is not a subscription to everything.
    /// This contract cannot serve it (INV-21).
    fn try_from(value: pb::SubscribeCommitsRequest) -> Result<Self, Self::Error> {
        let pb::SubscribeCommitsRequest {
            context,
            source,
            start,
            max_chunk_commits,
            consumer,
            __buffa_unknown_fields: _,
        } = value;
        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "a subscription names exactly one physical source",
            source,
        )?)?
        .into_inner();
        let start = Kernel::<FeedReadStart>::try_from(required(
            "start",
            "a subscription names where it begins",
            start,
        )?)?
        .into_inner();

        if let FeedReadStart::Resume(cursor) = &start
            && cursor.source() != &source
        {
            return Err(malformed(
                "start",
                "a subscription and its resume cursor name the same physical source",
            ));
        }
        let request = SubscribeCommits::new(source, start, max_chunk_commits);
        let subscription = match consumer {
            Some(consumer) if !consumer.is_empty() => {
                request.on_behalf_of(ConsumerId::new(consumer))
            }
            Some(_) => {
                return Err(malformed(
                    "consumer",
                    "a present subscription consumer identity is non-empty",
                ));
            }
            None => request,
        };
        Ok(Self(SubscriptionRequest {
            context,
            subscription,
        }))
    }
}

// ---------------------------------------------------------------------------
// Registration and retention
// ---------------------------------------------------------------------------

impl From<Kernel<&ProjectorRegistration>> for pb::ProjectorRegistration {
    fn from(value: Kernel<&ProjectorRegistration>) -> Self {
        let registration = value.0;
        Self {
            consumer: registration.consumer().as_str().to_owned(),
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(
                registration.source(),
            ))),
            policy: pb::ConsumerPolicy::from(Kernel(registration.policy())).into(),
            declared_lag_commits: registration.declared_lag_commits(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::ProjectorRegistration> for Kernel<ProjectorRegistration> {
    type Error = StateError;

    /// Reads what one projector declared about itself.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `policy` when the registration
    /// declares no retention policy or one this build does not know.
    fn try_from(value: pb::ProjectorRegistration) -> Result<Self, Self::Error> {
        let pb::ProjectorRegistration {
            consumer,
            source,
            policy,
            declared_lag_commits,
            __buffa_unknown_fields: _,
        } = value;
        let policy = consumer_policy("policy", policy)?;
        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "a projector registration names an exact physical source",
            source,
        )?)?
        .into_inner();
        if consumer.is_empty() {
            return Err(malformed(
                "consumer",
                "a projector names a non-empty consumer",
            ));
        }
        Ok(Self(ProjectorRegistration::new(
            ConsumerId::new(consumer),
            source,
            policy,
            declared_lag_commits,
        )))
    }
}

impl From<Kernel<&ProjectorStatus>> for pb::ProjectorStatus {
    fn from(value: Kernel<&ProjectorStatus>) -> Self {
        let status = value.0;
        Self {
            registration: buffa::MessageField::some(pb::ProjectorRegistration::from(Kernel(
                status.registration(),
            ))),
            acknowledged: buffa::MessageField::some(pb::FeedCursor::from(Kernel(
                status.acknowledged(),
            ))),
            evicted: status.is_evicted(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::ProjectorStatus> for Kernel<ProjectorStatus> {
    type Error = StateError;

    /// Reads what the feed reports about one projector.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `registration` when the status
    /// carries none, and whatever the registration conversion reports.
    fn try_from(value: pb::ProjectorStatus) -> Result<Self, Self::Error> {
        let pb::ProjectorStatus {
            registration,
            acknowledged,
            evicted,
            __buffa_unknown_fields: _,
        } = value;
        let registration = Kernel::<ProjectorRegistration>::try_from(required(
            "registration",
            "a projector status carries what the projector declared",
            registration,
        )?)?
        .into_inner();
        let acknowledged = Kernel::<FeedCursor>::try_from(required(
            "acknowledged",
            "a projector status carries the exact cursor it applied",
            acknowledged,
        )?)?
        .into_inner();
        if registration.source() != acknowledged.source() {
            return Err(malformed(
                "acknowledged",
                "a projector status and cursor name the same physical source",
            ));
        }
        let status = ProjectorStatus::new(registration, acknowledged);
        Ok(Self(if evicted { status.evicted() } else { status }))
    }
}

impl From<Kernel<&FeedRetention>> for pb::FeedRetention {
    fn from(value: Kernel<&FeedRetention>) -> Self {
        let retention = value.0;
        Self {
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(retention.source()))),
            earliest: retention.earliest().get(),
            head: retention.head().get(),
            held_by: retention
                .held_by_consumer()
                .map(|consumer| consumer.as_str().to_owned()),
            pinned_by: retention
                .pinned_by_snapshot()
                .map(|snapshot| snapshot.as_str().to_owned()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedRetention> for Kernel<FeedRetention> {
    type Error = StateError;

    fn try_from(value: pb::FeedRetention) -> Result<Self, Self::Error> {
        let pb::FeedRetention {
            source,
            earliest,
            head,
            held_by,
            pinned_by,
            __buffa_unknown_fields: _,
        } = value;
        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "feed retention names an exact physical source",
            source,
        )?)?
        .into_inner();
        let retention = FeedRetention::new(
            source,
            JournalPosition::new(earliest),
            JournalPosition::new(head),
        );
        let retention = match held_by {
            Some(consumer) if !consumer.is_empty() => retention.held_by(ConsumerId::new(consumer)),
            Some(_) => {
                return Err(malformed(
                    "held_by",
                    "a present retention consumer identity is non-empty",
                ));
            }
            None => retention,
        };
        Ok(Self(match pinned_by {
            Some(snapshot) if !snapshot.is_empty() => {
                retention.pinned_by(SnapshotId::new(snapshot))
            }
            Some(_) => {
                return Err(malformed(
                    "pinned_by",
                    "a present retention snapshot identity is non-empty",
                ));
            }
            None => retention,
        }))
    }
}

impl From<Kernel<&FeedCompaction>> for pb::FeedCompaction {
    fn from(value: Kernel<&FeedCompaction>) -> Self {
        let compaction = value.0;
        Self {
            retention: buffa::MessageField::some(pb::FeedRetention::from(Kernel(
                compaction.retention(),
            ))),
            evicted: compaction
                .evicted()
                .iter()
                .map(|consumer| consumer.as_str().to_owned())
                .collect(),
            receipt: buffa::MessageField::some(pb::Receipt::from(Kernel(compaction.receipt()))),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::FeedCompaction> for Kernel<FeedCompaction> {
    type Error = StateError;

    /// Reads what one compaction did.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `retention` when the result
    /// carries none, or naming `receipt` when the exact result has no durable
    /// settlement a retry could recover.
    fn try_from(value: pb::FeedCompaction) -> Result<Self, Self::Error> {
        let pb::FeedCompaction {
            retention,
            evicted,
            receipt,
            __buffa_unknown_fields: _,
        } = value;
        let retention = Kernel::<FeedRetention>::try_from(required(
            "retention",
            "a compaction reports what the partition now retains",
            retention,
        )?)?
        .into_inner();
        if evicted.iter().any(String::is_empty) {
            return Err(malformed(
                "evicted",
                "every evicted consumer identity is non-empty",
            ));
        }
        let evicted = evicted.into_iter().map(ConsumerId::new).collect();
        let receipt = Kernel::<Receipt>::try_from(required(
            "receipt",
            "a compaction result carries its durable settlement",
            receipt,
        )?)?
        .into_inner();
        Ok(Self(FeedCompaction::new(retention, evicted, receipt)))
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use super::*;
    use polyc_state::{
        consistency::Consistency,
        feed::{ATTESTATION_SIGNATURE_BYTES, ATTESTATION_SIGNER_BYTES},
        id::PartitionId,
        journal::{JournalAttestation, RecordKind, RecordTrust},
        receipt::CommitEvidence,
        revision::{CommitRoot, PartitionIncarnation},
    };

    fn metadata() -> CommandMetadata {
        CommandMetadata::new(
            CommandId::new("cmd-1"),
            feed::family(),
            ContentDigest::from_bytes([3; ContentDigest::LEN]),
            CommandScope::new(
                AggregateId::new("conv-1"),
                PartitionId::new("conv-1"),
                NamespaceId::new("default"),
            ),
            CommandEnvelope::new(
                Purpose::new("projection"),
                crate::state_audience(),
                feed_bounds(),
            ),
        )
        .with_fence(FencingToken::new(4))
    }

    fn receipt() -> Receipt {
        Receipt::committed(
            &metadata(),
            CommitEvidence::new()
                .with_snapshot(SnapshotId::new("feed:conv-1@3"))
                .with_position(JournalPosition::new(3)),
            Consistency::OrderedPerAggregate,
        )
    }

    fn source() -> JournalSource {
        JournalSource::new(
            PartitionId::new("conv-1"),
            PartitionIncarnation::from_bytes([7; PartitionIncarnation::LEN]),
        )
    }

    fn checkpoint(feed_position: u64, journal_position: u64) -> SourceCheckpoint {
        SourceCheckpoint::try_new(
            source(),
            JournalPosition::new(feed_position),
            JournalPosition::new(journal_position),
            journal_position,
            JournalAttestation::new(
                CommitRoot::from_bytes([journal_position as u8; CommitRoot::LEN]),
                journal_position + 1,
                vec![3; ATTESTATION_SIGNATURE_BYTES],
                vec![4; ATTESTATION_SIGNER_BYTES],
            ),
        )
        .expect("a complete test checkpoint")
    }

    fn cursor(position: u64) -> FeedCursor {
        if position == 0 {
            FeedCursor::origin(source())
        } else {
            FeedCursor::at(checkpoint(position, position + 4))
        }
    }

    fn entry() -> FeedRecord {
        FeedRecord::new(
            JournalPosition::new(3),
            CommitEnvelope::new(
                checkpoint(3, 7),
                CommandId::new("cmd-9"),
                ContentDigest::from_bytes([8; ContentDigest::LEN]),
                JournalPosition::new(5),
                JournalPosition::new(7),
                2,
            )
            .with_fence(FencingToken::new(2)),
            vec![
                JournalRecord::new(
                    JournalPosition::new(6),
                    RecordKind::new("turn_input"),
                    RecordTrust::TrustedUser,
                    b"one".to_vec(),
                ),
                JournalRecord::new(
                    JournalPosition::new(7),
                    RecordKind::new("turn_output"),
                    RecordTrust::QuarantinedContent,
                    b"two".to_vec(),
                ),
            ],
        )
    }

    #[test]
    fn a_feed_command_round_trips_every_field_it_carries() {
        let encoded = pb::FeedCommand::from(Kernel(&CreateSnapshot::new(metadata(), source())));
        let back = Kernel::<CommandMetadata>::try_from(encoded).unwrap().0;
        assert_eq!(back, metadata());
    }

    #[test]
    fn every_command_shape_encodes_through_the_same_metadata() {
        let expected = pb::FeedCommand::from(Kernel(&metadata()));
        let mut expected = expected;
        expected.source = buffa::MessageField::some(pb::JournalSource::from(Kernel(&source())));
        assert_eq!(
            pb::FeedCommand::from(Kernel(&CreateSnapshot::new(metadata(), source()))),
            expected
        );
        assert_eq!(
            pb::FeedCommand::from(Kernel(&CompactFeedPrefix::new(
                metadata(),
                source(),
                JournalPosition::new(4)
            ))),
            expected
        );
        assert_eq!(
            pb::FeedCommand::from(Kernel(&AcknowledgeProjectorCursor::new(
                metadata(),
                ConsumerId::new("search"),
                cursor(2)
            ))),
            expected
        );
        assert_eq!(
            pb::FeedCommand::from(Kernel(&RegisterProjector::new(
                metadata(),
                ProjectorRegistration::new(
                    ConsumerId::new("search"),
                    source(),
                    ConsumerPolicy::Required,
                    4,
                )
            ))),
            expected
        );
    }

    #[test]
    fn a_feed_entry_round_trips_its_envelope_and_every_record() {
        let encoded = pb::FeedRecord::from(Kernel(&entry()));
        let back = Kernel::<FeedRecord>::try_from(encoded).unwrap().0;
        assert_eq!(back, entry());
        assert!(back.is_self_consistent());
    }

    #[test]
    fn a_chunk_round_trips_its_records_cursor_and_reason_for_stopping() {
        for end in [
            polyc_state::stream::StreamEnd::More,
            polyc_state::stream::StreamEnd::Exhausted,
            polyc_state::stream::StreamEnd::Drained,
        ] {
            let cursor =
                FeedCursor::in_snapshot(SnapshotId::new("feed:test-snapshot"), checkpoint(3, 7));
            let chunk = FeedChunk::new(vec![entry()], cursor, end);
            let back = Kernel::<FeedChunk>::try_from(pb::FeedChunk::from(Kernel(&chunk)))
                .unwrap()
                .0;
            assert_eq!(back, chunk);
        }

        let origin = FeedChunk::new(
            Vec::new(),
            cursor(0),
            polyc_state::stream::StreamEnd::Exhausted,
        );
        let back = Kernel::<FeedChunk>::try_from(pb::FeedChunk::from(Kernel(&origin)))
            .unwrap()
            .0;
        assert_eq!(back, origin);
    }

    #[test]
    fn a_chunk_that_does_not_say_why_it_stopped_is_malformed() {
        let mut encoded = pb::FeedChunk::from(Kernel(&FeedChunk::new(
            Vec::new(),
            cursor(0),
            polyc_state::stream::StreamEnd::Exhausted,
        )));
        encoded.end = pb::StreamEnd::STREAM_END_UNSPECIFIED.into();
        assert!(matches!(
            Kernel::<FeedChunk>::try_from(encoded).unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "end"
        ));
    }

    #[test]
    fn a_snapshot_round_trips_and_refuses_an_identity_that_contradicts_itself() {
        let snapshot = FeedSnapshot::new(FeedAnchor::new(checkpoint(3, 9)), receipt());
        let encoded = pb::FeedSnapshot::from(Kernel(&snapshot));
        assert_eq!(encoded.snapshot, snapshot.id().as_str());
        let back = Kernel::<FeedSnapshot>::try_from(encoded.clone()).unwrap().0;
        assert_eq!(back, snapshot);

        let mut lying = encoded.clone();
        lying
            .checkpoint
            .as_option_mut()
            .expect("checkpoint")
            .feed_position = 4;
        assert!(matches!(
            Kernel::<FeedSnapshot>::try_from(lying).unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "snapshot"
        ));

        let mut renamed = encoded;
        renamed
            .checkpoint
            .as_option_mut()
            .expect("checkpoint")
            .source
            .as_option_mut()
            .expect("source")
            .partition = "conv-2".to_owned();
        assert!(matches!(
            Kernel::<FeedSnapshot>::try_from(renamed).unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "snapshot"
        ));
    }

    #[test]
    fn a_subscription_round_trips_its_partition_start_bound_and_consumer() {
        let declared = crate::DeclaredCall::bounded(
            crate::state_audience(),
            std::time::Duration::from_secs(1),
        );
        let request =
            SubscribeCommits::new(source(), FeedReadStart::Resume(Box::new(cursor(4))), 8)
                .on_behalf_of(ConsumerId::new("search"));
        let back = Kernel::<SubscriptionRequest>::try_from(pb::SubscribeCommitsRequest::from(
            Kernel((&declared, &request)),
        ))
        .unwrap()
        .into_inner();
        assert_eq!(back.subscription, request);
        assert!(back.context.as_option().is_some());

        let anonymous = SubscribeCommits::new(
            source(),
            FeedReadStart::Snapshot(SnapshotId::new("feed:test-snapshot")),
            8,
        );
        let encoded = pb::SubscribeCommitsRequest::from(Kernel((&declared, &anonymous)));
        assert!(encoded.consumer.is_none());
        let back = Kernel::<SubscriptionRequest>::try_from(encoded)
            .unwrap()
            .into_inner();
        assert_eq!(back.subscription, anonymous);
    }

    /// A subscription with no partition is not a subscription to everything.
    #[test]
    fn a_subscription_without_a_partition_or_a_start_is_malformed() {
        assert!(matches!(
            Kernel::<SubscriptionRequest>::try_from(pb::SubscribeCommitsRequest {
                context: buffa::MessageField::default(),
                source: buffa::MessageField::default(),
                start: buffa::MessageField::some(pb::FeedReadStart::from(Kernel(
                    &FeedReadStart::Resume(Box::new(cursor(0))),
                ))),
                max_chunk_commits: 4,
                consumer: None,
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
            .unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "source"
        ));
        assert!(matches!(
            Kernel::<SubscriptionRequest>::try_from(pb::SubscribeCommitsRequest {
                context: buffa::MessageField::default(),
                source: buffa::MessageField::some(pb::JournalSource::from(Kernel(&source()))),
                start: buffa::MessageField::default(),
                max_chunk_commits: 4,
                consumer: None,
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
            .unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "start"
        ));
    }

    #[test]
    fn a_projector_status_round_trips_including_its_eviction() {
        for policy in [ConsumerPolicy::Required, ConsumerPolicy::Optional] {
            let registration =
                ProjectorRegistration::new(ConsumerId::new("search"), source(), policy, 6);
            let live = ProjectorStatus::new(registration, cursor(2));
            for status in [live.clone(), live.evicted()] {
                let back =
                    Kernel::<ProjectorStatus>::try_from(pb::ProjectorStatus::from(Kernel(&status)))
                        .unwrap()
                        .0;
                assert_eq!(back, status);
            }
        }
    }

    /// An unset policy is refused. Defaulting either way is wrong, and both
    /// wrong answers are silent.
    #[test]
    fn a_registration_that_declares_no_policy_is_malformed() {
        let mut encoded = pb::ProjectorRegistration::from(Kernel(&ProjectorRegistration::new(
            ConsumerId::new("search"),
            source(),
            ConsumerPolicy::Required,
            6,
        )));
        encoded.policy = pb::ConsumerPolicy::CONSUMER_POLICY_UNSPECIFIED.into();
        assert!(matches!(
            Kernel::<ProjectorRegistration>::try_from(encoded).unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "policy"
        ));
    }

    #[test]
    fn retention_round_trips_with_and_without_the_consumer_holding_it() {
        let bare = FeedRetention::new(source(), JournalPosition::new(2), JournalPosition::new(9));
        for retention in [
            bare.clone(),
            bare.clone().held_by(ConsumerId::new("search")),
            bare.clone().pinned_by(SnapshotId::new("feed:conv-1@2")),
            bare.held_by(ConsumerId::new("search"))
                .pinned_by(SnapshotId::new("feed:conv-1@2")),
        ] {
            let back =
                Kernel::<FeedRetention>::try_from(pb::FeedRetention::from(Kernel(&retention)))
                    .unwrap()
                    .0;
            assert_eq!(back, retention);
        }
    }

    /// Every compaction result carries the settlement a retry recovers.
    #[test]
    fn a_compaction_round_trips_with_its_required_receipt() {
        let retention =
            FeedRetention::new(source(), JournalPosition::new(4), JournalPosition::new(9));
        let held_back = FeedCompaction::new(retention.clone(), Vec::new(), receipt());
        let back = Kernel::<FeedCompaction>::try_from(pb::FeedCompaction::from(Kernel(&held_back)))
            .unwrap()
            .0;
        assert_eq!(back, held_back);

        let recorded = FeedCompaction::new(
            retention,
            vec![ConsumerId::new("search"), ConsumerId::new("audit")],
            receipt(),
        );
        let back = Kernel::<FeedCompaction>::try_from(pb::FeedCompaction::from(Kernel(&recorded)))
            .unwrap()
            .0;
        assert_eq!(back.evicted().len(), 2);
        assert_eq!(back.retention(), recorded.retention());

        let mut missing = pb::FeedCompaction::from(Kernel(&recorded));
        missing.receipt = buffa::MessageField::default();
        assert!(matches!(
            Kernel::<FeedCompaction>::try_from(missing).unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "receipt"
        ));
    }
}