liminal-server 0.25.4

//! A foreign export's prepared frame settles against a real local subscription.

use super::{BridgeCounters, LinkBridge, Observation, PendingDelivery, ServerError, failure, lock};
use crate::auth_lys::SubjectBinding;
use crate::link_dial::DialedLink;
use crate::link_handshake::{LinkTransport, NOISE_MAX_MESSAGE};
use crate::server::connection::services::LiminalConnectionServices;
use liminal::protocol::{Frame, MessageEnvelope};
use std::sync::Arc;

#[derive(Debug)]
pub(super) struct PreparedCrossing {
    source: String,
    position: u64,
    subject: String,
    sealed: Vec<u8>,
}

impl PreparedCrossing {
    fn from_frame(source: String, position: u64, frame: Frame) -> Result<Self, ServerError> {
        let Frame::LinkMessage {
            flags: 0,
            stream_id: 1,
            subject,
            sealed,
        } = frame
        else {
            return Err(failure("prepared crossing must name initiator stream 1"));
        };
        for name in [&source, &subject] {
            SubjectBinding::new("wire".to_owned(), name.clone(), name.clone())
                .map_err(|error| failure(error.to_string()))?;
        }
        if sealed.is_empty() || sealed.len() > NOISE_MAX_MESSAGE {
            return Err(failure("prepared crossing has an invalid sealed size"));
        }
        Ok(Self {
            source,
            position,
            subject,
            sealed,
        })
    }
}

impl LinkBridge {
    /// Seals a foreign subject with the ordinary crossing body and Noise transport.
    /// # Errors
    /// The envelope cannot be encoded, exceeds the crossing limit, or sealing fails.
    pub fn prepare_crossing_for_test(
        transport: &mut LinkTransport,
        subject: String,
        position: u64,
        envelope: &MessageEnvelope,
    ) -> Result<Frame, ServerError> {
        let sealed = super::crossing::seal(position, envelope, |plain, sealed| {
            transport
                .seal(plain, sealed)
                .map_err(|error| failure(error.to_string()))
        })?
        .ok_or_else(|| failure("prepared crossing is too large"))?;
        Ok(Frame::LinkMessage {
            flags: 0,
            stream_id: 1,
            subject,
            sealed,
        })
    }

    /// Sends a hand-built foreign crossing as the first durable subscription delivery.
    ///
    /// The frame was sealed once with this dialed link's transport; its answer uses
    /// the ordinary bridge counters, lifecycle, and the source subscription's cursor.
    ///
    /// # Errors
    /// The frame is not an initiator-stream crossing, its names or size are invalid,
    /// its transport nonce is not the first, or bridge startup fails.
    pub fn start_with_prepared_crossing_for_test(
        link: DialedLink,
        services: Arc<LiminalConnectionServices>,
        source: String,
        position: u64,
        frame: Frame,
    ) -> Result<Self, ServerError> {
        let prepared = PreparedCrossing::from_frame(source, position, frame)?;
        if link.transport.sending_nonce() != 1 {
            return Err(failure("prepared crossing must be sealed exactly once"));
        }
        let peer = hex::encode(link.transport.peer().ed25519_public());
        Self::start_observing(
            link,
            services,
            peer,
            BridgeCounters::default(),
            |observation| {
                *lock(
                    &observation.prepared_crossing,
                    "prepared crossing installation",
                )? = Some(prepared);
                Ok(())
            },
        )
    }
}

pub(super) fn take(
    observation: &Observation,
    source: &mut PendingDelivery,
) -> Result<Option<Vec<u8>>, ServerError> {
    let mut prepared = lock(&observation.prepared_crossing, "prepared crossing delivery")?;
    let Some(first) = prepared.as_ref() else {
        return Ok(None);
    };
    if first.source != source.subject || Some(first.position) != source.position {
        return Err(failure(
            "prepared crossing does not match the first durable delivery",
        ));
    }
    let first = prepared
        .take()
        .ok_or_else(|| failure("prepared crossing disappeared"))?;
    drop(prepared);
    source.subject = first.subject;
    Ok(Some(first.sealed))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::link_handshake::{LinkHandshake, LocalMachine, NOISE_OVERHEAD};
    use liminal::protocol::{CausalContext, SchemaId};
    use lys_core::keys::Ed25519Identity;
    use zeroize::Zeroizing;

    #[test]
    fn prepared_crossing_accepts_the_real_sealers_output() -> Result<(), Box<dyn std::error::Error>>
    {
        let local = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([1; 32])));
        let peer = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([2; 32])));
        let mut initiator = LinkHandshake::initiator(&local, peer.machine())?;
        let mut responder = LinkHandshake::responder(&peer, local.machine())?;
        let mut bytes = [0; NOISE_OVERHEAD];
        let written = initiator.write(&mut bytes)?;
        responder.read(&bytes[..written])?;
        let written = responder.write(&mut bytes)?;
        initiator.read(&bytes[..written])?;
        let mut sender = initiator.finish()?;
        let mut receiver = responder.finish()?;
        let envelope = MessageEnvelope::new(
            SchemaId::new([0; 32]),
            CausalContext::independent(),
            b"{}".to_vec(),
        );
        let position = 7;
        let frame = LinkBridge::prepare_crossing_for_test(
            &mut sender,
            "liminal.foreign".to_owned(),
            position,
            &envelope,
        )?;
        let Frame::LinkMessage { sealed, .. } = &frame else {
            return Err("prepared frame is not a crossing".into());
        };
        assert!(
            sealed.len() <= NOISE_OVERHEAD,
            "fixture must expose the buffer allowance being used as a minimum"
        );
        let mut plain = vec![0; NOISE_MAX_MESSAGE];
        let written = receiver.open(sealed, &mut plain)?;
        let mut expected = position.to_be_bytes().to_vec();
        expected.extend(envelope.to_wire_bytes()?);
        assert_eq!(&plain[..written], expected);
        let prepared = PreparedCrossing::from_frame("orders.secret".to_owned(), position, frame)?;
        assert_eq!(prepared.subject, "liminal.foreign");
        assert_eq!(prepared.position, position);
        assert_eq!(sender.sending_nonce(), 1);
        Ok(())
    }
}