liminal-sdk 0.19.5

//! The one mapping from a publish reply frame to the SDK's publish result,
//! shared by the TCP, loopback and WebSocket transports so the three cannot
//! drift apart.
//!
//! A `PublishError` is classified by its reason code alone, never by its
//! message text: [`PUBLISH_UNRESOLVED_CODE`] is an unresolved publish and
//! becomes [`SdkError::PublishUnresolved`]; every other code keeps today's
//! [`SdkError::Backpressure`] text. A `PublishAck` becomes a [`DeliveryAck`]
//! whose delivered flag is the `PUBLISH_DELIVERED_FLAG` bit only (never the
//! durable-position bit) and whose durable position is exactly the one the ack
//! carried, never synthesised.

use alloc::format;
use alloc::string::{String, ToString};

use liminal::protocol::PUBLISH_DELIVERED_FLAG;
use liminal_protocol::reason_code::PUBLISH_UNRESOLVED_CODE;

use crate::{DeliveryAck, PressureResponse, SdkError};

/// Maps a `PublishError`'s reason code and message to the SDK error it means;
/// the message is carried verbatim and never consulted to classify.
pub(super) fn publish_error(reason_code: u16, message: Option<String>) -> SdkError {
    let detail = message.unwrap_or_else(|| "no detail".to_string());
    if reason_code == PUBLISH_UNRESOLVED_CODE {
        SdkError::PublishUnresolved {
            description: detail,
        }
    } else {
        SdkError::Backpressure {
            reason: format!("server rejected publish (reason {reason_code}): {detail}"),
        }
    }
}

/// Maps a `PublishAck`'s flags and durable position to a genuine delivery ack.
pub(super) const fn delivery_ack(flags: u8, durable_position: Option<u64>) -> DeliveryAck {
    DeliveryAck::new(
        PressureResponse::Accept,
        flags & PUBLISH_DELIVERED_FLAG != 0,
    )
    .with_durable_position(durable_position)
}

#[cfg(test)]
mod tests {
    use liminal::protocol::Frame;

    use super::*;

    /// The `DeliveryAck` a `PublishAck` frame maps to; `None` for any other frame.
    fn ack_of(frame: &Frame) -> Option<DeliveryAck> {
        match frame {
            Frame::PublishAck {
                flags,
                durable_position,
                ..
            } => Some(delivery_ack(*flags, *durable_position)),
            _ => None,
        }
    }

    #[test]
    fn ack_with_position_maps_to_delivery_ack_position() -> Result<(), SdkError> {
        let missing = || SdkError::Protocol {
            description: "a PublishAck must map to a DeliveryAck".to_string(),
        };
        let with_position = ack_of(&Frame::PublishAck {
            flags: 0x03,
            stream_id: 1,
            message_id: 7,
            durable_position: Some(7717),
        })
        .ok_or_else(missing)?;
        assert!(with_position.is_accepted());
        assert_eq!(with_position.durable_position(), Some(7717));

        let without_position = ack_of(&Frame::PublishAck {
            flags: 0x01,
            stream_id: 1,
            message_id: 7,
            durable_position: None,
        })
        .ok_or_else(missing)?;
        assert!(without_position.is_accepted());
        assert_eq!(without_position.durable_position(), None);
        Ok(())
    }

    #[test]
    fn delivery_ack_answers_no_position_until_given_one() {
        let ack = DeliveryAck::new(PressureResponse::Accept, true);
        assert_eq!(ack.durable_position(), None);
        assert_eq!(
            ack.with_durable_position(Some(5)).durable_position(),
            Some(5)
        );
    }

    #[test]
    fn unresolved_code_maps_to_publish_unresolved_and_others_to_backpressure() {
        assert_eq!(
            publish_error(PUBLISH_UNRESOLVED_CODE, Some("orders k1".to_string())),
            SdkError::PublishUnresolved {
                description: "orders k1".to_string()
            }
        );
        assert_eq!(
            publish_error(PUBLISH_UNRESOLVED_CODE, None),
            SdkError::PublishUnresolved {
                description: "no detail".to_string()
            }
        );
        assert_eq!(
            publish_error(0xFFFF, Some("boom".to_string())),
            SdkError::Backpressure {
                reason: "server rejected publish (reason 65535): boom".to_string()
            }
        );
    }
}