liminal-server 0.25.4

//! The bridge prepares a crossing once; the chosen path owns its encryption and recovery.
use super::{Carrier, CarrierError, CarrierReceive, LinkCarrier, WireCarrier};
use crate::console::records::Carrier as ObservedCarrier;
use crate::link::{LinkEnd, LinkLife};
use crate::link_dial::DialedLink;
use crate::link_handshake::{LinkTransport, MachineIdentity};
use crate::link_open::network::NetworkBackend;
use crate::link_open::{OpenError, Opened};
use std::sync::{Arc, Mutex};
mod stop;
pub use stop::Interrupt;

/// A bridge's configured prefixes and one common carrier owner.
#[derive(Debug)]
pub struct ChoiceLink {
    pub(crate) link_id: String,
    pub(crate) exports: Vec<String>,
    pub(crate) transport: Preparation,
    pub(crate) carrier: ChoiceCarrier,
}
impl ChoiceLink {
    /// Retains the selected carrier and its message-scoped recovery.
    /// # Errors
    /// The carrier cannot be interrupted or its stop adapter cannot start.
    pub fn chosen(link: Opened<NetworkBackend>) -> Result<Self, OpenError> {
        let peer = link.backend().peer();
        let link_id = link.description().link_id.clone();
        let exports = link.description().exports.clone();
        let stop = link.interrupt_handle()?;
        let interrupt = Interrupt::chosen(stop.clone())?;
        let last = link
            .carrier()
            .ok_or_else(|| OpenError("link_has_no_carrier".to_owned()))?;
        let life = link.life().unwrap_or_else(|| LinkLife::new(peer));
        let observed = Arc::new(Mutex::new(link.observed_carrier()));
        Ok(Self {
            link_id,
            exports,
            transport: Preparation::Chosen(peer),
            carrier: ChoiceCarrier {
                path: Path::Chosen(Box::new(link)),
                last,
                life: Some(life),
                observed,
                interrupt: Some(interrupt),
                answer: None,
                failure: std::sync::OnceLock::new(),
            },
        })
    }
}
impl From<DialedLink> for ChoiceLink {
    fn from(link: DialedLink) -> Self {
        let life = link.carrier.life().cloned();
        Self {
            link_id: link.link_id,
            exports: link.exports,
            transport: Preparation::Wire(link.transport),
            carrier: ChoiceCarrier {
                path: Path::Wire(link.carrier),
                last: Carrier::Wire,
                life,
                observed: Arc::new(Mutex::new(Some(ObservedCarrier::Wire))),
                interrupt: None,
                answer: None,
                failure: std::sync::OnceLock::new(),
            },
        }
    }
}
#[derive(Debug)]
pub(crate) enum Preparation {
    Wire(LinkTransport),
    Chosen(MachineIdentity),
}
impl Preparation {
    pub(crate) const fn peer(&self) -> &MachineIdentity {
        match self {
            Self::Wire(transport) => transport.peer(),
            Self::Chosen(peer) => peer,
        }
    }
    pub(crate) fn prepare(&mut self, plain: &[u8], output: &mut [u8]) -> Result<usize, OpenError> {
        match self {
            Self::Wire(transport) => transport
                .seal(plain, output)
                .map_err(|error| OpenError(error.to_string())),
            Self::Chosen(_) => {
                let target = output
                    .get_mut(..plain.len())
                    .ok_or_else(|| OpenError("crossing_buffer_too_small".to_owned()))?;
                target.copy_from_slice(plain);
                Ok(plain.len())
            }
        }
    }
}
#[derive(Debug)]
enum Path {
    Wire(WireCarrier),
    Chosen(Box<Opened<NetworkBackend>>),
}
#[derive(Debug)]
pub(crate) struct ChoiceCarrier {
    path: Path,
    last: Carrier,
    life: Option<LinkLife>,
    observed: Arc<Mutex<Option<ObservedCarrier>>>,
    interrupt: Option<Interrupt>,
    answer: Option<CarrierReceive>,
    /// The first failure the chosen path answered, by name.
    failure: std::sync::OnceLock<String>,
}
impl ChoiceCarrier {
    pub(crate) fn interrupt_handle(&mut self) -> Result<Interrupt, CarrierError> {
        match &self.path {
            Path::Wire(wire) => wire.interrupt_handle().map(Interrupt::wire),
            Path::Chosen(_) => self.interrupt.take().ok_or(CarrierError::Ended),
        }
    }
    pub(crate) const fn life(&self) -> Option<&LinkLife> {
        self.life.as_ref()
    }
    pub(crate) fn observe_carrier(&self) -> Arc<Mutex<Option<ObservedCarrier>>> {
        Arc::clone(&self.observed)
    }
    pub(crate) fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
        match &self.path {
            Path::Wire(wire) => wire.refuse_unsolicited(),
            Path::Chosen(link) => link
                .refuse_unsolicited()
                .map_err(|error| CarrierError::Lifecycle(error.0)),
        }
    }
}
impl LinkCarrier for ChoiceCarrier {
    fn carrier(&self) -> Carrier {
        self.last
    }
    fn ended(&self) -> Option<String> {
        match &self.path {
            Path::Wire(wire) => wire.ended(),
            Path::Chosen(_) => self.failure.get().cloned().or_else(|| {
                self.life
                    .as_ref()
                    .and_then(|life| life.reason().ok().flatten())
            }),
        }
    }
    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
        match &mut self.path {
            Path::Wire(wire) => wire.send(subject, sealed),
            Path::Chosen(link) => {
                let result = link.send(subject, sealed);
                if let Some(carrier) = link.carrier() {
                    self.last = carrier;
                }
                *self.observed.lock().map_err(|_| {
                    CarrierError::Lifecycle("carrier_observation_poisoned".to_owned())
                })? = link.observed_carrier();
                match result {
                    Ok(()) => {
                        let message_id = u64::from_be_bytes(
                            sealed
                                .get(..8)
                                .ok_or(CarrierError::UnexpectedFrame)?
                                .try_into()
                                .map_err(|_| CarrierError::UnexpectedFrame)?,
                        );
                        self.answer = Some(CarrierReceive::Acknowledged {
                            message_id,
                            durable_position: None,
                        });
                        Ok(())
                    }
                    Err(error) if error.0.starts_with("crossing_refused:") => {
                        let parts = error
                            .0
                            .strip_prefix("crossing_refused:")
                            .and_then(|value| value.split_once(':'))
                            .ok_or(CarrierError::UnexpectedFrame)?;
                        self.answer = Some(CarrierReceive::Refused {
                            reason_code: parts
                                .0
                                .parse()
                                .map_err(|_| CarrierError::UnexpectedFrame)?,
                            reason: Some(parts.1.to_owned()),
                        });
                        Ok(())
                    }
                    Err(error)
                        if link.stopped()
                            || matches!(
                                error.0.as_str(),
                                "network_stopped" | "udp_receiver_stopped"
                            ) =>
                    {
                        self.failure.get_or_init(|| error.0.clone());
                        Err(std::io::Error::new(std::io::ErrorKind::Interrupted, error.0).into())
                    }
                    Err(error) => {
                        self.failure.get_or_init(|| error.0.clone());
                        if let Some(life) = &self.life {
                            life.end(LinkEnd::Refused(error.0.clone()))
                                .map_err(CarrierError::Lifecycle)?;
                        }
                        Err(CarrierError::Lifecycle(error.0))
                    }
                }
            }
        }
    }
    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
        match &mut self.path {
            Path::Wire(wire) => wire.receive(),
            Path::Chosen(_) => self.answer.take().ok_or(CarrierError::UnexpectedFrame),
        }
    }
}