futu_cache/trd_cache/

order_list.rs

use std::collections::HashSet;

use super::*;

impl TrdCache {
    /// Mark an existing cached order as local Deleted(23) after a successful
    /// DeleteFailOrder operation.
    ///
    /// C++ keeps local orders (`bIsLocalOrder=true`) across `UpdateOrderList`
    /// backend refreshes. Rust must do the same for delete success, otherwise a
    /// backend refresh that omits the just-deleted failed order makes
    /// `order_list_query()` return empty while official C++ still shows
    /// Deleted(23).
    #[must_use]
    pub fn mark_order_deleted_local(
        &self,
        acc_id: u64,
        order_id: u64,
        order_id_ex: Option<&str>,
    ) -> bool {
        let trimmed_ex = order_id_ex.map(str::trim).filter(|s| !s.is_empty());
        let Some(mut entry) = self.orders.get_mut(&acc_id) else {
            return false;
        };
        let Some(order) = entry.iter_mut().find(|order| {
            if order_id != 0 && order.order_id == order_id {
                return true;
            }
            if let Some(ex) = trimmed_ex {
                order.backend_order_id.trim() == ex || order.order_id_ex.trim() == ex
            } else {
                false
            }
        }) else {
            return false;
        };

        order.order_status = 23;
        order.is_stub = false;
        order.is_local_order = true;
        order.stub_inserted_at_ms = 0;
        order.is_pending_broker_confirm = false;
        true
    }

    pub fn record_order_broker(
        &self,
        acc_id: AccKey,
        order_id_ex: impl AsRef<str>,
        broker_id: u32,
    ) {
        let order_id_ex = order_id_ex.as_ref();
        if order_id_ex.trim().is_empty() {
            return;
        }
        let order_id_ex = order_id_ex.to_string();
        let previous_acc = self
            .order_broker_accounts
            .insert(order_id_ex.clone(), acc_id);
        if let Some(old_acc_id) = previous_acc.filter(|old_acc_id| *old_acc_id != acc_id) {
            self.remove_order_broker_id_from_acc_index(old_acc_id, &order_id_ex);
        }
        self.order_broker_ids_by_acc
            .entry(acc_id)
            .or_default()
            .insert(order_id_ex.clone());
        self.order_brokers.insert(order_id_ex, broker_id);
    }

    fn remove_order_broker_id_from_acc_index(&self, acc_id: AccKey, order_id_ex: &str) {
        let mut remove_bucket = false;
        if let Some(mut ids) = self.order_broker_ids_by_acc.get_mut(&acc_id) {
            ids.remove(order_id_ex);
            remove_bucket = ids.is_empty();
        }
        if remove_bucket {
            self.order_broker_ids_by_acc
                .remove_if(&acc_id, |_, ids| ids.is_empty());
        }
    }

    pub(super) fn prune_order_brokers_for_acc(&self, acc_id: AccKey) {
        if self.order_brokers.is_empty() {
            return;
        }

        let stale_order_ids = {
            let orders = self.orders.get(&acc_id);
            let mut active_order_ids = HashSet::new();
            if let Some(ref entry) = orders {
                for order in entry.value() {
                    let order_id_ex = order.order_id_ex.trim();
                    if !order_id_ex.is_empty() {
                        active_order_ids.insert(order_id_ex);
                    }
                    let backend_order_id = order.backend_order_id.trim();
                    if !backend_order_id.is_empty() {
                        active_order_ids.insert(backend_order_id);
                    }
                }
            }

            let Some(order_ids) = self.order_broker_ids_by_acc.get(&acc_id) else {
                return;
            };
            order_ids
                .iter()
                .filter(|order_id_ex| !active_order_ids.contains(order_id_ex.as_str()))
                .cloned()
                .collect::<Vec<_>>()
        };

        for order_id_ex in stale_order_ids {
            let still_belongs_to_acc = self
                .order_broker_accounts
                .get(&order_id_ex)
                .is_some_and(|entry| *entry.value() == acc_id);
            if still_belongs_to_acc {
                self.order_broker_accounts.remove(&order_id_ex);
                self.order_brokers.remove(&order_id_ex);
            }
            self.remove_order_broker_id_from_acc_index(acc_id, &order_id_ex);
        }
    }

    /// v1.4.105 BUG-v1.4.104-001 (P0): broker async confirm 到达后清 pending 标志.
    ///
    /// 当 push notice_type=4/5/8/100 (ORDER_UPDATE / ORDER_LIST_UPDATE /
    /// TRADE_STATISTIC / ORDER_NTF) 到达对应 acc_id 时, 调本 fn 把所有
    /// `is_pending_broker_confirm=true` 的 order 翻成 `false`.
    ///
    /// 设计选择: 不按 order_id 精确匹配清 — push notice 通常不带具体 order_id,
    /// 只表 "本 acc 有 order 状态变化". 简化处理: acc 内任何 ORDER 类 push 到
    /// 即视为 broker 已开始处理本 acc 的 stub orders.
    /// 后续 query_orders refresh 会通过 `merge_preserving_stubs` 把 enriched
    /// 版本写入, 替换 stub.
    ///
    /// 返被清的 order 数 (caller 用于 audit log).
    pub fn clear_pending_confirm_for_acc(&self, acc_id: u64) -> usize {
        let mut cleared = 0;
        if let Some(mut entry) = self.orders.get_mut(&acc_id) {
            for o in entry.iter_mut() {
                if o.is_pending_broker_confirm {
                    o.is_pending_broker_confirm = false;
                    cleared += 1;
                }
            }
        }
        cleared
    }

    /// v1.4.106 codex 0226 F4 (P2): selective clear pending confirm by order_ids.
    ///
    /// `clear_pending_confirm_for_acc` 是 acc-level 全清, 但 ORDER push notify
    /// 在 backend 实际带具体 `order_ids` 时(notice_type=4 ORDER_UPDATE 通常
    /// 带), daemon 应**只**清对应订单的 pending flag, 而不是把同账户其他还没
    /// confirm 的 stub 一并误清.
    ///
    /// 触发场景 (`bridge/dispatcher.rs:251-268`):
    /// - notice_type=4/5/9 + 非空 `order_ids` (backend 真带 → 按订单清)
    /// - notice_type=4/5/9 + 空 `order_ids` → fall back to `clear_pending_confirm_for_acc`
    ///
    /// match 逻辑: `o.backend_order_id` / `o.order_id_ex` (alphanumeric backend
    /// szOrderID) 与 `order_ids` 任一相等. 不 match `o.order_id` (FTAPI u64
    /// hash) 因为 backend push 带的 `order_ids` 是 backend 原生 string id.
    ///
    /// 返被清的 order 数 (caller 用于 audit log).
    pub fn clear_pending_confirm_for_orders(&self, acc_id: u64, order_ids: &[String]) -> usize {
        if order_ids.is_empty() {
            return 0;
        }
        let mut cleared = 0;
        if let Some(mut entry) = self.orders.get_mut(&acc_id) {
            for o in entry.iter_mut() {
                let matches_backend_id = Self::order_backend_ids(o)
                    .any(|backend_id| order_ids.iter().any(|id| id.trim() == backend_id));
                if o.is_pending_broker_confirm && matches_backend_id {
                    o.is_pending_broker_confirm = false;
                    cleared += 1;
                }
            }
        }
        cleared
    }

    /// v1.4.105 BUG-v1.4.104-001 (P0): cleanup task 删超时未 confirm 的 pending stub.
    ///
    /// 触发: PlaceOrder spawn 一个 30s 延迟 task, 到点检查 (acc_id, order_id_ex)
    /// 对应的 stub 是否仍 `is_stub=true && is_pending_broker_confirm=true`.
    /// 若是 → 删 stub + warn (push channel 断 / broker 拒单未 push 的兜底).
    ///
    /// **不**简单调 STUB_TTL_MS evict — 那个是 query_orders merge 时的逻辑,
    /// 这里是主动 GC pending stub. 两者互补.
    ///
    /// 返 (purged: bool, reason: 描述), caller 写 audit log.
    pub fn purge_pending_stub_if_still_pending(
        &self,
        acc_id: u64,
        order_id: u64,
    ) -> Option<String> {
        if let Some(mut entry) = self.orders.get_mut(&acc_id) {
            let before = entry.len();
            let mut purged_code = None;
            entry.retain(|o| {
                let should_purge =
                    o.order_id == order_id && o.is_stub && o.is_pending_broker_confirm;
                if should_purge {
                    purged_code = Some(o.code.clone());
                }
                !should_purge
            });
            let after = entry.len();
            if before != after {
                // v1.4.111 P2-1 Tier 3 audit comment: purged_code 是 audit log
                // 字段, `Some(empty_string)` 跟 `Some("CODE")` 都表示 "purge 成功",
                // caller (e.g. post_ack.rs:50-80 stub cleanup) 不基于 code 内容
                // mutate state. 非 silent-success risk (audit verified).
                return Some(purged_code.unwrap_or_default());
            }
        }
        None
    }

    /// v1.4.83 §9 F6: 扫全 cache 查 orphan orders.
    ///
    /// **Orphan 定义**: `order_status ∈ {0, 1, 2, 4}` (未达到 Submitted=5
    /// 之前的 in-flight stub) **且** `create_timestamp.is_some()` **且**
    /// `now_secs - create_timestamp > threshold_secs`.
    ///
    /// 含义对应 C++ proto OrderStatus enum (Trd_Common.proto:108):
    /// - 0 = Unsubmitted (未提交) — 极端情况, daemon stub 修后不应该出现 (v1.4.103 P0 hotfix)
    /// - 1 = WaitingSubmit (等待提交) — 条件单 stub 初值, 等触发
    /// - 2 = Submitting (提交中) — 普通单 stub 初值 (v1.4.103 起)
    /// - 4 = TimeOut (处理超时) — 后端回 timeout, 状态未知
    ///
    /// **为什么需要**: v1.4.82 A2 PlaceOrder 成功后直接 upsert stub order
    /// 让 `/api/orders` 立刻可见 (BUG-60b0-002 fix). 后续 push notice_type=
    /// 4/5/8 / re-fetch 把 status 推到 5 (Submitted) / 10/11 (Filled).
    /// 若 push 通道断流 (§9 CMD3020 chain broken), stub 卡住 5min+ = orphan.
    ///
    /// **v1.4.103 P0 (BUG-WUZONG-001)**: stub status 从 0 (proto 定义为
    /// Unsubmitted "未提交", 触发客户端 retry 多下单) 改成 1/2 (WaitingSubmit/
    /// Submitting, 对齐 C++ NNProto_Trd_OrderOp.cpp:483-510). orphan 检测同步
    /// 扩展到 {0, 1, 2, 4} 全 in-flight 状态 — 老 daemon 留下来 status=0 的
    /// 卡死 stub 也能被检测到.
    ///
    /// 返 `Vec<OrphanOrder>`; caller 决定 log 级别 + metric bump.
    #[must_use]
    pub fn scan_orphan_orders(&self, now_secs: f64, threshold_secs: f64) -> Vec<OrphanOrder> {
        let mut orphans = Vec::new();
        for entry in self.orders.iter() {
            let acc_id = *entry.key();
            for order in entry.value().iter() {
                // v1.4.103 codex F2.4 (P2) round 2: 仅扫**真 stub** orders
                // (is_stub == true). 非 stub 的 backend 权威 orders 即使 status
                // 是 0/1/2/4 (in-flight) 也不算 orphan — 它们是 backend 主动
                // 持续推送的真实状态, daemon 不应认为是卡住. 之前 v1.4.103
                // P0 hotfix 加 status filter {0,1,2,4} 但漏 is_stub guard,
                // 导致 backend 长时 Submitting/TimeOut 真单 (5min+) 也被 log
                // 当 orphan / push 通道断流候选 → metrics 噪音.
                if !order.is_stub {
                    continue;
                }
                // v1.4.103 P0: in-flight 状态 (未到 Submitted=5 之前的过渡值)
                // 都视为 stub 卡住候选 — 0 (Unsubmitted) / 1 (WaitingSubmit) /
                // 2 (Submitting) / 4 (TimeOut).
                if !matches!(order.order_status, 0 | 1 | 2 | 4) {
                    continue;
                }
                // **v1.4.106 codex 0920 F7 (P2)**: PlaceOrder stub `create_timestamp=None`
                // (backend 还没 push 真 timestamp). 只用 `create_timestamp.is_some()`
                // 时所有 stub 都被 scanner skip → orphan 检测对最关键场景 (stub
                // 卡死) 失效.
                //
                // 修法: stub (`is_stub=true`) 优先用 `stub_inserted_at_ms`
                // 推算 age (转 secs); 非 stub 用 `create_timestamp` (与原行为
                // 一致, 但本 scanner 已加 `is_stub` guard, 不会 reach 此分支).
                //
                // `stub_inserted_at_ms` 在 PlaceOrder ack 后 upsert 时写, 所以
                // 真 stub 必有非零值. = 0 视为没初始化 (老 cache entry 兼容)
                // → skip.
                let age_secs = if order.is_stub {
                    if order.stub_inserted_at_ms == 0 {
                        // 老 cache entry 未携带 stub_inserted_at_ms (e.g. 升级前
                        // 持久化数据反序列化), skip 而非误报.
                        continue;
                    }
                    let stub_inserted_secs = (order.stub_inserted_at_ms as f64) / 1000.0;
                    let now_unix_secs = now_secs;
                    now_unix_secs - stub_inserted_secs
                } else {
                    // 非 stub: 走 create_timestamp 老路径 (本 scanner 已加
                    // is_stub guard, 此分支不应 reach 但保留兼容).
                    let Some(create_ts) = order.create_timestamp else {
                        continue;
                    };
                    now_secs - create_ts
                };
                if age_secs > threshold_secs {
                    orphans.push(OrphanOrder {
                        acc_id,
                        order_id: order.order_id,
                        order_id_ex: order.order_id_ex.clone(),
                        code: order.code.clone(),
                        age_secs,
                    });
                }
            }
        }
        orphans
    }
}

/// v1.4.83 §9 F6: orphan order 结构化报告.
#[derive(Debug, Clone)]
pub struct OrphanOrder {
    pub acc_id: u64,
    pub order_id: u64,
    pub order_id_ex: String,
    pub code: String,
    /// 距离 create_timestamp 的秒数
    pub age_secs: f64,
}