liminal-server 0.25.4

use super::{
    Crossing, DialedLink, LINK_CROSSING_REFUSED_CODE, PendingDelivery, ServerError, Wake, failure,
};
use crate::link_carrier::{CarrierError, CarrierReceive, LinkCarrier};
use crate::link_handshake::{NOISE_MAX_MESSAGE, NOISE_OVERHEAD};
use liminal::protocol::{CausalContext, MessageEnvelope};

pub(super) fn cross(
    link: &mut DialedLink,
    source: &mut PendingDelivery,
    wake: &Wake,
) -> Result<Crossing, ServerError> {
    let envelope = MessageEnvelope::new(
        source.schema,
        CausalContext::independent(),
        std::mem::take(&mut source.payload),
    );
    let position = source
        .position
        .ok_or_else(|| failure("exported message has no stable subscription position"))?;
    let Some(sealed) = seal(position, &envelope, |plain, sealed| {
        link.transport
            .prepare(plain, sealed)
            .map_err(|error| failure(error.to_string()))
    })?
    else {
        return Ok(Crossing::TooLarge);
    };
    exchange(link, &source.subject, &sealed, wake)
}

pub(super) fn seal(
    position: u64,
    envelope: &MessageEnvelope,
    prepare: impl FnOnce(&[u8], &mut [u8]) -> Result<usize, ServerError>,
) -> Result<Option<Vec<u8>>, ServerError> {
    let mut plain = position.to_be_bytes().to_vec();
    plain.extend(
        envelope
            .to_wire_bytes()
            .map_err(|error| failure(error.to_string()))?,
    );
    if plain.len().saturating_add(NOISE_OVERHEAD) > NOISE_MAX_MESSAGE {
        return Ok(None);
    }
    let mut sealed = vec![0; plain.len() + NOISE_OVERHEAD];
    let written = prepare(&plain, &mut sealed)?;
    sealed.truncate(written);
    Ok(Some(sealed))
}

pub(super) fn exchange(
    link: &mut DialedLink,
    subject: &str,
    sealed: &[u8],
    wake: &Wake,
) -> Result<Crossing, ServerError> {
    if let Err(error) = link.carrier.send(subject, sealed) {
        return interrupted(&error, wake, link.carrier.life());
    }
    let answer = match link.carrier.receive() {
        Ok(answer) => answer,
        Err(error) => return interrupted(&error, wake, link.carrier.life()),
    };
    match answer {
        CarrierReceive::Acknowledged { .. } => Ok(Crossing::Acknowledged),
        CarrierReceive::Refused {
            reason_code: LINK_CROSSING_REFUSED_CODE,
            reason,
        } => Ok(Crossing::Refused(reason)),
        CarrierReceive::End { reason } if wake.stopping() || reason == "peer_gone" => {
            wake.check()?;
            tracing::info!(%reason, "bridge carrier interrupted; subscription position retained");
            Ok(Crossing::Stopped)
        }
        CarrierReceive::End { reason } => Err(failure(format!(
            "link '{}' ended before its crossing answer: {reason}",
            link.link_id
        ))),
        CarrierReceive::Refused {
            reason_code,
            reason,
        } => Err(failure(format!(
            "link '{}' publish error {reason_code}: {reason:?}",
            link.link_id
        ))),
        CarrierReceive::Sealed { .. } => Err(failure(
            "export carrier received sealed traffic instead of a crossing answer",
        )),
    }
}

fn interrupted(
    error: &CarrierError,
    wake: &Wake,
    life: Option<&crate::link::LinkLife>,
) -> Result<Crossing, ServerError> {
    let peer_gone = life
        .map(crate::link::LinkLife::reason)
        .transpose()
        .map_err(failure)?
        .flatten()
        .as_deref()
        == Some("peer_gone");
    if (matches!(error, CarrierError::Io(_) | CarrierError::Ended) && wake.stopping())
        || (matches!(
            error,
            CarrierError::Io(_) | CarrierError::Ended | CarrierError::Lifecycle(_)
        ) && peer_gone)
    {
        wake.check()?;
        tracing::info!(%error, "bridge carrier interrupted; subscription position retained");
        Ok(Crossing::Stopped)
    } else {
        Err(failure(error.to_string()))
    }
}