futu_gateway_core/bridge/

qot_right_refresh.rs

//! QOT entitlement refresh.
//!
//! Mechanical split from `push_parser.rs`; exported types are re-exported by
//! `push_parser` to preserve existing integration-test and handler paths.

use std::sync::Arc;

use futu_backend::conn::BackendConn;
use futu_cache::qot_right::{QotRightBackendExtras, QotRightBackendUpdate, QotRightCache};

use super::{CacheBundle, GatewayBridge};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QotRightRequestMode {
    HighestPriority,
    WithPushedNotify,
}

#[derive(Debug, Clone)]
pub struct QotRightRefreshErr {
    pub stage: &'static str,
    pub detail: String,
}

impl std::fmt::Display for QotRightRefreshErr {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "qot_right refresh failed at {}: {}",
            self.stage, self.detail
        )
    }
}

impl std::error::Error for QotRightRefreshErr {}

pub use futu_cache::qot_right::QotRightRefreshReport;

pub(super) fn spawn_startup_refresh(
    backend: Arc<BackendConn>,
    caches: CacheBundle,
    epoch_uid: Option<u64>,
) -> tokio::task::JoinHandle<()> {
    caches.qot_right_cache.mark_pending();
    // F1: 首次 login 完成 → epoch 0→1 (后续 reconnect 继续递增).
    caches.qot_right_cache.advance_login_epoch(1, epoch_uid);
    tokio::spawn(async move {
        if let Err(e) = GatewayBridge::request_qot_right_v2(
            &backend,
            &caches.qot_right_cache,
            QotRightRequestMode::HighestPriority,
        )
        .await
        {
            tracing::warn!(error = %e, "startup qot_right refresh failed");
        }
    })
}

fn qot_right_quote_change_notify(
    qot_right_cache: &QotRightCache,
    mode: QotRightRequestMode,
) -> Option<futu_backend::proto_internal::ftcmd6651_qta_auth_chg::QuoteChangeNotify> {
    match mode {
        QotRightRequestMode::HighestPriority => None,
        QotRightRequestMode::WithPushedNotify => {
            qot_right_cache.pushed_quote_change_notify().map(|pn| {
                use futu_backend::proto_internal::ftcmd6651_qta_auth_chg as chg;
                chg::QuoteChangeNotify {
                    quote_change_list: pn
                        .change_items
                        .iter()
                        .map(|&(qt, before, after)| chg::QuoteChangeItem {
                            quote_type: Some(qt),
                            quote_before_change: Some(before),
                            quote_after_change: Some(after),
                            clt_type: None,
                            ip_area: None,
                        })
                        .collect(),
                }
            })
        }
    }
}

fn encode_qot_right_request(
    quote_change_notify: Option<
        futu_backend::proto_internal::ftcmd6651_qta_auth_chg::QuoteChangeNotify,
    >,
) -> Vec<u8> {
    use futu_backend::proto_internal::qta_auth;

    let req = qta_auth::QtaAuth6024Req {
        req_hk_auth: None,
        req_us_auth: None,
        req_cn_auth: None,
        // v1.4.44: proto 补齐后新增字段（device_type / quote_change_notify），
        // 沿用 C++ "空请求"语义；行情互踢重刷时才由 mode 注入 notify。
        device_type: None,
        quote_change_notify,
    };
    prost::Message::encode_to_vec(&req)
}

fn decode_qot_right_response(
    body_bytes: &[u8],
) -> Result<futu_backend::proto_internal::qta_auth::QtaAuth6024Rsp, String> {
    crate::handlers_shared::decode_srpc_or_direct(body_bytes, is_valid_qta_auth_6024_rsp)
        .map_err(|e| format!("decode failed (body_len={}): {}", body_bytes.len(), e))
}

fn is_valid_qta_auth_6024_rsp(r: &futu_backend::proto_internal::qta_auth::QtaAuth6024Rsp) -> bool {
    r.hk_qut_got_auth.is_some()
        || r.us_qut_got_auth.is_some()
        || r.cn_qut_got_auth.is_some()
        || r.shanghai_qut_auth.is_some()
        || r.shenzhen_qut_auth.is_some()
        || r.hk_option_auth.is_some()
        || r.hk_future_auth.is_some()
        || r.hk_option_orderbook_depth.is_some()
        || r.hk_future_orderbook_depth.is_some()
        || r.us_option_flag.is_some()
        || r.us_future_cme_cboe_auth.is_some()
        || r.us_future_auth.is_some()
        || r.us_lv2_arca_flag.is_some()
        || r.us_lv2_nyse_flag.is_some()
        || r.us_lv2_nasdaq_tv_flag.is_some()
        || r.us_lv2_edg_flag.is_some()
        || r.us_lv2_bzx_flag.is_some()
        || r.us_index_dow_jones_flag.is_some()
        || r.us_index_nasdaq_flag.is_some()
        || r.us_index_standard_poor_flag.is_some()
        || r.us_otc_deal_data_auth.is_some()
        || r.us_otc_order_book_auth.is_some()
        || r.sgx_future_auth.is_some()
        || r.sgx_stock_auth.is_some()
        || r.jp_future_auth.is_some()
        || r.jp_stock_tse_auth.is_some()
        || r.myx_auth.is_some()
        || r.digital_currency_auth.is_some()
        || r.digital_pt_orderbook_auth.is_some()
        || r.open_api_auth.is_some()
}

fn qot_right_backend_extras_from_rsp(
    r: &futu_backend::proto_internal::qta_auth::QtaAuth6024Rsp,
) -> QotRightBackendExtras {
    QotRightBackendExtras {
        us_lv2_flags: Some((
            r.us_lv2_arca_flag,
            r.us_lv2_nyse_flag,
            r.us_lv2_nasdaq_tv_flag,
            r.us_lv2_edg_flag,
            r.us_lv2_bzx_flag,
        )),
        us_index_flags: Some((
            r.us_index_dow_jones_flag,
            r.us_index_nasdaq_flag,
            r.us_index_standard_poor_flag,
        )),
        us_otc_auths: Some((r.us_otc_deal_data_auth, r.us_otc_order_book_auth)),
    }
}

fn qot_right_backend_update_from_rsp(
    r: &futu_backend::proto_internal::qta_auth::QtaAuth6024Rsp,
) -> QotRightBackendUpdate {
    let (sub_limit, kl_limit) = r
        .open_api_auth
        .as_ref()
        .map(|api_auth| (api_auth.sub_limit, api_auth.history_k_line_limit))
        .unwrap_or((None, None));
    // Ref: FutuOpenD/Src/NNProtoCenter/Quote/NNBiz_Qot_Right.cpp:396-478.
    // C++ checks each open_api_*_auth subfield before updating that market.
    let us_future_detail = r.us_future_auth.as_ref().and_then(|f| {
        let detail = (
            f.open_api_cme_auth,
            f.open_api_cbot_auth,
            f.open_api_nymex_auth,
            f.open_api_comex_auth,
        );
        if detail.0.is_none() && detail.1.is_none() && detail.2.is_none() && detail.3.is_none() {
            None
        } else {
            Some(detail)
        }
    });

    // Ref: FutuOpenD/Src/NNProtoCenter/Quote/NNBiz_Qot_Right.cpp:649-754,798-882,907-919.
    // These parsers run on every full CMD6024 rights reply and initialize the
    // new value to None before reading optional fields, so a missing field
    // clears stale SG/MY/JP/crypto rights instead of preserving the previous cache.
    let clear_missing_sg_future = Some(r.sgx_future_auth.unwrap_or(0));
    let clear_missing_jp_future = Some(r.jp_future_auth.unwrap_or(0));
    let clear_missing_sg_stock = Some(r.sgx_stock_auth.unwrap_or(0));
    let clear_missing_my_stock = Some(r.myx_auth.as_ref().and_then(|m| m.myx_auth).unwrap_or(0));
    let clear_missing_jp_stock = Some(r.jp_stock_tse_auth.unwrap_or(0));
    let clear_missing_crypto = Some(r.digital_currency_auth.unwrap_or(0));
    let clear_missing_crypto_pt_orderbook = Some(r.digital_pt_orderbook_auth.unwrap_or(0));

    QotRightBackendUpdate {
        hk_got: r.hk_qut_got_auth,
        us_got: r.us_qut_got_auth,
        cn_got: r.cn_qut_got_auth,
        sh_auth: r.shanghai_qut_auth,
        sz_auth: r.shenzhen_qut_auth,
        hk_option: r.hk_option_auth,
        hk_future: r.hk_future_auth,
        us_option: r.us_option_flag,
        us_future_cme_cboe: r.us_future_cme_cboe_auth,
        us_future_detail,
        sg_future: clear_missing_sg_future,
        jp_future: clear_missing_jp_future,
        sg_stock: clear_missing_sg_stock,
        my_stock: clear_missing_my_stock,
        jp_stock: clear_missing_jp_stock,
        digital_currency_auth: clear_missing_crypto,
        digital_pt_orderbook_auth: clear_missing_crypto_pt_orderbook,
        sub_limit,
        kl_limit,
        extras: qot_right_backend_extras_from_rsp(r),
    }
}

fn apply_qot_right_backend_response(
    qot_right_cache: &QotRightCache,
    rsp: &futu_backend::proto_internal::qta_auth::QtaAuth6024Rsp,
) {
    qot_right_cache.update_from_backend(qot_right_backend_update_from_rsp(rsp));
    qot_right_cache
        .set_orderbook_depths(rsp.hk_option_orderbook_depth, rsp.hk_future_orderbook_depth);
}

fn collect_decoded_fields(
    r: &futu_backend::proto_internal::qta_auth::QtaAuth6024Rsp,
) -> Vec<&'static str> {
    let mut v: Vec<&'static str> = Vec::new();
    if r.hk_qut_got_auth.is_some() {
        v.push("hk_qut_got_auth");
    }
    if r.us_qut_got_auth.is_some() {
        v.push("us_qut_got_auth");
    }
    if r.cn_qut_got_auth.is_some() {
        v.push("cn_qut_got_auth");
    }
    if r.shanghai_qut_auth.is_some() {
        v.push("shanghai_qut_auth");
    }
    if r.shenzhen_qut_auth.is_some() {
        v.push("shenzhen_qut_auth");
    }
    if r.hk_option_auth.is_some() {
        v.push("hk_option_auth");
    }
    if r.hk_future_auth.is_some() {
        v.push("hk_future_auth");
    }
    if r.hk_option_orderbook_depth.is_some() {
        v.push("hk_option_orderbook_depth");
    }
    if r.hk_future_orderbook_depth.is_some() {
        v.push("hk_future_orderbook_depth");
    }
    if r.us_option_flag.is_some() {
        v.push("us_option_flag");
    }
    if r.us_future_cme_cboe_auth.is_some() {
        v.push("us_future_cme_cboe_auth");
    }
    if r.us_future_auth.is_some() {
        v.push("us_future_auth");
    }
    if r.us_lv2_arca_flag.is_some() {
        v.push("us_lv2_arca_flag");
    }
    if r.us_lv2_nyse_flag.is_some() {
        v.push("us_lv2_nyse_flag");
    }
    if r.us_lv2_nasdaq_tv_flag.is_some() {
        v.push("us_lv2_nasdaq_tv_flag");
    }
    if r.us_lv2_edg_flag.is_some() {
        v.push("us_lv2_edg_flag");
    }
    if r.us_lv2_bzx_flag.is_some() {
        v.push("us_lv2_bzx_flag");
    }
    if r.us_index_dow_jones_flag.is_some() {
        v.push("us_index_dow_jones_flag");
    }
    if r.us_index_nasdaq_flag.is_some() {
        v.push("us_index_nasdaq_flag");
    }
    if r.us_index_standard_poor_flag.is_some() {
        v.push("us_index_standard_poor_flag");
    }
    if r.us_otc_deal_data_auth.is_some() {
        v.push("us_otc_deal_data_auth");
    }
    if r.us_otc_order_book_auth.is_some() {
        v.push("us_otc_order_book_auth");
    }
    if r.sgx_future_auth.is_some() {
        v.push("sgx_future_auth");
    }
    if r.sgx_stock_auth.is_some() {
        v.push("sgx_stock_auth");
    }
    if r.jp_future_auth.is_some() {
        v.push("jp_future_auth");
    }
    if r.jp_stock_tse_auth.is_some() {
        v.push("jp_stock_tse_auth");
    }
    if r.myx_auth.is_some() {
        v.push("myx_auth");
    }
    if r.digital_currency_auth.is_some() {
        v.push("digital_currency_auth");
    }
    if r.digital_pt_orderbook_auth.is_some() {
        v.push("digital_pt_orderbook_auth");
    }
    if r.open_api_auth.is_some() {
        v.push("open_api_auth");
    }
    v
}

impl GatewayBridge {
    /// 请求行情权限 (CMD 6024)
    /// 对齐 C++ RequestAPIQuota / RequestMkrQotRightWithPushedQotRight
    ///
    /// **v1.4.106 codex 1110 F7 [P2]**：升级为 `pub` 以供 RemoteCmd / TestCmd
    /// `request_highest_quote_right` 调用真实 backend 路径（不再 silent log no-op）。
    pub async fn request_qot_right(
        backend: &futu_backend::conn::BackendConn,
        qot_right_cache: &QotRightCache,
    ) {
        tracing::info!("CMD6024: requesting qot right...");

        let body = encode_qot_right_request(None);

        // C++ CMD ID = 6032 (NN_ProtoCmd_Qot_Request_Right)
        // proto 文件名虽然叫 FTCmdQtaAuth6024，但实际 CMD ID 是 6032
        match backend.request(6032, body).await {
            Ok(frame) => {
                let body_bytes = frame.body.as_ref();

                // CMD 6032 响应是裸 protobuf (已验证不需要 SRPC 解包)
                // v1.4.106 codex 1110 F3 [P2]: decode_srpc_or_direct 返 Result。
                // 此处是 background self-heal task，decode 失败时 warn + skip update（保留旧 cache 状态比 reset 到 default 更安全）。
                let rsp = match decode_qot_right_response(body_bytes) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::warn!(
                            error = %e,
                            body_len = body_bytes.len(),
                            "CMD6024 SRPC decode failed; preserving existing qot_right cache"
                        );
                        return;
                    }
                };

                let open_api_limits = rsp
                    .open_api_auth
                    .as_ref()
                    .map(|api_auth| (api_auth.sub_limit, api_auth.history_k_line_limit))
                    .unwrap_or((None, None));
                apply_qot_right_backend_response(qot_right_cache, &rsp);

                tracing::info!(
                    hk_got = rsp.hk_qut_got_auth.unwrap_or(0),
                    us_got = rsp.us_qut_got_auth.unwrap_or(0),
                    cn_got = rsp.cn_qut_got_auth.unwrap_or(0),
                    raw_us_lv2_arca = ?rsp.us_lv2_arca_flag,
                    raw_us_lv2_nyse = ?rsp.us_lv2_nyse_flag,
                    raw_us_lv2_nasdaq_tv = ?rsp.us_lv2_nasdaq_tv_flag,
                    raw_us_lv2_edg = ?rsp.us_lv2_edg_flag,
                    raw_us_lv2_bzx = ?rsp.us_lv2_bzx_flag,
                    raw_us_index_dow_jones = ?rsp.us_index_dow_jones_flag,
                    raw_us_index_nasdaq = ?rsp.us_index_nasdaq_flag,
                    raw_us_index_standard_poor = ?rsp.us_index_standard_poor_flag,
                    raw_us_otc_deal_data = ?rsp.us_otc_deal_data_auth,
                    raw_us_otc_order_book = ?rsp.us_otc_order_book_auth,
                    raw_us_future_cme_cboe = ?rsp.us_future_cme_cboe_auth,
                    raw_sgx_stock_auth = ?rsp.sgx_stock_auth,
                    raw_myx_auth = ?rsp.myx_auth.as_ref().and_then(|m| m.myx_auth),
                    raw_jp_stock_tse_auth = ?rsp.jp_stock_tse_auth,
                    raw_digital_currency_auth = ?rsp.digital_currency_auth,
                    raw_digital_pt_orderbook_auth = ?rsp.digital_pt_orderbook_auth,
                    hk_future = rsp.hk_future_auth.unwrap_or(0),
                    hk_option = rsp.hk_option_auth.unwrap_or(0),
                    sub_limit = open_api_limits.0.unwrap_or(0),
                    kl_limit = open_api_limits.1.unwrap_or(0),
                    "CMD6024 qot right updated"
                );
            }
            Err(e) => {
                tracing::warn!(error = %e, "CMD6024 request failed (using defaults)");
            }
        }
    }

    pub async fn request_qot_right_v2(
        backend: &futu_backend::conn::BackendConn,
        qot_right_cache: &QotRightCache,
        mode: QotRightRequestMode,
    ) -> Result<QotRightRefreshReport, QotRightRefreshErr> {
        qot_right_cache.mark_pending();

        let quote_change_notify = qot_right_quote_change_notify(qot_right_cache, mode);

        let mode_str = match mode {
            QotRightRequestMode::HighestPriority => "HighestPriority",
            QotRightRequestMode::WithPushedNotify => "WithPushedNotify",
        };
        tracing::info!(
            mode = mode_str,
            has_pushed_notify = quote_change_notify.is_some(),
            "CMD6032: requesting qot right (v2)..."
        );

        let body = encode_qot_right_request(quote_change_notify);

        let frame = match backend.request(6032, body).await {
            Ok(f) => f,
            Err(e) => {
                let err_msg = format!("backend request failed: {e}");
                qot_right_cache.mark_failed(err_msg.clone());
                return Err(QotRightRefreshErr {
                    stage: "request",
                    detail: err_msg,
                });
            }
        };

        let body_bytes = frame.body.as_ref();

        let rsp = match decode_qot_right_response(body_bytes) {
            Ok(r) => r,
            Err(err_msg) => {
                qot_right_cache.mark_failed(err_msg.clone());
                return Err(QotRightRefreshErr {
                    stage: "decode",
                    detail: err_msg,
                });
            }
        };

        let raw_hk_future_auth = rsp.hk_future_auth;
        let raw_hk_option_auth = rsp.hk_option_auth;
        let raw_hk_future_orderbook_depth = rsp.hk_future_orderbook_depth;
        let raw_hk_option_orderbook_depth = rsp.hk_option_orderbook_depth;
        let raw_us_lv2_arca = rsp.us_lv2_arca_flag;
        let raw_us_lv2_nyse = rsp.us_lv2_nyse_flag;
        let raw_us_lv2_nasdaq_tv = rsp.us_lv2_nasdaq_tv_flag;
        let raw_us_lv2_edg = rsp.us_lv2_edg_flag;
        let raw_us_lv2_bzx = rsp.us_lv2_bzx_flag;
        let raw_us_index_dow_jones = rsp.us_index_dow_jones_flag;
        let raw_us_index_nasdaq = rsp.us_index_nasdaq_flag;
        let raw_us_index_standard_poor = rsp.us_index_standard_poor_flag;
        let raw_us_otc_deal_data = rsp.us_otc_deal_data_auth;
        let raw_us_otc_order_book = rsp.us_otc_order_book_auth;
        let raw_us_future_cme_cboe = rsp.us_future_cme_cboe_auth;
        let raw_sgx_stock_auth = rsp.sgx_stock_auth;
        let raw_myx_auth = rsp.myx_auth.as_ref().and_then(|m| m.myx_auth);
        let raw_jp_stock_tse_auth = rsp.jp_stock_tse_auth;
        let raw_digital_currency_auth = rsp.digital_currency_auth;
        let raw_digital_pt_orderbook_auth = rsp.digital_pt_orderbook_auth;
        apply_qot_right_backend_response(qot_right_cache, &rsp);

        let decoded_fields = collect_decoded_fields(&rsp);
        let meta = qot_right_cache.meta_snapshot();
        let data = qot_right_cache.get();
        tracing::info!(
            mode = mode_str,
            decoded_fields = ?decoded_fields,
            raw_hk_future_auth = ?raw_hk_future_auth,
            raw_hk_option_auth = ?raw_hk_option_auth,
            raw_hk_future_orderbook_depth = ?raw_hk_future_orderbook_depth,
            raw_hk_option_orderbook_depth = ?raw_hk_option_orderbook_depth,
            raw_us_lv2_arca = ?raw_us_lv2_arca,
            raw_us_lv2_nyse = ?raw_us_lv2_nyse,
            raw_us_lv2_nasdaq_tv = ?raw_us_lv2_nasdaq_tv,
            raw_us_lv2_edg = ?raw_us_lv2_edg,
            raw_us_lv2_bzx = ?raw_us_lv2_bzx,
            raw_us_index_dow_jones = ?raw_us_index_dow_jones,
            raw_us_index_nasdaq = ?raw_us_index_nasdaq,
            raw_us_index_standard_poor = ?raw_us_index_standard_poor,
            raw_us_otc_deal_data = ?raw_us_otc_deal_data,
            raw_us_otc_order_book = ?raw_us_otc_order_book,
            raw_us_future_cme_cboe = ?raw_us_future_cme_cboe,
            raw_sgx_stock_auth = ?raw_sgx_stock_auth,
            raw_myx_auth = ?raw_myx_auth,
            raw_jp_stock_tse_auth = ?raw_jp_stock_tse_auth,
            raw_digital_currency_auth = ?raw_digital_currency_auth,
            raw_digital_pt_orderbook_auth = ?raw_digital_pt_orderbook_auth,
            hk_future_qot_right = data.hk_future_qot_right,
            hk_option_qot_right = data.hk_option_qot_right,
            us_qot_right = data.us_qot_right,
            us_index_qot_right = data.us_index_qot_right,
            us_otc_qot_right = data.us_otc_qot_right,
            us_cboe_future_qot_right = data.us_cboe_future_qot_right,
            sg_stock_qot_right = data.sg_stock_qot_right,
            my_stock_qot_right = data.my_stock_qot_right,
            jp_stock_qot_right = data.jp_stock_qot_right,
            cc_qot_right = data.cc_qot_right,
            cc_pt_orderbook_qot_right = data.cc_pt_orderbook_qot_right,
            hk_future_orderbook_depth = ?data.hk_future_orderbook_depth,
            hk_option_orderbook_depth = ?data.hk_option_orderbook_depth,
            freshness = ?meta.freshness,
            "CMD6032: qot right updated (API-facing cache values)"
        );

        Ok(QotRightRefreshReport {
            changed: true,
            fresh_at_ms: meta.last_refresh_at_ms,
            decoded_fields,
            freshness_after: meta.freshness,
            login_epoch: meta.login_epoch,
        })
    }
}