futu_server/

push.rs

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

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;

/// **防御深度**：即使客户端在订阅阶段某种方式绕过了 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]);
    /// 交易推送 (订单更新、成交更新等).
    ///
    /// `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,
    /// 外部推送接收器列表 (REST WebSocket, gRPC 等)
    external_sinks: Vec<Arc<dyn ExternalPushSink>>,
}

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 {
        Self {
            connections,
            subscriptions,
            metrics: None,
            push_serial_no: AtomicU32::new(0),
            external_sinks: Vec::new(),
        }
    }

    /// 设置监控指标引用
    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
    }

    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_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)
    }

    async fn send_client_frame(
        &self,
        tx: tokio::sync::mpsc::Sender<FutuFrame>,
        frame: FutuFrame,
        push_path: &'static str,
    ) {
        if let Err(e) = tx.send(frame).await {
            self.record_push_send_failure();
            tracing::warn!(push_path, error = %e, "client push send failed");
            return;
        }
        self.record_push();
    }

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

    /// 向指定连接推送（自动处理 AES 加密）
    pub async fn push_to_conn(&self, conn_id: u64, proto_id: u32, body: Vec<u8>) {
        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.send_client_frame(tx, frame, "push_to_conn").await;
        }
    }

    /// 向指定连接发送 quote 首推快照（带 qot:read 防御性过滤）。
    pub async fn push_qot_to_conn(&self, conn_id: u64, proto_id: u32, body: Vec<u8>) {
        let push = self.connections.get(&conn_id).and_then(|conn| {
            if !should_push_to(&conn, Scope::QotRead, "quote_first") {
                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");
        }
    }

    /// 向所有订阅了通知的连接广播（每个连接独立 AES 加密）
    pub async fn push_notify(&self, proto_id: u32, body: Vec<u8>) {
        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.tx.clone(), frame))
            })
            .collect();
        for (tx, frame) in pushes {
            self.send_client_frame(tx, frame, "push_notify").await;
        }
    }

    /// 向订阅了指定交易账户的所有连接推送
    pub async fn push_trd_acc(&self, acc_id: u64, proto_id: u32, body: Vec<u8>) {
        // 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.tx.clone(), frame))
            })
            .collect();
        for (tx, frame) in pushes {
            self.send_client_frame(tx, frame, "push_trd_acc").await;
        }
    }

    /// 向所有已连接的客户端广播（到价提醒等，不需要订阅通知）
    /// C++ 检查 IsConnSubRecvNotify，对齐使用 InitConnect.recvNotify。
    pub async fn push_broadcast(&self, proto_id: u32, body: Vec<u8>) {
        // 同时推送给外部接收器 (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.tx.clone(), frame))
            })
            .collect();
        for (tx, frame) in pushes {
            self.send_client_frame(tx, frame, "push_broadcast").await;
        }
    }

    /// 向**注册了 push** 的连接推送 quote (F4 P1 BLOCKER fix).
    ///
    /// **v1.4.106 codex 1131 F4 [P1]**: 改用 push_regs 三元 key 而非老的
    /// subscriber map. 老路径让仅订未 RegPush 的 conn 收到 quote push, 违反
    /// C++ `QotSubscribe::GetPushConn` 三元 key 路由 — F3 split state + F4
    /// push 端真过滤组合修复.
    ///
    /// **v1.4.110 broker-aware closeout**: `security_key` 是 push parser 传来的
    /// cache-key display string (`"market_code"` or `"market_code@b{id}"`).
    /// Dispatch 通过 cache-key bridge 查询 broker-aware push_regs, 不再直接调用
    /// legacy String facade.
    ///
    /// rehab_type 仅 KL 类有意义 (sub_type 6/7/8/9/10/11/12/13/15/16/17), 其它
    /// sub_type 应填 0. SubscriptionManager.get_qot_push_subscribers 内部对
    /// 非 KL 自动 normalize rehab=0.
    pub async fn push_qot(
        &self,
        security_key: &str,
        sub_type: i32,
        rehab_type: i32,
        proto_id: u32,
        body: Vec<u8>,
    ) {
        // 同时推送给外部接收器 (REST WebSocket, gRPC 等)
        // sink 内部按 (sec_key, sub_type, rehab_type) 自己过滤 conn 集合.
        for sink in &self.external_sinks {
            sink.on_quote_push(security_key, sub_type, rehab_type, proto_id, &body);
        }
        let body = Bytes::from(body);
        // **F4 P1**: push_regs 三元 key 查 conn allowlist (而非 subscriber set).
        let subscribers = self.subscriptions.get_qot_push_subscribers_by_cache_key(
            security_key,
            sub_type,
            rehab_type,
        );
        let body_sha1 = FutuFrame::body_sha1(&body);
        let pushes: Vec<_> = subscribers
            .into_iter()
            .filter_map(|conn_id| {
                let conn = self.connections.get(&conn_id)?;
                if !should_push_to(&conn, Scope::QotRead, "quote") {
                    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.tx.clone(), frame))
            })
            .collect();
        for (tx, frame) in pushes {
            self.try_send_qot_client_frame(tx, frame, sub_type, "push_qot");
        }
    }
}

#[cfg(test)]
mod tests;