futu_backend/trade_query/crypto_orders/

queries_orders.rs

//! trade_query/crypto_orders/queries_orders — query_crypto_orders / order_info / history_orders
//! (v1.4.110 CC Batch K: 拆自 crypto_orders.rs L57-292)

use futu_cache::trd_cache::{CachedOrder, OrderRelationSourceToken};
use futu_core::error::{FutuError, Result};
use futu_domain_trade_history::pagination::{
    CryptoReadCursorPageDecision, crypto_read_order_pagination_exceeded_like_cpp,
    decide_crypto_read_order_page_like_cpp,
};
use futu_domain_trade_history::status::{
    BackendHistoryMsgHeaderFacts, BackendHistoryStatusError,
    crypto_order_detail_response_status_like_cpp, crypto_order_read_response_status_like_cpp,
};
use futu_domain_trade_history::{
    plan_crypto_history_order_list_request_like_cpp, plan_crypto_order_detail_request_like_cpp,
    plan_crypto_order_list_request_like_cpp,
};

use super::super::*;

use crate::crypto_trade::{
    CryptoAccountContext, lookup_crypto_account_context, lookup_crypto_read_account_context,
};
use crate::proto_internal::trade_cmn;
use crate::trade_cmd::{CryptoTradeOperation, crypto_trade_command};

use super::projections::*;
use super::types::*;

pub async fn query_crypto_orders(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
) -> Result<Vec<CachedOrder>> {
    query_crypto_orders_with_source(
        backend,
        acc_id,
        trd_cache,
        OrderRelationSourceToken::unknown(backend.connection_generation()),
    )
    .await
}

pub async fn query_crypto_orders_with_source(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    source: OrderRelationSourceToken,
) -> Result<Vec<CachedOrder>> {
    let ctx = lookup_crypto_account_context(trd_cache, acc_id)?;
    query_crypto_orders_with_context(backend, acc_id, trd_cache, &ctx, source).await
}

/// Startup-only recent-order read. C++ permits this read before trade unlock;
/// an empty cipher is omitted from `CryptoMsgHeader` while strict/write paths
/// continue to require the cached cipher.
pub async fn query_crypto_orders_for_startup(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
) -> Result<Vec<CachedOrder>> {
    query_crypto_orders_for_startup_with_source(
        backend,
        acc_id,
        trd_cache,
        OrderRelationSourceToken::unknown(backend.connection_generation()),
    )
    .await
}

pub async fn query_crypto_orders_for_startup_with_source(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    source: OrderRelationSourceToken,
) -> Result<Vec<CachedOrder>> {
    let ctx = lookup_crypto_read_account_context(trd_cache, acc_id)?;
    query_crypto_orders_with_context(backend, acc_id, trd_cache, &ctx, source).await
}

async fn query_crypto_orders_with_context(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    ctx: &CryptoAccountContext,
    source: OrderRelationSourceToken,
) -> Result<Vec<CachedOrder>> {
    use prost::Message;

    let spec = crypto_trade_command(CryptoTradeOperation::Orders);
    let mut all_orders = Vec::new();
    let mut dropped_order_count = 0usize;
    let mut page_flag: Option<String> = None;

    for _ in 0..MAX_PAGES {
        let plan = plan_crypto_order_list_request_like_cpp(page_flag.as_deref());
        let req = inbound_read::OrderListReq {
            msg_header: Some(ctx.build_crypto_msg_header("order_list")),
            page_size: Some(plan.page_size),
            page_flag: plan.page_flag,
            list_type: Some(plan.list_type),
        };
        let resp = crate::command_runtime::execute_crypto_trade_command(
            backend,
            CryptoTradeOperation::Orders,
            None,
            bytes::Bytes::from(req.encode_to_vec()),
        )
        .await
        .inspect_err(|e| {
            tracing::warn!(
                cmd_id = spec.cmd,
                error = ?futu_core::log_redact::RedactedFutuError::new(e),
                "crypto order query failed"
            );
        })?;

        let parsed: inbound_read::OrderListRsp =
            Message::decode(resp.body.as_ref()).map_err(|e| {
                tracing::warn!(
                    cmd_id = spec.cmd,
                    body_len = resp.body.len(),
                    error_class = "protobuf_decode",
                    "crypto order query decode failed"
                );
                FutuError::Proto(e)
            })?;

        ensure_crypto_order_read_status("CryptoOrderListRsp", parsed.msg_header.as_ref(), acc_id)?;
        let raw_order_count = parsed.orders.len();
        let projected = parsed
            .orders
            .iter()
            .filter_map(project_crypto_order)
            .collect::<Vec<_>>();
        dropped_order_count =
            dropped_order_count.saturating_add(raw_order_count.saturating_sub(projected.len()));
        all_orders.extend(projected);
        match decide_crypto_read_order_page_like_cpp(
            "query_crypto_orders",
            parsed.completed,
            parsed.page_flag.as_deref(),
            all_orders.len(),
        ) {
            CryptoReadCursorPageDecision::Complete => {
                trd_cache.replace_order_snapshot_like_cpp_from_source(
                    acc_id,
                    &all_orders,
                    source,
                    dropped_order_count == 0,
                );
                tracing::debug!(count = all_orders.len(), "crypto orders queried");
                return Ok(all_orders);
            }
            CryptoReadCursorPageDecision::Continue { next_page_flag } => {
                page_flag = Some(next_page_flag);
            }
            CryptoReadCursorPageDecision::PartialMissingPageFlag { error_message, .. } => {
                return Err(FutuError::Codec(error_message));
            }
        }
    }

    Err(FutuError::Codec(
        crypto_read_order_pagination_exceeded_like_cpp("query_crypto_orders", MAX_PAGES),
    ))
}

/// Query crypto order details through CMD20625 and merge them into order cache.
///
/// C++ 10.5.6508 `NNProto_Trd_OrderCrypto.cpp:153-177` handles
/// `NotifyCryptoOrder` by calling `QueryOrderInfo` with the pushed string order
/// ids. Keep this detail path separate from the full recent-order list refresh
/// so push updates only refresh the orders named by backend.
pub async fn query_crypto_order_info(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    order_ids: &[String],
) -> Result<Vec<CachedOrder>> {
    query_crypto_order_info_with_source(
        backend,
        acc_id,
        trd_cache,
        order_ids,
        OrderRelationSourceToken::unknown(backend.connection_generation()),
    )
    .await
}

pub async fn query_crypto_order_info_with_source(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    order_ids: &[String],
    source: OrderRelationSourceToken,
) -> Result<Vec<CachedOrder>> {
    use prost::Message;

    let detail_plan =
        plan_crypto_order_detail_request_like_cpp(order_ids).map_err(FutuError::Codec)?;
    let requested = detail_plan.order_ids;

    let ctx = lookup_crypto_account_context(trd_cache, acc_id)?;
    let spec = crypto_trade_command(CryptoTradeOperation::OrderInfo);
    let req = inbound_read::OrderDetailReq {
        msg_header: Some(ctx.build_crypto_msg_header("order_detail")),
        order_ids: requested.clone(),
    };
    let resp = crate::command_runtime::execute_crypto_trade_command(
        backend,
        CryptoTradeOperation::OrderInfo,
        None,
        bytes::Bytes::from(req.encode_to_vec()),
    )
    .await
    .inspect_err(|e| {
        tracing::warn!(
            cmd_id = spec.cmd,
            order_ids = ?requested,
            error = ?futu_core::log_redact::RedactedFutuError::new(e),
            "crypto order detail query failed"
        );
    })?;

    let parsed: inbound_read::OrderDetailRsp =
        Message::decode(resp.body.as_ref()).map_err(|e| {
            tracing::warn!(
                cmd_id = spec.cmd,
                body_len = resp.body.len(),
                error_class = "protobuf_decode",
                "crypto order detail query decode failed"
            );
            FutuError::Proto(e)
        })?;
    ensure_crypto_order_detail_status(
        "CryptoOrderDetailRsp",
        parsed.msg_header.as_ref(),
        acc_id,
        requested.len(),
        parsed.orders.len(),
    )?;
    let orders: Vec<CachedOrder> = parsed
        .orders
        .iter()
        .filter_map(project_crypto_order)
        .collect();
    if orders.len() != parsed.orders.len() {
        trd_cache.mark_order_relation_snapshot_incomplete_from_source(acc_id, source);
    }
    for order in &orders {
        trd_cache.upsert_order_from_source(acc_id, order.clone(), source);
    }
    tracing::debug!(count = orders.len(), "crypto order details queried");
    Ok(orders)
}

/// Query crypto history orders through CMD20623.
///
/// C++ 10.5.6508 `NNProto_Trd_OrderCrypto.cpp:373-398` sends
/// `inbound_read::HistoryOrderListReq` with `start_time` / `end_time` already
/// in microseconds. Unlike the active order query, this read path does not
/// update the active order cache.
pub async fn query_crypto_history_orders(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    start_micros: u64,
    end_micros: u64,
) -> Result<Vec<CachedOrder>> {
    use prost::Message;

    let ctx = lookup_crypto_account_context(trd_cache, acc_id)?;
    let spec = crypto_trade_command(CryptoTradeOperation::HistoryOrders);
    let mut all_orders = Vec::new();
    let mut page_flag: Option<String> = None;

    for _ in 0..MAX_PAGES {
        let plan = plan_crypto_history_order_list_request_like_cpp(
            start_micros,
            end_micros,
            page_flag.as_deref(),
        );
        let req = inbound_read::HistoryOrderListReq {
            msg_header: Some(ctx.build_crypto_msg_header("history_order_list")),
            page_size: Some(plan.page_size),
            page_flag: plan.page_flag,
            start_time: Some(plan.start_time_micros),
            end_time: Some(plan.end_time_micros),
            symbol: plan.symbol,
            order_status: plan.order_status,
            currency: plan.currency,
            side: plan.side,
            query_word: plan.query_word,
            ord_type: plan.order_type,
        };
        let resp = crate::command_runtime::execute_crypto_trade_command(
            backend,
            CryptoTradeOperation::HistoryOrders,
            None,
            bytes::Bytes::from(req.encode_to_vec()),
        )
        .await
        .inspect_err(|e| {
            tracing::warn!(
                cmd_id = spec.cmd,
                error = ?futu_core::log_redact::RedactedFutuError::new(e),
                "crypto history order query failed"
            );
        })?;

        let parsed: inbound_read::HistoryOrderListRsp = Message::decode(resp.body.as_ref())
            .map_err(|e| {
                tracing::warn!(
                    cmd_id = spec.cmd,
                    body_len = resp.body.len(),
                    error_class = "protobuf_decode",
                    "crypto history order query decode failed"
                );
                FutuError::Proto(e)
            })?;

        ensure_crypto_order_read_status(
            "CryptoHistoryOrderListRsp",
            parsed.msg_header.as_ref(),
            acc_id,
        )?;
        all_orders.extend(parsed.orders.iter().filter_map(project_crypto_order));
        match decide_crypto_read_order_page_like_cpp(
            "query_crypto_history_orders",
            parsed.completed,
            parsed.page_flag.as_deref(),
            all_orders.len(),
        ) {
            CryptoReadCursorPageDecision::Complete => {
                tracing::debug!(count = all_orders.len(), "crypto history orders queried");
                return Ok(all_orders);
            }
            CryptoReadCursorPageDecision::Continue { next_page_flag } => {
                page_flag = Some(next_page_flag);
            }
            CryptoReadCursorPageDecision::PartialMissingPageFlag { error_message, .. } => {
                return Err(FutuError::Codec(error_message));
            }
        }
    }

    Err(FutuError::Codec(
        crypto_read_order_pagination_exceeded_like_cpp("query_crypto_history_orders", MAX_PAGES),
    ))
}

fn ensure_crypto_order_read_status(
    message_name: &str,
    msg_header: Option<&trade_cmn::CryptoMsgHeader>,
    acc_id: u64,
) -> Result<()> {
    crypto_order_read_response_status_like_cpp(
        message_name,
        msg_header.map(crypto_msg_header_facts),
        acc_id,
    )
    .map_err(crypto_order_status_error_to_futu_error)
}

fn ensure_crypto_order_detail_status(
    message_name: &str,
    msg_header: Option<&trade_cmn::CryptoMsgHeader>,
    acc_id: u64,
    expected_count: usize,
    actual_count: usize,
) -> Result<()> {
    crypto_order_detail_response_status_like_cpp(
        message_name,
        msg_header.map(crypto_msg_header_facts),
        acc_id,
        expected_count,
        actual_count,
    )
    .map_err(crypto_order_status_error_to_futu_error)
}

fn crypto_msg_header_facts(header: &trade_cmn::CryptoMsgHeader) -> BackendHistoryMsgHeaderFacts {
    BackendHistoryMsgHeaderFacts {
        account_id: header.account_id,
    }
}

fn crypto_order_status_error_to_futu_error(err: BackendHistoryStatusError) -> FutuError {
    if err.is_backend_error {
        FutuError::ServerError {
            ret_type: -1,
            msg: err.message,
        }
    } else {
        FutuError::Codec(err.message)
    }
}