polyc-state-connect 2026.9.0

//! Explicit outbound-spend wire mapping.
//!
//! One conversion per type, every field named. Nothing here spreads a default
//! over a struct: a field added on either side must be written down again in
//! both directions or this file stops compiling, which is the whole point.
//!
//! Base-unit amounts cross as their 16-byte big-endian encoding. Protobuf has
//! no 128-bit scalar, and a decimal string would put parsing between a caller
//! and a money figure; a fixed-width encoding either decodes to the exact
//! number or is refused.

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},
    revision::Revision,
    spend::{
        ConversationId, Reservation, ReservationId, SpendCommand, SpendLedger, SpendOperation,
        spend_scope,
    },
    versioned::{EntryExpectation, MAX_MUTATIONS_PER_TRANSACTION, MAX_TRANSACTION_PAYLOAD_BYTES},
};

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

/// Base units as the fixed 16 bytes they always occupy.
fn amount_to_wire(value: u128) -> Vec<u8> {
    value.to_be_bytes().to_vec()
}

/// Base units from exactly 16 bytes, refusing anything else.
fn amount_from_wire(field: &str, value: &[u8]) -> Result<u128, StateError> {
    Ok(u128::from_be_bytes(fixed_bytes::<16>(field, value)?))
}

fn expected_to_wire(value: EntryExpectation) -> pb::StateSpendExpectedEntry {
    use pb::__buffa::oneof::state_spend_expected_entry::Expected;
    let expected = match value {
        EntryExpectation::Absent => Expected::from(pb::StateSpendExpectedAbsent {
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        EntryExpectation::Revision(revision) => Expected::from(pb::StateSpendExpectedRevision {
            revision: revision.get(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
    };
    pb::StateSpendExpectedEntry {
        expected: Some(expected),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn expected_from_wire(value: pb::StateSpendExpectedEntry) -> Result<EntryExpectation, StateError> {
    use pb::__buffa::oneof::state_spend_expected_entry::Expected;
    match value.expected {
        Some(Expected::Absent(_)) => Ok(EntryExpectation::Absent),
        Some(Expected::Revision(value)) => {
            Ok(EntryExpectation::Revision(Revision::new(value.revision)))
        }
        None => Err(malformed(
            "expected",
            "a spend operation declares its exact row premise",
        )),
    }
}

fn reservation_to_wire(value: &Reservation) -> pb::StateSpendReservation {
    pb::StateSpendReservation {
        reservation_id: value.id().as_str().to_owned(),
        amount_base_units: amount_to_wire(value.amount()),
        expires_at_ms: value.expires_at_ms(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn reservation_from_wire(value: pb::StateSpendReservation) -> Result<Reservation, StateError> {
    let amount = amount_from_wire("amount_base_units", &value.amount_base_units)?;
    Ok(Reservation::new(
        ReservationId::new(value.reservation_id),
        amount,
        value.expires_at_ms,
    ))
}

pub(crate) fn ledger_to_wire(value: &SpendLedger) -> pb::StateSpendLedger {
    pb::StateSpendLedger {
        committed_total_base_units: amount_to_wire(value.committed_total()),
        settlement_count: value.settlement_count(),
        reservations: value
            .reservations()
            .iter()
            .map(reservation_to_wire)
            .collect(),
        fence: value.fence().get(),
        updated_at_ms: value.updated_at_ms(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn ledger_from_wire(value: pb::StateSpendLedger) -> Result<SpendLedger, StateError> {
    let committed_total = amount_from_wire(
        "committed_total_base_units",
        &value.committed_total_base_units,
    )?;
    let reservations = value
        .reservations
        .into_iter()
        .map(reservation_from_wire)
        .collect::<Result<Vec<_>, _>>()?;
    Ok(SpendLedger::from_parts(
        committed_total,
        value.settlement_count,
        reservations,
        value.fence,
        value.updated_at_ms,
    ))
}

pub(crate) fn operation_to_wire(value: &SpendOperation) -> pb::StateSpendOperation {
    use pb::__buffa::oneof::state_spend_operation::Operation;
    let operation = match value {
        SpendOperation::Reserve {
            now_ms,
            fence,
            cap,
            reservation,
            ledger,
            expected,
        } => Operation::from(pb::StateSpendReserve {
            now_ms: *now_ms,
            fence: *fence,
            cap_base_units: amount_to_wire(*cap),
            reservation: buffa::MessageField::some(reservation_to_wire(reservation)),
            ledger: buffa::MessageField::some(ledger_to_wire(ledger)),
            expected: buffa::MessageField::some(expected_to_wire(*expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        SpendOperation::Commit {
            now_ms,
            fence,
            reservation,
            actual,
            ledger,
            expected,
        } => Operation::from(pb::StateSpendCommit {
            now_ms: *now_ms,
            fence: *fence,
            reservation_id: reservation.as_str().to_owned(),
            actual_base_units: amount_to_wire(*actual),
            ledger: buffa::MessageField::some(ledger_to_wire(ledger)),
            expected: buffa::MessageField::some(expected_to_wire(*expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        SpendOperation::Release {
            now_ms,
            fence,
            reservation,
            ledger,
            expected,
        } => Operation::from(pb::StateSpendRelease {
            now_ms: *now_ms,
            fence: *fence,
            reservation_id: reservation.as_str().to_owned(),
            ledger: buffa::MessageField::some(ledger_to_wire(ledger)),
            expected: buffa::MessageField::some(expected_to_wire(*expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        SpendOperation::Reap {
            now_ms,
            fence,
            ledger,
            expected,
        } => Operation::from(pb::StateSpendReap {
            now_ms: *now_ms,
            fence: *fence,
            ledger: buffa::MessageField::some(ledger_to_wire(ledger)),
            expected: buffa::MessageField::some(expected_to_wire(*expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
    };
    pb::StateSpendOperation {
        operation: Some(operation),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn operation_from_wire(value: pb::StateSpendOperation) -> Result<SpendOperation, StateError> {
    use pb::__buffa::oneof::state_spend_operation::Operation;
    let ledger = |value| {
        required("ledger", "a spend operation carries its result", value).and_then(ledger_from_wire)
    };
    let expected = |value| {
        expected_from_wire(required(
            "expected",
            "a spend operation carries its premise",
            value,
        )?)
    };
    match value.operation {
        Some(Operation::Reserve(value)) => Ok(SpendOperation::Reserve {
            now_ms: value.now_ms,
            fence: value.fence,
            cap: amount_from_wire("cap_base_units", &value.cap_base_units)?,
            reservation: reservation_from_wire(required(
                "reservation",
                "a spend reservation carries what it holds",
                value.reservation,
            )?)?,
            ledger: ledger(value.ledger)?,
            expected: expected(value.expected)?,
        }),
        Some(Operation::Commit(value)) => Ok(SpendOperation::Commit {
            now_ms: value.now_ms,
            fence: value.fence,
            reservation: ReservationId::new(value.reservation_id),
            actual: amount_from_wire("actual_base_units", &value.actual_base_units)?,
            ledger: ledger(value.ledger)?,
            expected: expected(value.expected)?,
        }),
        Some(Operation::Release(value)) => Ok(SpendOperation::Release {
            now_ms: value.now_ms,
            fence: value.fence,
            reservation: ReservationId::new(value.reservation_id),
            ledger: ledger(value.ledger)?,
            expected: expected(value.expected)?,
        }),
        Some(Operation::Reap(value)) => Ok(SpendOperation::Reap {
            now_ms: value.now_ms,
            fence: value.fence,
            ledger: ledger(value.ledger)?,
            expected: expected(value.expected)?,
        }),
        None => Err(malformed(
            "operation",
            "a spend command names one operation",
        )),
    }
}

pub(crate) fn metadata_to_wire(command: &SpendCommand) -> pb::StateSpendCommandMetadata {
    let value = command.metadata();
    pb::StateSpendCommandMetadata {
        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(),
    }
}

pub(crate) fn command_from_wire(
    metadata: pb::StateSpendCommandMetadata,
    conversation_id: String,
    operation: pb::StateSpendOperation,
) -> Result<SpendCommand, StateError> {
    let namespace = NamespaceId::new(metadata.namespace);
    let digest = ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>(
        "digest",
        &metadata.digest,
    )?);
    Ok(SpendCommand::new(
        CommandMetadata::new(
            CommandId::new(metadata.command_id),
            polyc_state::spend::family(),
            digest,
            spend_scope(&namespace),
            CommandEnvelope::new(
                Purpose::new(metadata.purpose),
                Audience::new(metadata.command_audience),
                ResourceBounds::new(MAX_TRANSACTION_PAYLOAD_BYTES, MAX_MUTATIONS_PER_TRANSACTION),
            ),
        ),
        ConversationId::new(conversation_id),
        operation_from_wire(operation)?,
    ))
}

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

    const NOW: u64 = 1_700_000_000_000;

    fn ledger() -> SpendLedger {
        SpendLedger::from_parts(
            250,
            1,
            vec![Reservation::new(
                ReservationId::new("r-1"),
                u128::MAX,
                NOW + 60_000,
            )],
            7,
            NOW,
        )
    }

    #[test]
    fn every_operation_round_trips_through_the_wire() {
        let operations = [
            SpendOperation::Reserve {
                now_ms: NOW,
                fence: 7,
                cap: u128::MAX,
                reservation: Reservation::new(ReservationId::new("r-1"), 300, NOW + 60_000),
                ledger: ledger(),
                expected: EntryExpectation::Absent,
            },
            SpendOperation::Commit {
                now_ms: NOW + 1,
                fence: 7,
                reservation: ReservationId::new("r-1"),
                actual: 250,
                ledger: ledger(),
                expected: EntryExpectation::Revision(Revision::new(9)),
            },
            SpendOperation::Release {
                now_ms: NOW + 2,
                fence: 7,
                reservation: ReservationId::new("r-1"),
                ledger: ledger(),
                expected: EntryExpectation::Revision(Revision::new(11)),
            },
            SpendOperation::Reap {
                now_ms: NOW + 3,
                fence: 8,
                ledger: ledger(),
                expected: EntryExpectation::Revision(Revision::new(13)),
            },
        ];
        for operation in operations {
            let restored = operation_from_wire(operation_to_wire(&operation))
                .expect("an operation survives its own encoding");
            assert_eq!(restored, operation);
        }
    }

    #[test]
    fn the_ledger_round_trips_through_the_wire() {
        assert_eq!(
            ledger_from_wire(ledger_to_wire(&ledger())).expect("ledger round trip"),
            ledger()
        );
    }

    /// A money figure either decodes to the exact number or is refused. A
    /// short amount silently read as a smaller one is spend the ceiling never
    /// counts.
    #[test]
    fn a_truncated_amount_is_refused_rather_than_read_as_a_smaller_number() {
        let mut wire = ledger_to_wire(&ledger());
        wire.committed_total_base_units.truncate(8);
        assert!(
            ledger_from_wire(wire).is_err(),
            "a truncated committed total was accepted"
        );
        let mut wire = ledger_to_wire(&ledger());
        wire.reservations[0].amount_base_units.push(0);
        assert!(
            ledger_from_wire(wire).is_err(),
            "an overlong reservation amount was accepted"
        );
    }

    #[test]
    fn a_missing_oneof_and_a_missing_premise_fail_closed() {
        assert!(
            operation_from_wire(pb::StateSpendOperation::default()).is_err(),
            "an operation with no variant was accepted"
        );
        assert!(
            expected_from_wire(pb::StateSpendExpectedEntry::default()).is_err(),
            "a premise with no variant was accepted"
        );
        let reap = pb::StateSpendReap {
            now_ms: NOW,
            fence: 1,
            ledger: buffa::MessageField::some(ledger_to_wire(&ledger())),
            expected: buffa::MessageField::none(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        assert!(
            operation_from_wire(pb::StateSpendOperation {
                operation: Some(pb::__buffa::oneof::state_spend_operation::Operation::from(
                    reap
                )),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
            .is_err(),
            "a sweep with no premise was accepted"
        );
    }
}