futu_rest/routes/qot/

mod.rs

//! 行情 REST API 路由
//!
//! 所有行情相关接口通过 proto_request 适配到现有 handler。

use axum::extract::Json;
use axum::http::StatusCode;
use bytes::Bytes;
use prost::Message;
use serde_json::Value;

use futu_codec::header::ProtoFmtType;
use futu_proto::qot_get_basic_qot;
use futu_proto::qot_get_broker;
use futu_proto::qot_get_capital_distribution;
use futu_proto::qot_get_capital_flow;
use futu_proto::qot_get_code_change;
use futu_proto::qot_get_future_info;
use futu_proto::qot_get_holding_change_list;
use futu_proto::qot_get_ipo_list;
use futu_proto::qot_get_kl;
use futu_proto::qot_get_market_state;
use futu_proto::qot_get_option_chain;
use futu_proto::qot_get_option_expiration_date;
use futu_proto::qot_get_order_book;
use futu_proto::qot_get_owner_plate;
use futu_proto::qot_get_plate_security;
use futu_proto::qot_get_plate_set;
use futu_proto::qot_get_price_reminder;
use futu_proto::qot_get_reference;
use futu_proto::qot_get_rt;
use futu_proto::qot_get_security_snapshot;
use futu_proto::qot_get_static_info;
use futu_proto::qot_get_sub_info;
use futu_proto::qot_get_suspend;
use futu_proto::qot_get_ticker;
use futu_proto::qot_get_user_security;
use futu_proto::qot_get_warrant;
use futu_proto::qot_modify_user_security;
use futu_proto::qot_request_history_kl;
use futu_proto::qot_request_history_kl_quota;
use futu_proto::qot_request_rehab;
use futu_proto::qot_request_trade_date;
use futu_proto::qot_set_price_reminder;
use futu_proto::qot_stock_filter;
use futu_proto::qot_sub;
use futu_proto::used_quota;
use futu_server::conn::IncomingRequest;

use crate::adapter::{
    self, RestState, apply_known_field_aliases, expand_symbol_shorthand,
    maybe_wrap_flat_body_as_c2s, normalize_json_keys_snake_case,
};

type ApiResult = Result<Json<Value>, (StatusCode, Json<Value>)>;

/// v1.4.90 P0-B: REST 全 endpoint 共享 conn_id（替代之前每次请求 `next_conn_id()`
/// 自增 → 永久泄漏 sub-quota 直到耗尽 4000 上限）。
///
/// **背景**：v1.4.81 一直到 v1.4.89 之前，REST `proto_request` 每次都调
/// `state.next_conn_id()` 拿新 virtual conn_id（10_000_000 自增）。`SubscriptionManager`
/// 按 conn_id 记账：subscribe 把 (security, sub_type) 挂在 conn_id 下、quota
/// +=1；unsubscribe 从同一 conn_id 删除、quota -=1。但 REST `subscribe` 用 conn_id=X
/// → backend 订阅 ✓，下一次 REST `unsubscribe` 用 conn_id=Y → 找不到任何挂载 →
/// 静默 no-op，quota 永不释放。日积月累 4000 quota 耗尽，整个 daemon 不能再
/// 订阅任何东西。
///
/// **修法**：所有 REST sub-related endpoint（`subscribe` / `unsubscribe` /
/// `get_sub_info` / `query_subscription`）锁定到 `REST_SHARED_CONN`. lifecycle
/// 由 daemon 进程管，不随 REST 请求生灭。
///
/// **REST 视角合理性**：REST 是 stateless API，调用方不持续持有 daemon TCP
/// 连接，"连接 ID" 在 REST 层面没物理意义。把所有 REST 流量当作"REST gateway"
/// 这一个虚拟客户端的多次调用，对应一个固定 conn_id 是最干净的语义。v1.4.106
/// 起 quote / kline / orderbook / ticker / rt handler 也会检查
/// `SubscriptionManager::is_qot_subscribed(conn_id, ...)`，所以这些订阅门禁
/// 读路径也必须锁定到同一 `REST_SHARED_CONN`，否则 REST subscribe 后下一次
/// REST read 看不到同一个 conn 的订阅。
///
/// **取值选择**：`0xFFFF_FFFE_u64`（4_294_967_294）。`next_conn_id` 起点
/// 10_000_000，要 ~4.28B 次调用才碰到此值，daemon 早已重启。`0xFFFF_FFFF_u64`
/// 留给未来可能的 sentinel（如"all REST"广播）。
///
/// 对齐 MCP 路径行为：MCP 用单 `FutuClient` 复用底层 TCP，所有 MCP 请求自然
/// 共享同一个 daemon 分配的 conn_id，不存在该 bug。REST 现在显式实现同语义。
pub const REST_SHARED_CONN: u64 = 0xFFFF_FFFE;

/// v1.4.90 P0-B: 用 REST_SHARED_CONN 而非 `state.next_conn_id()` 派发请求。
///
/// 复刻 `adapter::proto_request_with_idempotency` 的 JSON normalize → encode →
/// dispatch → decode → JSON 流程，唯一差别：dispatch 时 `conn_id =
/// REST_SHARED_CONN`. 用于 sub-related endpoint 防 quota 泄漏（见
/// `REST_SHARED_CONN` 注释）。
///
/// 不复用 adapter::proto_request 是因为该函数硬编码 `state.next_conn_id()`，
/// 改 adapter 会越权（v1.4.90 多 agent 并行约定 agent C 改 adapter，不交叉）。
///
/// **codex 0522 F3 v1.4.106 (option B)**: 接 `Option<&CallerContext>` 让
/// gateway handler 知道 REST caller key 身份, 即使 conn_id 是
/// `REST_SHARED_CONN` 全局共享 (per-key 订阅配额 / cleanup / 审计). QOT
/// 行情 path 不直接受 `allowed_acc_ids` 约束, 但 `caller_key_id` 让未来
/// per-key 订阅 owner 模型 (e.g. QotSubscriptionState owner) 可识别 caller
/// 而不是看作 "REST 全局".
async fn proto_request_shared_conn<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    ctx: Option<&crate::caller_context::CallerContext>,
) -> ApiResult
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    // 1. JSON → protobuf 请求（与 adapter::proto_request_with_idempotency 同序）
    let req_msg: Req = if let Some(mut body) = json_body {
        normalize_json_keys_snake_case(&mut body);
        apply_known_field_aliases(&mut body);
        maybe_wrap_flat_body_as_c2s(&mut body);
        // 注：sub-related endpoint 不需要 trd header expand（trade-only path）
        expand_symbol_shorthand(&mut body)
            .map_err(|e| (StatusCode::BAD_REQUEST, Json(validation_error_body(e))))?;
        if let Some(spec) = futu_surface_spec::lookup_endpoint_by_proto_id(proto_id) {
            let validation_result = if let Some(c2s) = body.get_mut("c2s") {
                futu_surface_spec::validate_and_normalize(spec, c2s)
            } else {
                futu_surface_spec::validate_and_normalize(spec, &mut body)
            };
            if let Err(err) = validation_result {
                return Err(map_surface_spec_error(spec, err));
            }
        }
        serde_json::from_value(body).map_err(|e| {
            (
                StatusCode::BAD_REQUEST,
                Json(validation_error_body(format!("invalid request body: {e}"))),
            )
        })?
    } else {
        Req::default()
    };

    // 2. encode
    let body = Bytes::from(req_msg.encode_to_vec());

    // 3. dispatch with REST_SHARED_CONN（区别点）
    // codex 0522 F3 v1.4.106: 同时填 caller_key_id (per-call snapshot) 让
    // gateway handler 识别 REST caller 身份. allowed_acc_ids 仍 None
    // (QOT 行情不直接 acc-bound), 真填走 ctx.caller_allowed_acc_ids_arc.
    let incoming = IncomingRequest::builder(
        REST_SHARED_CONN,
        proto_id,
        state.next_serial(),
        ProtoFmtType::Protobuf,
        body,
    )
    .with_caller_scope(
        ctx.and_then(|c| c.caller_allowed_acc_ids_arc()),
        ctx.and_then(|c| c.caller_key_id()),
    )
    .build();
    let resp_bytes = state
        .router
        .dispatch(REST_SHARED_CONN, &incoming)
        .await
        .ok_or_else(|| {
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({
                    "error": "handler returned no response"
                })),
            )
        })?;

    // 4. decode
    let rsp_msg = Rsp::decode(Bytes::from(resp_bytes)).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": format!("failed to decode response: {e}")
            })),
        )
    })?;

    // 5. serialize JSON
    let mut json_rsp = serde_json::to_value(&rsp_msg).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": format!("failed to serialize response: {e}")
            })),
        )
    })?;

    // 6. err_code 前缀（与 adapter::maybe_wrap_err_code_prefix 同语义；该 helper
    //    在 adapter.rs 是 private，inline 一个等价小版本以满足"不碰 adapter" 约束）
    wrap_err_code_prefix_inline(&mut json_rsp);

    Ok(Json(json_rsp))
}

fn map_surface_spec_error(
    spec: &'static futu_surface_spec::EndpointSpec,
    err: futu_surface_spec::DispatchError,
) -> (StatusCode, Json<Value>) {
    let proto_id = spec
        .proto_id()
        .map(|id| id.to_string())
        .unwrap_or_else(|| "daemon-local".to_string());
    let ret_msg = format!(
        "{} (endpoint: {}, proto_id: {})",
        err, spec.canonical_name, proto_id
    );
    let mut body = validation_error_body(ret_msg);
    let machine_error_field = spec.runtime.error.machine_error_field;
    if let Some(obj) = body.as_object_mut() {
        obj.insert(
            machine_error_field.to_string(),
            serde_json::json!({
                "kind": "validation_error",
                "message": err.to_string(),
                "endpoint": spec.canonical_name,
                "proto_id": proto_id,
            }),
        );
    }
    (StatusCode::BAD_REQUEST, Json(body))
}

fn validation_error_body(message: impl Into<String>) -> Value {
    let message = message.into();
    serde_json::json!({
        "ret_type": -1,
        "ret_msg": message,
        "error": message,
    })
}

/// v1.4.90 P0-B: inline `adapter::maybe_wrap_err_code_prefix` 的等价实现。
///
/// adapter.rs 里的版本是 private（v1.4.34 BUG-2b 沉淀）。本 helper 为
/// proto_request_shared_conn 服务，避免修改 adapter（不交叉 agent C scope）。
///
/// 语义与 adapter 版本对齐：
/// - ret_type == 0 → 不动
/// - ret_type != 0 + err_code present → `[err_code=<X>] <ret_msg>`
/// - 幂等：已带 `[err_code=` 前缀的 ret_msg 不重复包
fn wrap_err_code_prefix_inline(v: &mut Value) {
    let obj = match v.as_object_mut() {
        Some(o) => o,
        None => return,
    };
    let is_err = obj
        .get("ret_type")
        .and_then(|t| t.as_i64())
        .map(|t| t != 0)
        .unwrap_or(false);
    if !is_err {
        return;
    }
    let raw_msg = obj
        .get("ret_msg")
        .and_then(|m| m.as_str())
        .unwrap_or("")
        .to_string();
    if raw_msg.starts_with("[err_code=") {
        return;
    }
    let err_code_label = match obj.get("err_code") {
        Some(Value::Number(n)) => n
            .as_i64()
            .map(|i| i.to_string())
            .unwrap_or_else(|| "none".to_string()),
        _ => "none".to_string(),
    };
    let new_msg = if raw_msg.is_empty() {
        format!("[err_code={err_code_label}]")
    } else {
        format!("[err_code={err_code_label}] {raw_msg}")
    };
    obj.insert("ret_msg".to_string(), Value::String(new_msg));
}

/// POST /api/subscribe — 订阅/退订行情
///
/// **v1.4.90 P0-B fix**: 用 `REST_SHARED_CONN` 替代 `state.next_conn_id()`,
/// 杜绝 quota 永久泄漏（详见 `REST_SHARED_CONN` 注释）.
///
/// **v1.4.104 eli S-005 (P1) fix**: REST 层显式检查 `is_sub_or_un_sub` 字段
/// 在 raw JSON body 是否 present. proto bool 字段 missing 时 prost 默认为
/// false → handler 走 unsub 路径; 但 unsub 路径对 invalid ticker silent
/// success (handler 注释 line 187-189: unsub by-design 不做 backend 解析).
/// agent 调用方默认只传 `symbols`, 不写 boolean → silent 没订阅.
///
/// 修法: REST adapter 入口先检查 `is_sub_or_un_sub` 在 body 里 (snake-case
/// 归一化后) 是否 present. 不 present → 400 提示用户必须显式传字段.
///
/// **v1.4.104 codex round 2 F5 (P2) fix**: REST 全部 caller 共用一个虚拟连接
/// `REST_SHARED_CONN=0xFFFFFFFE` (v1.4.90 P0-B 防 quota 泄漏的设计). 因此
/// `is_unsub_all=true` 调用会**清掉所有 REST callers** 的 QOT 订阅 (因为大家
/// 共享同一 conn_id 的 subscription bucket). 对于多 REST client 部署:
///
///   - 当前合约: `is_unsub_all=true` 在 REST 显式 reject, 仅返 400 解释
///     "REST is_unsub_all 是 process-wide 操作, 跨 caller 影响, 默认禁用".
///   - 可用替代: 显式列 `security_list` + `sub_type_list` 做单 symbol 退订;
///     或改用 MCP / gRPC / WS 这类 per-conn surface 执行 unsub_all.
///   - REST 目前没有 process-wide opt-in query, 也没有 admin clear endpoint;
///     后续若要支持 REST per-key unsub_all, 必须先补独立状态模型与公开契约。
///
/// 当前实装: 选项 A 保守 (拒+提示), 防止 caller A 意外清掉 caller B 的 subs.
/// MCP / gRPC / WS 各 caller 有自己 conn_id, 不受影响.
// Split: 1408 行 → 5 子文件 (contiguous fn 段)
mod misc;
mod quotes;
mod reference;
mod snapshot;
mod subscribe;

#[cfg(test)]
mod tests;

#[cfg(test)]
use misc::inject_default_is_req_all_conn;
#[cfg(test)]
use quotes::{annotate_quote_cache_miss, orderbook_loud_unsub_hint};
#[cfg(test)]
use snapshot::{
    augment_snapshot_with_exchange_code, augment_static_info_with_exchange_code,
    check_static_info_input,
};
#[cfg(test)]
use subscribe::body_has_sub_or_unsub_flag;

pub use misc::{
    get_risk_free_rate, get_spread_table, get_ticker_statistic, get_ticker_statistic_detail,
    list_plates, query_subscription, unsubscribe,
};
pub use quotes::{get_basic_qot, get_broker, get_kl, get_order_book, get_rt, get_ticker};
pub use reference::{
    get_capital_distribution, get_capital_flow, get_code_change, get_future_info,
    get_holding_change, get_ipo_list, get_market_state, get_option_chain,
    get_option_expiration_date, get_owner_plate, get_plate_security, get_plate_set,
    get_price_reminder, get_reference, get_suspend, get_used_quota, get_user_security, get_warrant,
    modify_user_security, request_history_kl, request_history_kl_quota, request_rehab,
    request_trading_days, set_price_reminder, stock_filter,
};
pub use snapshot::{get_snapshot, get_static_info};
pub use subscribe::{get_sub_info, subscribe};