futu_server/

push.rs

// 推送分发：三种推送模式

use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

use bytes::Bytes;
use dashmap::DashMap;
use futu_auth::Scope;
use futu_codec::frame::FutuFrame;
use tokio::sync::mpsc::error::TrySendError;

use crate::conn::ClientConn;
use crate::metrics::GatewayMetrics;
use crate::subscription::SubscriptionManager;

mod kline_delivery;
use kline_delivery::{KlineCursorKey, KlinePushCursor, RegisteredKlinePushCursor};

/// **防御深度**：即使客户端在订阅阶段某种方式绕过了 scope gate，推送时
/// 再按 client key 的 scope 过滤一次。
///
/// `conn.scopes` 语义：
/// - **空集** → legacy 模式（TCP listener / WS 未配 keys.json），全放行
/// - 非空集 → scope 模式，必须包含 `needed` 才推
fn should_push_to(conn: &ClientConn, needed: Scope, event_label: &str) -> bool {
    if conn.scopes.is_empty() {
        return true; // legacy 全放行
    }
    if conn.scopes.contains(&needed) {
        return true;
    }
    // 过滤掉 —— 记 metrics 便于运维发现"谁订阅了但 scope 不够"这种配置问题
    let key_id = conn.key_id.as_deref().unwrap_or("<none>");
    futu_auth::metrics::bump_ws_filtered(event_label, key_id);
    false
}

/// 外部推送接收器 trait
///
/// 允许外部模块（如 REST WebSocket）接收推送事件，
/// 不引入模块间循环依赖。
///
/// **v1.4.106 codex 1131 F4 [P1]**: `on_quote_push` 加 `rehab_type` 参数. KL 类
/// push 走非 0 rehab (forward / backward / 无), 其它 sub_type 走 0. Sink 实装
/// 应用 (sec_key, sub_type, rehab_type) 三元过滤 push 接收方 — 不再 broadcast
/// 给所有 quote-scope conn (老行为是 silent leak: 仅订未 RegPush 的 conn 也收
/// 到 quote push, 违反 C++ `QotSubscribe::GetPushConn` 三元 key 路由).
pub trait ExternalPushSink: Send + Sync {
    /// 行情推送 (rehab_type=0 for non-KL).
    fn on_quote_push(
        &self,
        sec_key: &str,
        sub_type: i32,
        rehab_type: i32,
        proto_id: u32,
        body: &[u8],
    );
    /// 广播推送 (到价提醒、系统通知等)
    fn on_broadcast_push(&self, proto_id: u32, body: &[u8]);
    /// UID-bound private user push. The public body has already removed UID.
    fn on_private_user_push(&self, _proto_id: u32, _body: &[u8]) {}
    /// 交易推送 (订单更新、成交更新等).
    ///
    /// `trd_market` 是 PushDispatcher 一次性 decode `body` 提取的
    /// `s2c.header.trd_market` 大写字符串 ("HK" / "US" / "CN" / ...), 为
    /// 4 surface (gRPC / REST WS / MCP) 复用避免各自 decode. 老 sink 实现可
    /// 忽略此参数 (只看 acc_id + proto_id + body), Layer 3 (allowed_markets)
    /// filter 直接从这里取 — 见 [`extract_trd_market_from_trade_body`].
    ///
    /// `None` = decode 失败 / proto_id 不识别 / market enum unknown — 老
    /// 路径下游应**不 trigger Layer 3 drop** (向后兼容 — pitfall #57
    /// backend-semantic 未真机验证前 default OFF behavior).
    fn on_trade_push(&self, acc_id: u64, proto_id: u32, body: &[u8], trd_market: Option<&str>);
}

/// v1.4.105 D3 (Phase 4) T-B: trade push body decode → trd_market 提取.
///
/// 4 surface (gRPC / REST WS / raw TCP WS / MCP) 共用同一 helper 而非各自
/// decode 一次, 避免 mapping 漂移 (与 futu-auth-pipeline::body_aware /
/// futu-rest::trd::trd_market_str 一致, 但本 crate 不能跨 dep 复用所以重复
/// 一份 — 跨 crate mismatch 会被 cross_surface_smoke 抓出).
///
/// caller (PushDispatcher) 在分发到 sink 前**只 decode 一次**, 把字符串塞
/// PushEventCtx.event_trd_market 让 TradePushFilter Layer 3 用
/// allowed_markets 校验.
///
/// 不识别 / decode 失败 / market enum unknown → None (Layer 3 不 trigger).
///
/// UNVERIFIED — 真机 verify 跨 market 推送流 (HK + US 双账户) 后才能升
/// confidence (per pitfall #57 backend-semantic risk).
#[must_use]
pub fn extract_trd_market_from_trade_body(proto_id: u32, body: &[u8]) -> Option<&'static str> {
    use prost::Message;
    let market_int = match proto_id {
        // TRD_UPDATE_ORDER (2208) → Trd_UpdateOrder.Response.s2c.header.trd_market
        2208 => {
            let resp = match futu_proto::trd_update_order::Response::decode(body) {
                Ok(resp) => resp,
                Err(e) => {
                    tracing::debug!(
                        proto_id,
                        body_len = body.len(),
                        error = %e,
                        "trade push body decode failed while extracting trd_market"
                    );
                    return None;
                }
            };
            resp.s2c?.header.trd_market
        }
        // TRD_UPDATE_ORDER_FILL (2218) → Trd_UpdateOrderFill.Response.s2c.header.trd_market
        2218 => {
            let resp = match futu_proto::trd_update_order_fill::Response::decode(body) {
                Ok(resp) => resp,
                Err(e) => {
                    tracing::debug!(
                        proto_id,
                        body_len = body.len(),
                        error = %e,
                        "trade push body decode failed while extracting trd_market"
                    );
                    return None;
                }
            };
            resp.s2c?.header.trd_market
        }
        // 未知 trade push proto_id → 不识别, 让 Layer 3 不 trigger
        _ => return None,
    };
    // Trd_Common.TrdMarket enum int → 大写字符串. 与 futu-rest::trd::trd_market_str
    // / futu-auth-pipeline::body_aware::trd_market_str 一致.
    match market_int {
        1 => Some("HK"),
        2 => Some("US"),
        3 => Some("CN"),
        4 => Some("HKCC"),
        5 => Some("FUTURES"),
        6 => Some("SG"),
        7 => Some("CRYPTO"),
        8 => Some("AU"),
        10 => Some("FUTURES_SIMULATE_HK"),
        11 => Some("FUTURES_SIMULATE_US"),
        12 => Some("FUTURES_SIMULATE_SG"),
        13 => Some("FUTURES_SIMULATE_JP"),
        15 => Some("JP"),
        111 => Some("MY"),
        112 => Some("CA"),
        113 => Some("HKFUND"),
        123 => Some("USFUND"),
        124 => Some("SGFUND"),
        125 => Some("MYFUND"),
        126 => Some("JPFUND"),
        _ => None,
    }
}

/// 推送分发器
pub struct PushDispatcher {
    connections: Arc<DashMap<u64, ClientConn>>,
    subscriptions: Arc<SubscriptionManager>,
    metrics: Option<Arc<GatewayMetrics>>,
    /// C++ `APIServerCS_Core.cpp:246-247` assigns a monotonic push serial
    /// number before sending each client push frame. It is only used by
    /// clients/CS reconciliation to identify push packets, not for backend
    /// replay.
    push_serial_no: AtomicU32,
    /// Per-connection EventContract push cursors.
    ///
    /// Frozen C++ keeps these in `IQotLastPushRecord` plus the
    /// `APIServer_Qot_EventContractPush` ticker/K-line maps. They cannot live
    /// in the shared quote cache because first push advances only the target
    /// connection's ticker cursor.
    event_contract_cursors: parking_lot::Mutex<EventContractPushCursors>,
    /// Ordinary KLine last-push state shared by first-push and live delivery.
    /// Physical connection generation is part of the key, so a replacement
    /// socket cannot inherit the prior socket's cursor. The cursor value also
    /// owns the current push-registration intent epoch, so unregister followed
    /// by re-register receives one new first-push without weakening per-intent
    /// duplicate suppression.
    kline_cursors: Arc<parking_lot::Mutex<HashMap<KlineCursorKey, RegisteredKlinePushCursor>>>,
    /// Canonical ordinary KLine cursor for broadcast-style external sinks.
    /// Native first/live delivery remains independently per connection.
    external_kline_cursors: parking_lot::Mutex<HashMap<(String, i32, i32), KlinePushCursor>>,
    /// 外部推送接收器列表 (REST WebSocket, gRPC 等)
    external_sinks: Vec<Arc<dyn ExternalPushSink>>,
    startup_readiness: crate::identity::StartupReadiness,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct EventContractKlineCursor {
    time_key: String,
    fingerprint: u64,
}

#[derive(Default)]
struct EventContractPushCursors {
    order_book_hashes: HashMap<(u64, String), u64>,
    ticker_sequences: HashMap<(u64, String), u64>,
    kline_points: HashMap<(u64, String, i32, i32), EventContractKlineCursor>,
}

type EventContractKlineUpdates = Vec<(i32, EventContractKlineCursor)>;

enum EventContractCursorUpdate {
    OrderBook(u64),
    Ticker(u64),
    Kline(EventContractKlineUpdates),
}

impl PushDispatcher {
    /// 创建推送分发器。`connections` 和 `subscriptions` 由
    /// [`super::listener::ApiServer`] 共享；外部 sink / metrics 可通过
    /// [`Self::with_metrics`] / [`Self::with_external_sink`] 后续注入。
    pub fn new(
        connections: Arc<DashMap<u64, ClientConn>>,
        subscriptions: Arc<SubscriptionManager>,
    ) -> Self {
        let kline_cursors = Arc::new(parking_lot::Mutex::new(HashMap::<
            KlineCursorKey,
            RegisteredKlinePushCursor,
        >::new()));
        let cursor_cleanup = Arc::clone(&kline_cursors);
        subscriptions.register_disconnect_observer(Arc::new(move |conn_id| {
            cursor_cleanup
                .lock()
                .retain(|key, _| key.conn_id != conn_id);
        }));
        Self {
            connections,
            subscriptions,
            metrics: None,
            push_serial_no: AtomicU32::new(0),
            event_contract_cursors: parking_lot::Mutex::new(EventContractPushCursors::default()),
            kline_cursors,
            external_kline_cursors: parking_lot::Mutex::new(HashMap::new()),
            external_sinks: Vec::new(),
            startup_readiness: crate::identity::StartupReadiness::default(),
        }
    }

    /// 设置监控指标引用
    pub fn with_metrics(mut self, metrics: Arc<GatewayMetrics>) -> Self {
        self.metrics = Some(metrics);
        self
    }

    /// 添加外部推送接收器（可多次调用注册多个）
    pub fn with_external_sink(mut self, sink: Arc<dyn ExternalPushSink>) -> Self {
        self.external_sinks.push(sink);
        self
    }

    pub fn with_startup_readiness(
        mut self,
        startup_readiness: crate::identity::StartupReadiness,
    ) -> Self {
        self.startup_readiness = startup_readiness;
        self
    }

    fn delivery_ready(&self) -> bool {
        self.startup_readiness.snapshot().state == crate::identity::StartupState::Ready
    }

    fn record_push(&self) {
        if let Some(ref m) = self.metrics {
            m.client_pushes_sent
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
    }

    fn record_push_send_failure(&self) {
        if let Some(ref m) = self.metrics {
            m.client_push_send_failures
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
    }

    fn record_ordinary_client_backpressure_disconnect(&self) {
        if let Some(ref m) = self.metrics {
            m.ordinary_client_push_backpressure_disconnects
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
    }

    fn record_qot_client_backpressure_drop(&self, sub_type: i32) {
        if let Some(ref m) = self.metrics {
            m.record_qot_client_push_backpressure_drop(sub_type);
        }
    }

    fn next_push_serial_no(&self) -> u32 {
        self.push_serial_no
            .fetch_add(1, Ordering::Relaxed)
            .wrapping_add(1)
    }

    fn try_send_ordinary_client_frame(
        &self,
        conn_id: u64,
        tx: tokio::sync::mpsc::Sender<FutuFrame>,
        frame: FutuFrame,
        push_path: &'static str,
    ) {
        match tx.try_send(frame) {
            Ok(()) => self.record_push(),
            Err(TrySendError::Full(_frame)) => {
                match self.subscriptions.request_client_close(conn_id) {
                    Some(true) => {
                        self.record_ordinary_client_backpressure_disconnect();
                        tracing::warn!(
                            conn_id,
                            push_path,
                            "ordinary client push queue is full; closing slow connection"
                        );
                    }
                    Some(false) => {}
                    None => {
                        let removed = self.connections.remove(&conn_id).is_some();
                        self.subscriptions.on_disconnect(conn_id);
                        if removed {
                            self.record_ordinary_client_backpressure_disconnect();
                        }
                        tracing::error!(
                            conn_id,
                            push_path,
                            removed,
                            "ordinary client push queue is full without registered close control; removed connection fail-closed"
                        );
                    }
                }
            }
            Err(TrySendError::Closed(_frame)) => {
                self.record_push_send_failure();
                tracing::warn!(
                    conn_id,
                    push_path,
                    "client push send failed because downstream channel is closed"
                );
            }
        }
    }

    fn try_send_qot_client_frame(
        &self,
        tx: tokio::sync::mpsc::Sender<FutuFrame>,
        frame: FutuFrame,
        sub_type: i32,
        push_path: &'static str,
    ) -> bool {
        match tx.try_send(frame) {
            Ok(()) => {
                self.record_push();
                true
            }
            Err(TrySendError::Full(_frame)) => {
                self.record_qot_client_backpressure_drop(sub_type);
                tracing::warn!(
                    push_path,
                    sub_type,
                    "client quote push dropped because downstream channel is full"
                );
                false
            }
            Err(TrySendError::Closed(_frame)) => {
                self.record_push_send_failure();
                tracing::warn!(
                    push_path,
                    "client quote push send failed because downstream channel is closed"
                );
                false
            }
        }
    }

    /// 向指定连接推送（自动处理 AES 加密）
    pub async fn push_to_conn(&self, conn_id: u64, proto_id: u32, body: Vec<u8>) {
        if !self.delivery_ready() {
            return;
        }
        let push = self.connections.get(&conn_id).map(|conn| {
            let frame = conn.make_frame(proto_id, self.next_push_serial_no(), Bytes::from(body));
            (conn.tx.clone(), frame)
        });
        if let Some((tx, frame)) = push {
            self.try_send_ordinary_client_frame(conn_id, tx, frame, "push_to_conn");
        }
    }

    /// Direct async delivery; generation is checked in the physical tx lookup.
    pub async fn push_qot_to_conn_generation(
        &self,
        conn_id: u64,
        expected_generation: u64,
        proto_id: u32,
        body: Vec<u8>,
    ) {
        if !self.delivery_ready() {
            return;
        }
        let push = self.connections.get(&conn_id).and_then(|conn| {
            if conn.session_generation != expected_generation
                || !should_push_to(&conn, Scope::QotRead, "indicator_direct")
            {
                return None;
            }
            let frame = conn.make_frame(proto_id, self.next_push_serial_no(), Bytes::from(body));
            Some((conn.tx.clone(), frame))
        });
        if let Some((tx, frame)) = push {
            self.try_send_qot_client_frame(tx, frame, 0, "push_qot_to_conn_generation");
        }
    }

    /// 向所有订阅了通知的连接广播（每个连接独立 AES 加密）
    pub async fn push_notify(&self, proto_id: u32, body: Vec<u8>) {
        if !self.delivery_ready() {
            return;
        }
        let body = Bytes::from(body);
        let body_sha1 = FutuFrame::body_sha1(&body);
        let pushes: Vec<_> = self
            .connections
            .iter()
            .filter_map(|entry| {
                let conn = entry.value();
                if !conn.recv_notify {
                    return None;
                }
                // 防御深度：订阅阶段应该已经挡了 qot:read 外的 key，这里再过滤一次
                if !should_push_to(conn, Scope::QotRead, "notify") {
                    return None;
                }
                let serial_no = self.next_push_serial_no();
                let frame = conn.make_frame_with_sha1(proto_id, serial_no, body.clone(), body_sha1);
                Some((conn.conn_id, conn.tx.clone(), frame))
            })
            .collect();
        for (conn_id, tx, frame) in pushes {
            self.try_send_ordinary_client_frame(conn_id, tx, frame, "push_notify");
        }
    }

    /// 向订阅了指定交易账户的所有连接推送
    pub async fn push_trd_acc(&self, acc_id: u64, proto_id: u32, body: Vec<u8>) {
        if !self.delivery_ready() {
            return;
        }
        // v1.4.105 D3 (Phase 4) T-B4: 一次 decode 提取 trd_market 给 sinks
        // 共用. 4 surface (gRPC / REST WS / 等) 复用同一字符串避免重复 decode.
        let trd_market = extract_trd_market_from_trade_body(proto_id, &body);
        // 同时推送给外部接收器 (REST WebSocket, gRPC 等)
        for sink in &self.external_sinks {
            sink.on_trade_push(acc_id, proto_id, &body, trd_market);
        }
        let body = Bytes::from(body);
        let body_sha1 = FutuFrame::body_sha1(&body);
        let subscribers = self.subscriptions.get_acc_subscribers(acc_id);
        let pushes: Vec<_> = subscribers
            .into_iter()
            .filter_map(|conn_id| {
                let conn = self.connections.get(&conn_id)?;
                // 防御深度：trade push 要求 acc:read
                if !should_push_to(&conn, Scope::AccRead, "trade") {
                    return None;
                }
                // codex round 1 F4 (P2) v1.4.105: Layer 1 — caller key
                // allowed_acc_ids push-time 硬过滤. 防 stale subscription /
                // KeyRecord reload 后 acc 范围窄化 / 历史 bug 留下的 conn→acc
                // 关系 让受限 key 仍收到非授权 acc 的 trade push.
                //
                // 设计同 futu-auth::Limits / KeyRecord:
                // - allowed_acc_ids None = 无限制 (legacy / unrestricted key) → 放行
                // - 非空 set + acc_id ∉ set → drop + metric
                // - 空 set = 无限制 (向后兼容); deny-all 用 sentinel {0}
                if let Some(allowed_accs) = conn.allowed_acc_ids.as_ref()
                    && !allowed_accs.is_empty()
                    && !allowed_accs.contains(&acc_id)
                {
                    let key_id = conn.key_id.as_deref().unwrap_or("<none>");
                    futu_auth::metrics::bump_ws_filtered("trade_acc_id", key_id);
                    return None;
                }
                // v1.4.105 D3 (Phase 4) T-B2: Layer 3 — caller key allowed_markets
                // 限制. trd_market None (decode 失败 / market 未知) → 不 trigger
                // drop (向后兼容 — pitfall #57 backend-semantic 未真机 verify
                // 前 default 不 drop, 防 false-negative 错过用户合法 push).
                // allowed_markets None / 空 set = 无限制.
                if let (Some(market), Some(allowed_mkts)) =
                    (trd_market, conn.allowed_markets.as_ref())
                    && !allowed_mkts.is_empty()
                    && !allowed_mkts.contains(market)
                {
                    let key_id = conn.key_id.as_deref().unwrap_or("<none>");
                    futu_auth::metrics::bump_ws_filtered("trade_market", key_id);
                    return None;
                }
                let serial_no = self.next_push_serial_no();
                let frame = conn.make_frame_with_sha1(proto_id, serial_no, body.clone(), body_sha1);
                Some((conn.conn_id, conn.tx.clone(), frame))
            })
            .collect();
        for (conn_id, tx, frame) in pushes {
            self.try_send_ordinary_client_frame(conn_id, tx, frame, "push_trd_acc");
        }
    }

    /// 向所有已连接的客户端广播（到价提醒等，不需要订阅通知）
    /// C++ 检查 IsConnSubRecvNotify，对齐使用 InitConnect.recvNotify。
    pub async fn push_broadcast(&self, proto_id: u32, body: Vec<u8>) {
        if !self.delivery_ready() {
            return;
        }
        // 同时推送给外部接收器 (REST WebSocket, gRPC 等)
        for sink in &self.external_sinks {
            sink.on_broadcast_push(proto_id, &body);
        }
        let body = Bytes::from(body);
        let body_sha1 = FutuFrame::body_sha1(&body);
        let pushes: Vec<_> = self
            .connections
            .iter()
            .filter_map(|entry| {
                let conn = entry.value();
                if !conn.recv_notify {
                    return None;
                }
                if !should_push_to(conn, Scope::QotRead, "broadcast") {
                    return None;
                }
                let serial_no = self.next_push_serial_no();
                let frame = conn.make_frame_with_sha1(proto_id, serial_no, body.clone(), body_sha1);
                Some((conn.conn_id, conn.tx.clone(), frame))
            })
            .collect();
        for (conn_id, tx, frame) in pushes {
            self.try_send_ordinary_client_frame(conn_id, tx, frame, "push_broadcast");
        }
    }

    pub fn push_private_user(&self, proto_id: u32, body: Vec<u8>) {
        if !self.delivery_ready() {
            return;
        }
        for sink in &self.external_sinks {
            sink.on_private_user_push(proto_id, &body);
        }
        let body = Bytes::from(body);
        let body_sha1 = FutuFrame::body_sha1(&body);
        let pushes: Vec<_> = self
            .connections
            .iter()
            .filter_map(|entry| {
                let conn = entry.value();
                if !conn.recv_notify
                    || conn.key_id.is_none()
                    || !conn.scopes.contains(&Scope::AccRead)
                {
                    return None;
                }
                let serial_no = self.next_push_serial_no();
                let frame = conn.make_frame_with_sha1(proto_id, serial_no, body.clone(), body_sha1);
                Some((conn.conn_id, conn.tx.clone(), frame))
            })
            .collect();
        for (conn_id, tx, frame) in pushes {
            self.try_send_ordinary_client_frame(conn_id, tx, frame, "push_private_user");
        }
    }

    fn push_event_contract_qot(
        &self,
        security_key: &str,
        sub_type: i32,
        rehab_type: i32,
        proto_id: u32,
        body: &[u8],
    ) {
        self.prune_event_contract_cursors();
        let subscribers = self.subscriptions.get_qot_push_subscribers_by_cache_key(
            security_key,
            sub_type,
            rehab_type,
        );
        for conn_id in subscribers {
            let Some(conn) = self.connections.get(&conn_id) else {
                continue;
            };
            if !should_push_to(&conn, Scope::QotRead, "quote") {
                continue;
            }
            let decision =
                self.event_contract_body_for_conn(conn_id, security_key, sub_type, proto_id, body);
            let Some((body, update)) = decision else {
                continue;
            };
            let body = Bytes::from(body);
            let frame = conn.make_frame_with_sha1(
                proto_id,
                self.next_push_serial_no(),
                body.clone(),
                FutuFrame::body_sha1(&body),
            );
            let tx = conn.tx.clone();
            drop(conn);
            if self.try_send_qot_client_frame(tx, frame, sub_type, "push_event_contract_qot") {
                self.commit_event_contract_cursor(conn_id, security_key, sub_type, update);
            }
        }
    }

    fn prune_event_contract_cursors(&self) {
        let mut cursors = self.event_contract_cursors.lock();
        cursors
            .order_book_hashes
            .retain(|(conn_id, _), _| self.connections.contains_key(conn_id));
        cursors
            .ticker_sequences
            .retain(|(conn_id, _), _| self.connections.contains_key(conn_id));
        cursors
            .kline_points
            .retain(|(conn_id, _, _, _), _| self.connections.contains_key(conn_id));
    }

    fn event_contract_body_for_conn(
        &self,
        conn_id: u64,
        security_key: &str,
        sub_type: i32,
        proto_id: u32,
        body: &[u8],
    ) -> Option<(Vec<u8>, EventContractCursorUpdate)> {
        match proto_id {
            futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_ORDER_BOOK => {
                let hash = push_body_hash(body);
                let repeated = self
                    .event_contract_cursors
                    .lock()
                    .order_book_hashes
                    .get(&(conn_id, security_key.to_owned()))
                    .is_some_and(|previous| *previous == hash);
                (!repeated).then(|| (body.to_vec(), EventContractCursorUpdate::OrderBook(hash)))
            }
            futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_TICKER => {
                let previous = self
                    .event_contract_cursors
                    .lock()
                    .ticker_sequences
                    .get(&(conn_id, security_key.to_owned()))
                    .copied()
                    .unwrap_or(0);
                filter_event_contract_ticker_body(body, previous)
                    .map(|(body, sequence)| (body, EventContractCursorUpdate::Ticker(sequence)))
            }
            futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_KLINE => {
                let cursors = self.event_contract_cursors.lock();
                filter_event_contract_kline_body(body, |direction| {
                    cursors
                        .kline_points
                        .get(&(conn_id, security_key.to_owned(), sub_type, direction))
                        .cloned()
                })
                .map(|(body, updates)| (body, EventContractCursorUpdate::Kline(updates)))
            }
            _ => None,
        }
    }

    fn commit_event_contract_cursor(
        &self,
        conn_id: u64,
        security_key: &str,
        sub_type: i32,
        update: EventContractCursorUpdate,
    ) {
        let mut cursors = self.event_contract_cursors.lock();
        match update {
            EventContractCursorUpdate::OrderBook(hash) => {
                cursors
                    .order_book_hashes
                    .insert((conn_id, security_key.to_owned()), hash);
            }
            EventContractCursorUpdate::Ticker(sequence) => {
                cursors
                    .ticker_sequences
                    .entry((conn_id, security_key.to_owned()))
                    .and_modify(|current| *current = (*current).max(sequence))
                    .or_insert(sequence);
            }
            EventContractCursorUpdate::Kline(updates) => {
                for (direction, cursor) in updates {
                    cursors.kline_points.insert(
                        (conn_id, security_key.to_owned(), sub_type, direction),
                        cursor,
                    );
                }
            }
        }
    }
}

fn push_body_hash(body: &[u8]) -> u64 {
    let mut hasher = DefaultHasher::new();
    body.hash(&mut hasher);
    hasher.finish()
}

fn event_contract_ticker_cursor_from_body(proto_id: u32, body: &[u8]) -> Option<(String, u64)> {
    use prost::Message;

    if proto_id != futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_TICKER {
        return None;
    }
    let response = futu_proto::qot_update_event_contract_ticker::Response::decode(body).ok()?;
    let item = response.s2c?.ticker_list.into_iter().next()?;
    let sequence = item
        .ticker_list
        .iter()
        .filter_map(|point| point.sequence.as_deref()?.parse::<u64>().ok())
        .max()?;
    Some((format!("{}_{}", item.code.market, item.code.code), sequence))
}

fn filter_event_contract_ticker_body(body: &[u8], previous: u64) -> Option<(Vec<u8>, u64)> {
    use prost::Message;

    let mut response = futu_proto::qot_update_event_contract_ticker::Response::decode(body).ok()?;
    let s2c = response.s2c.as_mut()?;
    let mut newest = previous;
    for item in &mut s2c.ticker_list {
        item.ticker_list.retain(|point| {
            let Some(sequence) = point
                .sequence
                .as_deref()
                .and_then(|value| value.parse::<u64>().ok())
            else {
                return false;
            };
            if sequence <= previous {
                return false;
            }
            newest = newest.max(sequence);
            true
        });
    }
    s2c.ticker_list.retain(|item| !item.ticker_list.is_empty());
    (!s2c.ticker_list.is_empty()).then(|| (response.encode_to_vec(), newest))
}

fn filter_event_contract_kline_body(
    body: &[u8],
    mut previous_for_direction: impl FnMut(i32) -> Option<EventContractKlineCursor>,
) -> Option<(Vec<u8>, EventContractKlineUpdates)> {
    use prost::Message;

    let mut response = futu_proto::qot_update_event_contract_kline::Response::decode(body).ok()?;
    let s2c = response.s2c.as_mut()?;
    let mut updates = Vec::new();
    for item in &mut s2c.kline_list {
        let direction = item.pre_side.unwrap_or(0);
        let mut cursor = previous_for_direction(direction);
        if item.kline_list.is_empty() {
            let fingerprint = push_body_hash(&item.encode_to_vec());
            if cursor
                .as_ref()
                .is_some_and(|previous| previous.fingerprint == fingerprint)
            {
                continue;
            }
            updates.push((
                direction,
                EventContractKlineCursor {
                    time_key: cursor
                        .as_ref()
                        .map(|previous| previous.time_key.clone())
                        .unwrap_or_default(),
                    fingerprint,
                },
            ));
            continue;
        }
        item.kline_list.retain(|point| {
            let fingerprint = push_body_hash(&point.encode_to_vec());
            if cursor.as_ref().is_some_and(|previous| {
                previous.time_key > point.time_key
                    || (previous.time_key == point.time_key && previous.fingerprint == fingerprint)
            }) {
                return false;
            }
            cursor = Some(EventContractKlineCursor {
                time_key: point.time_key.clone(),
                fingerprint,
            });
            true
        });
        if let Some(cursor) = cursor
            && !item.kline_list.is_empty()
        {
            updates.push((direction, cursor));
        }
    }
    s2c.kline_list.retain(|item| {
        !item.kline_list.is_empty()
            || updates
                .iter()
                .any(|(direction, _)| *direction == item.pre_side.unwrap_or(0))
    });
    (!s2c.kline_list.is_empty() && !updates.is_empty()).then(|| (response.encode_to_vec(), updates))
}

#[cfg(test)]
mod tests;