polyc-state-connect 2026.9.0

//! Explicit persona-memory snapshot wire mapping.
//!
//! One conversion per type, every field named, and nothing spread over a
//! default: a field added on either side has to be written down again in both
//! directions or this file stops compiling.
//!
//! The snapshot record itself crosses as the kernel's own encoded bytes. That
//! is the one deliberate opacity here. [`MemorySnapshot`] already has a
//! canonical encoding carrying its own format revision, and
//! [`MemorySnapshot::decode`] already refuses a record that violates any bound
//! [`MemorySnapshot::taken`] enforces — so the bytes are validated at the
//! boundary either way, and restating the index's fields as protobuf would put
//! a second encoding beside the first for the two to disagree about.

use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    command::{CommandEnvelope, CommandMetadata, ResourceBounds},
    digest::ContentDigest,
    error::StateError,
    id::{Audience, CommandId, NamespaceId, Purpose},
    persona_memory::{
        MemoryPartitionId, MemorySnapshot,
        store::{SnapshotCommand, SnapshotOperation, snapshot_scope},
    },
    revision::{CommitRoot, JournalHead, JournalPosition, Revision},
    versioned::{EntryExpectation, MAX_MUTATIONS_PER_TRANSACTION, MAX_TRANSACTION_PAYLOAD_BYTES},
};

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

pub(super) fn head_to_wire(head: JournalHead) -> pb::StatePersonaMemoryJournalHead {
    pb::StatePersonaMemoryJournalHead {
        position: head.position().get(),
        root: head.root().map(|root| root.as_bytes().to_vec()),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(super) fn head_from_wire(
    value: pb::StatePersonaMemoryJournalHead,
) -> Result<JournalHead, StateError> {
    let root = match value.root {
        Some(bytes) => Some(CommitRoot::from_bytes(fixed_bytes::<{ CommitRoot::LEN }>(
            "root", &bytes,
        )?)),
        None => None,
    };
    Ok(JournalHead::new(JournalPosition::new(value.position), root))
}

pub(super) fn operation_to_wire(
    operation: &SnapshotOperation,
) -> pb::StatePersonaMemorySnapshotOperation {
    use pb::__buffa::oneof::state_persona_memory_snapshot_operation::Operation;
    let operation = match operation {
        SnapshotOperation::Store { snapshot, replaces } => {
            Operation::from(pb::StatePersonaMemorySnapshotStore {
                record: snapshot.encode(),
                replaces_entry_revision: match replaces {
                    EntryExpectation::Absent => None,
                    EntryExpectation::Revision(revision) => Some(revision.get()),
                },
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
        SnapshotOperation::Discard { at } => {
            Operation::from(pb::StatePersonaMemorySnapshotDiscard {
                observed: buffa::MessageField::some(head_to_wire(*at)),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    };
    pb::StatePersonaMemorySnapshotOperation {
        operation: Some(operation),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(super) fn operation_from_wire(
    value: pb::StatePersonaMemorySnapshotOperation,
) -> Result<SnapshotOperation, StateError> {
    use pb::__buffa::oneof::state_persona_memory_snapshot_operation::Operation;
    match value.operation {
        Some(Operation::Store(store)) => Ok(SnapshotOperation::Store {
            snapshot: Box::new(MemorySnapshot::decode(&store.record)?),
            replaces: store
                .replaces_entry_revision
                .map_or(EntryExpectation::Absent, |revision| {
                    EntryExpectation::Revision(Revision::new(revision))
                }),
        }),
        Some(Operation::Discard(discard)) => Ok(SnapshotOperation::Discard {
            at: head_from_wire(required(
                "observed",
                "a discard names the head the rewritten partition commits now",
                discard.observed,
            )?)?,
        }),
        None => Err(malformed(
            "operation",
            "a snapshot command names one known operation",
        )),
    }
}

pub(super) fn metadata_to_wire(
    command: &SnapshotCommand,
) -> pb::StatePersonaMemorySnapshotCommandMetadata {
    let value = command.metadata();
    pb::StatePersonaMemorySnapshotCommandMetadata {
        command_id: value.command_id().as_str().to_owned(),
        namespace: value.scope().namespace().as_str().to_owned(),
        purpose: value.envelope().purpose().as_str().to_owned(),
        command_audience: value.envelope().audience().as_str().to_owned(),
        digest: value.digest().as_bytes().to_vec(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Rebuilds the command a peer sent.
///
/// The identity travels rather than being re-derived here, and that is what
/// makes the refusal reachable: a listener that re-sealed every arrival would
/// mint a fresh identity for a write that reused an old one, and the reuse
/// would land as a new command instead of being refused as the conflict it is.
pub(super) fn command_from_wire(
    metadata: pb::StatePersonaMemorySnapshotCommandMetadata,
    partition: String,
    operation: pb::StatePersonaMemorySnapshotOperation,
) -> Result<SnapshotCommand, StateError> {
    let namespace = NamespaceId::new(metadata.namespace);
    let digest = ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>(
        "digest",
        &metadata.digest,
    )?);
    Ok(SnapshotCommand::new(
        CommandMetadata::new(
            CommandId::new(metadata.command_id),
            polyc_state::persona_memory::store::family(),
            digest,
            snapshot_scope(&namespace),
            CommandEnvelope::new(
                Purpose::new(metadata.purpose),
                Audience::new(metadata.command_audience),
                ResourceBounds::new(MAX_TRANSACTION_PAYLOAD_BYTES, MAX_MUTATIONS_PER_TRANSACTION),
            ),
        ),
        MemoryPartitionId::new(partition),
        operation_from_wire(operation)?,
    ))
}