liminal-rs 0.16.0

use crate::protocol::{
    DELIVER_MESSAGE_ID_FLAG, DeliveredId, DisconnectReason, Frame, FrameType, LinkEnd, LinkOffer,
    LinkOpenOutcome, MessageEnvelope, MessageId, ProtocolError, ProtocolVersion, StreamPressure,
    WorkerActivityDescriptor, WorkerRegisterOutcome, WorkerRegistration,
    frame::{
        DISCONNECT_REASON_BODY_VERSION, DISCONNECT_REASON_NAME_MAX, LINK_END_EXPORT,
        LINK_END_IMPORT, LINK_OPEN_ACK_HELD, LINK_OPEN_ACK_OPENED, LINK_OPEN_ACK_REFUSED,
        PUBLISH_DURABLE_POSITION_FLAG, PUBLISH_IDEMPOTENCY_KEY_FLAG, WORKER_REGISTER_ACK_ACCEPTED,
        WORKER_REGISTER_ACK_REJECTED,
    },
};

use super::payload::PayloadReader;

pub(super) fn decode_known_payload(
    frame_type: FrameType,
    flags: u8,
    stream_id: u32,
    payload: &[u8],
) -> Result<Frame, ProtocolError> {
    let mut reader = PayloadReader::new(payload);
    let frame = match frame_type {
        FrameType::Connect
        | FrameType::ConnectAck
        | FrameType::ConnectError
        | FrameType::Disconnect
        | FrameType::Ping
        | FrameType::Pong => decode_control_payload(frame_type, flags, &mut reader)?,
        FrameType::Subscribe
        | FrameType::SubscribeAck
        | FrameType::SubscribeError
        | FrameType::Unsubscribe => {
            decode_subscription_payload(frame_type, flags, stream_id, &mut reader)?
        }
        FrameType::Publish | FrameType::PublishAck | FrameType::PublishError => {
            decode_publish_payload(frame_type, flags, stream_id, &mut reader)?
        }
        FrameType::ConversationOpen
        | FrameType::ConversationMessage
        | FrameType::ConversationClose
        | FrameType::ConversationError => {
            decode_conversation_payload(frame_type, flags, stream_id, &mut reader)?
        }
        FrameType::Accept | FrameType::Defer | FrameType::Reject => {
            decode_pressure_payload(frame_type, flags, stream_id, &mut reader)?
        }
        FrameType::Push | FrameType::PushReply => {
            decode_push_payload(frame_type, flags, stream_id, &mut reader)?
        }
        FrameType::WorkerRegister => decode_worker_register_payload(flags, &mut reader)?,
        FrameType::WorkerRegisterAck => decode_worker_register_ack_payload(flags, &mut reader)?,
        FrameType::LinkOpen => decode_link_open_payload(flags, stream_id, &mut reader)?,
        FrameType::LinkOpenAck => decode_link_open_ack_payload(flags, stream_id, &mut reader)?,
        FrameType::LinkMessage => Frame::LinkMessage {
            flags,
            stream_id,
            subject: reader.read_string_field()?,
            sealed: reader.read_bytes_field()?,
        },
        FrameType::Deliver => decode_deliver_payload(flags, stream_id, &mut reader)?,
        FrameType::DisconnectReason => decode_disconnect_reason_payload(flags, &mut reader)?,
        FrameType::Unknown(type_id) => Frame::Unknown {
            type_id,
            flags,
            stream_id,
            payload: payload.to_vec(),
        },
    };
    reader.finish()?;
    Ok(frame)
}

fn read_protocol_version(reader: &mut PayloadReader<'_>) -> Result<ProtocolVersion, ProtocolError> {
    Ok(ProtocolVersion::new(reader.read_u16()?, reader.read_u16()?))
}

fn decode_control_payload(
    frame_type: FrameType,
    flags: u8,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    match frame_type {
        FrameType::Connect => Ok(Frame::Connect {
            flags,
            min_version: read_protocol_version(reader)?,
            max_version: read_protocol_version(reader)?,
            auth_token: reader.read_bytes_field()?,
        }),
        FrameType::ConnectAck => Ok(Frame::ConnectAck {
            flags,
            selected_version: read_protocol_version(reader)?,
            capabilities: reader.read_u32()?,
        }),
        FrameType::ConnectError => Ok(Frame::ConnectError {
            flags,
            reason_code: reader.read_u16()?,
            message: reader.read_optional_string()?,
        }),
        FrameType::Disconnect => Ok(Frame::Disconnect { flags }),
        FrameType::Ping => Ok(Frame::Ping { flags }),
        FrameType::Pong => Ok(Frame::Pong { flags }),
        _ => Err(ProtocolError::codec("frame type was not a control frame")),
    }
}

fn decode_subscription_payload(
    frame_type: FrameType,
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    match frame_type {
        FrameType::Subscribe => {
            let channel = reader.read_string_field()?;
            let accepted_schemas = reader.read_schema_ids_field()?;
            let max_in_flight = reader.read_u32()?;
            StreamPressure::new(max_in_flight)?;
            Ok(Frame::Subscribe {
                flags,
                stream_id,
                channel,
                accepted_schemas,
                max_in_flight,
            })
        }
        FrameType::SubscribeAck => Ok(Frame::SubscribeAck {
            flags,
            stream_id,
            subscription_id: reader.read_u64()?,
            selected_schema: reader.read_schema_id()?,
        }),
        FrameType::SubscribeError => Ok(Frame::SubscribeError {
            flags,
            stream_id,
            reason_code: reader.read_u16()?,
            message: reader.read_optional_string()?,
        }),
        FrameType::Unsubscribe => Ok(Frame::Unsubscribe {
            flags,
            stream_id,
            subscription_id: reader.read_u64()?,
        }),
        _ => Err(ProtocolError::codec(
            "frame type was not a subscription frame",
        )),
    }
}

fn decode_publish_payload(
    frame_type: FrameType,
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    match frame_type {
        FrameType::Publish => {
            let channel = reader.read_string_field()?;
            let envelope = MessageEnvelope::deserialize(&reader.read_bytes_field()?)?;
            // The trailing idempotency-key field is present ONLY when the publisher
            // set PUBLISH_IDEMPOTENCY_KEY_FLAG; otherwise the payload ends here and
            // `reader.finish()` confirms no trailing bytes, exactly as before.
            let idempotency_key = if flags & PUBLISH_IDEMPOTENCY_KEY_FLAG == 0 {
                None
            } else {
                Some(reader.read_string_field()?)
            };
            Ok(Frame::Publish {
                flags,
                stream_id,
                channel,
                envelope,
                idempotency_key,
            })
        }
        FrameType::PublishAck => {
            let message_id = reader.read_u64()?;
            // The trailing durable position is present ONLY when the server set
            // PUBLISH_DURABLE_POSITION_FLAG; otherwise the payload ends here and
            // `reader.finish()` refuses trailing bytes, exactly as before.
            let durable_position = if flags & PUBLISH_DURABLE_POSITION_FLAG == 0 {
                None
            } else {
                Some(reader.read_u64()?)
            };
            Ok(Frame::PublishAck {
                flags,
                stream_id,
                message_id,
                durable_position,
            })
        }
        FrameType::PublishError => Ok(Frame::PublishError {
            flags,
            stream_id,
            reason_code: reader.read_u16()?,
            message: reader.read_optional_string()?,
        }),
        _ => Err(ProtocolError::codec("frame type was not a publish frame")),
    }
}

fn decode_conversation_payload(
    frame_type: FrameType,
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    let conversation_id = reader.read_u64()?;
    match frame_type {
        FrameType::ConversationOpen => Ok(Frame::ConversationOpen {
            flags,
            stream_id,
            conversation_id,
            subject: reader.read_string_field()?,
        }),
        FrameType::ConversationMessage => Ok(Frame::ConversationMessage {
            flags,
            stream_id,
            conversation_id,
            envelope: MessageEnvelope::deserialize(&reader.read_bytes_field()?)?,
        }),
        FrameType::ConversationClose => Ok(Frame::ConversationClose {
            flags,
            stream_id,
            conversation_id,
            reason_code: reader.read_optional_u16()?,
            message: reader.read_optional_string()?,
        }),
        FrameType::ConversationError => Ok(Frame::ConversationError {
            flags,
            stream_id,
            conversation_id,
            reason_code: reader.read_u16()?,
            message: reader.read_optional_string()?,
        }),
        _ => Err(ProtocolError::codec(
            "frame type was not a conversation frame",
        )),
    }
}

fn decode_push_payload(
    frame_type: FrameType,
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    let correlation_id = reader.read_u64()?;
    let payload = reader.read_bytes_field()?;
    match frame_type {
        FrameType::Push => Ok(Frame::Push {
            flags,
            stream_id,
            correlation_id,
            payload,
        }),
        FrameType::PushReply => Ok(Frame::PushReply {
            flags,
            stream_id,
            correlation_id,
            payload,
        }),
        _ => Err(ProtocolError::codec("frame type was not a push frame")),
    }
}

fn decode_deliver_payload(
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    let delivery_seq = reader.read_u64()?;
    let envelope = MessageEnvelope::deserialize(&reader.read_bytes_field()?)?;
    // The trailing pair is present ONLY when the server set
    // DELIVER_MESSAGE_ID_FLAG; otherwise the payload ends here and
    // `reader.finish()` refuses trailing bytes, exactly as before.
    let message = if flags & DELIVER_MESSAGE_ID_FLAG == 0 {
        None
    } else {
        Some(DeliveredId {
            boot: reader.read_u64()?,
            id: reader.read_u64()?,
        })
    };
    Ok(Frame::Deliver {
        flags,
        stream_id,
        delivery_seq,
        envelope,
        message,
    })
}

fn decode_pressure_payload(
    frame_type: FrameType,
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    match frame_type {
        FrameType::Accept => Ok(Frame::Accept {
            flags,
            stream_id,
            referenced_message_id: MessageId::new(reader.read_string_field()?),
        }),
        FrameType::Defer => Ok(Frame::Defer {
            flags,
            stream_id,
            referenced_message_id: MessageId::new(reader.read_string_field()?),
            reason: reader.read_optional_string()?,
        }),
        FrameType::Reject => Ok(Frame::Reject {
            flags,
            stream_id,
            referenced_message_id: MessageId::new(reader.read_string_field()?),
            reason: reader.read_optional_string()?,
        }),
        _ => Err(ProtocolError::codec("frame type was not a pressure frame")),
    }
}

fn decode_worker_register_payload(
    flags: u8,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    let namespaces = reader.read_string_vec_field()?;
    let task_queue = reader.read_string_field()?;
    // `node` carries a presence byte, so `None` and `Some("")` stay distinct.
    let node = reader.read_optional_string()?;
    let activity_types = reader.read_string_vec_field()?;
    let identity = reader.read_string_field()?;
    // This sniff consumes WorkerRegister's one trailing-bytes extension slot. Any
    // future field appended to this frame must use a ProtocolVersion gate, never a
    // second sniff: another optional tail is indistinguishable from census bytes.
    let activities = if reader.is_finished() {
        Vec::new()
    } else {
        let count = usize::try_from(reader.read_u32()?)
            .map_err(|_| ProtocolError::codec("activity descriptor count cannot fit usize"))?;
        let mut activities = Vec::new();
        activities
            .try_reserve(count)
            .map_err(|_| ProtocolError::codec("activity descriptor allocation failed"))?;
        for _ in 0..count {
            activities.push(WorkerActivityDescriptor {
                name: reader.read_string_field()?,
                input_schema_json: reader.read_string_field()?,
                output_schema_json: reader.read_string_field()?,
            });
        }
        activities
    };
    Ok(Frame::WorkerRegister {
        flags,
        registration: WorkerRegistration {
            namespaces,
            task_queue,
            node,
            activity_types,
            identity,
            activities,
        },
    })
}

fn decode_worker_register_ack_payload(
    flags: u8,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    let status = reader.read_u8()?;
    let outcome = match status {
        WORKER_REGISTER_ACK_ACCEPTED => WorkerRegisterOutcome::Accepted,
        WORKER_REGISTER_ACK_REJECTED => WorkerRegisterOutcome::Rejected {
            reason: reader.read_string_field()?,
        },
        _ => {
            return Err(ProtocolError::codec(
                "worker register ack status byte was invalid",
            ));
        }
    };
    Ok(Frame::WorkerRegisterAck { flags, outcome })
}

fn decode_link_open_payload(
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    let link_id = reader.read_string_field()?;
    let end = match reader.read_u8()? {
        LINK_END_EXPORT => LinkEnd::Export,
        LINK_END_IMPORT => LinkEnd::Import,
        _ => return Err(ProtocolError::codec("link end byte was invalid")),
    };
    let machine = reader.read_array()?;
    let handshake = reader.read_bytes_field()?;
    Ok(Frame::LinkOpen {
        flags,
        stream_id,
        offer: LinkOffer {
            link_id,
            end,
            machine,
            handshake,
        },
    })
}

fn decode_link_open_ack_payload(
    flags: u8,
    stream_id: u32,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    let outcome = match reader.read_u8()? {
        LINK_OPEN_ACK_OPENED => LinkOpenOutcome::Opened {
            handshake: reader.read_bytes_field()?,
        },
        LINK_OPEN_ACK_HELD => LinkOpenOutcome::Held {
            grant: reader.read_string_field()?,
            draft: reader.read_string_field()?,
        },
        LINK_OPEN_ACK_REFUSED => LinkOpenOutcome::Refused {
            reason: reader.read_string_field()?,
        },
        _ => {
            return Err(ProtocolError::codec(
                "link open ack status byte was invalid",
            ));
        }
    };
    Ok(Frame::LinkOpenAck {
        flags,
        stream_id,
        outcome,
    })
}

/// ACCESS-005 R5: the reason-bearing close. Every field is checked before a
/// frame is built: the body version is the one this implementation reads,
/// the name is within its bound, and the code and name name one registered
/// reason together. Anything else is a named codec error, never a silent
/// close; trailing bytes are refused by the caller's `finish`.
fn decode_disconnect_reason_payload(
    flags: u8,
    reader: &mut PayloadReader<'_>,
) -> Result<Frame, ProtocolError> {
    if reader.read_u8()? != DISCONNECT_REASON_BODY_VERSION {
        return Err(ProtocolError::codec(
            "disconnect reason body version is not supported",
        ));
    }
    let code = reader.read_u16()?;
    let name = reader.read_string_field()?;
    if name.len() > DISCONNECT_REASON_NAME_MAX {
        return Err(ProtocolError::codec(
            "disconnect reason name exceeds its bound",
        ));
    }
    let reason = DisconnectReason::from_wire(code, &name).ok_or_else(|| {
        ProtocolError::codec("disconnect reason code and name are not one registered reason")
    })?;
    Ok(Frame::DisconnectReason { flags, reason })
}