futu_gateway_core/bridge/

push_parser.rs

//! bridge push dispatcher core（v1.4.60 Phase E1 从 bridge/mod.rs 抽出）
//!
//! v1.4.109 开始把各 push 类型的纯解析 / 投影函数拆到同目录 sibling modules，
//! 但仍通过 Rust 多 `impl GatewayBridge` 块挂在同一个类型上，保持调用点行为不变。
//!
//! 当前本文件保留 quote push bit 分发逻辑；具体
//! price/orderbook/ticker/kline/rt/basic_qot/broker_queue/trade_notify/msg_center
//! 解析在相应 `bridge/push_*.rs` 文件中，qot-right 刷新在
//! `bridge/qot_right_refresh.rs` 中，投递/丢弃 metric 在
//! `bridge/push_delivery.rs` 中，字节级去重在 `bridge/push_dedup.rs` 中。

use futu_cache::qot_right::{QotRightCache, QotRightData};
use futu_cache::static_data::CachedSecurityInfo;
use futu_core::qot_stock_key::QotSecurityKey;
use futu_server::subscription::SubscriptionManager;
use std::time::{Duration, SystemTime};

use super::push_dedup::is_duplicate_push;
use super::push_delivery::try_send_qot_push_with_metric;
use super::{GatewayBridge, PushEvent};

pub use super::qot_right_push::QotRightPushDedup;
pub use super::qot_right_refresh::{
    QotRightRefreshErr, QotRightRefreshReport, QotRightRequestMode,
};

struct QuotePushBitContext<'a> {
    source: &'static str,
    stock_id: u64,
    /// v1.4.110 codex Phase 4 Slice 7: SecurityQuote.broker_id (从 push wire 读).
    /// 用于 crypto LV2 path 写入 broker-level cache. `None` = 普通股 (公共 cache).
    broker_id: Option<u32>,
    qot_sec_key: &'a QotSecurityKey,
    cache_key: &'a str,
    sec_info: &'a CachedSecurityInfo,
    bit: u32,
    sub_prob: i64,
    data: &'a [u8],
    f3c_recv_at: SystemTime,
    count_delay: bool,
    qot_cache: &'a futu_cache::qot_cache::QotCache,
    qot_right: &'a QotRightData,
    /// v1.4.110 codex Phase 4 Slice 7: crypto LV2 多交易所缓存. crypto LV2 push
    /// 路径 (bit==17 + is_crypto) 写入 exchange-level cache + 反向索引重建 broker cache.
    crypto_exchange_cache: &'a futu_cache::crypto_exchange_cache::CryptoExchangeCache,
    push_tx: &'a tokio::sync::mpsc::Sender<PushEvent>,
    metrics: &'a futu_server::metrics::GatewayMetrics,
}

#[cfg(test)]
#[path = "push_parser_tests.rs"]
mod push_parser_tests;

impl GatewayBridge {
    fn process_quote_push_bit(ctx: QuotePushBitContext<'_>) {
        let QuotePushBitContext {
            source,
            stock_id,
            broker_id,
            qot_sec_key,
            cache_key,
            sec_info,
            bit,
            sub_prob,
            data,
            f3c_recv_at,
            count_delay,
            qot_cache,
            qot_right,
            crypto_exchange_cache,
            push_tx,
            metrics,
        } = ctx;

        tracing::debug!(
            stock_id,
            %cache_key,
            bit,
            data_len = data.len(),
            source,
            "CMD6212 processing bit"
        );

        // v1.4.90 P2-A: 字节级 dedup. WS subscribe 后某些 sub_type
        // (KL_Month / 特定股票) 收到字节一模一样的 push N 条 →
        // 同 (proto_id=6212, sec_key, hash(bit + data)) 5s 内 skip.
        if is_duplicate_push(6212, cache_key, bit, sub_prob, data) {
            tracing::debug!(
                stock_id,
                %cache_key,
                bit,
                sub_prob,
                data_len = data.len(),
                source,
                "v1.4.90 P2-A: duplicate push within 5s window — skipped"
            );
            return;
        }

        match bit {
            0 => {
                // SBIT_PRICE: 解析 + 缓存 + 构建 FTAPI push
                if let Some(event) = Self::parse_price_to_push(
                    data,
                    cache_key,
                    sec_info,
                    qot_cache,
                    f3c_recv_at,
                    count_delay,
                ) {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            1 => {
                // SBIT_STOCK_STATE: 停牌状态 → 更新缓存 + 发推送
                Self::parse_stock_state(data, cache_key, stock_id, qot_cache);
                // C++ 每个 bit 处理完都推送当前累积状态
                if let Some(event) =
                    Self::build_basic_qot_push_from_cache(cache_key, sec_info, qot_cache)
                {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            3 => {
                // SBIT_ORDER_BOOK: 买卖盘
                if let Some(event) = Self::parse_order_book_to_push_with_rights(
                    data,
                    cache_key,
                    sec_info,
                    qot_cache,
                    qot_right,
                    sub_prob,
                    f3c_recv_at,
                    count_delay,
                ) {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            17 | 39 => {
                // SBIT_US_LV2_ORDER / SBIT_MEGER_LV2_ORDER: US/crypto/futures
                // LV2 合单摆盘。C++ QotRealTimeData.cpp 同一 parser 处理这两类 bit。
                //
                // v1.4.110 codex Phase 4 Slice 7: crypto + broker_id Some(N) 时走
                // exchange-cache 路径 (写 (stock_id, lv2_prob) → exchange cache,
                // 反向索引找受影响 broker, per-broker 40 档 merge → broker cache).
                // 普通股 (USLv2 / 期货) 仍走原 `parse_us_lv2_order_book_to_push`.
                if futu_cache::qot_right::is_crypto_market(sec_info) {
                    // v1.4.110 R6-2: crypto LV2 一次 exchange-level push 需 fan-out
                    // 给所有订阅同交易所的 broker → parser 返 Vec<PushEvent>, 逐个发.
                    for event in Self::parse_crypto_lv2_order_book_to_push(
                        data,
                        stock_id,
                        broker_id,
                        cache_key,
                        sec_info,
                        qot_cache,
                        crypto_exchange_cache,
                        f3c_recv_at,
                        count_delay,
                    ) {
                        try_send_qot_push_with_metric(push_tx, event, metrics);
                    }
                } else if let Some(event) = Self::parse_us_lv2_order_book_to_push(
                    data,
                    cache_key,
                    sec_info,
                    qot_cache,
                    crypto_exchange_cache,
                    f3c_recv_at,
                    count_delay,
                ) {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            5 => {
                // SBIT_DEAL_STATISTICS: 成交量/额/OHLC → 更新缓存 + 发推送
                Self::parse_deal_statistics(data, cache_key, qot_cache);
                // C++ 每个 bit 处理完都推送当前累积状态
                if let Some(event) =
                    Self::build_basic_qot_push_from_cache(cache_key, sec_info, qot_cache)
                {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            9 => {
                // SBIT_HK_BROKER_QUEUE: 港股经纪队列
                if let Some(event) = Self::parse_broker_queue_to_push(
                    data,
                    cache_key,
                    sec_info,
                    qot_cache,
                    f3c_recv_at,
                    count_delay,
                ) {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            13 => {
                // v1.4.72 BUG-006 L2 fix (external reviewer v1.4.69 P1):
                // SBIT_US_PREMARKET_AFTERHOURS_DETAIL - 美股盘前/盘后/夜盘
                // OHLCV → 更新缓存。后续 Price/DealStats 推送会从 cache
                // 带上 overnight 字段，下游 subscriber 收到完整数据。
                Self::parse_us_pre_after_detail(data, cache_key, qot_cache);
                if let Some(event) =
                    Self::build_basic_qot_push_from_cache(cache_key, sec_info, qot_cache)
                {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            35 => {
                // SBIT_TICK: 逐笔成交
                if let Some(event) = Self::parse_ticker_to_push(
                    data,
                    cache_key,
                    sec_info,
                    qot_cache,
                    f3c_recv_at,
                    count_delay,
                ) {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            20 => {
                // SBIT_TIME_SHARING: 分时数据
                if let Some(event) = Self::parse_rt_to_push(data, qot_sec_key, sec_info, qot_cache)
                {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            21..=31 | 36 | 37 | 40 | 41 => {
                // SBIT_KLINE_*: K 线数据. Most values are 21..=31, but C++
                // keeps 10/120/180/240 minute cycles on non-contiguous bits
                // 40/36/41/37. Ref:
                // FutuOpenD/Src/NNProtoFile/Server/PB/Quote/FTCmdStockQuoteSubData.proto:92-98.
                if let Some(event) =
                    Self::parse_kline_to_push(data, bit, cache_key, sec_info, qot_cache)
                {
                    try_send_qot_push_with_metric(push_tx, event, metrics);
                }
            }
            _ => {
                tracing::debug!(bit, source, "CMD6212: unhandled bit");
            }
        }
    }

    /// Apply a CMD6824 `FetchQuoteRsp` body to the QOT cache using the same
    /// BitQuote projections as the normal CMD6212 push path.
    ///
    /// C++ `NNBiz_Qot_StockSnapshot::OnReply_PullBasicQot` handles the direct
    /// `Pull_SubData` reply by raising `NN_OMEvent_Qot_Refresh_RealTimeData`;
    /// waiting API requests can therefore complete from the reply itself even
    /// when no later push arrives. This helper preserves that cache-fill
    /// behavior without emitting an FTAPI push event for a request/response
    /// reply.
    pub fn apply_quote_fetch_response_to_cache(
        body: &[u8],
        qot_sec_key: &QotSecurityKey,
        sec_info: &CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
        qot_right: &QotRightData,
        f3c_recv_at: SystemTime,
    ) -> bool {
        use futu_backend::proto_internal::ft_cmd_stock_quote_fetch;

        let fetch_rsp: ft_cmd_stock_quote_fetch::FetchQuoteRsp = match prost::Message::decode(body)
        {
            Ok(rsp) => rsp,
            Err(err) => {
                tracing::debug!(
                    body_len = body.len(),
                    error = %err,
                    "CMD6824 FetchQuoteRsp decode failed"
                );
                return false;
            }
        };

        let requested_broker = qot_sec_key.stock_key.broker_id_or_zero();
        let mut applied = false;
        for sec_quote in &fetch_rsp.security_qta_list {
            if sec_quote.security_id != Some(sec_info.stock_id) {
                continue;
            }
            let response_broker = sec_quote
                .broker_id
                .and_then(|id| (id > 0).then_some(id as u32))
                .unwrap_or(0);
            if response_broker != requested_broker {
                tracing::debug!(
                    stock_id = sec_info.stock_id,
                    requested_broker,
                    response_broker,
                    "CMD6824 FetchQuoteRsp broker mismatch, skipping"
                );
                continue;
            }

            let cache_key = qot_sec_key.cache_key();
            for bit_quote in &sec_quote.bit_qta_list {
                let bit = bit_quote.bit.unwrap_or(u32::MAX);
                let data = match &bit_quote.data {
                    Some(data) => data.as_slice(),
                    None => continue,
                };

                match bit {
                    0 => {
                        if Self::parse_price_to_push(
                            data,
                            cache_key.as_str(),
                            sec_info,
                            qot_cache,
                            f3c_recv_at,
                            false,
                        )
                        .is_some()
                        {
                            applied = true;
                        }
                    }
                    1 => {
                        Self::parse_stock_state(
                            data,
                            cache_key.as_str(),
                            sec_info.stock_id,
                            qot_cache,
                        );
                        applied = true;
                    }
                    3 => {
                        if Self::parse_order_book_to_push_with_rights(
                            data,
                            cache_key.as_str(),
                            sec_info,
                            qot_cache,
                            qot_right,
                            bit_quote.prob.unwrap_or(0),
                            f3c_recv_at,
                            false,
                        )
                        .is_some()
                        {
                            applied = true;
                        }
                    }
                    5 => {
                        Self::parse_deal_statistics(data, cache_key.as_str(), qot_cache);
                        applied = true;
                    }
                    13 => {
                        Self::parse_us_pre_after_detail(data, cache_key.as_str(), qot_cache);
                        applied = true;
                    }
                    _ => {}
                }
            }
        }

        applied
    }

    /// 处理后端 CMD 6212 行情推送
    /// 解析 QuotePush → 更新 qot_cache → 通过 channel 发送 FTAPI 推送事件
    /// 返 `true` = decode 成功, `false` = body 无法解析为 QuotePush. caller 据此调
    /// `push_health.record_parse_error()` 触发 F3/F4 threshold.
    ///
    /// v1.4.106 codex 1140 F8: 加 `metrics` 参数, 替换所有 `let _ =
    /// push_tx.try_send(event)` 为 `try_send_with_metric()`. channel full /
    /// closed 时 record_qot_push_dropped(sub_type) + warn log, 不再 silent.
    ///
    /// **v1.4.110 Phase 2 Slice 5**: 读 `SecurityQuote.broker_id` (FTCmdStockQuoteSubData.proto:284),
    /// 用 `quote_push_targets_for_stock_key(stock_id, broker_id)` 重建 broker-aware
    /// `QotSecurityKey`. Cache 写入 + PushEvent routing 走 `QotSecurityKey::cache_key()`:
    /// - `broker_id=None` (普通股, 当前 backend 默认): `cache_key()` == public_sec_key,
    ///   与升级前行为完全等价
    /// - `broker_id=Some(N)` (crypto multi-broker): `cache_key()` = `"market_code@b{N}"`,
    ///   不同 broker 同 stock_id 写入独立 cache 桶, 不互相覆盖
    ///
    /// 对齐 C++ `NNBiz_Qot_PushQot.cpp:220-269` 从 SecurityQuote.broker_id 重建
    /// `StockKey(stock_id, broker_id)`.
    /// v1.4.110 codex Phase 4 Slice 7: 改 `pub` 以便 integration test 可以直接
    /// 触发 CMD6212 push pipeline (cross-crate test from gateway-qot/tests).
    /// 真生产路径仍由 push_callback 调.
    pub fn handle_quote_push(
        body: &[u8],
        static_cache: &futu_cache::static_data::StaticDataCache,
        qot_cache: &futu_cache::qot_cache::QotCache,
        qot_right_cache: &QotRightCache,
        crypto_exchange_cache: &futu_cache::crypto_exchange_cache::CryptoExchangeCache,
        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
        metrics: &futu_server::metrics::GatewayMetrics,
    ) -> bool {
        Self::handle_quote_push_with_recv_time(
            body,
            static_cache,
            qot_cache,
            qot_right_cache,
            crypto_exchange_cache,
            push_tx,
            metrics,
            SystemTime::now(),
        )
    }

    pub fn handle_quote_push_with_recv_time(
        body: &[u8],
        static_cache: &futu_cache::static_data::StaticDataCache,
        qot_cache: &futu_cache::qot_cache::QotCache,
        qot_right_cache: &QotRightCache,
        crypto_exchange_cache: &futu_cache::crypto_exchange_cache::CryptoExchangeCache,
        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
        metrics: &futu_server::metrics::GatewayMetrics,
        f3c_recv_at: SystemTime,
    ) -> bool {
        Self::handle_quote_push_with_recv_time_and_subscriptions(
            body,
            static_cache,
            qot_cache,
            qot_right_cache,
            crypto_exchange_cache,
            push_tx,
            metrics,
            f3c_recv_at,
            None,
        )
    }

    pub fn handle_quote_push_with_recv_time_and_subscriptions(
        body: &[u8],
        static_cache: &futu_cache::static_data::StaticDataCache,
        qot_cache: &futu_cache::qot_cache::QotCache,
        qot_right_cache: &QotRightCache,
        crypto_exchange_cache: &futu_cache::crypto_exchange_cache::CryptoExchangeCache,
        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
        metrics: &futu_server::metrics::GatewayMetrics,
        f3c_recv_at: SystemTime,
        subscriptions: Option<&SubscriptionManager>,
    ) -> bool {
        use futu_backend::proto_internal::ft_cmd_stock_quote_sub;

        let push: ft_cmd_stock_quote_sub::QuotePush = match prost::Message::decode(body) {
            Ok(p) => p,
            Err(_) => {
                tracing::debug!(body_len = body.len(), "CMD6212 QuotePush decode failed");
                return false;
            }
        };

        tracing::debug!(
            sec_count = push.security_qta_list.len(),
            "CMD6212 push parsed"
        );
        let qot_right = qot_right_cache.get();

        for sec_quote in &push.security_qta_list {
            let stock_id = match sec_quote.security_id {
                Some(id) => id,
                None => continue,
            };

            // v1.4.110 Phase 2 Slice 5: 从 SecurityQuote.broker_id 派生
            // broker-aware key. proto field i32, 转 Option<NonZeroU32>
            // (C++ NN_BrokerID_Unknown=0 ≡ no-broker).
            let broker_id_raw = sec_quote.broker_id.unwrap_or(0);
            let broker_id: Option<std::num::NonZeroU32> = if broker_id_raw > 0 {
                std::num::NonZeroU32::new(broker_id_raw as u32)
            } else {
                None
            };

            let targets = static_cache
                .quote_push_targets_for_stock_key(stock_id, broker_id)
                .into_iter()
                .map(|(qot_sec_key, sec_info)| {
                    let cache_key = qot_sec_key.cache_key();
                    (qot_sec_key, cache_key, sec_info)
                })
                .collect::<Vec<_>>();
            if targets.is_empty() {
                tracing::debug!(
                    stock_id,
                    broker_id = ?broker_id,
                    "v1.4.106 audit 1148 F5: CMD6212 push stock_id not in id_to_key \
                     and no future main-link alias found (resolver miss). bump counter."
                );
                static_cache
                    .mkt_id_refresh_failed_total
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                // 注: 没 sec_key 无法 mark_stale_mkt_id (那需 key string), 仅 counter。
                // 下次 backend 推同 stock_id 时再 retry — backend 通常会 (subscribe 仍 alive)。
                continue;
            }

            for bit_quote in &sec_quote.bit_qta_list {
                let bit = bit_quote.bit.unwrap_or(u32::MAX);
                let sub_prob = bit_quote.prob.unwrap_or(0);
                let data = match &bit_quote.data {
                    Some(d) => d.as_slice(),
                    None => continue,
                };

                for (qot_sec_key, cache_key, sec_info) in &targets {
                    // v1.4.110 Phase 2 Slice 5: cache 写入 + PushEvent routing
                    // 都用 `cache_key()` — broker_id=None 时退化到 public_sec_key
                    // (零行为变化), Some(N) 时走 "market_code@b{N}" 独立 cache 桶.
                    // v1.4.110 codex Phase 4 Slice 7: broker_id u32 from NonZeroU32.
                    let bid_u32 = broker_id.map(|b| b.get());
                    let count_delay = qot_push_delay_count_allowed(subscriptions, qot_sec_key, bit);
                    Self::process_quote_push_bit(QuotePushBitContext {
                        source: "push",
                        stock_id,
                        broker_id: bid_u32,
                        qot_sec_key,
                        cache_key: cache_key.as_str(),
                        sec_info: sec_info.as_ref(),
                        bit,
                        sub_prob,
                        data,
                        f3c_recv_at,
                        count_delay,
                        qot_cache,
                        qot_right: &qot_right,
                        crypto_exchange_cache,
                        push_tx,
                        metrics,
                    });
                }
            }
        }
        // v1.4.84 §9 Phase 2.6: 全部 bit 处理完成（即使部分 sec_quote 没找到
        // stock_id 也算 decode 成功 — proto 解析层面 OK）
        true
    }
}

fn qot_push_delay_count_allowed(
    subscriptions: Option<&SubscriptionManager>,
    qot_sec_key: &QotSecurityKey,
    bit: u32,
) -> bool {
    let Some(sub_type) = qot_push_delay_sub_type(bit) else {
        return false;
    };
    subscriptions
        .map(|manager| {
            manager.qot_sub_elapsed_at_least_broker(qot_sec_key, sub_type, Duration::from_secs(3))
        })
        .unwrap_or(true)
}

fn qot_push_delay_sub_type(bit: u32) -> Option<i32> {
    match bit {
        0 => Some(1),           // SubType_Basic
        3 | 17 | 39 => Some(2), // SubType_OrderBook
        9 => Some(14),          // SubType_Broker
        35 => Some(4),          // SubType_Ticker
        _ => None,
    }
}

// ===== v1.4.77 E1 push_parser tests =====
//
// 背景：v3 code review master report §9.2 F-GW-E2E-002 建议给 push_parser.rs
// 10 个 `parse_*` 函数加单测（覆盖 bridge → cache 路径各 SBIT subtype）。
//
// 设计：每个 parse_* 都是 `impl GatewayBridge` 内的 associated fn（无 self，
// 可直接 `GatewayBridge::parse_*()` 调用）。测试策略：
//
// 1. 构造 minimal proto message（只填关键字段）+ prost encode → bytes
// 2. 创建 empty `QotCache`
// 3. 调 `GatewayBridge::parse_*(bytes, sec_key, &qot_cache)` 或相应签名
// 4. 断言 cache state 正确改变（常是 cache.basic_qot.get(sec_key).some_field）
//
// 对齐 CLAUDE.md 坑 #24 "backend 分 cmd_id channel"：任何 parse_* 悄悄漏
// field 会导致 cache 永远空，这里加单测防回归。
//
// Note: v1.4.77 起只覆盖最常用的 parse_*（stock_state / deal_statistics /
// us_pre_after_detail）。其余 parse_price / parse_order_book / parse_broker /
// parse_ticker / parse_kline / parse_rt 留 v1.4.78
// 继续扩（每个 parse_ 单测 scope 较大 ~30-50 LoC，单版避免 500+ LoC）。

// ===== v1.4.88 F-GW-E2E-002: push_parser 深度覆盖 (~40 新 tests) =====
//
// 背景: v3 code review master report §9.2 F-GW-E2E-002 建议 push_parser.rs 的
// 10 个 `parse_*` fn 全面单测覆盖。v1.4.77 E1 + v1.4.78 B1 + v1.4.80 B1 已达
// 10/10 fn 覆盖 (21 tests)，但部分 fn 只有 smoke/malformed 覆盖，缺乏 "normal
// path + 字段级断言 + cache merge 语义 + 边界条件"。
//
// 本版 F-GW-E2E-002 补齐每个 parse_* 的深度验证 (每个 fn 3-5 new tests):
//   - normal path: 填全字段 → 断言 cache 多字段 (price/volume/timestamp 等)
//     + 断言 PushEvent 结构 (proto_id / sub_type)
//   - cache merge: existing cache + 新 push → 验证 "非零字段替换 / 零字段保留"
//   - boundary: 空 list / None 字段 / zero / 负值 → 各 fn 独立逻辑
//
// 对齐 CLAUDE.md 坑 #24 "backend 分 cmd_id channel" — 漏掉 parse_* 某字段会
// 导致 cache 永远空且 handler 静默返 empty。这里深度 assert 防回归。