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 connection_lifecycle;
mod disconnected_cleanup;
mod push_regs;
mod qot_commit;
mod session_detail;
mod unsubscribe_all_commit;
mod views;

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

use dashmap::DashMap;
use futu_core::qot_stock_key::QotSecurityKey;
pub use futu_domain_qot_subscription::QOT_MIN_UNSUB_ELAPSED_SECS;
use futu_domain_qot_subscription::{
    CryptoSubscriptionProbe, UnsubscribeAllGlobalEmptyProbe,
    is_crypto_stock_broker_globally_unsubscribed, is_crypto_stock_globally_unsubscribed,
    plan_unsubscribe_all_global_empty_keys, qot_min_unsub_freshness_from_elapsed_secs,
};
use parking_lot::RwLock;

use crate::conn::ClientCloseControl;

pub type ConnectionDisconnectObserver = Arc<dyn Fn(u64) + Send + Sync>;
pub type ConnectionOpenObserver = Arc<dyn Fn(u64, u64) + Send + Sync>;

/// 订阅管理器
pub struct SubscriptionManager {
    connection_open_observers: RwLock<Vec<ConnectionOpenObserver>>,
    /// Runtime owners that must cancel connection-scoped work before a dead
    /// TCP/WS sink can receive any late result. Observers receive only the
    /// opaque connection generation id and must be idempotent.
    disconnect_observers: RwLock<Vec<ConnectionDisconnectObserver>>,

    /// Per-connection close controls shared by listeners and push fanout.
    ///
    /// This lives beside connection lifecycle state instead of inside the
    /// public `ClientConn`, preserving the latter's external struct-literal
    /// construction contract.
    client_close_controls: DashMap<u64, ClientCloseControl>,
    /// Exact physical connection generation used by async QOT work leases.
    connection_generations: DashMap<u64, u64>,

    /// 通知订阅：哪些连接订阅了系统通知
    notify_subs: RwLock<HashSet<u64>>,

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

    /// C++ `APIServerCS_PageReq` equivalent: opaque 16-byte page keys are
    /// owned by the client connection that received them.
    api_page_req_keys: RwLock<HashMap<u64, HashSet<[u8; 16]>>>,

    /// **行情订阅** (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 / 4=OVERNIGHT (rejected).
    qot_sub_sessions: RwLock<QotSessionState>,

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

    /// 全局订阅时间 (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,
    qot_owner_token_high_water: AtomicU64,
    qot_owner_tokens: RwLock<HashMap<(QotSecurityKey, i32, u64), u64>>,
    qot_push_intent_high_water: AtomicU64,
}

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

type QotPushIntentOwners = HashMap<u64, u64>;
type QotPushIntentRoutes = HashMap<(i32, i32), QotPushIntentOwners>;
type QotPushIntentIndex = HashMap<String, QotPushIntentRoutes>;

/// Exact push-registration intent captured for asynchronous first-push work.
///
/// The opaque epoch distinguishes unregister -> re-register ABA transitions;
/// the connection generation and session fence late results for a replaced
/// client sink or a changed KLine aggregate.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct QotPushRegistrationLease {
    key: QotSecurityKey,
    sub_type: i32,
    rehab_type: i32,
    conn_id: u64,
    intent_epoch: u64,
    connection_generation: u64,
    conn_session: i32,
}

impl QotPushRegistrationLease {
    #[must_use]
    pub fn connection_generation(&self) -> u64 {
        self.connection_generation
    }

    #[must_use]
    pub(crate) fn intent_epoch(&self) -> u64 {
        self.intent_epoch
    }

    #[must_use]
    pub fn matches_delivery(
        &self,
        conn_id: u64,
        connection_generation: u64,
        sec_key: &str,
        sub_type: i32,
        rehab_type: i32,
    ) -> bool {
        self.conn_id == conn_id
            && self.connection_generation == connection_generation
            && self.key.cache_key() == sec_key
            && self.sub_type == sub_type
            && self.rehab_type == rehab_type
    }
}

#[derive(Default)]
struct QotSessionState {
    by_key: HashMap<(QotSecurityKey, i32), HashMap<u64, i32>>,
    /// Hot push-route mirror keyed by the exact cache-key display carried by
    /// `PushEvent`. Both views are mutated under this one owner.
    by_cache_key: HashMap<(String, i32), HashMap<u64, i32>>,
}

/// **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 {
            connection_open_observers: RwLock::new(Vec::new()),
            disconnect_observers: RwLock::new(Vec::new()),
            client_close_controls: DashMap::new(),
            connection_generations: DashMap::new(),
            notify_subs: RwLock::new(HashSet::new()),
            trd_acc_subs: RwLock::new(HashMap::new()),
            api_page_req_keys: RwLock::new(HashMap::new()),
            qot_subs: RwLock::new(HashMap::new()),
            qot_push_regs: RwLock::new(QotPushRegistrations::default()),
            qot_sub_sessions: RwLock::new(QotSessionState::default()),
            qot_orderbook_detail: RwLock::new(HashMap::new()),
            qot_broker_detail: RwLock::new(HashMap::new()),
            qot_sub_times: RwLock::new(HashMap::new()),
            qot_disconnected_conns: RwLock::new(HashSet::new()),
            qot_disconnect_sync_generation: AtomicU64::new(0),
            qot_owner_token_high_water: AtomicU64::new(0),
            qot_owner_tokens: RwLock::new(HashMap::new()),
            qot_push_intent_high_water: 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 register_api_page_req_key(&self, conn_id: u64, key: &[u8]) -> bool {
        let Ok(key) = <[u8; 16]>::try_from(key) else {
            return false;
        };
        self.api_page_req_keys
            .write()
            .entry(conn_id)
            .or_default()
            .insert(key);
        true
    }

    #[must_use]
    pub fn is_api_page_req_key_registered(&self, conn_id: u64, key: &[u8]) -> bool {
        let Ok(key) = <[u8; 16]>::try_from(key) else {
            return false;
        };
        self.api_page_req_keys
            .read()
            .get(&conn_id)
            .is_some_and(|keys| keys.contains(&key))
    }

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

    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 {
        qot_commit::subscribe_broker(self, conn_id, Self::broker_key(sec_key), sub_type)
    }

    /// 退订并返结构化结果. 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 {
        qot_commit::unsubscribe_broker(self, conn_id, Self::broker_key(sec_key), sub_type)
    }

    /// 是否 (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| {
                qot_min_unsub_freshness_from_elapsed_secs(instant.elapsed().as_secs()).min_elapsed
            })
            .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_freshness_from_elapsed_secs(instant.elapsed().as_secs())
                    .remaining_secs
            })
            .unwrap_or(0)
    }

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

    #[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)> {
        unsubscribe_all_commit::collect_global_empty(self, conn_id)
    }

    /// 清理已断开且已满足 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)> {
        disconnected_cleanup::cleanup_due(self)
    }

    /// **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 qot = self.qot_subs.read();
        let probes = qot
            .iter()
            .map(|((key, sub_type), set)| UnsubscribeAllGlobalEmptyProbe {
                key: key.cache_key(),
                sub_type: *sub_type,
                conn_is_subscribed: set.contains(&conn_id),
                subscriber_count: set.len(),
            });
        plan_unsubscribe_all_global_empty_keys(probes)
    }

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

    /// Snapshot active owners for generation-fenced async quote work.
    pub fn qot_owner_lease_broker(
        &self,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> Vec<(u64, u64, i32, u64)> {
        let disconnected = self.qot_disconnected_conns.read();
        let mut owners = self
            .get_qot_subscribers_broker(sec_key, sub_type)
            .into_iter()
            .filter(|conn_id| !disconnected.contains(conn_id))
            .filter_map(|conn_id| {
                let connection_generation = self
                    .connection_generations
                    .get(&conn_id)
                    .map(|generation| *generation)
                    .unwrap_or(0);
                let owner_token = self
                    .qot_owner_tokens
                    .read()
                    .get(&(Self::broker_key(sec_key), sub_type, conn_id))
                    .copied()?;
                Some((
                    conn_id,
                    connection_generation,
                    self.get_conn_session_broker(conn_id, sec_key, sub_type),
                    owner_token,
                ))
            })
            .collect::<Vec<_>>();
        owners.sort_unstable();
        owners
    }

    pub fn qot_owner_lease_is_current(
        &self,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        captured: &[(u64, u64, i32, u64)],
        request_section: Option<i32>,
    ) -> bool {
        self.with_current_qot_owner_lease(sec_key, sub_type, captured, request_section, || ())
            .is_some()
    }

    pub fn with_current_qot_owner_lease<R>(
        &self,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        captured: &[(u64, u64, i32, u64)],
        request_section: Option<i32>,
        publish: impl FnOnce() -> R,
    ) -> Option<R> {
        let key = Self::broker_key(sec_key);
        let qot = self.qot_subs.read();
        let subscribers = qot.get(&(key.clone(), sub_type))?;
        let disconnected = self.qot_disconnected_conns.read();
        let tokens = self.qot_owner_tokens.read();
        let sessions = self.qot_sub_sessions.read();
        let session_map = sessions.by_key.get(&(key.clone(), sub_type));
        let current = captured.iter().any(
            |(conn_id, connection_generation, captured_session, owner_token)| {
                if !subscribers.contains(conn_id) || disconnected.contains(conn_id) {
                    return false;
                }
                let current_connection_generation = self
                    .connection_generations
                    .get(conn_id)
                    .map(|generation| *generation)
                    .unwrap_or(0);
                let same_connection = current_connection_generation == *connection_generation;
                let same_owner = tokens
                    .get(&(key.clone(), sub_type, *conn_id))
                    .is_some_and(|token| *token == *owner_token);
                let session = session_map
                    .and_then(|map| map.get(conn_id))
                    .copied()
                    .unwrap_or(1);
                same_connection
                    && same_owner
                    && session == *captured_session
                    && request_section.is_none_or(|section| match section {
                        2 | 3 => matches!(session, 2 | 3),
                        5 => session == 3,
                        _ => matches!(session, 0..=3),
                    })
            },
        );
        current.then(publish)
    }

    fn next_qot_owner_token(&self) -> u64 {
        let mut current = self.qot_owner_token_high_water.load(Ordering::SeqCst);
        loop {
            let next = if current == u64::MAX { 1 } else { current + 1 };
            match self.qot_owner_token_high_water.compare_exchange(
                current,
                next,
                Ordering::SeqCst,
                Ordering::SeqCst,
            ) {
                Ok(_) => return next,
                Err(observed) => current = observed,
            }
        }
    }

    pub(super) fn next_qot_push_intent_epoch(&self) -> u64 {
        let mut current = self.qot_push_intent_high_water.load(Ordering::SeqCst);
        loop {
            let next = if current == u64::MAX { 1 } else { current + 1 };
            match self.qot_push_intent_high_water.compare_exchange(
                current,
                next,
                Ordering::SeqCst,
                Ordering::SeqCst,
            ) {
                Ok(_) => return next,
                Err(observed) => current = observed,
            }
        }
    }

    pub(super) fn assign_qot_owner_token(&self, key: &QotSecurityKey, sub_type: i32, conn_id: u64) {
        let token = self.next_qot_owner_token();
        self.qot_owner_tokens
            .write()
            .insert((Self::broker_key(key), sub_type, conn_id), token);
    }

    pub(super) fn remove_qot_owner_token(&self, key: &QotSecurityKey, sub_type: i32, conn_id: u64) {
        self.qot_owner_tokens
            .write()
            .remove(&(Self::broker_key(key), sub_type, conn_id));
    }

    pub(super) fn remove_all_qot_owner_tokens(&self, conn_id: u64) {
        self.qot_owner_tokens
            .write()
            .retain(|(_, _, owner), _| *owner != conn_id);
    }

    /// **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 都退掉才能安全清.
    ///
    /// 返 `true` ⟺ 该 stock_id 在 `qot_subs` 中无任何带 subscriber 的 entry.
    pub fn crypto_stock_globally_unsubscribed(&self, stock_id: u64) -> bool {
        let qot = self.qot_subs.read();
        let probes = qot.iter().map(|((key, _sub_type), subs)| {
            CryptoSubscriptionProbe::from_runtime_facts(
                key.stock_key.stock_id,
                key.stock_key.broker_id,
                subs.len(),
            )
        });
        is_crypto_stock_globally_unsubscribed(stock_id, probes)
    }

    /// 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` 不适用).
    ///
    /// 返 `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();
        let probes = qot.iter().map(|((key, _sub_type), subs)| {
            CryptoSubscriptionProbe::from_runtime_facts(
                key.stock_key.stock_id,
                key.stock_key.broker_id,
                subs.len(),
            )
        });
        is_crypto_stock_broker_globally_unsubscribed(stock_id, target_broker, probes)
    }

    // ===== 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,
    ) {
        let previous = self.get_conn_session_broker(conn_id, sec_key, sub_type);
        session_detail::set_conn_session(
            self,
            conn_id,
            Self::broker_key(sec_key),
            sub_type,
            session,
        );
        if previous != session && self.is_qot_subscribed_broker(conn_id, sec_key, sub_type) {
            self.assign_qot_owner_token(sec_key, sub_type, conn_id);
        }
    }

    pub fn get_global_session_broker(&self, sec_key: &QotSecurityKey, sub_type: i32) -> i32 {
        session_detail::global_session(self, Self::broker_key(sec_key), sub_type)
    }

    /// 获取单连接订阅 session（没有显式记录时按 C++ 默认 RTH）。
    pub fn get_conn_session_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        sub_type: i32,
    ) -> i32 {
        session_detail::conn_session(self, conn_id, Self::broker_key(sec_key), sub_type)
    }

    /// Lookup the FTAPI connection session from the internal cache-key carried
    /// by a live push. Missing entries retain the C++ default RTH session.
    pub fn get_conn_session_by_cache_key(
        &self,
        conn_id: u64,
        cache_key: &str,
        sub_type: i32,
    ) -> i32 {
        self.qot_sub_sessions
            .read()
            .by_cache_key
            .get(&(cache_key.to_owned(), sub_type))
            .and_then(|sessions| sessions.get(&conn_id))
            .copied()
            .unwrap_or(1)
    }

    pub fn set_conn_orderbook_detail_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        detail: bool,
    ) {
        session_detail::set_conn_orderbook_detail(self, conn_id, Self::broker_key(sec_key), detail);
    }

    pub fn is_global_orderbook_detail_broker(&self, sec_key: &QotSecurityKey) -> bool {
        session_detail::global_orderbook_detail(self, Self::broker_key(sec_key))
    }

    pub fn set_conn_broker_detail_broker(
        &self,
        conn_id: u64,
        sec_key: &QotSecurityKey,
        detail: bool,
    ) {
        session_detail::set_conn_broker_detail(self, conn_id, Self::broker_key(sec_key), detail);
    }

    pub fn is_global_broker_detail_broker(&self, sec_key: &QotSecurityKey) -> bool {
        session_detail::global_broker_detail(self, Self::broker_key(sec_key))
    }

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

    pub fn register_connection_open_observer(&self, observer: ConnectionOpenObserver) {
        self.connection_open_observers.write().push(observer);
    }

    pub(crate) fn on_connect(&self, conn_id: u64, session_generation: u64) {
        self.connection_generations
            .insert(conn_id, session_generation);
        let observers = self.connection_open_observers.read().clone();
        for observer in observers {
            observer(conn_id, session_generation);
        }
    }

    pub fn register_disconnect_observer(&self, observer: ConnectionDisconnectObserver) {
        self.disconnect_observers.write().push(observer);
    }

    pub(crate) fn register_client_close_control(
        &self,
        conn_id: u64,
        close_control: ClientCloseControl,
    ) {
        self.client_close_controls.insert(conn_id, close_control);
    }

    /// Returns `None` when the listener never registered a close control,
    /// otherwise whether this call performed the first open-to-closing
    /// transition.
    pub(crate) fn request_client_close(&self, conn_id: u64) -> Option<bool> {
        self.client_close_controls
            .get(&conn_id)
            .map(|control| control.request_close())
    }

    pub(crate) fn remove_client_close_control(&self, conn_id: u64) {
        self.client_close_controls.remove(&conn_id);
    }

    pub fn on_disconnect(&self, conn_id: u64) -> Vec<(String, i32)> {
        connection_lifecycle::on_disconnect(self, conn_id)
    }
}

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;