futu_rest/adapter/

proto_request.rs

//! Split from adapter.rs: proto_request.
//!
//! pub items: proto_request,proto_request_with_filter,proto_request_with_idempotency,proto_request_with_idempotency_and_caller,proto_request_with_ctx.

use axum::Json;
use axum::http::StatusCode;
use serde_json::Value;

use super::*;

///
/// 泛型参数:
/// - `Req`: protobuf 请求类型 (prost::Message + serde::Deserialize)
/// - `Rsp`: protobuf 响应类型 (prost::Message + serde::Serialize)
///
/// 流程: JSON → Req → encode → dispatch(proto_id) → decode → Rsp → JSON
pub async fn proto_request<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    proto_request_internal::<Req, Rsp>(state, proto_id, json_body, None, None).await
}

/// 把 `DispatchError` 翻译成 REST 400 response (HTTP + JSON body).
///
/// pitfall #45 loud error: 不返 silent empty ret_type=0, 返清晰 400 + ret_msg.
pub(super) fn map_dispatch_error(
    spec: &'static EndpointSpec,
    err: 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 status = StatusCode::from_u16(spec.runtime.error.validation_http_status)
        .unwrap_or(StatusCode::BAD_REQUEST);
    let machine_error_field = spec.runtime.error.machine_error_field;
    let mut body = serde_json::json!({
        "ret_type": -1,
        "ret_msg": ret_msg,
    });
    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,
            }),
        );
    }
    (status, Json(body))
}

pub(super) 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.104 阶段 7-1: 走 FilterRegistry 的 proto_request 变体.
///
/// 跟 [`proto_request`] / [`proto_request_with_idempotency`] 流程一致, 但在
/// `decode protobuf 响应` 之前**插入 FilterRegistry::apply** 一步, 按
/// `allowed_acc_ids` filter 受限 key 的响应 acc_list (proto 2001 TRD_GET_ACC_LIST
/// 等). 与 gRPC server.rs / WS ws_listener.rs 同源 (单一 registry).
///
/// `allowed_acc_ids = None` 时 filter no-op (legacy / 无限制 key).
///
/// **codex 0522 F2 v1.4.106**: 推荐改用 [`proto_request_with_ctx`], 这个
/// helper 改为内部包装, 丢失 `caller_key_id`. 保留作 backward-compat 给
/// 已有的 acc_list filter call site 不破坏 (route 层迁移到 ctx 后此函数
/// 全部 caller 应该归零, 保留一段过渡期再删).
pub async fn proto_request_with_filter<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    allowed_acc_ids: Option<&std::collections::HashSet<u64>>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    // codex 0522 F1 v1.4.106: 构 minimal CallerContext 把 allowed_acc_ids 同时
    // 接进 IncomingRequest.caller_allowed_acc_ids + FilterRegistry. caller_key_id
    // 仍 None (老 call site 没传). 推荐 caller 迁移到 proto_request_with_ctx.
    let ctx = if let Some(allowed) = allowed_acc_ids {
        crate::caller_context::CallerContext {
            key_id: None,
            allowed_acc_ids: Some(std::sync::Arc::new(allowed.clone())),
        }
    } else {
        crate::caller_context::CallerContext::legacy()
    };
    proto_request_internal::<Req, Rsp>(state, proto_id, json_body, None, Some(&ctx)).await
}

/// v1.4.38 Phase 4: 支持 `Idempotency-Key` header 的 proto_request。
/// 老 call site 继续用 `proto_request`(header=None), 新写 trade endpoint
/// 用 `proto_request_with_idempotency` 从 axum HeaderMap 提取 header 后传入。
pub async fn proto_request_with_idempotency<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    idempotency_key: Option<String>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    proto_request_internal::<Req, Rsp>(state, proto_id, json_body, idempotency_key, None).await
}

/// v1.4.106 codex 0920 F1 (P1): 支持 caller key id 的 idempotency 变体.
/// 让 cache namespace 跨 caller 隔离 — 不同 caller 用同 Idempotency-Key
/// **不能** 跨 caller 命中老 response (避免跨账户数据泄漏 + 重复下单).
///
/// **call site**: REST trade endpoint (place / modify / cancel / reconfirm)
/// 在 `rec: Option<Extension<Arc<KeyRecord>>>` 抽 caller_key_id 后传入.
pub async fn proto_request_with_idempotency_and_caller<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    idempotency_key: Option<String>,
    caller_key_id: Option<String>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    let ctx = caller_key_id.map(|k| crate::caller_context::CallerContext {
        key_id: Some(k),
        allowed_acc_ids: None,
    });
    proto_request_internal::<Req, Rsp>(state, proto_id, json_body, idempotency_key, ctx.as_ref())
        .await
}

/// codex 0522 F1 v1.4.106 (推荐 API): 带 `CallerContext` 的 proto_request.
///
/// 与 `proto_request_with_idempotency` / `proto_request_with_filter` 的关系:
/// 后两者只接 `idempotency_key` 或 `allowed_acc_ids` 单一维度, 本函数接完整
/// `CallerContext` (含 `key_id` + `allowed_acc_ids`), 把 caller scope **同时**
/// 接到三个下游消费点:
///
/// 1. `IncomingRequest.caller_allowed_acc_ids` (dispatch handler 的 per-acc
///    enforce, defense-in-depth)
/// 2. `IncomingRequest.caller_key_id` (per-key 配额 / cleanup / 审计入口)
/// 3. `FilterRegistry::apply` (响应 acc_list 过滤)
///
/// 任一接错 → `dispatch handler` 看到 `None` 就 silent bypass (codex F1 audit
/// 实锤的 v1.4.105 之前 REST regression).
///
/// `ctx = None` 等价 `legacy mode` (无 caller key, 无 acc 限制) — 通常仅
/// 测试 / 显式 unauthenticated route 用. 真 route handler 应该总是构 ctx
/// 从 `Extension<Arc<KeyRecord>>` 抽出来 (`CallerContext::from_key_record`).
pub async fn proto_request_with_ctx<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    idempotency_key: Option<String>,
    ctx: Option<&crate::caller_context::CallerContext>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    proto_request_internal::<Req, Rsp>(state, proto_id, json_body, idempotency_key, ctx).await
}

/// v1.4.104 阶段 7-1 + codex 0522 F1 v1.4.106: 内部统一实现, 接 `CallerContext`
/// 替代单 `allowed_acc_ids` 入参. 调度时同时填 `IncomingRequest.caller_key_id`
/// 与 `caller_allowed_acc_ids`, 响应过滤共用同一个 `allowed_acc_ids` 借引用 ——
/// 单一来源, 防 routes 层手写多套 caller scope check 漂移 (codex F2).
async fn proto_request_internal<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    idempotency_key: Option<String>,
    ctx: Option<&crate::caller_context::CallerContext>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    // 1. JSON → protobuf 请求
    // v1.4.45: normalize camelCase keys to snake_case（兼容 FTAPI 官方文档的
    // camelCase 字段名，如 accID/trdEnv/filterConditions）—— 见 normalize_json_keys_snake_case 注释
    // v1.4.68: alias 常见 SDK 惯用字段名到 proto 字段（如 max_count → max_ack_kl_num）
    // v1.4.73 BUG-005: auto-wrap flat body 到 `{c2s: ...}` 结构 —— 用户习惯用
    // 扁平 body（对齐 Python SDK 风格），但 proto Request struct 要求嵌套 `c2s`。
    let req_msg: Req = if let Some(mut body) = json_body {
        normalize_json_keys_snake_case(&mut body);
        apply_known_field_aliases(&mut body);
        // v1.4.73 BUG-005: flat body → {c2s: body} 自动包装（见 fn 注释）
        maybe_wrap_flat_body_as_c2s(&mut body);
        // v1.4.90 P2-D: trade endpoint 用户常传 flat `acc_id` / `trd_env`
        // / `trd_market` / `jp_acc_type`, proto 期望 `c2s.header.{...}`.
        // 在 c2s 已建好但尚未 expand symbol shorthand 时补 header 嵌套.
        maybe_expand_flat_trd_header(&mut body);
        // v1.4.73 BUG-005: history-kline 用户习惯 "symbol": "HK.00700" 代替
        // 嵌套 security:{market,code}，adapter 解析 symbol 前缀 + code
        // v1.4.90 P0-C: expand 路径加 MAX_SYMBOLS_PER_REQUEST 检查, 超返 400.
        expand_symbol_shorthand(&mut body)
            .map_err(|e| (StatusCode::BAD_REQUEST, Json(validation_error_body(e))))?;
        // v1.4.110 Layer 2: spec-driven validation 兜底 — 所有 transformations
        // 完成后, 按 proto_id lookup EndpointSpec, 调 validate_and_normalize.
        // body 已是 `{c2s: {...}}` (或 c2s 字段在顶层, 取决于 endpoint), spec
        // 验 c2s 内部. 未 declare spec 的 endpoint 跳 (向后兼容).
        if let Some(spec) = futu_surface_spec::lookup_endpoint_by_proto_id(proto_id) {
            // 多数 endpoint body 是 {c2s: {actual_content}}, validate c2s 内容.
            // 但有 endpoint 没 wrap c2s (eg. empty Request struct), validate body 自身.
            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_dispatch_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 为 protobuf bytes
    let body = Bytes::from(req_msg.encode_to_vec());

    // 3. 构造 IncomingRequest 调用现有 handler
    // codex 0522 F1 v1.4.106: caller_allowed_acc_ids + caller_key_id 同时从
    // CallerContext 拿, 单一来源, 不再分别写 None.
    let incoming = IncomingRequest::builder(
        state.next_conn_id(),
        proto_id,
        state.next_serial(),
        ProtoFmtType::Protobuf,
        body,
    )
    .with_idempotency_key(idempotency_key)
    .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(incoming.conn_id, &incoming)
        .await
        .ok_or_else(|| {
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({
                    "error": "handler returned no response"
                })),
            )
        })?;

    // 3.5. v1.4.104 阶段 7-1: response filter (cross-surface 共享 FilterRegistry).
    //      proto 2001 TRD_GET_ACC_LIST 默认注册, allowed_acc_ids 非空时 filter
    //      响应 acc_list, 受限 key 不能跨账户 enumerate. legacy / 无限制 key /
    //      非注册 proto_id → no-op 返原 bytes 不动.
    // codex 0522 F2 v1.4.106: 同一 ctx.allowed_acc_ids 借引用 (与
    // IncomingRequest.caller_allowed_acc_ids 是同一份 Arc), 单一来源.
    let resp_bytes = state.filter_registry.apply(
        proto_id,
        resp_bytes,
        ctx.and_then(|c| c.allowed_acc_ids_borrow()),
    );

    // 4. decode protobuf 响应
    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. 序列化为 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. v1.4.34 BUG-2b 修：给所有错误响应的 ret_msg 加 `[err_code=X]` 前缀
    //    历史：v1.4.27 server_err() helper 在 futu-trd 客户端 lib 里包了 CLI/gRPC/MCP
    //    三个 surface，漏了 REST 这个 surface。eli 4 次独立复现；v1.4.30 / 31 /
    //    32 / 33 都没修。根因：REST 走 proto_request 直接序列化响应，没经过 server_err
    //    helper 层。v1.4.34 直接在 JSON 层包一下，不动 daemon response（daemon 要给
    //    CLI 客户端留原始 err_code 字段，CLI 自己会包）。
    maybe_wrap_err_code_prefix(&mut json_rsp);

    Ok(Json(json_rsp))
}

/// v1.4.34 BUG-2b：REST 响应的 ret_msg 包 `[err_code=X]` 前缀。
///
/// 与 `futu_trd::server_err()` 同语义，但作用在已序列化的 JSON 上：
///
/// - `ret_type == 0`：成功，不动
/// - `ret_type != 0` 且 `err_code` 非 null：`[err_code=<code>] <原 ret_msg>`
/// - `ret_type != 0` 且 `err_code` 缺省：`[err_code=none] <原 ret_msg>`
/// - `ret_type != 0` 且 ret_msg 空：只留 `[err_code=<X>]` 带方括号的标签
///
/// 幂等（已经带 `[err_code=` 前缀的 ret_msg 不重复包）。方括号位置精确，既
/// 便于客户端 grep，也不误伤"错误描述里恰好有方括号"的正常 msg。
///
/// 只对**顶层**的 `ret_type / ret_msg / err_code` 生效；嵌套在 s2c 里的状态字段
/// 不动。
pub(super) fn maybe_wrap_err_code_prefix(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;
    }
    // 读当前 msg（可能是 null）
    let raw_msg = obj
        .get("ret_msg")
        .and_then(|m| m.as_str())
        .unwrap_or("")
        .to_string();
    // 幂等：已带前缀就不动（多轮 middleware 不应该双包）
    if raw_msg.starts_with("[err_code=") {
        return;
    }
    // 读 err_code（可能是 null / 整数）
    let err_code_label = match obj.get("err_code") {
        Some(Value::Number(n)) => n
            .as_i64()
            .map(|i| i.to_string())
            .unwrap_or("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));
}