futu_cache/trd_cache/

order_state.rs

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use super::*;

impl TrdCache {
    pub fn get_cipher(&self, acc_id: u64) -> Option<Vec<u8>> {
        self.ciphers.get(&acc_id).map(|v| v.clone())
    }

    pub fn set_cipher(&self, acc_id: u64, cipher: Vec<u8>) {
        self.ciphers.insert(acc_id, cipher);
    }

    /// v1.4.73 A2 BUG-008 fix: 读当前账户的 cipher state version（用于 unlock idem_key）。
    ///
    /// 首次访问 acc_id 会初始化为 0。后续每次 lock 清 cipher 会 `fetch_add(1)`。
    /// `idem_key` 构造时把这个 version 纳入 hash → cipher 清后 version 递增 →
    /// 同 body 的 idem_key 不同 → cache miss → 真执行 unlock（或 backend 真校验）。
    pub fn get_cipher_state_version(&self, acc_id: u64) -> u64 {
        let entry = self
            .cipher_state_versions
            .entry(acc_id)
            .or_insert_with(|| Arc::new(AtomicU64::new(0)));
        entry.load(Ordering::SeqCst)
    }

    /// v1.4.73 A2 BUG-008 fix: lock 清 cipher 时调，递增 version → 让下次 unlock
    /// 同 body 得 cache miss。
    ///
    /// 返回 new version（递增后值），便于调用方 log。
    #[must_use]
    pub fn bump_cipher_state_version(&self, acc_id: u64) -> u64 {
        let entry = self
            .cipher_state_versions
            .entry(acc_id)
            .or_insert_with(|| Arc::new(AtomicU64::new(0)));
        entry.fetch_add(1, Ordering::SeqCst) + 1
    }

    /// v1.4.106 codex 0226 F1+F2: PlaceOrder 解析到 `OrderNewRsp.action.order_confirm`
    /// 时调用, 保存上下文用于后续 `Trd_ReconfirmOrder` 构造 backend `OrderConfirmReq`.
    ///
    /// `now_ms` 由 caller 传入 (便于单测注入固定时钟); 真实路径用
    /// `SystemTime::now()`.
    pub fn store_pending_order_confirm(
        &self,
        acc_id: u64,
        ftapi_order_id: u64,
        mut ctx: OrderConfirmContext,
        now_ms: u64,
    ) {
        ctx.inserted_at_ms = now_ms;
        let key = OrderConfirmKey::new(acc_id, ftapi_order_id);
        self.pending_order_confirms.insert(key, ctx);
    }

    /// v1.4.106 codex 0226 F1+F2: ReconfirmOrder handler 入口 lookup, 取出
    /// `(acc_id, ftapi_order_id)` 对应 OrderConfirmContext.
    ///
    /// 返 `None`: cache miss (PlaceOrder 没存 / TTL 过期 / 已被消费). caller 必须
    /// 早 reject loud, **不**允许 silent fallback (避免反模式 D / silent-success).
    ///
    /// `now_ms` 检查 TTL: `now - ctx.inserted_at_ms > ORDER_CONFIRM_CONTEXT_TTL_MS`
    /// 视为 stale → return None + remove (proactive GC).
    pub fn get_pending_order_confirm(
        &self,
        acc_id: u64,
        ftapi_order_id: u64,
        now_ms: u64,
    ) -> Option<OrderConfirmContext> {
        let key = OrderConfirmKey::new(acc_id, ftapi_order_id);
        let ctx = self.pending_order_confirms.get(&key)?.value().clone();
        if now_ms.saturating_sub(ctx.inserted_at_ms) > ORDER_CONFIRM_CONTEXT_TTL_MS {
            // Stale → proactive GC
            self.pending_order_confirms.remove(&key);
            return None;
        }
        Some(ctx)
    }

    /// v1.4.106 codex 0226 F1+F2: ReconfirmOrder backend 成功 (`OrderConfirmRsp.result==0`)
    /// 后调用, 从 cache 删除 (一次性消费, 防重复 confirm).
    ///
    /// 返 `true` 表示真有删除发生; `false` = 已被其他路径消费 / 过期 GC.
    pub fn remove_pending_order_confirm(&self, acc_id: u64, ftapi_order_id: u64) -> bool {
        let key = OrderConfirmKey::new(acc_id, ftapi_order_id);
        self.pending_order_confirms.remove(&key).is_some()
    }

    /// v1.4.106 codex 0226 F1+F2: GC stale OrderConfirmContext entries.
    ///
    /// 用于定时清理 (push dispatcher 收到 ORDER 类 push 时顺便扫一次), 防止
    /// stale ctx 累积. 返回清理掉的条目数.
    pub fn purge_stale_order_confirms(&self, now_ms: u64) -> usize {
        let mut purged = Vec::new();
        for entry in self.pending_order_confirms.iter() {
            if now_ms.saturating_sub(entry.value().inserted_at_ms) > ORDER_CONFIRM_CONTEXT_TTL_MS {
                purged.push(*entry.key());
            }
        }
        let n = purged.len();
        for key in purged {
            self.pending_order_confirms.remove(&key);
        }
        n
    }

    /// Store backend write `req_id -> order_id_ex` for C++ order op-result
    /// push parity.
    ///
    /// Ref:
    /// - `NNProto_Trd_OnPush.cpp:22-27`
    /// - `NNProto_Trd_OrderOpReal.cpp:217,352`
    ///
    /// Empty values are ignored because C++ only maps meaningful backend
    /// request/order pairs. Return value mirrors best-effort insert success.
    pub fn store_order_op_req_order(&self, acc_id: u64, req_id: &str, order_id_ex: &str) -> bool {
        let req_id = req_id.trim();
        let order_id_ex = order_id_ex.trim();
        if req_id.is_empty() || order_id_ex.is_empty() {
            return false;
        }

        self.order_op_req_orders
            .insert(
                OrderOpReqKey::new(acc_id, req_id.to_string()),
                order_id_ex.to_string(),
            )
            .is_none()
    }

    /// Consume backend order id by operation `req_id`.
    ///
    /// C++ `FindOrderIDByReqID` erases the pair after lookup and clears the
    /// helper map once it reaches 512 entries. Keep the same bounded,
    /// best-effort behavior; a miss only means the push may have come from
    /// another terminal/client and the dispatcher may fall back to full list.
    pub fn take_order_id_by_op_req_id(&self, acc_id: u64, req_id: &str) -> Option<String> {
        let req_id = req_id.trim();
        if req_id.is_empty() {
            return None;
        }

        let key = OrderOpReqKey::new(acc_id, req_id.to_string());
        let value = self
            .order_op_req_orders
            .remove(&key)
            .map(|(_, value)| value);
        if self.order_op_req_orders.len() >= ORDER_OP_REQ_ORDER_MAP_CLEAR_LIMIT {
            self.order_op_req_orders.clear();
        }
        value
    }

    /// v1.4.106 codex 0554 F1 [P1]: 原子性清空所有 cipher + 同步 bump 各账户的
    /// `cipher_state_version`。
    ///
    /// 起源：`/api/admin/reload` 之前的实现是
    /// `bridge.caches.trd_cache.ciphers.clear()` 直接动 `DashMap`，但 **没** bump
    /// `cipher_state_version`。这与 v1.4.73 A2 BUG-008 修复的语义不一致：
    /// lock-trade 路径里 `ciphers.remove()` 之后必跟 `bump_cipher_state_version()`，
    /// 防止旧 idempotency cache entry（unlock idem_key 含 cipher_state_version
    /// hash）在 cipher 被清后仍命中返 stale "cached success"，导致
    /// step 4 / step 5 silent regression。
    ///
    /// admin/reload 漏 bump 的具体后果：
    /// - reload 清光 ciphers
    /// - 客户端再调 unlock-trade 同 body → idem_key 命中（cipher_state_version
    ///   未变）→ 返 stale 成功 → cipher cache 仍空 → place-order `-401` 解锁失败
    ///
    /// 本 helper 把两步打包，**禁止外部直接 `cache.ciphers.clear()`**（那条
    /// 路径 silent skip bump，复活 BUG-008）。所有清 cipher 的 control-plane
    /// 路径（reload / admin / 未来若加更多）必须走本 helper。
    ///
    /// 返回 `(cleared_count, bumped_versions)`：
    /// - `cleared_count`：清掉的 cipher 数（即 reload 前已解锁账户数）
    /// - `bumped_versions`：每个被清 acc_id 的 (acc_id, new_version) 列表，
    ///   便于 log + 客户端调试 idem_key 失效原因
    ///
    /// 与 lock-trade 路径的 bump 行为一致：仅对**实际清掉 cipher** 的 acc_id
    /// 递增 version；从未解锁的账户 cipher_state_version 保持不变。
    ///
    /// 并发：`DashMap::iter()` 期间其他线程的 `ciphers.remove()` /
    /// `ciphers.insert()` 可能 race，但本 helper 用 `remove(&key)` 逐个清，
    /// 拿到 `Some(_)` 才 bump，保证 `version` 单调递增 + 与 `ciphers` 实际
    /// 状态一致。`SeqCst` 保证 bump 对所有后续 `get_cipher_state_version()`
    /// 立即可见。
    #[must_use]
    pub fn clear_all_ciphers_and_bump_versions(&self) -> (usize, Vec<(u64, u64)>) {
        // 先收集 acc_ids（避免持 DashMap iter guard 时 mutate map → deadlock）
        let acc_ids: Vec<u64> = self.ciphers.iter().map(|e| *e.key()).collect();
        let mut cleared_count = 0usize;
        let mut bumped_versions: Vec<(u64, u64)> = Vec::with_capacity(acc_ids.len());
        for acc_id in acc_ids {
            if self.ciphers.remove(&acc_id).is_some() {
                cleared_count += 1;
                let new_ver = self.bump_cipher_state_version(acc_id);
                bumped_versions.push((acc_id, new_ver));
            }
        }
        (cleared_count, bumped_versions)
    }
}