futu_server/

subscription.rs

// 订阅管理：行情订阅 + 交易账户推送订阅 + 通知订阅
//
// v1.4.106 codex 1131 F3 [P1] (BLOCKER fix): C++ 把"backend 订阅状态"
// (`m_setSub`) 与"push 注册状态" (`m_mapRegPush`) 分开维护. Rust 之前
// 把两者塞同一 `qot_subs` map → `Qot_RegQotPush(register=true)` 制造
// 假订阅, `Qot_RegQotPush(register=false)` 误删真订阅, GetSubInfo 把
// push 注册当订阅项报告. 本版彻底拆开:
//   - `qot_subs: HashMap<(SecurityKey, SubType), HashSet<ConnId>>`
//     = desired sub state, 决定 backend CMD 6211 desired set + GetSubInfo
//   - `qot_push_regs: HashMap<(SecurityKey, SubType, RehabType), HashSet<ConnId>>`
//     = push 注册 state, 决定 PushDispatcher 路由对象, 不影响订阅
// C++ 对照:
//   - QotSubscribe.cpp:107-108 m_setSub[(stockID, subType)].insert(connID)
//   - QotSubscribe.cpp:489-490 m_mapRegPush[(stockID, subType, rehabType)].insert(connID)
//   - QotSubscribe.cpp:639-649 GetPushConn(stockID, subType, rehabType)
//
// v1.4.110 codex Phase 3 closeout (broker-aware key model): C++ QOT 模块
// (subscription / cache / push / quota) 以 `StockKey` (=stockID + 可选
// brokerID) 为 first-class identity. Rust 旧 String facade 已从生产
// `SubscriptionManager` 删除；内部状态直接以 `QotSecurityKey` keyed.
//
// C++ 对照 (QotSubscribe.h):
//   - Map_t<SubType, Set_t<StockKey>> m_mapSub  (per-conn)
//   - Set_t<(StockKey, SubType)>      m_setSub  (global)
//   - Map_t<(StockKey, SubType, RehabType), Set<ConnID>> m_mapRegPush
//   - Map_t<(ConnID, StockKey), bool> m_mapConnOrderBookDetail / BrokerDetail

use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use futu_core::qot_stock_key::QotSecurityKey;
use parking_lot::RwLock;

/// 订阅管理器
pub struct SubscriptionManager {
    /// 通知订阅：哪些连接订阅了系统通知
    notify_subs: RwLock<HashSet<u64>>,

    /// 交易账户推送订阅：acc_id → Set<conn_id>
    trd_acc_subs: RwLock<HashMap<u64, HashSet<u64>>>,

    /// **行情订阅** (desired sub state, 对齐 C++ `m_setSub`):
    ///   key = (QotSecurityKey, sub_type), val = 订阅 conn 集合.
    qot_subs: RwLock<HashMap<(QotSecurityKey, i32), HashSet<u64>>>,

    /// **行情 push 注册** (对齐 C++ `m_mapRegPush`):
    ///   key = (QotSecurityKey, sub_type, rehab_type), val = 注册接收 push 的 conn 集合.
    /// rehab_type 仅 KL 类有效 (None=0 / Forward=1 / Backword=2 / N/A=0 for non-KL).
    qot_push_regs: RwLock<HashMap<(QotSecurityKey, i32, i32), HashSet<u64>>>,

    /// **每 (security, sub_type) 的 desired session** (对齐 C++
    /// `m_mapConnTickerSession` / `m_mapConnKLRTSession` global view):
    /// max(per-conn session) 决定 backend desired session.
    /// session: 0=Unknown / 1=RTH / 2=ETH / 3=ALL / 5=OVERNIGHT (rejected).
    qot_sub_sessions: RwLock<HashMap<(QotSecurityKey, i32), HashMap<u64, i32>>>,

    /// **每 (security) 的 OrderBook detail flag** (对齐 C++
    /// `m_mapConnOrderBookDetail`): 一旦有 conn 要 detail, 全局走 detail.
    qot_orderbook_detail: RwLock<HashMap<QotSecurityKey, HashMap<u64, bool>>>,

    /// **每 (security) 的 Broker detail flag** (对齐 C++
    /// `m_mapConnBrokerDetail`).
    qot_broker_detail: RwLock<HashMap<QotSecurityKey, HashMap<u64, bool>>>,

    /// **总 quota 上限** (对齐 C++ `INNData_APIInterLimit::GetSubQuota()`):
    /// 启动 hardcode 4000 fallback, backend SubscribeSetRsp.max_sub_count 下发
    /// 后 setter 更新. 不再用静态 const.
    total_quota: RwLock<u32>,

    /// 全局订阅时间 (key = (QotSecurityKey, sub_type)).
    /// 对齐 C++ `m_mapSubTime`: 只有全局第一次订阅该 SubKey 或 backend
    /// 属性升级重新拉取时才刷新；退订前至少等待 `QOT_MIN_UNSUB_ELAPSED_SECS`.
    qot_sub_times: RwLock<HashMap<(QotSecurityKey, i32), Instant>>,

    /// 已断开的 conn_id，但其 QOT 订阅还没达到 C++ 最短退订窗口。
    ///
    /// 对齐 C++ `QotSubscribe::ClearConnSubInfo`: 断线时 push 注册立即清，
    /// 但 `m_setSub` 只有在 `IsSubTimeEnoughToUnSub` 后才移除；没到窗口的
    /// 连接由后续定时清理再次尝试。
    qot_disconnected_conns: RwLock<HashSet<u64>>,

    /// 断线延迟清理导致 global desired set 变化的 generation。
    ///
    /// server 层没有 backend 句柄，不能在 `on_disconnect` 里直接发 CMD6211。
    /// 这里仅记录“需要 gateway 同步”的单调计数，gateway 后台任务看到变化后
    /// 发当前 desired set。
    qot_disconnect_sync_generation: AtomicU64,
}

/// 总订阅额度上限 fallback (启动时未从 backend 拉到真值前用此).
///
/// 对齐 C++ `QotSubscribe.cpp:1132` `GetUserSubQuota()` 默认 4000.
/// 真实 quota 由 backend `SubscribeSetRsp.max_sub_count` 在 CMD 6211 响应
/// 里下发, daemon 收到后调 `set_total_quota_from_backend(value)` 更新
/// `total_quota`.
pub const TOTAL_QUOTA: u32 = 4000;

/// C++ `QotSubscribe.cpp:9` uses 60 seconds and
/// `IsSubTimeEnoughToUnSub` checks `>= SubAtLeastTime - 1`.
pub const QOT_MIN_UNSUB_ELAPSED_SECS: u64 = 59;

/// **subscribe_qot 返回的 commit 结果** (用于 quota 维度精确计算 — 对齐 C++
/// `m_setSub` 全局唯一计 quota).
///
/// C++ 对照: QotSubscribe.cpp:84-111. `bNoSub = m_setSub.count(pairSubKey) == 0`,
/// 仅当全局 set 不存在该 key 时才 `UseQuota()`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubResult {
    /// 全局首次订阅该 (security, sub_type) — quota 应 +1.
    NewGlobal,
    /// 全局已有订阅, 本 conn 是新加入 — quota 不变.
    AlreadyGlobal,
    /// (conn_id, security, sub_type) 已在 set 中, 重复订阅 — quota 不变.
    NoChange,
}

/// **unsubscribe_qot 返回的 commit 结果**.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UnsubResult {
    /// 最后一个 conn 退订 → 全局 set 删除该 key, **caller 必须发 backend
    /// fresh CMD 6211 with new desired set** (drop 该 (stock_id, sub_type)).
    LastSubscriber,
    /// 还有其他 conn 订阅该 (security, sub_type), backend 不需退订.
    StillSubscribed,
    /// (conn_id, security, sub_type) 之前未订阅, silent no-op (caller 决定
    /// 是否 loud reject).
    NotSubscribed,
}

impl SubscriptionManager {
    pub fn new() -> Self {
        Self {
            notify_subs: RwLock::new(HashSet::new()),
            trd_acc_subs: RwLock::new(HashMap::new()),
            qot_subs: RwLock::new(HashMap::new()),
            qot_push_regs: RwLock::new(HashMap::new()),
            qot_sub_sessions: RwLock::new(HashMap::new()),
            qot_orderbook_detail: RwLock::new(HashMap::new()),
            qot_broker_detail: RwLock::new(HashMap::new()),
            total_quota: RwLock::new(TOTAL_QUOTA),
            qot_sub_times: RwLock::new(HashMap::new()),
            qot_disconnected_conns: RwLock::new(HashSet::new()),
            qot_disconnect_sync_generation: AtomicU64::new(0),
        }
    }

    // ===== 通知订阅 =====

    pub fn subscribe_notify(&self, conn_id: u64) {
        self.notify_subs.write().insert(conn_id);
    }

    pub fn unsubscribe_notify(&self, conn_id: u64) {
        self.notify_subs.write().remove(&conn_id);
    }

    pub fn is_subscribed_notify(&self, conn_id: u64) -> bool {
        self.notify_subs.read().contains(&conn_id)
    }

    // ===== 交易账户推送 =====

    pub fn subscribe_trd_acc(&self, conn_id: u64, acc_id: u64) {
        self.trd_acc_subs
            .write()
            .entry(acc_id)
            .or_default()
            .insert(conn_id);
    }

    pub fn unsubscribe_trd_acc(&self, conn_id: u64, acc_id: u64) {
        if let Some(subs) = self.trd_acc_subs.write().get_mut(&acc_id) {
            subs.remove(&conn_id);
        }
    }

    pub fn get_acc_subscribers(&self, acc_id: u64) -> Vec<u64> {
        self.trd_acc_subs
            .read()
            .get(&acc_id)
            .map(|s| s.iter().copied().collect())
            .unwrap_or_default()
    }

    // ===== 行情订阅 (subscribers, F3 split-state) =====

    /// 生成行情订阅 key.
    pub fn make_qot_key(market: i32, code: &str, sub_type: i32) -> String {
        format!("{market}_{code}:{sub_type}")
    }

    #[inline]
    fn broker_key(sec_key: &QotSecurityKey) -> QotSecurityKey {
        sec_key.clone()
    }

    /// **v1.4.106 codex 1131 F1+F5 [P1+P2]**: 订阅行情. 返 [`SubResult`]
    /// 表示是否新加全局订阅 (caller 据此累 quota).
    /// 重复订阅 (同 conn_id 同 key) 不影响 set, 不影响 quota.
    /// **NOTE**: caller 必须先 backend ack-then-commit (F1) — 本方法仅写
    /// local state. 失败 caller 应调 `unsubscribe_qot_broker` 回滚.
    pub fn subscribe_qot_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> SubResult {
        self.subscribe_qot_inner(conn_id, Self::broker_key(sec_key), sub_type)
    }

    fn subscribe_qot_inner(&self, conn_id: u64, key: QotSecurityKey, sub_type: i32) -> SubResult {
        self.qot_disconnected_conns.write().remove(&conn_id);
        let mut qot = self.qot_subs.write();
        let map_key = (key.clone(), sub_type);
        let entry = qot.entry(map_key.clone()).or_default();
        let was_empty_global = entry.is_empty();
        let inserted = entry.insert(conn_id);
        if was_empty_global && inserted {
            self.qot_sub_times.write().insert(map_key, Instant::now());
        }
        if !inserted {
            SubResult::NoChange
        } else if was_empty_global {
            SubResult::NewGlobal
        } else {
            SubResult::AlreadyGlobal
        }
    }

    /// 退订并返结构化结果. caller 据 `LastSubscriber` 决定是否发 backend
    /// fresh CMD 6211 with new desired set.
    pub fn unsubscribe_qot_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> UnsubResult {
        self.unsubscribe_qot_inner(conn_id, Self::broker_key(sec_key), sub_type)
    }

    fn unsubscribe_qot_inner(
        &self,
        conn_id: u64,
        key: QotSecurityKey,
        sub_type: i32,
    ) -> UnsubResult {
        let map_key = (key.clone(), sub_type);
        let became_empty;
        let was_member;
        {
            let mut qot = self.qot_subs.write();
            let entry = match qot.get_mut(&map_key) {
                Some(e) => e,
                None => return UnsubResult::NotSubscribed,
            };
            was_member = entry.remove(&conn_id);
            became_empty = entry.is_empty();
            if became_empty {
                qot.remove(&map_key);
            }
        }
        if !was_member {
            return UnsubResult::NotSubscribed;
        }
        if became_empty {
            self.qot_sub_times.write().remove(&map_key);
        }
        // 清 session/detail per-conn entry (对齐 C++ QotSubscribe::UnSub line 251-292).
        {
            let mut sess = self.qot_sub_sessions.write();
            if let Some(e) = sess.get_mut(&map_key) {
                e.remove(&conn_id);
                if e.is_empty() {
                    sess.remove(&map_key);
                }
            }
        }
        if sub_type == sub_type_orderbook() {
            let mut ob = self.qot_orderbook_detail.write();
            if let Some(e) = ob.get_mut(&key) {
                e.remove(&conn_id);
                if e.is_empty() {
                    ob.remove(&key);
                }
            }
        }
        if sub_type == sub_type_broker() {
            let mut br = self.qot_broker_detail.write();
            if let Some(e) = br.get_mut(&key) {
                e.remove(&conn_id);
                if e.is_empty() {
                    br.remove(&key);
                }
            }
        }
        if became_empty {
            UnsubResult::LastSubscriber
        } else {
            UnsubResult::StillSubscribed
        }
    }

    /// 是否 (conn_id, key, sub_type) 已订阅.
    pub fn is_qot_subscribed_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> bool {
        self.qot_subs
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .is_some_and(|subs| subs.contains(&conn_id))
    }

    /// v1.4.106 codex 1131 F3 [P1]: 全局 (ignore conn) 是否有订阅 — RegQotPush
    /// 的 precondition. 对齐 C++ `QotSubscribe::IsSub(stockID, subType)`.
    pub fn is_globally_subscribed_broker(&self, sec_key: &QotSecurityKey, sub_type: i32) -> bool {
        self.qot_subs
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .is_some_and(|subs| !subs.is_empty())
    }

    /// min-unsub window for broker-aware subscription keys.
    pub fn qot_min_unsub_elapsed_broker(&self, sec_key: &QotSecurityKey, sub_type: i32) -> bool {
        self.qot_sub_times
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .map(|instant| instant.elapsed() >= Duration::from_secs(QOT_MIN_UNSUB_ELAPSED_SECS))
            .unwrap_or(true)
    }

    /// remaining min-unsub window for broker-aware keys.
    pub fn qot_min_unsub_remaining_secs_broker(
        &self,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> u64 {
        self.qot_sub_times
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .map(|instant| QOT_MIN_UNSUB_ELAPSED_SECS.saturating_sub(instant.elapsed().as_secs()))
            .unwrap_or(0)
    }

    /// 断线延迟清理后需要 gateway 同步 CMD6211 的 generation。
    pub fn qot_disconnect_sync_generation(&self) -> u64 {
        self.qot_disconnect_sync_generation.load(Ordering::SeqCst)
    }

    fn bump_qot_disconnect_sync_generation(&self) {
        self.qot_disconnect_sync_generation
            .fetch_add(1, Ordering::SeqCst);
    }

    fn conn_has_qot_subs(&self, conn_id: u64) -> bool {
        self.qot_subs
            .read()
            .values()
            .any(|subs| subs.contains(&conn_id))
    }

    #[doc(hidden)]
    pub fn backdate_qot_sub_time_broker_for_test(
        &self,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        elapsed: Duration,
    ) {
        let map_key = (Self::broker_key(sec_key), sub_type);
        let instant = Instant::now()
            .checked_sub(elapsed)
            .unwrap_or_else(Instant::now);
        self.qot_sub_times.write().insert(map_key, instant);
    }

    /// v1.4.106 codex 1131 F2: clear all qot subs for a single conn_id.
    /// 返 (sec_key, sub_type) 列表 of "本 conn 退订后变成全局空的" — caller
    /// 据此构 backend new desired set. 返的 sec_key 是 cache_key display string
    /// (`"market_code"` or `"market_code@b{id}"`).
    pub fn unsubscribe_all_qot_collect_global_empty(&self, conn_id: u64) -> Vec<(String, i32)> {
        let mut became_empty: Vec<(String, i32)> = Vec::new();
        let keys_to_check: Vec<(QotSecurityKey, i32)> = self
            .qot_subs
            .read()
            .iter()
            .filter(|(_, set)| set.contains(&conn_id))
            .map(|(k, _)| k.clone())
            .collect();
        {
            let mut qot = self.qot_subs.write();
            for k in keys_to_check {
                if let Some(set) = qot.get_mut(&k) {
                    set.remove(&conn_id);
                    if set.is_empty() {
                        became_empty.push((k.0.cache_key(), k.1));
                        let removed_key = k.clone();
                        qot.remove(&k);
                        self.qot_sub_times.write().remove(&removed_key);
                    }
                }
            }
        }
        // session/detail/push_regs cleanup for this conn.
        {
            let mut sess = self.qot_sub_sessions.write();
            sess.retain(|_, m| {
                m.remove(&conn_id);
                !m.is_empty()
            });
        }
        {
            let mut ob = self.qot_orderbook_detail.write();
            ob.retain(|_, m| {
                m.remove(&conn_id);
                !m.is_empty()
            });
        }
        {
            let mut br = self.qot_broker_detail.write();
            br.retain(|_, m| {
                m.remove(&conn_id);
                !m.is_empty()
            });
        }
        {
            let mut pr = self.qot_push_regs.write();
            pr.retain(|_, set| {
                set.remove(&conn_id);
                !set.is_empty()
            });
        }
        became_empty
    }

    /// 清理已断开且已满足 C++ 最短退订窗口的 QOT conn。
    ///
    /// 返回本次清理后 global desired set 变空的 `(sec_key, sub_type)` 列表
    /// (sec_key 是 cache_key display string: `"market_code"` or
    /// `"market_code@b{id}"`).
    /// 若列表非空，会 bump `qot_disconnect_sync_generation`，由 gateway 后台
    /// 任务负责把新的全局 desired set 发到 backend。
    pub fn cleanup_due_disconnected_qot(&self) -> Vec<(String, i32)> {
        let disconnected: Vec<u64> = self.qot_disconnected_conns.read().iter().copied().collect();
        if disconnected.is_empty() {
            return Vec::new();
        }

        let mut to_remove: Vec<(u64, QotSecurityKey, i32)> = Vec::new();
        {
            let qot = self.qot_subs.read();
            let sub_times = self.qot_sub_times.read();
            for conn_id in &disconnected {
                for ((key, sub_type), subs) in qot.iter() {
                    if !subs.contains(conn_id) {
                        continue;
                    }
                    let elapsed_ok = sub_times
                        .get(&(key.clone(), *sub_type))
                        .map(|instant| {
                            instant.elapsed() >= Duration::from_secs(QOT_MIN_UNSUB_ELAPSED_SECS)
                        })
                        .unwrap_or(true);
                    if elapsed_ok {
                        to_remove.push((*conn_id, key.clone(), *sub_type));
                    }
                }
            }
        }

        let mut became_empty = Vec::new();
        for (conn_id, key, sub_type) in to_remove {
            let display_key = key.cache_key();
            if matches!(
                self.unsubscribe_qot_inner(conn_id, key, sub_type),
                UnsubResult::LastSubscriber
            ) {
                became_empty.push((display_key, sub_type));
            }
        }

        {
            let mut disconnected = self.qot_disconnected_conns.write();
            disconnected.retain(|conn_id| self.conn_has_qot_subs(*conn_id));
        }

        if !became_empty.is_empty() {
            self.bump_qot_disconnect_sync_generation();
        }
        became_empty
    }

    /// **v1.4.106 codex 0631 F1 [P1]**: ack-then-commit `unsub_all` 的"干跑"半段.
    ///
    /// 计算: **若**本 conn 退订全部, 哪些 `(sec_key, sub_type)` 在 global
    /// desired set 中**变空** (= backend 该真退). **不修 state, 不动 detail
    /// flag, 不动 push_regs**. 用在 ack-then-commit pipeline:
    ///
    /// `dry_run -> submit_global_desired_set (backend ack) -> commit (清 state)`
    ///
    /// backend reject → caller 不调 `commit`, state 保留 → 客户端可重试幂等.
    /// 老 `unsubscribe_all_qot_collect_global_empty` 是先清后算 — 失败时
    /// state 已 mutate, 不能 rollback (split-brain 风险). 本 helper 替代.
    pub fn unsubscribe_all_qot_dry_run(&self, conn_id: u64) -> Vec<(String, i32)> {
        let mut became_empty: Vec<(String, i32)> = Vec::new();
        let qot = self.qot_subs.read();
        for ((k, sub_type), set) in qot.iter() {
            if !set.contains(&conn_id) {
                continue;
            }
            // 退订本 conn 后该 set 是否变空? set.len() == 1 + contains(&conn_id) → 退后空.
            if set.len() == 1 {
                became_empty.push((k.cache_key(), *sub_type));
            }
        }
        became_empty
    }

    /// **v1.4.106 codex 0631 F1 [P1]**: ack-then-commit `unsub_all` 的"提交"半段.
    /// 等价于老 `unsubscribe_all_qot_collect_global_empty` (语义不变, 仅在
    /// backend ack OK 后才调). 同时清 session / detail / push_regs.
    pub fn unsubscribe_all_qot_commit(&self, conn_id: u64) -> Vec<(String, i32)> {
        self.unsubscribe_all_qot_collect_global_empty(conn_id)
    }

    /// 获取订阅了指定行情的连接列表 (subscribers, **不**用作 push 路由).
    /// 用于 `apply_unsubscribe_delta` 判断 broker-aware key 上是否还有其他
    /// conn 订阅 (last-subscriber gate for desired-set remove).
    pub fn get_qot_subscribers_broker(&self, sec_key: &QotSecurityKey, sub_type: i32) -> Vec<u64> {
        self.qot_subs
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .map(|s| s.iter().copied().collect())
            .unwrap_or_default()
    }

    /// **v1.4.110 codex audit Round3 P2 #21**: 给定 `stock_id`, 判断该 stock 是否
    /// **全局**已无任何 conn 订阅 (跨所有 broker_id + 所有 sub_type).
    ///
    /// 用途: `Qot_Sub` 退订路径在 commit 之后判断 crypto symbol 是否真正全空,
    /// 决定是否调 `CryptoExchangeCache::clear_stock(stock_id)` 清 stale entry —
    /// 因为 `crypto_exchange_cache` 按 `stock_id` keyed (broker 无关), 只有该
    /// stock 全 broker 全 sub_type 都退掉才能安全清.
    ///
    /// 扫 `qot_subs` 找任意 key 满足 `QotStockKey.stock_id == stock_id`;
    /// broker 1007 退订但 1008 仍订阅时返 `false` (不能清).
    ///
    /// 返 `true` ⟺ 该 stock_id 在 `qot_subs` 中无任何带 subscriber 的 entry.
    pub fn crypto_stock_globally_unsubscribed(&self, stock_id: u64) -> bool {
        let qot = self.qot_subs.read();
        !qot.iter()
            .any(|((key, _sub_type), subs)| !subs.is_empty() && key.stock_key.stock_id == stock_id)
    }

    /// v1.4.110 R6-8: `(stock_id, broker_id)`-级版 `crypto_stock_globally_unsubscribed`.
    ///
    /// 用途: 部分 broker 退订时, 判断某具体 `(stock_id, broker_id)` 是否已无任何
    /// conn 订阅 → 决定是否调 `CryptoExchangeCache::clear_stock_broker` 清该
    /// broker 的 stale `by_broker` entry (整 stock 仍有别的 broker 在订时
    /// `clear_stock` 不适用).
    ///
    /// 按 `stock_key.stock_id` + `stock_key.broker_id` 双字段匹配.
    ///
    /// 返 `true` ⟺ 该 `(stock_id, broker_id)` 在 `qot_subs` 无任何带 subscriber 的 entry.
    pub fn crypto_stock_broker_globally_unsubscribed(&self, stock_id: u64, broker_id: u32) -> bool {
        let target_broker = std::num::NonZeroU32::new(broker_id);
        let qot = self.qot_subs.read();
        !qot.iter().any(|((key, _sub_type), subs)| {
            !subs.is_empty()
                && key.stock_key.stock_id == stock_id
                && key.stock_key.broker_id == target_broker
        })
    }

    // ===== 行情 push 注册 (qot_push_regs, F3 独立) =====

    /// **v1.4.106 codex 1131 F3 [P1]**: 注册接收 push (对齐 C++ `RegPush`).
    /// 仅写本 map, **不动 `qot_subs`**. caller (RegQotPushHandler) 必须在
    /// 调本方法前确认已订阅 (`is_globally_subscribed_broker`).
    pub fn register_push_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        rehab_type: i32,
    ) {
        self.register_push_inner(conn_id, Self::broker_key(sec_key), sub_type, rehab_type)
    }

    fn register_push_inner(
        &self,
        conn_id: u64,
        key: QotSecurityKey,
        sub_type: i32,
        rehab_type: i32,
    ) {
        let effective_rehab = if is_kl_sub_type(sub_type) {
            rehab_type
        } else {
            0
        };
        self.qot_push_regs
            .write()
            .entry((key, sub_type, effective_rehab))
            .or_default()
            .insert(conn_id);
    }

    /// **v1.4.106 codex 1131 F3 [P1]**: 取消 push 注册 — 不删 `qot_subs`.
    pub fn unregister_push_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        rehab_type: i32,
    ) {
        self.unregister_push_inner(conn_id, Self::broker_key(sec_key), sub_type, rehab_type)
    }

    fn unregister_push_inner(
        &self,
        conn_id: u64,
        key: QotSecurityKey,
        sub_type: i32,
        rehab_type: i32,
    ) {
        let effective_rehab = if is_kl_sub_type(sub_type) {
            rehab_type
        } else {
            0
        };
        let map_key = (key, sub_type, effective_rehab);
        let mut pr = self.qot_push_regs.write();
        if let Some(set) = pr.get_mut(&map_key) {
            set.remove(&conn_id);
            if set.is_empty() {
                pr.remove(&map_key);
            }
        }
    }

    /// **v1.4.106 codex 1131 F4 [P1]**: push delivery filter. 对齐 C++
    /// `QotSubscribe::GetPushConn`.
    pub fn get_qot_push_subscribers_broker(
        &self,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        rehab_type: i32,
    ) -> Vec<u64> {
        self.get_qot_push_subscribers_inner(Self::broker_key(sec_key), sub_type, rehab_type)
    }

    /// Lookup by internal cache-key display
    /// string carried by push events (`"market_code"` or `"market_code@b{id}"`).
    ///
    /// `PushEvent::QuotePush` does not carry `stock_id`, so dispatch cannot
    /// reconstruct a full `QotSecurityKey`. Instead, match the already stored
    /// subscription keys by their cache-key display string.
    pub fn get_qot_push_subscribers_by_cache_key(
        &self,
        cache_key: &str,
        sub_type: i32,
        rehab_type: i32,
    ) -> Vec<u64> {
        if QotSecurityKey::parse_cache_key(cache_key).is_none() {
            return Vec::new();
        }

        let effective_rehab = if is_kl_sub_type(sub_type) {
            rehab_type
        } else {
            0
        };
        let regs = self.qot_push_regs.read();
        let mut out = Vec::new();
        for ((key, stored_sub_type, stored_rehab), subscribers) in regs.iter() {
            if *stored_sub_type != sub_type
                || *stored_rehab != effective_rehab
                || key.cache_key() != cache_key
            {
                continue;
            }
            out.extend(subscribers.iter().copied());
        }

        out.sort_unstable();
        out.dedup();
        out
    }

    fn get_qot_push_subscribers_inner(
        &self,
        key: QotSecurityKey,
        sub_type: i32,
        rehab_type: i32,
    ) -> Vec<u64> {
        let effective_rehab = if is_kl_sub_type(sub_type) {
            rehab_type
        } else {
            0
        };
        self.qot_push_regs
            .read()
            .get(&(key, sub_type, effective_rehab))
            .map(|s| s.iter().copied().collect())
            .unwrap_or_default()
    }

    /// **v1.4.106 codex 1131 F3 [P1]**: 是否注册过 push (任意 rehab).
    pub fn is_push_registered_any_rehab_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> bool {
        let broker_key = Self::broker_key(sec_key);
        let pr = self.qot_push_regs.read();
        pr.iter().any(|((k, st, _rehab), set)| {
            k == &broker_key && *st == sub_type && set.contains(&conn_id)
        })
    }

    // ===== Session / Detail (per-(security, sub_type) global aggregator) =====

    pub fn set_conn_session_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        session: i32,
    ) {
        self.qot_sub_sessions
            .write()
            .entry((Self::broker_key(sec_key), sub_type))
            .or_default()
            .insert(conn_id, session);
    }

    pub fn get_global_session_broker(&self, sec_key: &QotSecurityKey, sub_type: i32) -> i32 {
        self.qot_sub_sessions
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .map(|m| m.values().copied().max().unwrap_or(1))
            .unwrap_or(1)
    }

    /// 获取单连接订阅 session（没有显式记录时按 C++ 默认 RTH）。
    pub fn get_conn_session_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> i32 {
        self.qot_sub_sessions
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .and_then(|m| m.get(&conn_id).copied())
            .unwrap_or(1)
    }

    pub fn set_conn_orderbook_detail_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        detail: bool,
    ) {
        self.qot_orderbook_detail
            .write()
            .entry(Self::broker_key(sec_key))
            .or_default()
            .insert(conn_id, detail);
    }

    pub fn is_global_orderbook_detail_broker(&self, sec_key: &QotSecurityKey) -> bool {
        self.qot_orderbook_detail
            .read()
            .get(&Self::broker_key(sec_key))
            .map(|m| m.values().any(|&d| d))
            .unwrap_or(false)
    }

    pub fn set_conn_broker_detail_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        detail: bool,
    ) {
        self.qot_broker_detail
            .write()
            .entry(Self::broker_key(sec_key))
            .or_default()
            .insert(conn_id, detail);
    }

    pub fn is_global_broker_detail_broker(&self, sec_key: &QotSecurityKey) -> bool {
        self.qot_broker_detail
            .read()
            .get(&Self::broker_key(sec_key))
            .map(|m| m.values().any(|&d| d))
            .unwrap_or(false)
    }

    // ===== Quota =====

    /// per-conn used quota — count 该 conn 的 (security_key, sub_type) 对数.
    pub fn get_conn_used_quota(&self, conn_id: u64) -> u32 {
        self.qot_subs
            .read()
            .iter()
            .filter(|(_, set)| set.contains(&conn_id))
            .count() as u32
    }

    /// **全局 used quota** — 对齐 C++ `m_nAllUsedQuota` (累加全局唯一
    /// SubKey count). 不重复计 conn — 多 conn 订同 (stock, sub_type) 全局算 1.
    pub fn get_total_used_quota(&self) -> u32 {
        self.qot_subs.read().len() as u32
    }

    /// 获取 total quota (动态, backend 下发后更新).
    pub fn get_total_quota(&self) -> u32 {
        *self.total_quota.read()
    }

    /// **v1.4.106 codex 1131 F5 [P2]**: backend 下发的 quota 真值 setter.
    pub fn set_total_quota_from_backend(&self, value: u32) {
        let mut q = self.total_quota.write();
        if *q != value {
            tracing::info!(
                old = *q,
                new = value,
                "v1.4.106 codex 1131 F5: total_quota updated from backend"
            );
            *q = value;
        }
    }

    pub fn get_remain_quota(&self) -> u32 {
        let total = self.get_total_quota();
        let used = self.get_total_used_quota();
        total.saturating_sub(used)
    }

    // ===== Conn-level views =====

    pub fn get_conn_qot_subs(&self, conn_id: u64) -> HashMap<i32, Vec<String>> {
        let qot = self.qot_subs.read();
        let mut result: HashMap<i32, Vec<String>> = HashMap::new();
        for ((key, sub_type), conn_ids) in qot.iter() {
            if conn_ids.contains(&conn_id) {
                result.entry(*sub_type).or_default().push(key.cache_key());
            }
        }
        result
    }

    pub fn get_all_qot_conn_ids(&self) -> HashSet<u64> {
        let qot = self.qot_subs.read();
        let mut ids = HashSet::new();
        for conn_ids in qot.values() {
            ids.extend(conn_ids);
        }
        ids
    }

    /// v1.4.106 codex 0932 F5 [P2]: 获取所有连接 ID（有交易账户订阅的）.
    pub fn get_all_trd_conn_ids(&self) -> HashSet<u64> {
        let trd = self.trd_acc_subs.read();
        let mut ids = HashSet::new();
        for conn_ids in trd.values() {
            ids.extend(conn_ids);
        }
        ids
    }

    /// v1.4.106 codex 1131 F2 [P1]: 计算全局 desired set.
    /// 返 (sec_key_display, sub_type), sec_key_display 是 cache_key 形态
    /// (`"market_code"` or `"market_code@b{id}"`).
    pub fn compute_global_desired_set(&self) -> Vec<(String, i32)> {
        let qot = self.qot_subs.read();
        let mut out = Vec::with_capacity(qot.len());
        for (k, sub_type) in qot.keys() {
            out.push((k.cache_key(), *sub_type));
        }
        out
    }
    /// **v1.4.106 codex 0631 F2 [P2]**: shared global desired-set keys helper.
    ///
    /// 返当前全集 `Vec<(sec_key, sub_type)>` (与 `compute_global_desired_set`
    /// 等价, 命名对齐 codex 0631 audit 习惯). 共享给 SubHandler / unsub /
    /// unsub_all / resubscribe_quotes 三路径, 不再每条 sub 单独发 delta —
    /// 对齐 CMD 6211 set-state 协议 (per-market full set replaces).
    ///
    /// caller 应做的事:
    /// 1. 调本 fn 拿当前全集
    /// 2. 应用 delta (sub: 加; unsub: 减; unsub_all: 移除本 conn 独占的)
    /// 3. resolve sec_key → stock_id (cache); cache miss → loud warn (本 fn 不
    ///    替 caller 决定 — 由 caller 选择 fail-loud 还是 partial-submit)
    /// 4. 调 `submit_global_desired_set(NEW set)` 让 backend ack
    ///
    /// 命名: keys = `(sec_key, sub_type)` 二元组, 不是 `(stock_id, market)` —
    /// stock_id resolve 是 caller 责任 (因为依赖 StaticDataCache).
    #[inline]
    pub fn qot_global_desired_keys(&self) -> Vec<(String, i32)> {
        self.compute_global_desired_set()
    }

    // ===== 连接断开清理 =====

    pub fn on_disconnect(&self, conn_id: u64) -> Vec<(String, i32)> {
        self.notify_subs.write().remove(&conn_id);

        {
            let mut trd = self.trd_acc_subs.write();
            for subs in trd.values_mut() {
                subs.remove(&conn_id);
            }
        }

        {
            let mut pr = self.qot_push_regs.write();
            pr.retain(|_, set| {
                set.remove(&conn_id);
                !set.is_empty()
            });
        }

        if self.conn_has_qot_subs(conn_id) {
            self.qot_disconnected_conns.write().insert(conn_id);
        }
        self.cleanup_due_disconnected_qot()
    }
}

impl Default for SubscriptionManager {
    fn default() -> Self {
        Self::new()
    }
}

#[inline]
fn sub_type_orderbook() -> i32 {
    2
}

#[inline]
fn sub_type_broker() -> i32 {
    14
}

/// 是否 KL 类 — 对齐 C++ `IsKLSubType`.
#[inline]
fn is_kl_sub_type(sub_type: i32) -> bool {
    matches!(sub_type, 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 15 | 16 | 17)
}

#[cfg(test)]
mod tests;