futu_backend/trade_query/orders/

query_orders.rs

//! trade_query/orders/query_orders — query_orders + query_orders_inner (real + sim 双路径)
//! (v1.4.110 CC Batch J: 拆自 orders.rs L336-666)

use futu_cache::trd_cache::TrdCache;
use futu_core::error::{FutuError, Result};

use super::super::common::{hash_str_to_u64, parse_open_d_text};
use super::super::*;
use crate::conn::BackendConn;
use crate::proto_internal::{order_sys_interface, sim_odr_sys_cmn, sim_order_sys_interface};

use super::types::QueryErrorMode;

use super::builders::*;
use super::helpers::*;
use super::status_helpers::*;

pub async fn query_orders(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
) -> Result<Vec<futu_cache::trd_cache::CachedOrder>> {
    query_orders_inner(backend, acc_id, trd_cache, QueryErrorMode::LenientCacheRead).await
}

/// v1.4.109: strict 版 query_orders for write/push safety paths.
///
/// 与 lenient 版区别: decode 失败 / `result != 0` → 返 `Err` (会被 retry 重试).
/// 不再 silent 当 "Ok(empty)" → silent-success 反模式 D (CLAUDE.md #45) 的修复.
///
/// Callers that need a trustworthy order baseline before mutating order state
/// must use this strict variant, not the read-path lenient cache-miss helper.
pub async fn query_orders_strict(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
) -> Result<Vec<futu_cache::trd_cache::CachedOrder>> {
    query_orders_inner(
        backend,
        acc_id,
        trd_cache,
        QueryErrorMode::StrictPushRefresh,
    )
    .await
}

/// v1.4.106 codex 0932 F3 [P2]: strict 版 query_orders for push-refresh path.
///
/// **4 attempts 全失败**: caller (dispatcher) 应触发 F4 `record_f2_exhausted` +
/// log warn. 不再 silent 当成功流转.
pub async fn query_orders_strict_for_push_refresh(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
) -> Result<Vec<futu_cache::trd_cache::CachedOrder>> {
    query_orders_strict(backend, acc_id, trd_cache).await
}

fn require_real_cipher_for_strict_trade_query(
    acc_id: u64,
    trd_cache: &TrdCache,
    op_name: &str,
) -> Result<Vec<u8>> {
    let cipher = trd_cache.get_cipher(acc_id).unwrap_or_default();
    if cipher.is_empty() {
        return Err(FutuError::ServerError {
            ret_type: -401,
            msg: format!(
                "{op_name}: real account cipher missing; trade is not unlocked (acc_id={acc_id})"
            ),
        });
    }
    Ok(cipher)
}

pub fn order_proto_to_cached_like_cpp(
    trd_env: i32,
    o: &crate::proto_internal::odr_sys_cmn::Order,
) -> Option<futu_cache::trd_cache::CachedOrder> {
    order_proto_to_cached_with_jp_acc_type_like_cpp(trd_env, o, None)
}

pub fn order_proto_to_cached_with_jp_acc_type_like_cpp(
    trd_env: i32,
    o: &crate::proto_internal::odr_sys_cmn::Order,
    jp_acc_type: Option<i32>,
) -> Option<futu_cache::trd_cache::CachedOrder> {
    use futu_cache::trd_cache::CachedOrder;

    let trd_market = backend_order_unpacked_trd_market_like_cpp(trd_env, o)?;
    let order_id_ex = o.order_id.clone().unwrap_or_default();
    // v1.4.106 codex 0226 F6: backend `order_id` (= szOrderID) 是 alphanumeric 字符串,
    // 不是 numeric. C++ 用 `HashStrToU64(szOrderID)` 派生 FTAPI numeric orderID。
    let order_id: u64 = if order_id_ex.is_empty() {
        0
    } else {
        hash_str_to_u64(&order_id_ex)
    };
    let create_timestamp = o.create_time.map(|t| t as f64 / 1_000_000.0);
    let update_timestamp = o.update_time.map(|t| t as f64 / 1_000_000.0);
    let order_status = backend_order_status_to_ftapi(o.status.unwrap_or(0));
    let trd_side = o.side.unwrap_or(0) as i32;
    let security_type = o.security_type.map(|v| v as i32);
    let order_type = backend_order_type_to_ftapi(o.r#type.unwrap_or(0), trd_market, security_type);
    let backend_order_id = order_id_ex.clone();
    let order_version = o.version.unwrap_or(0) as i32;
    let exchange_code = o.exchange_code.unwrap_or(0) as i32;
    let exchange = o.exchange.clone().unwrap_or_default();
    let security_type = security_type.unwrap_or(0);
    let extracted_remark = match parse_open_d_text(o.text.as_deref()) {
        Some((_local_id, user_remark)) => Some(user_remark),
        None => o.text.clone(),
    };

    let order_trade_time_type = o.sup.as_ref().and_then(|s| s.order_trade_time_type);

    Some(CachedOrder {
        order_id,
        order_id_ex,
        code: o.symbol.as_ref().cloned().unwrap_or_default(),
        name: o.stock_name.as_ref().cloned().unwrap_or_default(),
        trd_side,
        order_type,
        order_status,
        qty: parse_backend_decimal(&o.qty),
        price: parse_backend_decimal(&o.price),
        fill_qty: parse_backend_decimal(&o.cum_qty),
        fill_avg_price: parse_backend_decimal(&o.avg_fill_price),
        create_time: String::new(),
        update_time: String::new(),
        last_err_msg: o.last_err_msg.clone(),
        sec_market: None,
        create_timestamp,
        update_timestamp,
        remark: extracted_remark,
        time_in_force: None,
        fill_outside_rth: None,
        aux_price: None,
        trail_type: None,
        trail_value: None,
        trail_spread: None,
        currency: o.currency.map(|c| c as i32),
        trd_market,
        backend_order_id,
        order_version,
        exchange_code,
        exchange,
        security_type,
        session: futu_trd::projection::order_trade_time_type_to_session(order_trade_time_type),
        order_trade_time_type,
        jp_acc_type,
        is_stub: false,
        is_local_order: false,
        stub_inserted_at_ms: 0,
        is_pending_broker_confirm: false,
    })
}

pub(super) fn sim_order_proto_to_cached_like_cpp(
    trd_env: i32,
    o: &sim_odr_sys_cmn::Order,
) -> Option<futu_cache::trd_cache::CachedOrder> {
    use futu_cache::trd_cache::CachedOrder;

    o.order_id.as_ref()?;
    o.side?;
    let trd_market = if trd_env == 1 {
        let trd_market = backend_real_market_to_trd_market_like_cpp(o.market?);
        if !is_valid_real_trd_market_like_cpp(trd_market) {
            return None;
        }
        Some(trd_market)
    } else {
        o.market.map(map_backend_market_to_trd_market)
    };
    o.symbol.as_ref()?;
    o.create_time?;
    o.status?;

    let order_id_ex = o.order_id.clone().unwrap_or_default();
    let order_id: u64 = if order_id_ex.is_empty() {
        0
    } else {
        hash_str_to_u64(&order_id_ex)
    };
    let create_timestamp = o.create_time.map(|t| t as f64 / 1_000_000.0);
    let update_timestamp = o.update_time.map(|t| t as f64 / 1_000_000.0);
    let order_status = backend_order_status_to_ftapi(o.status.unwrap_or(0));
    let trd_side = o.side.unwrap_or(0) as i32;
    let security_type = o.security_type.map(|v| v as i32);
    let order_type = backend_order_type_to_ftapi(o.r#type.unwrap_or(0), trd_market, security_type);
    let backend_order_id = order_id_ex.clone();
    let order_version = o.version.unwrap_or(0) as i32;
    let exchange_code = o.exchange_code.unwrap_or(0) as i32;
    let exchange = o.exchange.clone().unwrap_or_default();
    let security_type = security_type.unwrap_or(0);
    let extracted_remark = match parse_open_d_text(o.text.as_deref()) {
        Some((_local_id, user_remark)) => Some(user_remark),
        None => o.text.clone(),
    };

    let order_trade_time_type = o.sup.as_ref().and_then(|s| s.order_trade_time_type);

    Some(CachedOrder {
        order_id,
        order_id_ex,
        code: o.symbol.as_ref().cloned().unwrap_or_default(),
        name: o.stock_name.as_ref().cloned().unwrap_or_default(),
        trd_side,
        order_type,
        order_status,
        qty: parse_backend_decimal(&o.qty),
        price: parse_backend_decimal(&o.price),
        fill_qty: parse_backend_decimal(&o.cum_qty),
        fill_avg_price: parse_backend_decimal(&o.avg_fill_price),
        create_time: String::new(),
        update_time: String::new(),
        last_err_msg: o.last_err_msg.clone(),
        sec_market: None,
        create_timestamp,
        update_timestamp,
        remark: extracted_remark,
        time_in_force: None,
        fill_outside_rth: None,
        aux_price: None,
        trail_type: None,
        trail_value: None,
        trail_spread: None,
        currency: o.currency.map(|c| c as i32),
        trd_market,
        backend_order_id,
        order_version,
        exchange_code,
        exchange,
        security_type,
        session: futu_trd::projection::order_trade_time_type_to_session(order_trade_time_type),
        order_trade_time_type,
        jp_acc_type: o.sub_account.map(|v| v as i32),
        is_stub: false,
        is_local_order: false,
        stub_inserted_at_ms: 0,
        is_pending_broker_confirm: false,
    })
}

/// Strict order detail query for C++ push-refresh parity.
///
/// C++ receives `NOTICE_TYPE_ORDER_UPDATE` / `NOTICE_TYPE_ORDER_OP_RESULT` and
/// calls `QueryOrderInfo` (4707/14707) with the pushed order IDs, rather than
/// running a full `OrderListReq` (4708/14708). This matters for terminal states
/// such as Deleted(23), which can be visible through the detail path while the
/// current-order list omits them.
pub async fn query_order_info_strict_for_push_refresh(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    order_ids: &[String],
    security_type: Option<u32>,
    exchange: Option<&str>,
) -> Result<Vec<futu_cache::trd_cache::CachedOrder>> {
    use prost::Message;

    let filtered_ids: Vec<String> = order_ids
        .iter()
        .filter(|id| !id.is_empty())
        .cloned()
        .collect();
    if filtered_ids.is_empty() {
        return Ok(vec![]);
    }

    let (trd_env, _enabled_markets) = derive_acc_query_context(trd_cache, acc_id);
    let cmd_id = order_info_cmd_for_env(trd_env);
    let real_cipher = if trd_env == 1 {
        require_real_cipher_for_strict_trade_query(acc_id, trd_cache, "QueryOrderInfo")?
    } else {
        Vec::new()
    };
    let req =
        build_order_info_req_base(acc_id, security_type, exchange, &filtered_ids, real_cipher);
    let resp = backend.request(cmd_id, req.encode_to_vec()).await?;
    let (orders, received): (Vec<futu_cache::trd_cache::CachedOrder>, usize) = if trd_env == 1 {
        let parsed: order_sys_interface::OrderDetailRsp = Message::decode(resp.body.as_ref())?;
        if let Err(status_err) = backend_order_info_status_like_cpp(
            parsed.result,
            parsed.msg_header.as_ref(),
            parsed.err_msg.as_deref(),
            acc_id,
            filtered_ids.len(),
            parsed.orders.len(),
        ) {
            return Err(futu_core::error::FutuError::ServerError {
                ret_type: status_err.result,
                msg: format!("{} (acc_id={acc_id})", status_err.message),
            });
        }
        let received = parsed.orders.len();
        let orders = parsed
            .orders
            .iter()
            .filter_map(|order| {
                let jp_acc_type = order
                    .sub_account_id
                    .and_then(|id| trd_cache.jp_acc_type_for_sub_account(acc_id, id));
                order_proto_to_cached_with_jp_acc_type_like_cpp(trd_env, order, jp_acc_type)
            })
            .collect();
        (orders, received)
    } else {
        let parsed: sim_order_sys_interface::OrderDetailRsp = Message::decode(resp.body.as_ref())?;
        if let Err(status_err) = backend_order_info_status_by_account_like_cpp(
            parsed.result,
            parsed
                .msg_header
                .as_ref()
                .and_then(|header| header.account_id),
            parsed.err_msg.as_deref(),
            acc_id,
            filtered_ids.len(),
            parsed.orders.len(),
        ) {
            return Err(futu_core::error::FutuError::ServerError {
                ret_type: status_err.result,
                msg: format!("{} (acc_id={acc_id})", status_err.message),
            });
        }
        let received = parsed.orders.len();
        let orders = parsed
            .orders
            .iter()
            .filter_map(|order| sim_order_proto_to_cached_like_cpp(trd_env, order))
            .collect();
        (orders, received)
    };
    let mut applied_orders = Vec::with_capacity(orders.len());
    for order in orders {
        if trd_cache.upsert_order(acc_id, order.clone()) {
            applied_orders.push(order);
        }
    }
    tracing::debug!(
        acc_id,
        cmd_id,
        requested = filtered_ids.len(),
        received,
        applied = applied_orders.len(),
        "order details queried"
    );
    Ok(applied_orders)
}

pub async fn query_orders_inner(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    error_mode: QueryErrorMode,
) -> Result<Vec<futu_cache::trd_cache::CachedOrder>> {
    use prost::Message;

    // v1.4.106 codex 0955 F4: req shape 对齐 C++ FillOrderListReq.
    // v1.4.106 14708 audit: C++ QueryOrderList_Inner sim 分支使用
    // sim_order_sys_interface::OrderListReq + registry Orders.sim_cmd。
    let (trd_env, enabled_markets) = derive_acc_query_context(trd_cache, acc_id);
    // C++ `M_SendProto_SetReqMsgHeader` always copies
    // `INNData_Trd_AccList::GetAccCipher(m_nAccID)` into real trade query
    // headers. Deleted(23) rows observed after daemon restart are sensitive to
    // this real-order-list request shape: sending an always-empty cipher can
    // return a valid but empty backend snapshot.
    let real_cipher = if trd_env == 1 && error_mode.is_strict() {
        require_real_cipher_for_strict_trade_query(acc_id, trd_cache, "QueryOrderList")?
    } else {
        trd_cache.get_cipher(acc_id).unwrap_or_default()
    };
    let real_req = build_order_list_req_base(acc_id, trd_env, &enabled_markets, real_cipher);
    let sim_req = build_sim_order_list_req_base(acc_id, &enabled_markets);

    let mut all_orders = Vec::new();
    // C++ first page calls `FillOrderListReq(req, "")`, so the optional
    // page_flag is present with an empty string on the wire.
    let mut page_flag: Option<String> = Some(String::new());
    // v1.4.106 codex 0955 F5: 显式 track completion. partial pagination
    // (completed=false 但 page_flag 缺失 / 超 max_pages 未 completed) 必须
    // 返 Err, 不能写 cache (stub orders 会被 partial 列表抹掉, 历史教训
    // 见坑 #44 v1.4.73-90 saga).
    const MAX_PAGES: usize = 100;
    let mut completed = false;

    for _page in 0..MAX_PAGES {
        let spec = trade_query_command(TradeQueryOperation::Orders);
        let (cmd_id, req_body) = if trd_env == 1 {
            let mut paged_req = real_req.clone();
            paged_req.page_flag = page_flag.clone();
            (spec.real_cmd, paged_req.encode_to_vec())
        } else {
            let mut paged_req = sim_req.clone();
            paged_req.page_flag = page_flag.clone();
            (spec.sim_cmd, paged_req.encode_to_vec())
        };

        let resp = match backend.request(cmd_id, req_body).await {
            Ok(r) => r,
            Err(e) => {
                tracing::debug!(acc_id, cmd_id, error = %e, "order query failed");
                return Err(e);
            }
        };

        let parsed: order_sys_interface::OrderListRsp = match Message::decode(resp.body.as_ref()) {
            Ok(p) => p,
            Err(e) => {
                // v1.4.106 codex 0932 F3 [P2]: strict 模式 → 返 Err 让上层 retry,
                // 不 silent 当 Ok(empty). lenient (read path) 保留老行为.
                if error_mode.is_strict() {
                    tracing::warn!(
                        acc_id,
                        error = %e,
                        "v1.4.106 F3 strict: order response decode failed → \
                         returning Err to trigger retry (push-refresh path)"
                    );
                    return Err(futu_core::error::FutuError::from(e));
                }
                tracing::debug!(acc_id, error = %e, "order response decode failed (lenient)");
                return Ok(vec![]);
            }
        };

        if let Err(status_err) = backend_order_list_status_like_cpp(
            parsed.result,
            parsed.msg_header.as_ref(),
            parsed.err_msg.as_deref(),
            acc_id,
        ) {
            // v1.4.110: C++ CheckRspHeaderAndGetSvrRet rejects missing or
            // mismatched msg_header.account_id. Treat structural response
            // issues as retryable even on lenient read paths so they cannot
            // masquerade as a legitimate empty order snapshot.
            if error_mode.is_strict() || !status_err.is_backend_error {
                tracing::warn!(
                    acc_id,
                    result = status_err.result,
                    error = %status_err.message,
                    "v1.4.110: order query status/header invalid -> returning Err \
                     to trigger retry (push-refresh path)"
                );
                return Err(futu_core::error::FutuError::ServerError {
                    ret_type: status_err.result,
                    msg: format!("{} (acc_id={acc_id})", status_err.message),
                });
            }
            return Ok(vec![]);
        }

        for o in &parsed.orders {
            let jp_acc_type = o
                .sub_account_id
                .and_then(|id| trd_cache.jp_acc_type_for_sub_account(acc_id, id));
            if let Some(order) =
                order_proto_to_cached_with_jp_acc_type_like_cpp(trd_env, o, jp_acc_type)
            {
                all_orders.push(order);
            }
        }

        // v1.4.106 codex 0955 F5: 显式分三档处理:
        // - completed=true → 全部拉完, 写 cache.
        // - completed=false + page_flag 非空 → 继续 paginate.
        // - completed=false + page_flag 空/缺 → partial failure, 不写 cache, 返 Err.
        let parsed_completed = parsed.completed.unwrap_or(false);
        if parsed_completed {
            completed = true;
            break;
        }
        match parsed.page_flag {
            Some(ref pf) if !pf.is_empty() => {
                page_flag = Some(pf.clone());
            }
            _ => {
                // partial + page_flag 缺 → 不能继续, backend 状态机异常.
                tracing::warn!(
                    acc_id,
                    partial_count = all_orders.len(),
                    "v1.4.106 audit 0955 F5: backend completed=false 但 page_flag \
                     缺失/空 — partial response, 不写 cache 防 stub 被抹"
                );
                return Err(futu_core::error::FutuError::Codec(
                    "query_orders: partial response (completed=false + missing page_flag)"
                        .to_string(),
                ));
            }
        }
    }

    // v1.4.106 codex 0955 F5: 超 MAX_PAGES 仍未 completed → partial, 不写 cache.
    if !completed {
        tracing::warn!(
            acc_id,
            max_pages = MAX_PAGES,
            partial_count = all_orders.len(),
            "v1.4.106 audit 0955 F5: pagination 超 MAX_PAGES 仍未 completed — \
             partial response, 不写 cache 防 stub 被抹"
        );
        return Err(futu_core::error::FutuError::Codec(format!(
            "query_orders: pagination exceeded {MAX_PAGES} pages without completed=true"
        )));
    }

    tracing::debug!(acc_id, count = all_orders.len(), "orders queried");
    // v1.4.93 S3 BUG-5318-009 真修：用 merge_preserving_stubs 代替 update_orders
    // (= orders.insert 整覆盖)。
    //
    // 历史坑（v1.4.90 S BUG-e4da-009 fix 漏修这一层）：
    // - v1.4.90 把 place_order.rs / modify_order.rs 三个 *outer* callsite 改成
    //   merge，但 query_orders 内部仍 update_orders → orders.insert 整覆盖。
    // - 实际执行顺序：
    //   1. PlaceOrder handler upsert stub → cache: [stub]
    //   2. spawn task 调 query_orders →
    //   3. backend 返 vec![] (just-placed not yet authoritative) →
    //   4. *trd_cache.update_orders(acc_id, vec![])* → cache: []  ← STUB 被抹
    //   5. query_orders 返 Ok(vec![]) →
    //   6. outer caller cache.merge_preserving_stubs(acc_id, vec![]) →
    //      existing_stubs = [] → cache 仍 []  ← merge 来晚了
    //   7. log "merged ... count=0"
    // - 端到端：stub 在 spawn 内被 query_orders 自己抹掉，外层 merge 是 dead code
    //   from the stub's perspective.
    //
    // 真机证据（claude-5318 v1.4.90 real-money US.F PlaceOrder, 2026-04-25）：
    // - daemon log: place_order.rs:436 "stub upsert (order_id=15250071496077083016)"
    // - daemon log: place_order.rs:502 "merge_preserving_stubs ... count=0"
    // - /api/orders 0/5/30/60s 全空
    //
    // 修法：把 update_orders（仅 caller = 此处）改为 merge_preserving_stubs。
    // - 仍保持 query_orders 的"populate cache as side-effect"语义
    // - 但 backend 的权威列表与 fresh stub 共存（< 30s TTL）
    // - 外层 callsite 的 merge_preserving_stubs 现在是幂等冗余（无害）
    //
    // 对 dispatcher.rs:228 push handler 调用：仅迭代返回的 Vec，不依赖 cache
    // 状态，行为不变。
    //
    // 同坑 #44 "同 bug 跨版本 regression 链": v1.4.73→89 7 版 + v1.4.90 8th 版的
    // "每版补一层没补根因"。本版（v1.4.93 S3）真补到 inner helper 这一层。
    trd_cache.merge_preserving_stubs(acc_id, all_orders.clone());
    Ok(all_orders)
}