futu_gateway_core/

handlers_sys.rs

// 系统业务处理器 — GetUserInfo / GetDelayStatistics / Verification / TestCmd / RemoteCmd

use std::sync::Arc;

use async_trait::async_trait;

use futu_cache::user_profile::{UserProfile, UserProfileCache};
use futu_core::error::FutuError;
use futu_core::proto_id;
use futu_server::conn::IncomingRequest;
use futu_server::router::{RequestHandler, RequestRouter};

use crate::bridge::{BrokerRuntime, CacheBundle, GatewayBridge};

mod commands;

use commands::{RemoteCmdHandler, TestCmdHandler};

const DEFAULT_USER_INFO_FLAG: i32 = 0xFFFFF7FFu32 as i32;
const USER_INFO_FIELD_BASIC: i32 = 1;
const USER_INFO_FIELD_API: i32 = 2;
const USER_INFO_FIELD_QOT_RIGHT: i32 = 4;
const USER_INFO_FIELD_DISCLAIMER: i32 = 8;
const USER_INFO_FIELD_UPDATE: i32 = 16;

/// 注册系统处理器
pub fn register_handlers(router: &Arc<RequestRouter>, bridge: &GatewayBridge) {
    register_handlers_inner(router, bridge, None);
}

/// 注册系统处理器，并把 daemon-owned shutdown signal 注入给 TestCmd。
///
/// C++ `APIServer_TestCmd.cpp` 的 `exit` 先回成功包，随后发
/// `API_OMEvent_Command / GTW_Command_Exit` 给网关主事件循环。Rust 生产路径
/// 等价接入 `futu-opend` phase4 的统一 shutdown channel，避免业务 handler
/// 直接 `process::exit` 绕过 surface graceful shutdown。
pub fn register_handlers_with_shutdown(
    router: &Arc<RequestRouter>,
    bridge: &GatewayBridge,
    shutdown_tx: tokio::sync::watch::Sender<bool>,
) {
    register_handlers_inner(router, bridge, Some(shutdown_tx));
}

fn register_handlers_inner(
    router: &Arc<RequestRouter>,
    bridge: &GatewayBridge,
    shutdown_tx: Option<tokio::sync::watch::Sender<bool>>,
) {
    let caches = bridge.caches();
    router.register(
        proto_id::GET_USER_INFO,
        Arc::new(GetUserInfoHandler {
            caches: caches.clone(),
            broker_runtime: bridge.broker_runtime().clone(),
        }),
    );
    router.register(
        proto_id::GET_DELAY_STATISTICS,
        Arc::new(GetDelayStatisticsHandler),
    );
    router.register(proto_id::VERIFICATION, Arc::new(VerificationHandler));
    // v1.4.106 codex 1110 F7 [P2]: TestCmd / RemoteCmd 拿到 backend + qot_right_cache
    // 来真发 CMD6024 (request_qot_right) — silent-success 反模式修复。
    router.register(
        proto_id::TEST_CMD,
        Arc::new(TestCmdHandler {
            caches: caches.clone(),
            broker_runtime: bridge.broker_runtime().clone(),
            shutdown_tx,
        }),
    );
    router.register(
        proto_id::REMOTE_CMD,
        Arc::new(RemoteCmdHandler {
            caches: caches.clone(),
            broker_runtime: bridge.broker_runtime().clone(),
        }),
    );

    // v1.4.98 T2-8 (mobile-source-audit Phase 2): NN+MM token 状态查询 (cmd 1326).
    router.register(
        proto_id::GET_TOKEN_STATE,
        Arc::new(GetTokenStateHandler {
            caches: caches.clone(),
            broker_runtime: bridge.broker_runtime().clone(),
        }),
    );

    tracing::debug!("system handlers registered");
}

// ===== GetUserInfo (1005) =====
// C++ 从多个后端聚合查询（Profile、QotRight、Disclaimer、Update）。
// Rust: user_id 从登录缓存获取，行情权限从 CMD 6024 响应缓存获取。
struct GetUserInfoHandler {
    caches: CacheBundle,
    broker_runtime: BrokerRuntime,
}

#[async_trait]
impl RequestHandler for GetUserInfoHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::get_user_info::Request =
            match crate::handlers_shared::decode_request_or_error(
                request.body.as_ref(),
                "sys",
                "GetUserInfo",
            ) {
                Ok(req) => req,
                Err(resp) => return Some(resp),
            };
        let flag = req.c2s.flag.unwrap_or(DEFAULT_USER_INFO_FLAG);

        let include_basic = has_user_info_field(flag, USER_INFO_FIELD_BASIC);
        let include_api = has_user_info_field(flag, USER_INFO_FIELD_API);
        let include_qot_right = has_user_info_field(flag, USER_INFO_FIELD_QOT_RIGHT);
        let include_disclaimer = has_user_info_field(flag, USER_INFO_FIELD_DISCLAIMER);
        let include_update = has_user_info_field(flag, USER_INFO_FIELD_UPDATE);

        let login_state = self.caches.login_cache.get_login_state();
        let user_id = login_state.as_ref().map(|s| s.user_id as i64).unwrap_or(0);
        let user_id_u64 = u64::try_from(user_id).unwrap_or(0);
        let user_attribution = login_state.as_ref().and_then(|s| s.user_attribution);
        if user_id_u64 != 0
            && include_basic
            && let Err(e) = refresh_user_profile_cache(
                &self.broker_runtime,
                &self.caches.user_profile_cache,
                user_id_u64,
            )
            .await
            && !matches!(e, FutuError::Timeout)
        {
            return Some(crate::handlers_shared::make_error_response(
                -1,
                &format!("GetUserInfo QueryUserProfile failed: {e}"),
            ));
        }
        let profile = self.caches.user_profile_cache.get(user_id_u64);
        let nick_name = profile
            .as_ref()
            .map(|profile| profile.nick_name.clone())
            .unwrap_or_default();
        let avatar_url = profile
            .as_ref()
            .map(|profile| profile.avatar_url.clone())
            .unwrap_or_default();

        // v1.4.106 codex 1217 F4 [P2] (5月1日): 不再 silent 暴露 default quota / Unknown 权限.
        //
        // freshness=Fresh → 返实值; Pending/Unknown/Failed → 全部字段 0/false + ret_msg
        // 提示客户端 "数据未就绪". Stale → 保留旧 data 但 ret_msg 标 stale.
        let qr = self.caches.qot_right_cache.get();
        let freshness = self.caches.qot_right_cache.freshness();
        let meta = self.caches.qot_right_cache.meta_snapshot();
        let needs_qot_cache = include_api || include_qot_right;
        let (use_real, ret_msg_v) = match &freshness {
            futu_cache::qot_right::QotRightFreshness::Fresh => (true, None),
            futu_cache::qot_right::QotRightFreshness::Stale => (
                true,
                needs_qot_cache.then(|| format!(
                    "qot_right cache stale (login_epoch={}, backend_generation={}); refresh in flight",
                    meta.login_epoch, meta.backend_generation,
                )),
            ),
            futu_cache::qot_right::QotRightFreshness::Pending => (
                false,
                needs_qot_cache.then(|| {
                    "qot_right refresh pending (CMD6032 in flight, retry shortly)".to_string()
                }),
            ),
            futu_cache::qot_right::QotRightFreshness::Unknown => (
                false,
                needs_qot_cache.then(|| {
                    "qot_right cache uninitialized; daemon may still be booting".to_string()
                }),
            ),
            futu_cache::qot_right::QotRightFreshness::Failed { error, since_ms } => (
                false,
                needs_qot_cache.then(|| format!(
                    "qot_right refresh failed: {} (since_ms={}); call request_highest_quote_right to retry",
                    error, since_ms,
                )),
            ),
        };
        let pick_r = |v: i32| if use_real { v } else { 0 };
        let pick_b = |v: bool| if use_real { v } else { false };
        let pick_q = |v: i32| if use_real { v } else { 0 };

        let s2c = futu_proto::get_user_info::S2c {
            nick_name: include_basic.then_some(nick_name),
            avatar_url: include_basic.then_some(avatar_url),
            api_level: None,
            hk_qot_right: include_qot_right.then_some(pick_r(qr.hk_qot_right)),
            us_qot_right: include_qot_right.then_some(pick_r(qr.api_us_qot_right)),
            // Ref: FutuOpenD/Src/APIServer/Business/APIServer_GetUserInfo.cpp:425-429.
            // C++ fills shQotRight/szQotRight, not legacy cnQotRight.
            cn_qot_right: None,
            is_need_agree_disclaimer: include_disclaimer.then_some(false),
            user_id: include_basic.then_some(user_id),
            update_type: include_update.then_some(0), // UpdateType_None
            web_key: None,
            web_jump_url_head: None,
            hk_option_qot_right: include_qot_right.then_some(pick_r(qr.hk_option_qot_right)),
            has_us_option_qot_right: include_qot_right
                .then_some(pick_b(qr.has_us_option_qot_right)),
            hk_future_qot_right: include_qot_right.then_some(pick_r(qr.hk_future_qot_right)),
            sub_quota: include_api.then_some(pick_q(qr.sub_quota)),
            history_kl_quota: include_api.then_some(pick_q(qr.history_kl_quota)),
            // Ref: FutuOpenD/Src/APIServer/Business/APIServer_GetUserInfo.cpp:435-436.
            // The aggregate US future field is commented out in C++; newer per-exchange
            // fields are filled below.
            us_future_qot_right: None,
            us_option_qot_right: include_qot_right.then_some(pick_r(qr.us_option_qot_right)),
            user_attribution: (include_basic || user_id_u64 == 0)
                .then_some(user_attribution)
                .flatten(),
            update_whats_new: None,
            us_index_qot_right: include_qot_right.then_some(pick_r(qr.us_index_qot_right)),
            us_otc_qot_right: include_qot_right.then_some(pick_r(qr.us_otc_qot_right)),
            us_cme_future_qot_right: include_qot_right
                .then_some(pick_r(qr.us_cme_future_qot_right)),
            us_cbot_future_qot_right: include_qot_right
                .then_some(pick_r(qr.us_cbot_future_qot_right)),
            us_nymex_future_qot_right: include_qot_right
                .then_some(pick_r(qr.us_nymex_future_qot_right)),
            us_comex_future_qot_right: include_qot_right
                .then_some(pick_r(qr.us_comex_future_qot_right)),
            us_cboe_future_qot_right: include_qot_right
                .then_some(pick_r(qr.us_cboe_future_qot_right)),
            sg_future_qot_right: include_qot_right.then_some(pick_r(qr.sg_future_qot_right)),
            jp_future_qot_right: include_qot_right.then_some(pick_r(qr.jp_future_qot_right)),
            is_app_nn_or_mm: include_basic.then_some(true), // NN 版本
            sh_qot_right: include_qot_right.then_some(pick_r(qr.sh_qot_right)),
            sz_qot_right: include_qot_right.then_some(pick_r(qr.sz_qot_right)),
            extra: None,
            cc_qot_right: include_qot_right.then_some(pick_r(qr.cc_qot_right)),
            sg_stock_qot_right: include_qot_right.then_some(pick_r(qr.sg_stock_qot_right)),
            my_stock_qot_right: include_qot_right.then_some(pick_r(qr.my_stock_qot_right)),
            jp_stock_qot_right: include_qot_right.then_some(pick_r(qr.jp_stock_qot_right)),
        };

        let resp = futu_proto::get_user_info::Response {
            ret_type: 0,
            ret_msg: ret_msg_v,
            err_code: None,
            s2c: Some(s2c),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

fn has_user_info_field(flag: i32, field: i32) -> bool {
    (flag & field) != 0
}

async fn refresh_user_profile_cache(
    broker_runtime: &BrokerRuntime,
    user_profile_cache: &UserProfileCache,
    user_id: u64,
) -> Result<(), FutuError> {
    let Some(backend) = broker_runtime.platform_backend() else {
        return Ok(());
    };

    match futu_backend::user_profile::query_user_profile(&backend, user_id).await {
        Ok(profile) => {
            user_profile_cache.set(UserProfile {
                user_id: profile.user_id,
                nick_name: profile.nick_name,
                avatar_url: profile.avatar_url,
            });
            Ok(())
        }
        Err(e) => {
            tracing::warn!(
                user_id,
                error = %e,
                "CMD7506 QueryUserProfile failed; using cached profile if available"
            );
            Err(e)
        }
    }
}

// ===== GetDelayStatistics (1007) =====
// C++ 从本地 INNData_ProtoDelay 收集延迟统计数据。
// Ref: FutuOpenD/Src/APIServer/Business/APIServer_GetDelayStatistics.cpp:71
// Rust delay_stats 接入 ReqReply + QOT push 聚合计数 + PlaceOrder once-detail。
struct GetDelayStatisticsHandler;

#[async_trait]
impl RequestHandler for GetDelayStatisticsHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::get_delay_statistics::Request =
            match crate::handlers_shared::decode_request_or_error(
                request.body.as_ref(),
                "sys",
                "GetDelayStatistics",
            ) {
                Ok(req) => req,
                Err(resp) => return Some(resp),
            };

        let include_req_reply = req
            .c2s
            .type_list
            .contains(&(futu_proto::get_delay_statistics::DelayStatisticsType::ReqReply as i32));
        let include_qot_push = req
            .c2s
            .type_list
            .contains(&(futu_proto::get_delay_statistics::DelayStatisticsType::QotPush as i32));
        let include_place_order = req
            .c2s
            .type_list
            .contains(&(futu_proto::get_delay_statistics::DelayStatisticsType::PlaceOrder as i32));

        let s2c = futu_proto::get_delay_statistics::S2c {
            qot_push_statistics_list: if include_qot_push {
                futu_core::delay_stats::snapshot_qot_push_statistics(
                    req.c2s.qot_push_stage.unwrap_or(0),
                    &req.c2s.segment_list,
                )
                .into_iter()
                .map(|item| futu_proto::get_delay_statistics::DelayStatistics {
                    qot_push_type: item.qot_push_type,
                    item_list: item
                        .item_list
                        .into_iter()
                        .map(
                            |bucket| futu_proto::get_delay_statistics::DelayStatisticsItem {
                                begin: bucket.begin,
                                end: bucket.end,
                                count: bucket.count,
                                proportion: bucket.proportion,
                                cumulative_ratio: bucket.cumulative_ratio,
                            },
                        )
                        .collect(),
                    delay_avg: item.delay_avg_ms,
                    count: item.count,
                })
                .collect()
            } else {
                vec![]
            },
            req_reply_statistics_list: if include_req_reply {
                futu_core::delay_stats::snapshot_req_reply_statistics()
                    .into_iter()
                    .map(
                        |item| futu_proto::get_delay_statistics::ReqReplyStatisticsItem {
                            proto_id: item.proto_id as i32,
                            count: item.count as i32,
                            total_cost_avg: item.total_cost_avg_ms,
                            open_d_cost_avg: item.open_d_cost_avg_ms,
                            net_delay_avg: item.net_delay_avg_ms,
                            is_local_reply: item.is_local_reply,
                        },
                    )
                    .collect()
            } else {
                vec![]
            },
            place_order_statistics_list: if include_place_order {
                futu_core::delay_stats::snapshot_place_order_statistics()
                    .into_iter()
                    .map(
                        |item| futu_proto::get_delay_statistics::PlaceOrderStatisticsItem {
                            order_id: item.order_id,
                            total_cost: item.total_cost_ms,
                            open_d_cost: item.open_d_cost_ms,
                            net_delay: item.net_delay_ms,
                            update_cost: item.update_cost_ms,
                        },
                    )
                    .collect()
            } else {
                vec![]
            },
        };

        let resp = futu_proto::get_delay_statistics::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(s2c),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== Verification (1006) =====
// C++ 处理图形/短信验证码的请求和验证。
// Ref: FutuOpenD/Src/APIServer/Business/APIServer_Verification.cpp:191
// Rust 尚未接入 INNLoginCenter 验证码状态机；这里必须 fail-loud，避免
// 用户/SDK 误以为验证码请求或输入已生效。
struct VerificationHandler;

#[async_trait]
impl RequestHandler for VerificationHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let _req: futu_proto::verification::Request =
            match crate::handlers_shared::decode_request_or_error(
                request.body.as_ref(),
                "sys",
                "Verification",
            ) {
                Ok(req) => req,
                Err(resp) => return Some(resp),
            };

        let resp = futu_proto::verification::Response {
            ret_type: -1,
            ret_msg: Some(
                "Verification requires the C++ INNLoginCenter captcha/SMS flow; \
                 Rust OpenD does not implement this flow yet"
                    .to_string(),
            ),
            err_code: None,
            s2c: None,
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetTokenState (1326) — v1.4.98 T2-8 (mobile-source-audit Phase 2) =====
//
// 来源: moomoo `Moomoo/Src/FTBase/FTBaseBusiness/Settings/Controller/Token/FuTuToken.cpp:19`
// cmd 1326 `CS_CMDID_NewToken_GetStateInfo` + proto
// `moomoo/.../Proto/futu_token.proto::GetTokenListRequest/Response`.
//
// **Use case**: pitfall #15 实证 "moomoo token = 富途令牌的海外版本", 但 daemon
// 之前没暴露查询 endpoint. 用户 unlock-trade 失败 -20011 时第一线诊断 — 调
// `/api/token-state` 查 NN/MM 双系 token 绑定/启用情况, 决定 TOTP secret 该来
// 自哪个 app.
//
// **三铁律自检** (per pitfall #51 v1.4.94 mobile-driven extension 修订):
// - 来源真实: ✅ 引用 moomoo proto file:line + cmd_id
// - OpenD 兼容: ✅ ADD-only (`/api/token-state` 新 endpoint, 不动现有)
// - 失败 loud: ✅ backend 拒绝时返清晰 ret_type + msg
//
// **Channel**: platform TCP (无 encryption, 4 个 uint32 简单 response).
// **真机 verify**: 标 UNVERIFIED (per pitfall #57) — 保留为
// real-machine-required，跨地区 sim 账户验证有正证据后再升 confidence.
struct GetTokenStateHandler {
    caches: CacheBundle,
    broker_runtime: BrokerRuntime,
}

#[async_trait]
impl RequestHandler for GetTokenStateHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        // Decode daemon-side wire envelope (DaemonGetTokenStateReq)
        let req: futu_backend::proto_internal::futu_token_state::DaemonGetTokenStateReq =
            match prost::Message::decode(request.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(conn_id, error = %e, "GetTokenState: decode req failed");
                    return Some(crate::handlers_shared::make_error_response(
                        -1,
                        "invalid GetTokenState request body",
                    ));
                }
            };

        // uid 从 login_cache 取 (caller 不需要传 — daemon 单租户语义)
        let user_id = match self.caches.login_cache.get_login_state() {
            Some(s) => s.user_id as u64,
            None => {
                return Some(crate::handlers_shared::make_error_response(
                    -1,
                    "GetTokenState: not logged in (login_cache empty). \
                     Run f3clogin via futu-opend before querying token state.",
                ));
            }
        };

        let app_id_str = req.c2s.app_id.unwrap_or_else(|| "all".to_string());
        let app_id = if app_id_str.is_empty() {
            "all".to_string()
        } else {
            app_id_str
        };

        let backend = match self.broker_runtime.platform_backend() {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetTokenState: no backend connection");
                return Some(crate::handlers_shared::make_error_response(
                    -1,
                    "no backend connection (try again after backend reconnect)",
                ));
            }
        };

        // Build moomoo native GetTokenListRequest (uid + app_id bytes)
        let backend_req = futu_backend::proto_internal::futu_token_state::GetTokenListRequest {
            uid: Some(user_id),
            app_id: Some(app_id.as_bytes().to_vec()),
        };
        let body = prost::Message::encode_to_vec(&backend_req);

        tracing::debug!(
            conn_id,
            uid = user_id,
            app_id = %app_id,
            "sending CMD1326 GetTokenListRequest"
        );

        let resp_frame = match backend
            .request(proto_id::GET_TOKEN_STATE as u16, body)
            .await
        {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD1326 backend request failed");
                return Some(crate::handlers_shared::make_error_response(
                    -1,
                    &format!("GetTokenState: backend cmd 1326 failed: {e}"),
                ));
            }
        };

        let backend_rsp: futu_backend::proto_internal::futu_token_state::GetTokenListResponse =
            match crate::handlers_shared::decode_backend_proto(&resp_frame.body) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD1326 decode response failed");
                    return Some(crate::handlers_shared::make_error_response(
                        -1,
                        "GetTokenState: failed to decode backend response",
                    ));
                }
            };

        // v1.4.106 codex 1110 F6 [P2] 修：err_code 缺失不能当成功
        //
        // 老代码 `err_code.unwrap_or(0)` 把 backend response 的 err_code=None 当成功。
        // 实际：backend envelope 异常 / 空 default response decode 出来 err_code 为 None
        // 时，handler 会把 4 个 token 状态字段（nn_token_enable / nn_token_bind /
        // mm_token_enable / mm_token_bind，全 None）原样塞进 ret_type=0 响应。
        // unlock-trade 诊断如果依赖 token-state，会被引导成 "token 未绑定 / 未启用"
        // 假象，而不是 "查询失败" 真相。
        //
        // 严格成功条件：err_code == Some(0)。
        let err_code = backend_rsp.err_code;
        if err_code != Some(0) {
            let err_msg = backend_rsp.err_msg.as_deref().unwrap_or("");
            let detail = match err_code {
                Some(c) => format!("backend cmd 1326 err: {err_msg} (code={c})"),
                None => format!(
                    "backend cmd 1326 response missing err_code (raw err_msg='{err_msg}'); cannot determine token state"
                ),
            };
            tracing::warn!(
                conn_id,
                err_code = ?err_code,
                err_msg,
                "CMD1326 backend response not Some(0): refusing silent success"
            );
            return Some(prost::Message::encode_to_vec(
                &futu_backend::proto_internal::futu_token_state::DaemonGetTokenStateRsp {
                    ret_type: -1,
                    ret_msg: Some(detail),
                    err_code,
                    s2c: None,
                },
            ));
        }

        // 成功响应中四个 token 状态字段全缺 → 视作 partial/unknown，不静默成功
        let all_token_fields_missing = backend_rsp.nn_token_enable.is_none()
            && backend_rsp.nn_token_bind.is_none()
            && backend_rsp.mm_token_enable.is_none()
            && backend_rsp.mm_token_bind.is_none();
        if all_token_fields_missing {
            tracing::warn!(
                conn_id,
                "CMD1326 backend success (err_code=0) but all 4 token state fields missing — refusing silent ret_type=0"
            );
            return Some(prost::Message::encode_to_vec(
                &futu_backend::proto_internal::futu_token_state::DaemonGetTokenStateRsp {
                    ret_type: -1,
                    ret_msg: Some(
                        "backend cmd 1326 success but token state fields all missing; cannot return reliable token-state".to_string(),
                    ),
                    err_code: Some(0),
                    s2c: None,
                },
            ));
        }

        // Build daemon-side wire response (DaemonGetTokenStateRsp)
        let daemon_rsp = futu_backend::proto_internal::futu_token_state::DaemonGetTokenStateRsp {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(
                futu_backend::proto_internal::futu_token_state::daemon_get_token_state_rsp::S2c {
                    nn_token_enable: backend_rsp.nn_token_enable,
                    nn_token_bind: backend_rsp.nn_token_bind,
                    mm_token_enable: backend_rsp.mm_token_enable,
                    mm_token_bind: backend_rsp.mm_token_bind,
                },
            ),
        };
        Some(prost::Message::encode_to_vec(&daemon_rsp))
    }
}