futu_rest/routes/trd/

sub_acc_push.rs

//! REST trade account push subscription route and helpers.

use std::sync::Arc;

use axum::extract::{Extension, Json, State};
use axum::http::StatusCode;
use serde_json::Value;

use futu_auth::KeyRecord;

use crate::adapter::RestState;

use super::ApiResult;

/// v1.4.90 P1-C 修：POST /api/sub-acc-push — 订阅交易账户 push（订单 / 成交变动）
///
/// **历史背景**（外部 tester v1.4.69-89 报告 6 版 silent stub）：
/// - 老版 handler 是单 `proto_request` 透传 wrapper，无 input validation、
///   无 audit log、无 session_id token —— 用户 POST 后 daemon log 看不到任何
///   "本 endpoint 被调用过" 的痕迹（仅 gateway 侧 SubAccPushHandler 默默更
///   新 `SubscriptionManager.subscribe_trd_acc(conn_id, acc_id)`），
///   "ret=0 + s2c={}" 看似成功但下游全无可观测信号。
///
/// **本次 (v1.4.90 P1-C) 对齐 MCP `futu_sub_acc_push`**：
/// 1. **Loud input validation**（acc_id_list 必须非空 + 每个 id 非零；C++
///    `IAPIServer_Trd_SubAccPush::OnClientReq_SubAccPush` 等价 check）
/// 2. **Audit log** `tracing::info!` 留 endpoint + count + 头几个 acc_id 痕迹，
///    给 push_health 看不到的部分提供可观测性
/// 3. **Backend CMD forward 经 `proto_request`**（即 dispatch 到 gateway
///    `SubAccPushHandler` 完成 `SubscriptionManager` 注册 + return ret=0）
/// 4. **Response augment**：在标准 proto Response (`ret_type/s2c{}`) 上补
///    `subscribed_acc_ids` + `session_id` + `unsub_hint`，对齐 MCP
///    `{ok, subscribed_acc_ids, session_id, unsub_hint}` 输出形状。
///    REST stateless，session_id 用 sorted acc_ids hash 派生（确定性 token，
///    可拿来 POST `/api/unsub-acc-push`）
///
/// **架构说明**：本 daemon 内 gateway local `SubAccPushHandler` 即对齐
/// C++ APIServer 的 `Trd_SubAccPush` 处理点 —— 没有"再下一层 backend"
/// 的 CMD 需要 forward。MCP 通过 TCP `client.request(TRD_SUB_ACC_PUSH)`
/// 也是落到同一个 gateway handler。所以 "CMD forward" 的语义是 REST →
/// `proto_request` → `RequestRouter::dispatch` → `SubAccPushHandler`。
pub async fn sub_acc_push(
    State(state): State<RestState>,
    rec: Option<Extension<Arc<KeyRecord>>>,
    Json(mut body): Json<Value>,
) -> ApiResult {
    // v1.4.103 (codex 56 F1 / 58 F5 — B9): legacy 模式 (无 keys.json / 无 Bearer)
    // 直接 reject. 之前 legacy 模式返 ret_type=0 + session_id 给客户端 "看似成功",
    // 但**不写** rest_acc_subscriptions 状态 (legacy WS 连接也无 key_id, sub
    // state filter 跳过) — 用户实际看到 revoke 成功仍按 legacy 全量 trade push
    // 行为继续收, silent inconsistency. 改为 loud reject 让用户知道 sub state
    // 需要 keys.json + Bearer 才工作.
    if rec.is_none() {
        futu_auth::audit::reject(
            "rest",
            "/api/sub-acc-push",
            "<legacy>",
            "sub-acc-push not supported in legacy mode (no keys.json)",
        );
        return Err((
            StatusCode::FORBIDDEN,
            Json(serde_json::json!({
                "error": "/api/sub-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 BUG-009: legacy mode previously returned silent success without persisting sub state. Now loud-reject to surface the limitation."
            })),
        ));
    }
    // v1.4.102 codex 43 F1 (P2): normalize 在 extract_acc_id_list 之前.
    // 之前 extractor 只看 snake_case `acc_id_list`, 但官方 proto 是 camelCase
    // `accIDList`. adapter 后续会 normalize, 但本 extractor 先跑会误 reject 空
    // → 400 silent. 现在 normalize 进去, extractor 看到 acc_id_list.
    crate::adapter::normalize_json_keys_snake_case(&mut body);
    // 1. Loud input validation（解析 acc_id_list；畸形 / 空 / 全零 → 400）
    let acc_ids = match 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/sub-acc-push", &key_id, &reason);
            return Err((
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({
                    "error": format!("/api/sub-acc-push: {reason}")
                })),
            ));
        }
    };
    if let Err(reason) = 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/sub-acc-push", &key_id, reason);
        return Err((
            StatusCode::BAD_REQUEST,
            Json(serde_json::json!({
                "error": format!("/api/sub-acc-push: {reason}")
            })),
        ));
    }
    // v1.4.102 codex 46 F4 (P1): sub-acc-push 必须对 acc_id_list 每个 id 跑
    // allowed_acc_ids 限额. 之前只 validate 非空, 但 key 限到 acc A 的用户传
    // [A, B, C] 仍能 subscribe 别人账户的 trade push stream — 越权.
    //
    // codex 0522 F2 v1.4.106: 走共享 helper `check_per_acc_rate_for_caller`,
    // 与 `/api/unsub-acc-push` 同源 (单一逻辑, 防漂移).
    check_per_acc_rate_for_caller(
        &state.counters,
        rec.as_deref().map(|r| r.as_ref()),
        &acc_ids,
        "/api/sub-acc-push",
    )?;

    // 2. Audit log：留 endpoint + count + head acc_ids 痕迹（push_health 不
    // 跟踪 subscribe，本 log 是 silent-stub 的可观测性补丁）
    let head: Vec<u64> = acc_ids.iter().take(3).copied().collect();
    tracing::info!(
        target: futu_auth::audit::TARGET,
        iface = "rest",
        endpoint = "/api/sub-acc-push",
        count = acc_ids.len(),
        head_acc_ids = ?head,
        "v1.4.90 P1-C: sub-acc-push CMD forward → SubAccPushHandler"
    );

    // 3. v1.4.102 codex 51 F2 (P2): REST 不再 dispatch 到 gateway
    // SubAccPushHandler — 那条路径 register ephemeral REST conn_id 到
    // SubscriptionManager (per-conn_id), 而 unsub 用新 conn_id 删不到, 留
    // stale entries 让 push_trd_acc 一直扫. REST sub state map (per-key) 已
    // 是 REST 层 truly canonical 真相源, 直接 return success + 让 augment
    // 段 (后) 写 state map.
    let mut resp = Json(serde_json::json!({
        "ret_type": 0,
        "ret_msg": serde_json::Value::Null,
        "err_code": serde_json::Value::Null,
        "s2c": {}
    }));
    // body 不再下发给 gateway, 但仍要消费防 unused warning (REST 已经从 body
    // 提了 acc_ids).
    let _ = body;

    // 4. 仅在 backend 返成功 (ret_type=0) 时 augment session_id / unsub_hint
    //    + v1.4.102 codex 46 F2/F3 / 47 F3 / 48 F3 (P1): 写入 REST sub state
    //    map 让 WS push delivery 过滤生效. **mutate-after-daemon-success** —
    //    proto_request 已 await 完, ret_type=0 才 register, 失败时不留 local
    //    state (避免 daemon 失败但 REST WS state 残留导致行为不一致).
    if let Some(obj) = resp.0.as_object_mut() {
        let ret_type_ok = obj.get("ret_type").and_then(|v| v.as_i64()).unwrap_or(-1) == 0;
        if ret_type_ok {
            // codex 47 F3 / 48 F3: 仅 daemon-success 后 register local state.
            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| {
                        let entry = subs.entry(key_id).or_default();
                        for &acc_id in &acc_ids {
                            entry.insert(acc_id);
                        }
                    },
                );
            }
            let session_id = derive_sub_acc_push_session_id(&acc_ids);
            obj.insert("subscribed_acc_ids".to_string(), serde_json::json!(acc_ids));
            obj.insert("session_id".to_string(), Value::String(session_id.clone()));
            obj.insert(
                "unsub_hint".to_string(),
                Value::String(format!(
                    "POST /api/unsub-acc-push with same acc_id_list to revoke (REST is stateless, session_id=\"{session_id}\" is a deterministic token derived from sorted acc_ids)"
                )),
            );
        }
    }
    Ok(resp)
}

/// v1.4.90 P1-C helper：从 JSON body 抽 `c2s.acc_id_list`（兼容 flat body）。
///
/// 容忍点：
/// - `body.c2s.acc_id_list` 嵌套（标准 proto 结构）
/// - `body.acc_id_list` 扁平（adapter 之后会 wrap 成 c2s，本 helper 提前看一眼）
/// - 数组元素必须是 JSON 正整数；畸形元素 fail-closed，不静默过滤
pub(crate) fn extract_acc_id_list(body: &Value) -> Result<Vec<u64>, String> {
    let raw = body
        .pointer("/c2s/acc_id_list")
        .or_else(|| body.pointer("/acc_id_list"));
    let Some(raw) = raw else {
        return Ok(Vec::new());
    };
    let Some(arr) = raw.as_array() else {
        return Err("c2s.acc_id_list must be an array of positive integer acc_id values".into());
    };

    let mut acc_ids = Vec::with_capacity(arr.len());
    for (idx, value) in arr.iter().enumerate() {
        let Some(acc_id) = value.as_u64() else {
            return Err(format!(
                "c2s.acc_id_list[{idx}] must be a positive integer acc_id"
            ));
        };
        acc_ids.push(acc_id);
    }
    Ok(acc_ids)
}

/// v1.4.90 P1-C helper：sub-acc-push 入参校验。
///
/// C++ `IAPIServer_Trd_SubAccPush::OnClientReq_SubAccPush` 没有显式校验
/// 空 list / 0 acc_id（默默 no-op），但 silent no-op 是用户视角的 bug
/// （tester 报告 "看似 ret=0 但什么都没订上"）。Rust REST 侧 fail-fast。
pub(crate) fn validate_sub_acc_push_acc_ids(acc_ids: &[u64]) -> Result<(), &'static str> {
    if acc_ids.is_empty() {
        return Err("c2s.acc_id_list is required and must not be empty");
    }
    if acc_ids.contains(&0) {
        return Err(
            "c2s.acc_id_list contains zero — call /api/list-accounts to discover real acc_id values",
        );
    }
    Ok(())
}

/// codex 0522 F2 v1.4.106: 共享的 per-acc rate / scope check helper.
///
/// `/api/sub-acc-push` 与 `/api/unsub-acc-push` 之前在两个文件 (trd.rs +
/// sys.rs) 各写一份相同的 for-loop 跑 `check_full_skip_rate` per acc_id.
/// codex F2 audit 指出: REST caller scope 没统一入口, 多套手写 = 漂移风险.
/// 本 helper 是单一来源 — 两处 callers 都调它, 任一改逻辑都同步生效.
///
/// 接 `CallerContext` 而不是 `&KeyRecord`: 让 caller side 总是先 build ctx
/// (per-call snapshot), 与 `IncomingRequest.caller_*` 同源数据不漂移.
///
/// `endpoint` 用作 audit reject label (e.g. "/api/sub-acc-push").
///
/// **行为**:
/// - `ctx.key_id = None` (legacy mode): no-op 返 Ok (route handler 已在 sub
///   入口 reject legacy, 但本 helper 防御性兼容)
/// - `acc_ids` 任一不通过 → audit reject + 返 Err 含 status code (从 outcome
///   的 http_status_code 派生, 默认 403)
/// - 全通过 → Ok
pub(crate) fn check_per_acc_rate_for_caller(
    counters: &futu_auth::RuntimeCounters,
    rec: Option<&futu_auth::KeyRecord>,
    acc_ids: &[u64],
    endpoint: &'static str,
) -> Result<(), (StatusCode, Json<Value>)> {
    let Some(key_rec) = rec else {
        // legacy mode — caller handler 已 reject 过, 但防御性兼容. no-op.
        return Ok(());
    };
    for &acc_id in acc_ids {
        let ctx = futu_auth::CheckCtx {
            market: String::new(),
            symbol: String::new(),
            order_value: None,
            trd_side: None,
            acc_id: Some(acc_id),
            mutation_no_exposure: false,
            currency: None,
        };
        let now = chrono::Utc::now();
        let outcome = counters.check_full_skip_rate(&key_rec.id, &key_rec.limits(), &ctx, now);
        if let Some(reason) = outcome.reason() {
            futu_auth::audit::reject(
                "rest",
                endpoint,
                &key_rec.id,
                &format!("limit: {reason} (acc_id={acc_id})"),
            );
            let status =
                StatusCode::from_u16(outcome.http_status_code()).unwrap_or(StatusCode::FORBIDDEN);
            return Err((
                status,
                Json(serde_json::json!({
                    "error": format!("{endpoint}: limit check failed for acc_id={acc_id}: {reason}")
                })),
            ));
        }
    }
    Ok(())
}

/// v1.4.90 P1-C helper：从 sorted acc_ids 派生确定性 session_id。
///
/// REST stateless 没有 in-process push 注册表（MCP 有 `Arc<Mutex<HashMap<>>>`），
/// session_id 由 sorted acc_ids hash 派生，**输入相同 → 输出相同**。
/// 用户存下 session_id 之后通过 `/api/unsub-acc-push` 用相同 acc_ids
/// 派生出相同 session_id 即可撤销。
///
/// 注：这不是为了"防 hallucination"或"鉴权"，仅作为输出 token 让用户
/// 体感对齐 MCP `futu_sub_acc_push` 的 `{session_id, unsub_hint}` 形状。
fn derive_sub_acc_push_session_id(acc_ids: &[u64]) -> String {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    let mut sorted = acc_ids.to_vec();
    sorted.sort_unstable();
    sorted.dedup();
    let mut hasher = DefaultHasher::new();
    sorted.hash(&mut hasher);
    let hash = hasher.finish();
    format!("rest-sub-{hash:016x}")
}

/// codex 0522 F2 v1.4.106: caller scope invariant tests for sub/unsub helpers.
///
/// 验证 `check_per_acc_rate_for_caller` 给定同一 `KeyRecord` 的两个不同
/// endpoint label (sub / unsub) 看到同一份 allowed set。即使 helper
/// 被加新的 endpoint 调用，`KeyRecord.allowed_acc_ids` 仍是 single source
/// of truth，helper 不能引入 per-endpoint 漂移。
#[cfg(test)]
mod tests_v1_4_106_caller_ctx_invariant;

#[cfg(test)]
mod acc_id_list_tests;

#[cfg(test)]
mod tests_v1_4_90_p1_c_sub_acc_push;