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

mod push_regs;
mod views;

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<QotPushRegistrations>,

    /// **每 (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,
}

#[derive(Default)]
struct QotPushRegistrations {
    by_tuple: HashMap<(QotSecurityKey, i32, i32), HashSet<u64>>,
    qot_push_regs_by_cache_key: HashMap<String, HashMap<(i32, i32), HashSet<u64>>>,
}

/// 总订阅额度上限 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(QotPushRegistrations::default()),
            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> {
        match self.trd_acc_subs.read().get(&acc_id) {
            Some(subscribers) => subscribers.iter().copied().collect(),
            None => Vec::new(),
        }
    }

    // ===== 行情订阅 (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.by_tuple.retain(|_, set| {
                set.remove(&conn_id);
                !set.is_empty()
            });
            pr.qot_push_regs_by_cache_key.retain(|_, by_sub| {
                by_sub.retain(|_, set| {
                    set.remove(&conn_id);
                    !set.is_empty()
                });
                !by_sub.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> {
        match self
            .qot_subs
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
        {
            Some(subscribers) => subscribers.iter().copied().collect(),
            None => Vec::new(),
        }
    }

    /// **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
        })
    }

    // ===== 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)
    }

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

    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.by_tuple.retain(|_, set| {
                set.remove(&conn_id);
                !set.is_empty()
            });
            pr.qot_push_regs_by_cache_key.retain(|_, by_sub| {
                by_sub.retain(|_, set| {
                    set.remove(&conn_id);
                    !set.is_empty()
                });
                !by_sub.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
}

#[cfg(test)]
mod tests;