futu_rest/routes/

sys.rs

//! 系统 REST API 路由

use std::sync::Arc;

use axum::Extension;
use axum::extract::{Json, Query, State};
use axum::http::StatusCode;
use bytes::Bytes;
use futu_codec::header::ProtoFmtType;
use futu_server::conn::IncomingRequest;
use prost::Message;
use serde::Deserialize;
use serde_json::Value;

use futu_core::proto_id;
use futu_proto::get_delay_statistics;
use futu_proto::get_global_state;
use futu_proto::get_user_info;
use futu_proto::test_cmd;
use futu_surface_spec::endpoints::get_delay_statistics::default_request_body_json;
// v1.4.98 T2-8 (mobile-source-audit Phase 2): NN+MM token 状态查询
use futu_backend::proto_internal::futu_token_state;
use futu_qot::quote_rights::SYS_QUERY_GET_QUOTE_RIGHTS_PROFILE;

use crate::adapter::{self, RestState};

type ApiResult = Result<Json<Value>, (StatusCode, Json<Value>)>;
type RawApiResult = Result<adapter::RawJson, (StatusCode, Json<Value>)>;

/// GET /api/global-state — 获取全局状态
///
/// v1.4.110 Layer 2: spec validation 由 proto_request_internal 自动注入
/// (proto_id-based lookup), 此 route 走老 proto_request 调用即可.
pub async fn get_global_state(State(state): State<RestState>) -> RawApiResult {
    adapter::proto_request_raw::<get_global_state::Request, get_global_state::Response>(
        &state,
        proto_id::GET_GLOBAL_STATE,
        None,
    )
    .await
}

/// GET /api/user-info — 获取用户信息
pub async fn get_user_info(State(state): State<RestState>) -> RawApiResult {
    adapter::proto_request_raw::<get_user_info::Request, get_user_info::Response>(
        &state,
        proto_id::GET_USER_INFO,
        None,
    )
    .await
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct QuoteRightsQuery {
    refresh: Option<bool>,
}

/// GET /api/quote-rights — C++ OpenD GUI 风格行情权限概览
pub async fn get_quote_rights(
    State(state): State<RestState>,
    Query(query): Query<QuoteRightsQuery>,
) -> ApiResult {
    if query.refresh.unwrap_or(false) {
        let req = test_cmd::Request {
            c2s: test_cmd::C2s {
                cmd: "request_highest_quote_right".to_string(),
                param_str: None,
                param_bytes: None,
            },
        };
        let resp: test_cmd::Response = dispatch_proto(
            &state,
            proto_id::TEST_CMD,
            req,
            "request_highest_quote_right",
        )
        .await?;
        if resp.ret_type != 0 {
            return Err(api_error(
                StatusCode::BAD_GATEWAY,
                format_sys_command_error_message(
                    "request_highest_quote_right",
                    resp.ret_type,
                    resp.ret_msg.as_deref(),
                ),
            ));
        }
    }

    let req = test_cmd::Request {
        c2s: test_cmd::C2s {
            cmd: SYS_QUERY_GET_QUOTE_RIGHTS_PROFILE.to_string(),
            param_str: None,
            param_bytes: None,
        },
    };
    let resp: test_cmd::Response = dispatch_proto(
        &state,
        proto_id::TEST_CMD,
        req,
        SYS_QUERY_GET_QUOTE_RIGHTS_PROFILE,
    )
    .await?;
    if resp.ret_type != 0 {
        return Err(api_error(
            StatusCode::BAD_GATEWAY,
            format_sys_command_error_message(
                SYS_QUERY_GET_QUOTE_RIGHTS_PROFILE,
                resp.ret_type,
                resp.ret_msg.as_deref(),
            ),
        ));
    }
    let json = resp.s2c.and_then(|s| s.result_str).ok_or_else(|| {
        api_error(
            StatusCode::BAD_GATEWAY,
            format!("{SYS_QUERY_GET_QUOTE_RIGHTS_PROFILE}: missing result_str"),
        )
    })?;
    serde_json::from_str::<Value>(&json).map(Json).map_err(|e| {
        api_error(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("parse {SYS_QUERY_GET_QUOTE_RIGHTS_PROFILE}: {e}"),
        )
    })
}

fn format_sys_command_error_message(label: &str, ret_type: i32, ret_msg: Option<&str>) -> String {
    let ret_msg = ret_msg
        .filter(|msg| !msg.is_empty())
        .unwrap_or("<missing ret_msg>");
    format!("{label} ret_type={ret_type} msg={ret_msg}")
}

async fn dispatch_proto<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    req: Req,
    label: &str,
) -> Result<Rsp, (StatusCode, Json<Value>)>
where
    Req: Message,
    Rsp: Message + Default,
{
    let incoming = IncomingRequest::builder(
        state.next_conn_id(),
        proto_id,
        state.next_serial(),
        ProtoFmtType::Protobuf,
        Bytes::from(req.encode_to_vec()),
    )
    .build();
    let resp_bytes = state
        .router
        .dispatch(incoming.conn_id, &incoming)
        .await
        .ok_or_else(|| {
            api_error(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("{label}: handler returned no response"),
            )
        })?;
    Rsp::decode(Bytes::from(resp_bytes)).map_err(|e| {
        api_error(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("decode {label}: {e}"),
        )
    })
}

fn api_error(status: StatusCode, message: String) -> (StatusCode, Json<Value>) {
    (
        status,
        Json(serde_json::json!({
            "ret_type": -1,
            "ret_msg": message,
        })),
    )
}

/// GET /api/delay-statistics — 获取延迟统计（无 body, 使用 backend 默认过滤）
pub async fn get_delay_statistics(State(state): State<RestState>) -> RawApiResult {
    let body = default_request_body_json();
    adapter::proto_request_raw::<get_delay_statistics::Request, get_delay_statistics::Response>(
        &state,
        proto_id::GET_DELAY_STATISTICS,
        Some(body),
    )
    .await
}

/// v1.4.83 §6 Phase 1.4: POST /api/delay-statistics — 带 body 过滤
///
/// 双 tester v1.4.81 §6 报告 `{"type_list":[1,2,3]}` POST 返 None (原 route
/// 只注册了 GET). 本版加 POST 支持 type_list / qot_push_stage / segment_list
/// 过滤（proto `GetDelayStatistics.C2S` 字段齐全）.
pub async fn get_delay_statistics_post(
    State(state): State<RestState>,
    Json(body): Json<Value>,
) -> RawApiResult {
    adapter::proto_request_raw::<get_delay_statistics::Request, get_delay_statistics::Response>(
        &state,
        proto_id::GET_DELAY_STATISTICS,
        Some(body),
    )
    .await
}

/// v1.4.74 A2 BUG-013 fix: GET /api/ping — Futu-specific health check
///
/// 对齐 MCP `futu_ping`。返回 `{ok: bool, gateway: string, version: string}`。
/// 不同于 `/health`（进程 alive 就 200），本 endpoint 检查 gateway dispatch
/// 层是否 ready（能接受新请求）。
///
/// 对齐 Python SDK 层 `/api/ping` 风格。
pub async fn ping(State(state): State<RestState>) -> ApiResult {
    // 简单 routing ping：router.dispatch 返回 None = 不 OK，Some = OK
    // 用 GET_GLOBAL_STATE 作 canary（最轻量的 proto）
    let ok = adapter::proto_request::<get_global_state::Request, get_global_state::Response>(
        &state,
        proto_id::GET_GLOBAL_STATE,
        None,
    )
    .await
    .is_ok();

    Ok(Json(serde_json::json!({
        "ok": ok,
        "version": env!("CARGO_PKG_VERSION"),
        "gateway": "futu-opend-rs",
    })))
}

/// v1.4.74 A2 BUG-013 fix: GET|POST /api/push-subscriber-info — push 订阅者列表
///
/// **v1.4.83 §9 Phase 2 F5 实装**（双 tester v1.4.81 §9 CMD3020 chain recovery
/// 核心）：
///
/// - **Push stream 真实健康状态** (`push_stream_healthy`)：基于
///   last_push_received_at + consecutive_parse_errors + circuit breaker 综合判定
/// - **Last push received** (`last_push_received_at_ms`)：Unix ms，0=启动后未收过
/// - **Consecutive parse errors**: F3/F4 触发阈值 (5/20)
/// - **Circuit breaker state** (`is_circuit_tripped_now` + trips count)
/// - **Orphan orders detected** (F6): 卡住订单计数
/// - **Re-subscribe count** (F3)
/// - **Request backend liveness** (`backend_connected`): Platform request path
///   当前是否仍持有可用 TCP connection。push stream healthy 不代表普通 QOT
///   request 一定可发；这个字段用于区分 push path 与 request path 健康分裂。
///
/// **provider 未注入时**（某些 test / embedded 场景没 GatewayBridge）返回 503。
/// 生产 daemon 启动路径一定注入 provider；缺失时不能伪装成 `ret_type=0`。
///
/// **MCP-centric 备注**：原 `futu_push_subscriber_info` MCP tool 查 MCP
/// session 级 rmcp Peer 注册表（v1.4.58 Phase A）。REST 侧无 session 概念,
/// 本 endpoint 返的是 daemon 级 push 通道健康 (tester 的实际诉求).
pub async fn push_subscriber_info(
    State(state): State<RestState>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
    // v1.4.83 §9 F5: 优先返真实 health snapshot
    if let Some(ref provider) = state.push_health_snapshot_provider {
        let health = provider();
        return Ok(Json(serde_json::json!({
            "ret_type": 0,
            "ret_msg": "success",
            "push_health": health,
            "recommendations": [
                {
                    "purpose": "查订阅列表 + 全局 quota",
                    "endpoint": "POST /api/query-subscription -d '{}'",
                    "note": "v1.4.83 起默认 all-conn 视图"
                },
                {
                    "purpose": "接收 push 数据（quote / tick / orderbook 等）",
                    "endpoint": "WebSocket /ws (支持 Bearer Token 握手)"
                }
            ],
        })));
    }
    // provider 未注入: loud fail. 生产 daemon 应由 startup/phase4.rs 注入
    // `push_health_snapshot_provider`; 缺失说明 wiring 有问题，不能以 ret_type=0
    // 伪成功，否则自动化脚本会误判已拿到真实健康状态。
    Err((
        StatusCode::SERVICE_UNAVAILABLE,
        Json(serde_json::json!({
            "ret_type": -1,
            "ret_msg": "push health snapshot provider not wired (internal setup bug)",
            "recommendations": [
                {
                    "purpose": "查订阅列表 + 全局 quota",
                    "endpoint": "POST /api/query-subscription -d '{}'"
                },
                {
                    "purpose": "接收 push 数据",
                    "endpoint": "WebSocket /ws"
                }
            ],
        })),
    ))
}

/// v1.4.74 A2 BUG-013 + v1.4.102 codex 44 F1 / 46 F5 (P1/P2) fix:
/// POST /api/unsub-acc-push — 真撤账户 push 订阅 + 同 sub-acc-push 严格 validation.
///
/// **历史**: v1.4.74 这条路由直接 forward 到 `TRD_SUB_ACC_PUSH`, 但 backend
/// proto 没有 `is_sub` 字段, daemon `SubAccPushHandler` 一律调
/// `subscribe_trd_acc` → 实际**重新订阅** (silent regression).
///
/// **v1.4.102 修法**:
/// - codex 44 F1: 改路由到 daemon-internal `TRD_UNSUB_ACC_PUSH_INTERNAL`,
///   gateway 加 dedicated `UnsubAccPushHandler` 调 `unsubscribe_trd_acc`.
/// - codex 46 F5: 加 `extract_acc_id_list` + `validate_sub_acc_push_acc_ids` +
///   per-acc allowed_acc_ids 限额 check (与 sub-acc-push 对称, 防 silent
///   no-op `{}` 接 ret_type=0 的反模式 D).
///
/// body proto 仍 reuse `Trd_SubAccPush.Request` (`acc_id_list` 字段).
pub async fn unsub_acc_push(
    State(state): State<RestState>,
    rec: Option<Extension<Arc<futu_auth::KeyRecord>>>,
    Json(mut body): Json<Value>,
) -> ApiResult {
    // v1.4.103 (codex 56 F1 / 58 F5 — B9): legacy 模式 (无 keys.json / 无 Bearer)
    // 直接 reject. 与 /api/sub-acc-push 对称 — legacy 模式没 sub state 可 unsub,
    // 之前返 ret_type=0 silent 等于客户端误以为撤了实际没撤.
    if rec.is_none() {
        futu_auth::audit::reject(
            "rest",
            "/api/unsub-acc-push",
            "<legacy>",
            "unsub-acc-push not supported in legacy mode (no keys.json)",
        );
        return Err((
            axum::http::StatusCode::FORBIDDEN,
            Json(serde_json::json!({
                "error": "/api/unsub-acc-push: legacy mode (no keys.json) does not support per-key sub state. \
                          Configure keys.json and pass Bearer token to enable.",
                "ret_type": -1,
                "hint": "v1.4.103 B9: legacy mode previously returned silent success without revoking. Now loud-reject to surface the limitation."
            })),
        ));
    }
    // codex 43 F1 + 44 F2 (normalize-first + strict 兼容): 同 sub-acc-push.
    crate::adapter::normalize_json_keys_snake_case(&mut body);

    // codex 46 F5 (P2): 验 acc_id_list 非空 (与 sub-acc-push 对称).
    let acc_ids = match crate::routes::trd::extract_acc_id_list(&body) {
        Ok(acc_ids) => acc_ids,
        Err(reason) => {
            let key_id = rec
                .as_deref()
                .map(|r| r.as_ref().id.clone())
                .unwrap_or_else(|| "<legacy>".to_string());
            futu_auth::audit::reject("rest", "/api/unsub-acc-push", &key_id, &reason);
            return Err((
                axum::http::StatusCode::BAD_REQUEST,
                Json(serde_json::json!({
                    "error": format!("/api/unsub-acc-push: {reason}")
                })),
            ));
        }
    };
    if let Err(reason) = crate::routes::trd::validate_sub_acc_push_acc_ids(&acc_ids) {
        let key_id = rec
            .as_deref()
            .map(|r| r.as_ref().id.clone())
            .unwrap_or_else(|| "<legacy>".to_string());
        futu_auth::audit::reject("rest", "/api/unsub-acc-push", &key_id, reason);
        return Err((
            axum::http::StatusCode::BAD_REQUEST,
            Json(serde_json::json!({
                "error": format!("/api/unsub-acc-push: {reason}")
            })),
        ));
    }

    // codex 46 F4 (P1): per-acc allowed_acc_ids 限额 check (与 sub-acc-push 对称).
    //
    // codex 0522 F2 v1.4.106: 走共享 helper `check_per_acc_rate_for_caller`
    // (定义在 routes/trd.rs), 与 `/api/sub-acc-push` 同源.
    crate::routes::trd::check_per_acc_rate_for_caller(
        &state.counters,
        rec.as_deref().map(|r| r.as_ref()),
        &acc_ids,
        "/api/unsub-acc-push",
    )?;

    // v1.4.102 codex 51 F2 (P2): REST unsub 也跳 dispatch — 直接改 state map.
    // 之前 dispatch 到 UnsubAccPushHandler 调 unsubscribe_trd_acc(conn_id, ...),
    // 但 REST conn_id 是临时的, 不是当年 sub 时的; 删不到. REST state map
    // 才是 REST 层真相源.
    let daemon_resp: Json<serde_json::Value> = Json(serde_json::json!({
        "ret_type": 0,
        "ret_msg": serde_json::Value::Null,
        "err_code": serde_json::Value::Null,
        "s2c": {}
    }));

    // v1.4.102 codex 46 F2/F3 + 48 F2 (P1): 删除 REST sub state map 里的 entries.
    // **codex 48 F2 (P1) tombstone fix**: 之前 set 空时 subs.remove(&key_id),
    // 但 WS delivery 把 missing key 当 "未 sub-acc-push, 全开 push" backward-compat
    // → unsub last acc 反而**重启全 push**. 现在: 保留**空 HashSet** 作 tombstone,
    // WS filter Layer 2 看到 entry 存在但空 → 全拒.
    if let Some(rec_ref) = rec.as_deref() {
        let key_id = rec_ref.as_ref().id.clone();
        crate::adapter::with_rest_acc_subscriptions_write(&state.rest_acc_subscriptions, |subs| {
            // codex 48 F2: 即使 key 之前没 sub-acc-push 过, unsub 也要建 tombstone
            // (空 HashSet) 让 WS filter 拒绝全 push (用户主动 opt-in 拒接).
            let entry = subs.entry(key_id).or_default();
            for &acc_id in &acc_ids {
                entry.remove(&acc_id);
            }
            // 不删 key, 留空 set 作 tombstone (= "已显式 unsub all")
        });
    }

    Ok(daemon_resp)
}

/// v1.4.98 T2-8 (mobile-source-audit Phase 2): NN+MM token 状态查询.
///
/// **POST /api/token-state** (无 body / 可选 `{"c2s":{"app_id":"nn"|"mm"|"all"}}`).
/// **GET /api/token-state?app_id=nn|mm|all** — query param 同样可指定.
///
/// 返 NN (Futu Token app) + MM (moomoo Token app) 两边 token 启用 + 绑定 4 字段
/// (1=已绑定/已启用, 0=未绑定/未启用).
///
/// **Use case**: pitfall #15 "moomoo token = 富途令牌的海外版本" 实证后, 用户
/// 调 `/api/unlock-trade` 失败 -20011 时, 第一线诊断: `curl /api/token-state` 看
/// 双系绑定情况, 决定 TOTP secret 该来自哪个 app.
///
/// **codex 2026-04-27 audit fix**: 之前 docstring 声称支持 `?app_id=nn` 但
/// handler 只接 body 不读 query, 真机调 GET ?app_id=nn → app_id="all" silent
/// 错. 加 Query extractor map → c2s.app_id 真传到 backend.
///
/// **真机 verify**: T2-8 自测 PASS (NN/MM 4 字段返).
///
/// **v1.4.99 codex F5 fix (P2, 2026-04-27)**: `deny_unknown_fields` —
/// 之前 typo `?app_idd=nn` silent 默认到 `all` (typo 字段被忽略, 走 daemon
/// default). strict-fields middleware 只对 POST body 跑, GET query 不被
/// 验证. 加 `deny_unknown_fields` 让 typo 立即返 400, 与 POST 表面对齐.
/// (per pitfall #45 silent-success anti-pattern, 子模式: silent default).
#[derive(Debug, Deserialize, Default)]
#[serde(deny_unknown_fields)]
pub struct TokenStateQuery {
    /// app_id filter: "nn" | "mm" | "all" (default "all")
    pub app_id: Option<String>,
}

pub async fn get_token_state(
    State(state): State<RestState>,
    Query(q): Query<TokenStateQuery>,
    body: Option<Json<Value>>,
) -> RawApiResult {
    // 优先级: body 显式 > query param > daemon default "all"
    let body_val = match body {
        Some(Json(mut v)) => {
            // 若 body 没传 app_id, 用 query param 填充
            if let Some(qs_app) = q.app_id.as_ref()
                && let Some(map) = v.as_object_mut()
            {
                let c2s_has = map
                    .get("c2s")
                    .and_then(|c| c.as_object())
                    .is_some_and(|c| c.contains_key("app_id") || c.contains_key("appId"));
                let top_has = map.contains_key("app_id") || map.contains_key("appId");
                if !c2s_has && !top_has {
                    map.entry("c2s".to_string())
                        .or_insert_with(|| serde_json::json!({}))
                        .as_object_mut()
                        .map(|c| {
                            c.insert(
                                "app_id".to_string(),
                                serde_json::Value::String(qs_app.clone()),
                            )
                        });
                }
            }
            Some(v)
        }
        None => {
            // 无 body — 仅用 query param (or daemon default 'all')
            q.app_id
                .as_ref()
                .map(|app_id| serde_json::json!({"c2s": {"app_id": app_id}}))
        }
    };
    adapter::proto_request_raw::<
        futu_token_state::DaemonGetTokenStateReq,
        futu_token_state::DaemonGetTokenStateRsp,
    >(&state, proto_id::GET_TOKEN_STATE, body_val)
    .await
}

#[cfg(test)]
mod tests;