core/server/src/http/reply.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! Committed-reply decoding and classification: result-section grading, typed
//! payload decoders, partition-reply status classification, and the eviction /
//! login error mappings the write and auth paths render through.

use iggy_binary_protocol::consensus::{
    Command, EvictionHeader, HEADER_SIZE, result_code, result_section_len,
};
use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::ConsumerGroupDetailsResponse;
use iggy_binary_protocol::responses::messages::SendMessagesResponse;
use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse;
use iggy_binary_protocol::responses::streams::get_stream::GetStreamResponse;
use iggy_binary_protocol::responses::topics::get_topic::GetTopicResponse;
use iggy_binary_protocol::responses::users::get_user::UserDetailsResponse;
use iggy_binary_protocol::{GenericHeader, ReplyHeader, WireDecode};
use iggy_common::{
    ConsumerGroupDetails, IggyError, StreamDetails, TopicDetails, UserInfoDetails,
    eviction_reason_to_error,
};
use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use tracing::warn;

use crate::dispatch::login_error::LoginRegisterError;
use crate::http::error::{PartitionWriteError, WriteError};

/// Discriminate a partition write reply. Partition replies carry no result
/// section - a denial is empty-bodied and a committed body, where there is one,
/// is the bare typed payload - so the discriminators live in the header.
/// `status` is read first, and that order is load-bearing: every pre-commit
/// denial names itself there (a namespace that does not resolve or is not
/// routable, a dispatch-time authorization denial, the partition primary's
/// delete-of-missing-offset rejection), and each renders through the legacy
/// `IggyError -> status` map, which is what keeps an unresolvable namespace a
/// 404. Only with status 0 does `op` split the rest: a committed reply is
/// built from its prepare header, whose op (the partition group's commit
/// number) is always >= 1, so a 0 there is an ack with nothing committed
/// behind it and must not grade as success.
///
/// Grading only, and it hands the graded header back: the committed body is
/// read separately (see [`send_confirmations`]) because it is
/// operation-specific - consumer-offset writes commit empty, a produce commits
/// its confirmations - and reading it needs the header's `size`. Returning it
/// keeps that a single parse whose failure mode is already decided here, rather
/// than a second one whose fallback would have to invent a body.
pub(in crate::http) fn classify_partition_reply(
    reply: &Frozen<MESSAGE_ALIGN>,
) -> Result<ReplyHeader, PartitionWriteError> {
    let header = reply
        .as_slice()
        .get(..HEADER_SIZE)
        .and_then(|bytes| bytemuck::checked::try_from_bytes::<ReplyHeader>(bytes).ok())
        .ok_or(PartitionWriteError::Rejected(IggyError::InvalidCommand))?;
    if header.command != Command::Reply {
        return Err(PartitionWriteError::Rejected(IggyError::InvalidCommand));
    }
    if header.status != 0 {
        return Err(PartitionWriteError::Rejected(IggyError::from_code(
            header.status,
        )));
    }
    if header.op == 0 {
        return Err(PartitionWriteError::NotFound);
    }
    Ok(*header)
}

/// The per-partition commit confirmations carried by a graded `SendMessages`
/// reply, or `None` when the reply has no readable confirmation.
///
/// `None` is a normal outcome, never a failure: a peer whose partition plane
/// does not stamp the payload commits with an empty body, and a body this build
/// cannot decode is a shape mismatch. Neither unmakes the commit, so both fall
/// back to an empty confirmation list rather than failing a write that already
/// landed.
///
/// `header` is the one [`classify_partition_reply`] graded, so this reads the
/// body without re-deriving its extent.
pub(in crate::http) fn send_confirmations(
    reply: &Frozen<MESSAGE_ALIGN>,
    header: &ReplyHeader,
) -> Option<SendMessagesResponse> {
    let body = partition_reply_body(reply, header);
    if body.is_empty() {
        return None;
    }
    match SendMessagesResponse::decode_from(body) {
        Ok(confirmations) => Some(confirmations),
        Err(error) => {
            warn!(
                ?error,
                "server HTTP: undecodable send_messages commit confirmation"
            );
            None
        }
    }
}

/// A partition reply's body past the header, bounded by the header's `size`
/// rather than by the buffer length: `size` is the frame's authoritative
/// extent, and the typed decoders reject trailing bytes.
fn partition_reply_body<'a>(reply: &'a Frozen<MESSAGE_ALIGN>, header: &ReplyHeader) -> &'a [u8] {
    reply
        .as_slice()
        .get(HEADER_SIZE..header.size as usize)
        .unwrap_or_default()
}

/// Classify a committed reply's leading result section and return the typed
/// payload slice on success. Mirrors the SDK's `split_metadata_result`:
/// `Some(0)` is success and the payload follows the result section; a nonzero
/// first result is a committed business rejection carrying an `IggyError` code
/// (e.g. a duplicate token name); a missing or short section is a malformed
/// committed reply, mapped to an error rather than a false success. Shared by
/// [`submit_write`] and [`create_pat`] so a committed rejection can never render
/// as a 2xx.
pub(in crate::http) fn committed_payload(
    reply: &Message<GenericHeader>,
) -> Result<&[u8], WriteError> {
    let reply_body = reply_body(reply);
    match result_code(reply_body) {
        Some(0) => {
            let payload_start = result_section_len(reply_body)
                .ok_or(WriteError::Rejected(IggyError::InvalidCommand))?;
            reply_body
                .get(payload_start..)
                .ok_or(WriteError::Rejected(IggyError::InvalidCommand))
        }
        Some(code) => Err(WriteError::Rejected(IggyError::from_code(code))),
        None => Err(WriteError::Rejected(IggyError::InvalidCommand)),
    }
}

/// The transient variant of a reply-shaped pre-consensus rejection frame
/// (`[count=1][index=0][code]`, see `build_result_rejection_reply`), or `None`
/// for a committed outcome. Either transient means the op did not commit, so
/// the write path must replay the same request id rather than grade it as a
/// committed result or advance the session gate. The two codes are kept
/// distinct because they exhaust differently: `TransientNotAccepted` never
/// entered the pipeline and is safe to re-issue anywhere, while
/// `TransientNotCommitted` may still commit and only a same-session same-id
/// replay is safe.
pub(in crate::http) fn transient_code(reply: &Message<GenericHeader>) -> Option<IggyError> {
    match result_code(reply_body(reply)) {
        Some(code) if code == IggyError::TransientNotCommitted.as_code() => {
            Some(IggyError::TransientNotCommitted)
        }
        Some(code) if code == IggyError::TransientNotAccepted.as_code() => {
            Some(IggyError::TransientNotAccepted)
        }
        _ => None,
    }
}

/// The reply body past the generic header, bounded by the header's `size`.
fn reply_body(reply: &Message<GenericHeader>) -> &[u8] {
    let size = reply.header().size as usize;
    reply.as_slice().get(HEADER_SIZE..size).unwrap_or_default()
}

/// Decode the `GetStreamResponse` payload of a committed create-stream reply into
/// `StreamDetails`. `payload` is the slice past the result section that
/// [`submit_write`] already validated as a success.
pub(in crate::http) fn decode_stream_details(payload: &[u8]) -> Result<StreamDetails, WriteError> {
    let response = GetStreamResponse::decode_from(payload)
        .map_err(|_| WriteError::Rejected(IggyError::InvalidCommand))?;
    StreamDetails::try_from(response).map_err(WriteError::Rejected)
}

/// Decode the `GetTopicResponse` payload of a committed create-topic reply into
/// `TopicDetails`. `payload` is the slice past the result section that
/// [`submit_write`] already validated as a success.
pub(in crate::http) fn decode_topic_details(payload: &[u8]) -> Result<TopicDetails, WriteError> {
    let response = GetTopicResponse::decode_from(payload)
        .map_err(|_| WriteError::Rejected(IggyError::InvalidCommand))?;
    TopicDetails::try_from(response).map_err(WriteError::Rejected)
}

/// Decode the `UserDetailsResponse` payload of a committed create-user reply into
/// `UserInfoDetails`. `payload` is the slice past the result section that
/// [`submit_write`] already validated as a success.
pub(in crate::http) fn decode_user_details(payload: &[u8]) -> Result<UserInfoDetails, WriteError> {
    let response = UserDetailsResponse::decode_from(payload)
        .map_err(|_| WriteError::Rejected(IggyError::InvalidCommand))?;
    UserInfoDetails::try_from(response).map_err(WriteError::Rejected)
}

/// Decode the `ConsumerGroupDetailsResponse` payload of a committed
/// create-consumer-group reply into `ConsumerGroupDetails`. `payload` is the
/// slice past the result section that [`submit_write`] already validated as a
/// success. The wire-to-domain conversion is infallible.
pub(in crate::http) fn decode_consumer_group_details(
    payload: &[u8],
) -> Result<ConsumerGroupDetails, WriteError> {
    let response = ConsumerGroupDetailsResponse::decode_from(payload)
        .map_err(|_| WriteError::Rejected(IggyError::InvalidCommand))?;
    Ok(ConsumerGroupDetails::from(response))
}

/// Extract the raw one-time token from a [`build_raw_pat_reply`] output. The
/// spliced reply is framed like any committed metadata reply (a success result
/// section, then the `RawPersonalAccessTokenResponse`) so the SDK's
/// `split_metadata_result` can decode it; strip the section the same way here.
pub(in crate::http) fn decode_raw_pat_token(
    reply: &Message<GenericHeader>,
) -> Result<String, WriteError> {
    let payload = committed_payload(reply)?;
    let response = RawPersonalAccessTokenResponse::decode_from(payload)
        .map_err(|_| WriteError::Rejected(IggyError::InvalidCommand))?;
    Ok(response.token.to_string())
}

/// Map an eviction frame to the same typed [`IggyError`] the SDK's
/// `decode_eviction` produces, so an HTTP caller sees the identical status a TCP
/// caller would (session-terminal reasons render as 401 -> re-authenticate).
/// Reuses the shared [`EvictionHeader`] primitive rather than hand-decoding
/// offsets; an unreadable frame falls back to re-authentication.
pub(in crate::http) fn eviction_error(reply: &Message<GenericHeader>) -> IggyError {
    let Some(eviction) = reply
        .as_slice()
        .get(..HEADER_SIZE)
        .and_then(|bytes| bytemuck::checked::try_from_bytes::<EvictionHeader>(bytes).ok())
    else {
        return IggyError::Unauthenticated;
    };
    eviction_reason_to_error(
        eviction.reason,
        eviction.server_protocol_version,
        eviction.server_protocol_version_min,
    )
}

/// Grade a credential-verification failure onto the `IggyError` the legacy
/// HTTP error map already maps to a status + body, so `CustomError` renders
/// what the SDKs are tested against. `verify_*` only yield the first three
/// variants; the tail is unreachable but kept terminal (401) for the
/// `#[non_exhaustive]` enum.
pub(in crate::http) const fn login_error_to_iggy(error: &LoginRegisterError) -> IggyError {
    match error {
        LoginRegisterError::InvalidCredentials => IggyError::InvalidCredentials,
        LoginRegisterError::InvalidToken => IggyError::InvalidPersonalAccessToken,
        LoginRegisterError::UserInactive => IggyError::UserInactive,
        _ => IggyError::Unauthenticated,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use bytes::Bytes;
    use iggy_binary_protocol::Operation;
    use iggy_binary_protocol::PrepareHeader;
    use iggy_binary_protocol::WireEncode;
    use iggy_binary_protocol::responses::messages::SendMessagesConfirmationResponse;

    use crate::responses::{
        NonReplicatedResponse, build_deny_reply, build_empty_reply, build_reply_from_bytes,
        build_reply_with_body,
    };

    use crate::http::wire::build_request_message;

    // The dispatch-time deny frame: an empty body plus `ReplyHeader.status` set
    // to the rule's error code, the request-level channel the SDK peeks before
    // body decode. `build_empty_reply` (status 0) is the ok-shaped counterpart.
    #[test]
    fn deny_reply_stamps_status_over_an_empty_body() {
        let request = build_request_message(Operation::SendMessages, 42, 7, 3, &[]);
        let status = IggyError::Unauthorized.as_code();
        let reply = build_deny_reply(request.header(), 42, 0, 9, status);
        assert_eq!(
            reply.header().size as usize,
            HEADER_SIZE,
            "deny body is empty (header-only)"
        );
        assert_eq!(
            reply.header().status,
            status,
            "status carries the deny code"
        );
        assert_ne!(status, 0, "a deny status is nonzero so the SDK peek fires");
        assert_eq!(reply.header().command, Command::Reply);
        // Echoes the request so the SDK routes it back to the waiting slot.
        assert_eq!(reply.header().request, request.header().request);
        assert_eq!(reply.header().operation, request.header().operation);
    }

    // Enforces the status contract: every ok-path reply builder leaves status
    // 0, so a future builder that leaks a nonzero reserved tail (or sets
    // status by mistake) trips here rather than silently shipping a fake
    // denial. The deny builders (build_deny_reply here, the partition
    // primary's consensus::build_deny_reply_from_request) are the only
    // nonzero-status writers.
    #[test]
    fn only_the_deny_builder_emits_a_nonzero_status() {
        let request = build_request_message(Operation::CreateStream, 42, 7, 3, &[]);
        let header = request.header();
        let commit = 9;
        let body = Bytes::from_static(b"body");

        // Reply builders, all funnelled through build_reply_with_body.
        for status in [
            build_reply_with_body(header, 42, 7, commit, 0, |_| {})
                .header()
                .status,
            build_empty_reply(header, 42, 7, commit).header().status,
            build_reply_from_bytes(header, 42, 7, commit, &body)
                .header()
                .status,
            NonReplicatedResponse::Empty
                .into_reply(header, 42, 7, commit)
                .header()
                .status,
            NonReplicatedResponse::Bytes(body.clone())
                .into_reply(header, 42, 7, commit)
                .header()
                .status,
        ] {
            assert_eq!(status, 0, "ok-path reply builders must leave status 0");
        }

        // The consensus plane's committed-reply builder (plane_helpers path).
        let prepare = PrepareHeader {
            command: Command::Prepare,
            operation: Operation::SendMessages,
            client: 42,
            op: 1,
            request: 1,
            ..Default::default()
        };
        let committed = consensus::build_reply_message(&prepare, &Bytes::new());
        assert_eq!(
            committed.header().status,
            0,
            "committed replies carry status 0"
        );

        // The intentional deny writer stamps nonzero.
        assert_ne!(
            build_deny_reply(header, 42, 7, commit, IggyError::Unauthorized.as_code())
                .header()
                .status,
            0,
            "the deny builder must stamp a nonzero status"
        );
    }

    fn frozen(reply: Message<iggy_binary_protocol::ReplyHeader>) -> Frozen<MESSAGE_ALIGN> {
        reply.into_generic().into_frozen()
    }

    #[test]
    fn committed_partition_reply_classifies_as_success() {
        let prepare = PrepareHeader {
            command: Command::Prepare,
            operation: Operation::SendMessages,
            client: 42,
            op: 1,
            request: 1,
            ..Default::default()
        };
        let reply = frozen(consensus::build_reply_message(&prepare, &Bytes::new()));
        assert!(classify_partition_reply(&reply).is_ok());
    }

    /// The 204 path of the offset write routes: a committed
    /// `StoreConsumerOffset` reply (op >= 1) classifies as success.
    #[test]
    fn committed_offset_write_reply_classifies_as_success() {
        let prepare = PrepareHeader {
            command: Command::Prepare,
            operation: Operation::StoreConsumerOffset,
            client: 42,
            op: 3,
            request: 2,
            ..Default::default()
        };
        let reply = frozen(consensus::build_reply_message(&prepare, &Bytes::new()));
        assert!(classify_partition_reply(&reply).is_ok());
    }

    /// An ack with no commit number behind it: status 0 and `op` 0. A routing
    /// failure names itself in `status`, so whatever emits this shape never
    /// reached the partition plane, and grading it as success would report a
    /// write that never happened.
    #[test]
    fn status_zero_op_zero_ack_classifies_as_not_found() {
        let request = build_request_message(Operation::SendMessages, 42, 7, 1, &[]);
        let reply = frozen(build_empty_reply(request.header(), 42, 0, 9));
        assert!(matches!(
            classify_partition_reply(&reply),
            Err(PartitionWriteError::NotFound)
        ));
    }

    #[test]
    fn non_reply_frame_classifies_as_rejected() {
        let request = build_request_message(Operation::SendMessages, 42, 7, 1, &[]);
        assert!(matches!(
            classify_partition_reply(&request.into_generic().into_frozen()),
            Err(PartitionWriteError::Rejected(IggyError::InvalidCommand))
        ));
    }

    /// The pre-consensus retry frame must classify as transient so the write
    /// path replays the same request id instead of advancing the session gate
    /// and grading it as a committed rejection (which rendered as a terminal
    /// HTTP error).
    #[test]
    fn transient_rejection_frame_classifies_as_transient() {
        let request = build_request_message(Operation::CreateStream, 42, 7, 3, &[]);
        let reply = consensus::build_result_rejection_reply(
            request.header(),
            9,
            IggyError::TransientNotCommitted.as_code(),
        )
        .into_generic();
        assert_eq!(
            transient_code(&reply),
            Some(IggyError::TransientNotCommitted)
        );
        assert!(matches!(
            committed_payload(&reply),
            Err(WriteError::Rejected(IggyError::TransientNotCommitted))
        ));

        let not_accepted = consensus::build_result_rejection_reply(
            request.header(),
            9,
            IggyError::TransientNotAccepted.as_code(),
        )
        .into_generic();
        assert_eq!(
            transient_code(&not_accepted),
            Some(IggyError::TransientNotAccepted)
        );
    }

    /// Genuine committed outcomes must advance the gate, so neither a success
    /// reply nor a committed business rejection may trip the replay
    /// classification.
    #[test]
    fn committed_replies_do_not_classify_as_transient() {
        let request = build_request_message(Operation::CreateStream, 42, 7, 3, &[]);
        // `[count=1][index=0][result=0]` success section, then the payload.
        let mut body = Vec::new();
        for word in [1u32, 0, 0] {
            body.extend_from_slice(&word.to_le_bytes());
        }
        body.extend_from_slice(b"payload");
        let success =
            build_reply_from_bytes(request.header(), 42, 7, 9, &Bytes::from(body)).into_generic();
        assert_eq!(transient_code(&success), None);
        let Ok(payload) = committed_payload(&success) else {
            panic!("success section must grade ok");
        };
        assert_eq!(payload, b"payload");

        let rejected = consensus::build_result_rejection_reply(
            request.header(),
            9,
            IggyError::UserAlreadyExists.as_code(),
        )
        .into_generic();
        assert_eq!(transient_code(&rejected), None);
        assert!(matches!(
            committed_payload(&rejected),
            Err(WriteError::Rejected(IggyError::UserAlreadyExists))
        ));
    }

    fn send_reply(body: &Bytes) -> Frozen<MESSAGE_ALIGN> {
        let prepare = PrepareHeader {
            command: Command::Prepare,
            operation: Operation::SendMessages,
            client: 42,
            op: 1,
            request: 1,
            ..Default::default()
        };
        frozen(consensus::build_reply_message(&prepare, body))
    }

    /// Walks the same order the write path does: grade first, then read the
    /// body with the header that grading returned.
    fn graded_send_confirmations(body: &Bytes) -> Option<SendMessagesResponse> {
        let reply = send_reply(body);
        let header = classify_partition_reply(&reply).expect("committed send reply grades ok");
        send_confirmations(&reply, &header)
    }

    #[test]
    fn committed_send_reply_yields_its_confirmations() {
        let response = SendMessagesResponse {
            confirmations: vec![SendMessagesConfirmationResponse {
                stream_id: 3,
                topic_id: 5,
                partition_id: 7,
                base_offset: 41,
            }],
        };
        assert_eq!(
            graded_send_confirmations(&response.to_bytes()),
            Some(response)
        );
    }

    /// A zero-count body is a committed batch that reports no offsets, which is
    /// distinct from the empty body below: it decodes, so the response carries
    /// an empty confirmation list rather than falling back to no body at all.
    #[test]
    fn zero_count_commit_body_yields_empty_confirmations() {
        let empty = SendMessagesResponse {
            confirmations: Vec::new(),
        };
        assert_eq!(graded_send_confirmations(&empty.to_bytes()), Some(empty));
    }

    /// A peer whose partition plane does not stamp the payload commits with an
    /// empty body. The write still landed, so this is `None`, not an error, and
    /// the handler answers 201 with an empty confirmation list.
    #[test]
    fn empty_commit_body_yields_no_confirmations() {
        assert_eq!(graded_send_confirmations(&Bytes::new()), None);
    }

    /// Same fallback for a body this build cannot read: a shape mismatch must
    /// not fail a produce that already committed.
    #[test]
    fn undecodable_commit_body_yields_no_confirmations() {
        let truncated = Bytes::from_static(&[1, 0, 0, 0, 9]);
        assert_eq!(graded_send_confirmations(&truncated), None);
    }

    /// The partition primary's typed pre-commit deny (delete of a missing
    /// consumer offset) rides `ReplyHeader.status` and must classify as the
    /// mapped `IggyError`, not as the generic op-0 not-found.
    #[test]
    fn status_bearing_deny_reply_classifies_as_typed_rejection() {
        let request = build_request_message(Operation::DeleteConsumerOffset, 42, 7, 1, &[]);
        let mut deny = build_empty_reply(request.header(), 42, 0, 9);
        let header = bytemuck::checked::try_from_bytes_mut::<ReplyHeader>(
            &mut deny.as_mut_slice()[..HEADER_SIZE],
        )
        .expect("empty reply header is valid");
        header.status = IggyError::ConsumerOffsetNotFound(0).as_code();
        assert!(matches!(
            classify_partition_reply(&frozen(deny)),
            Err(PartitionWriteError::Rejected(
                IggyError::ConsumerOffsetNotFound(_)
            ))
        ));
    }
}