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 axum::http::header::CONTENT_TYPE;
use axum::response::{IntoResponse, Response};
use serde_json::Value;

use super::*;

mod decode;
pub(crate) use decode::{
    JsonRequestMode, decode_json_request, decode_json_request_with_surface_spec,
};

#[derive(Debug, Clone)]
pub struct RawJson {
    bytes: Bytes,
}

impl RawJson {
    pub fn new(bytes: impl Into<Bytes>) -> Self {
        Self {
            bytes: bytes.into(),
        }
    }
}

impl IntoResponse for RawJson {
    fn into_response(self) -> Response {
        ([(CONTENT_TYPE, "application/json")], self.bytes).into_response()
    }
}

///
/// 泛型参数:
/// - `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, None).await
}

pub(crate) async fn proto_request_with_surface_spec_and_idempotency<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    idempotency_key: Option<String>,
    surface_spec: &'static EndpointSpec,
) -> 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,
        Some(surface_spec),
    )
    .await
}

pub async fn proto_request_raw<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
) -> Result<RawJson, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    proto_request_raw_internal::<Req, Rsp>(
        state,
        proto_id,
        json_body,
        None,
        None,
        JsonRequestMode::GenericRest,
    )
    .await
}

pub async fn proto_request_raw_spec_body<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
) -> Result<RawJson, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    proto_request_raw_internal::<Req, Rsp>(
        state,
        proto_id,
        json_body,
        None,
        None,
        JsonRequestMode::RawSpecBody,
    )
    .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())),
            has_auth_setup_scope: false,
            is_loopback: false,
            legacy_local_mode: false,
        }
    } else {
        crate::caller_context::CallerContext::legacy()
    };
    proto_request_internal::<Req, Rsp>(state, proto_id, json_body, None, Some(&ctx), None).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, 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,
        has_auth_setup_scope: false,
        is_loopback: false,
        legacy_local_mode: false,
    });
    proto_request_internal::<Req, Rsp>(
        state,
        proto_id,
        json_body,
        idempotency_key,
        ctx.as_ref(),
        None,
    )
    .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, None).await
}

async fn proto_request_raw_internal<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    idempotency_key: Option<String>,
    ctx: Option<&crate::caller_context::CallerContext>,
    mode: JsonRequestMode,
) -> Result<RawJson, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    let req_msg: Req = decode_json_request(proto_id, json_body, mode)?;
    let conn_id = state.next_conn_id();
    let serial_no = state.next_serial();
    let body = futu_server::trade_packet_id::fill_omitted_trade_packet_id_bytes(
        proto_id,
        req_msg.encode_to_vec(),
        conn_id,
        serial_no,
    )
    .map(Bytes::from)
    .map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": format!("failed to prepare trade PacketID: {e}")
            })),
        )
    })?;

    let incoming =
        IncomingRequest::builder(conn_id, proto_id, serial_no, 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()),
            )
            .with_auth_setup_admission(
                ctx.is_some_and(|c| c.caller_has_auth_setup_scope()),
                ctx.is_some_and(|c| c.caller_is_loopback()),
                ctx.is_some_and(|c| c.caller_legacy_local_mode()),
            )
            .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"
                })),
            )
        })?;

    let resp_bytes = state.filter_registry.apply(
        proto_id,
        resp_bytes,
        ctx.and_then(|c| c.allowed_acc_ids_borrow()),
    );

    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}")
            })),
        )
    })?;

    raw_json_from_proto_response(&rsp_msg)
}

pub(crate) fn raw_json_from_proto_response<Rsp>(
    rsp_msg: &Rsp,
) -> Result<RawJson, (StatusCode, Json<Value>)>
where
    Rsp: serde::Serialize,
{
    let encoded = encode_proto_response_raw_or_error_value(rsp_msg).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": e
            })),
        )
    })?;
    match encoded {
        ProtoJsonBody::Raw(raw) => Ok(RawJson::new(raw)),
        ProtoJsonBody::Value(value) => {
            let raw = serde_json::to_vec(&value).map_err(|e| {
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    Json(serde_json::json!({
                        "error": format!("failed to serialize response: {e}")
                    })),
                )
            })?;
            Ok(RawJson::new(raw))
        }
    }
}

/// 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>,
    surface_spec: Option<&'static EndpointSpec>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)>
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    // 1. JSON → protobuf 请求. Empty body still flows through EndpointSpec
    // validation; otherwise required fields could silently become
    // `Req::default()` and bypass REST contract checks.
    let req_msg: Req = decode_json_request_with_surface_spec(
        proto_id,
        json_body,
        JsonRequestMode::GenericRest,
        surface_spec,
    )?;

    // 2. encode 为 protobuf bytes
    let conn_id = state.next_conn_id();
    let serial_no = state.next_serial();
    let body = futu_server::trade_packet_id::fill_omitted_trade_packet_id_bytes(
        proto_id,
        req_msg.encode_to_vec(),
        conn_id,
        serial_no,
    )
    .map(Bytes::from)
    .map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": format!("failed to prepare trade PacketID: {e}")
            })),
        )
    })?;

    // 3. 构造 IncomingRequest 调用现有 handler
    // codex 0522 F1 v1.4.106: caller_allowed_acc_ids + caller_key_id 同时从
    // CallerContext 拿, 单一来源, 不再分别写 None.
    let incoming =
        IncomingRequest::builder(conn_id, proto_id, serial_no, 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()),
            )
            .with_auth_setup_admission(
                ctx.is_some_and(|c| c.caller_has_auth_setup_scope()),
                ctx.is_some_and(|c| c.caller_is_loopback()),
                ctx.is_some_and(|c| c.caller_legacy_local_mode()),
            )
            .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。external reviewer 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))
}

pub(super) enum ProtoJsonBody {
    Raw(Bytes),
    Value(Value),
}

pub(super) fn encode_proto_response_raw_or_error_value<Rsp>(
    rsp_msg: &Rsp,
) -> Result<ProtoJsonBody, String>
where
    Rsp: serde::Serialize,
{
    let raw =
        serde_json::to_vec(rsp_msg).map_err(|e| format!("failed to serialize response: {e}"))?;
    if serialized_ret_type(&raw) == Some(0) {
        return Ok(ProtoJsonBody::Raw(Bytes::from(raw)));
    }

    let mut json_rsp =
        serde_json::to_value(rsp_msg).map_err(|e| format!("failed to serialize response: {e}"))?;
    maybe_wrap_err_code_prefix(&mut json_rsp);
    Ok(ProtoJsonBody::Value(json_rsp))
}

fn serialized_ret_type(raw: &[u8]) -> Option<i64> {
    #[derive(serde::Deserialize)]
    struct RetOnly {
        ret_type: Option<i64>,
    }

    serde_json::from_slice::<RetOnly>(raw)
        .ok()
        .and_then(|ret| ret.ret_type)
}

/// 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(crate) 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));
}