ibapi 4.1.0

//! Common message routing logic for sync and async implementations

use log::warn;

use crate::errors::Error;
use crate::messages::{is_informational_code, routes_by_request_id, IncomingMessages, Notice, ResponseMessage};

use super::RoutedItem;

/// Represents how a message should be routed
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum RoutingDecision {
    /// Route by request ID
    ByRequestId(i32),
    /// Route by order ID
    ByOrderId(i32),
    /// Route by message type to shared channel
    ByMessageType(IncomingMessages),
    /// Route to shared message channel
    SharedMessage(IncomingMessages),
    /// Special handling for error messages
    Error(DecodedError),
    /// Shutdown signal
    Shutdown,
}

/// Decoded contents of an Error wire message (type 4), populated regardless of
/// wire format. Carries both warnings ([`crate::messages::WARNING_CODE_RANGE`])
/// and hard errors.
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct DecodedError {
    pub request_id: i32,
    pub error_code: i32,
    pub error_message: String,
    /// Milliseconds since Unix epoch; `None` for old-format text messages without an error_time field.
    pub error_time: Option<i64>,
    pub advanced_order_reject_json: String,
}

impl Default for DecodedError {
    fn default() -> Self {
        Self {
            request_id: UNSPECIFIED_REQUEST_ID,
            error_code: 0,
            error_message: String::new(),
            error_time: None,
            advanced_order_reject_json: String::new(),
        }
    }
}

/// Decode the protobuf Error envelope. Defaults match the text-path accessors:
/// missing id → `UNSPECIFIED_REQUEST_ID`, missing error_code → 0,
/// missing strings → empty, missing error_time → `None`.
pub(crate) fn decode_error_envelope(raw_bytes: &[u8]) -> Option<DecodedError> {
    let envelope: crate::proto::ErrorMessage = prost::Message::decode(raw_bytes).ok()?;
    Some(DecodedError {
        request_id: envelope.id.unwrap_or(UNSPECIFIED_REQUEST_ID),
        error_code: envelope.error_code.unwrap_or(0),
        error_message: envelope.error_msg.unwrap_or_default(),
        error_time: envelope.error_time,
        advanced_order_reject_json: envelope.advanced_order_reject_json.unwrap_or_default(),
    })
}

fn is_order_message(message_type: IncomingMessages) -> bool {
    matches!(
        message_type,
        IncomingMessages::OrderStatus
            | IncomingMessages::OrderBound
            | IncomingMessages::OpenOrder
            | IncomingMessages::OpenOrderEnd
            | IncomingMessages::CompletedOrder
            | IncomingMessages::CompletedOrdersEnd
            | IncomingMessages::ExecutionData
            | IncomingMessages::ExecutionDataEnd
            | IncomingMessages::CommissionsReport
    )
}

fn is_shared_message(message_type: IncomingMessages) -> bool {
    matches!(
        message_type,
        IncomingMessages::ManagedAccounts | IncomingMessages::NextValidId | IncomingMessages::CurrentTime
    )
}

/// Determine how to route an incoming message
pub(crate) fn determine_routing(message: &ResponseMessage) -> RoutingDecision {
    let message_type = message.message_type();

    if message_type == IncomingMessages::Shutdown {
        return RoutingDecision::Shutdown;
    }

    if message_type == IncomingMessages::Error {
        let raw_bytes = message.raw_bytes();
        let decoded = raw_bytes.and_then(decode_error_envelope).unwrap_or_else(|| {
            if let Some(bytes) = raw_bytes {
                warn!(
                    "error frame did not decode as an ErrorMessage proto ({} bytes); routing it with default fields",
                    bytes.len()
                );
            }
            DecodedError::default()
        });
        return RoutingDecision::Error(decoded);
    }

    // ResponseMessage::{order_id, request_id} are proto-aware, so the same
    // dispatch handles text and protobuf wire formats.
    if is_order_message(message_type) {
        return RoutingDecision::ByOrderId(message.order_id().unwrap_or(-1));
    }
    if is_shared_message(message_type) {
        return RoutingDecision::SharedMessage(message_type);
    }
    if let Some(request_id) = message.request_id() {
        return RoutingDecision::ByRequestId(request_id);
    }
    RoutingDecision::ByMessageType(message_type)
}

/// `true` when a subscription keyed by `request_id` can receive `message_type`.
///
/// Two ways in, matching the arm order of [`determine_routing`]: order-scoped
/// types arrive over the order-id channel, and everything else needs a
/// `text_request_id_field` entry.
///
/// `Error` is not one of them, and used to be exempted here so that decoders
/// declaring it would not trip the guard — const declares, guard exempts,
/// circular. [`determine_routing`] classifies `Error` before the allow-list is
/// consulted, so it reaches a subscription as `RoutedItem::Error`/`Notice` and
/// never as a `Response` to decode. The declarations were removed instead; see
/// `test_no_decoder_declares_dispatcher_intercepted_types`.
fn routable_to_request_id_subscription(message_type: IncomingMessages) -> bool {
    is_order_message(message_type) || routes_by_request_id(message_type)
}

/// The first type in `message_types` that a `request_id`-keyed subscription
/// declares but could never receive; `None` when all of them reach it.
///
/// Guards a failure that is otherwise silent end to end: with no
/// `text_request_id_field` entry [`determine_routing`] falls through to
/// `ByMessageType`, the message goes to a shared channel nobody subscribed to,
/// and the subscription simply never yields that variant. `MessageBusStub`
/// tests inject below the dispatcher, so they stay green. See
/// `docs/rules/wire/proto-aware-accessors.md`.
pub(crate) fn first_unroutable_by_request_id(message_types: &[IncomingMessages]) -> Option<IncomingMessages> {
    message_types.iter().copied().find(|&kind| !routable_to_request_id_subscription(kind))
}

/// Routing strategy for order-related messages.
/// Describes which channel keys to try and in what order.
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum OrderRoutingStrategy {
    /// Deliver only to the order-update stream, never a client-scoped raw order channel.
    OrderUpdateOnly,
    /// Try order_id channel, then request_id channel. Store execution_id mapping.
    ExecutionData,
    /// Try order_id channel, then request_id channel.
    ExecutionDataEnd,
    /// Try order_id channel, then shared channel.
    OrderOrShared,
    /// Route via execution_id only.
    ByExecutionId,
    /// Route to shared channel only.
    SharedOnly,
    /// Route by order_id only.
    ByOrderId,
}

/// Determine the routing strategy for an order-related message type.
pub(crate) fn order_routing_strategy(message_type: IncomingMessages) -> OrderRoutingStrategy {
    match message_type {
        IncomingMessages::OrderBound => OrderRoutingStrategy::OrderUpdateOnly,
        IncomingMessages::ExecutionData => OrderRoutingStrategy::ExecutionData,
        IncomingMessages::ExecutionDataEnd => OrderRoutingStrategy::ExecutionDataEnd,
        IncomingMessages::OpenOrder | IncomingMessages::OrderStatus => OrderRoutingStrategy::OrderOrShared,
        IncomingMessages::CommissionsReport => OrderRoutingStrategy::ByExecutionId,
        IncomingMessages::CompletedOrder | IncomingMessages::OpenOrderEnd | IncomingMessages::CompletedOrdersEnd => OrderRoutingStrategy::SharedOnly,
        _ => OrderRoutingStrategy::ByOrderId,
    }
}

/// Request ID for unspecified errors
pub(crate) const UNSPECIFIED_REQUEST_ID: i32 = -1;

/// Notice to copy onto the order-update stream for an error frame, or `None`
/// when the frame is not order-bound: request-less frames and frames whose id
/// is owned by a data-request subscription carry nothing an order consumer
/// should see.
pub(crate) fn order_update_notice(payload: &DecodedError, id_owned_by_data_request: bool) -> Option<Notice> {
    (payload.request_id != UNSPECIFIED_REQUEST_ID && !id_owned_by_data_request).then(|| Notice::from(payload.clone()))
}

/// The outcome of classifying an inbound error frame.
///
/// The *policy* (which arm applies) is centralised here; each transport
/// provides only the runtime-specific delivery in its `route_error_message`.
#[derive(Debug)]
pub(crate) enum ErrorDisposition {
    /// Log + `NoticeStream` only (request-less warning).
    NoticeOnly(Notice),
    /// `NoticeStream` + fail-fast fan-out to in-flight one-shot shared
    /// requests (request-less hard error).
    NoticeAndFailOneShots(Notice, Error),
    /// Deliver `RoutedItem` to the subscription that owns `request_id`.
    Route(i32, RoutedItem),
}

/// Classify an inbound error frame into the action each transport must take.
///
/// Extracts the common four-arm policy so that `sync::route_error_message` and
/// `async::route_error_message` are thin runtime-specific delivery shells.
pub(crate) fn classify_error(payload: DecodedError) -> ErrorDisposition {
    let request_id = payload.request_id;
    // Informational frames (the same rule as `Notice::is_informational`) are
    // delivered as a `Notice`: they neither terminate the subscription nor,
    // when request-less, fail the pending one-shots. Code 0 - a code-less
    // frame, or the fallback for an undecodable one - is informational.
    let is_informational = is_informational_code(payload.error_code, &payload.error_message);

    if request_id == UNSPECIFIED_REQUEST_ID {
        let notice = Notice::from(payload.clone());
        if is_informational {
            ErrorDisposition::NoticeOnly(notice)
        } else {
            ErrorDisposition::NoticeAndFailOneShots(notice, Error::from(payload))
        }
    } else {
        let item = if is_informational {
            RoutedItem::Notice(Notice::from(payload))
        } else {
            RoutedItem::Error(Error::from(payload))
        };
        ErrorDisposition::Route(request_id, item)
    }
}

#[cfg(test)]
#[path = "routing_tests.rs"]
mod tests;