futu_mcp/

state.rs

//! 共享状态：网关连接 + 订阅状态 + 授权

use std::collections::HashMap;
use std::sync::Arc;

use anyhow::{Context, Result, anyhow};
use futu_auth::{KeyRecord, KeyStore, RuntimeCounters};
use futu_net::client::{ClientConfig, FutuClient, ReconnectingClient};
use futu_net::reconnect::ReconnectPolicy;
use futu_qot::types::Security;
use rmcp::{RoleServer, service::Peer};
use tokio::sync::Mutex;

mod push_filter;
mod push_subscribers;
#[cfg(test)]
mod tests;

use push_filter::{TradePushDecode, classify_trade_push, trd_market_int_to_str};
#[cfg(test)]
use push_filter::{
    extract_acc_id_and_market_from_push, is_trade_push_proto_id, subscriber_should_receive,
    subscriber_should_receive_with_market, subscriber_visible_to_caller,
};
use push_subscribers::PushSubscriber;

const MCP_CONNECT_TOTAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
const MCP_CONNECT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(200);

/// v1.4.38 Phase 5 helper: bytes → base64 (用于 push body 安全包进 JSON)
fn base64_encode_bytes(bytes: &[u8]) -> String {
    use base64::Engine as _;
    base64::engine::general_purpose::STANDARD.encode(bytes)
}

struct PushDelivery {
    peer: Peer<RoleServer>,
    data: serde_json::Value,
    session_id: String,
    owner_key_id: Option<String>,
    proto_id: u32,
}

/// MCP server 运行时状态
#[derive(Clone)]
pub struct ServerState {
    /// [`Inner`] 共享可变状态（gateway 地址 + 懒加载的 [`FutuClient`]）
    inner: Arc<Mutex<Inner>>,
    /// 是否启用交易写工具（place/modify/cancel）。默认 false。旧开关，仅当
    /// `key_store.is_configured() == false` 时生效。
    enable_trading: bool,
    /// 是否允许对 real 环境下单。默认 false。旧开关，同上。
    allow_real_trading: bool,
    /// keys.json 加载的 KeyStore。`is_configured()` 为 true 时走 scope 授权模式。
    key_store: Arc<KeyStore>,
    /// 调用方传入的 API Key 对应的记录；None 表示未提供 key。
    authed_key: Option<Arc<KeyRecord>>,
    /// 交易密码所属登录账号。用于 `futu_unlock_trade` 从账号级 keychain
    /// `trade-password.<login-account>` 读取密码；None 时走 legacy/global/env 兼容路径。
    trade_pwd_account: Option<String>,
    /// 限额运行时（日累计计数器）
    counters: Arc<RuntimeCounters>,
    /// v1.4.38 Phase 5: MCP push 订阅者注册表（session_uuid → subscriber）。
    /// `futu_sub_acc_push` 工具在 HTTP 模式下调用时注册当前 session，daemon
    /// push 到 MCP 后按 acc_id filter 向注册的 peer 发
    /// `notify_logging_message`（server-initiated notification）。
    push_subscribers: Arc<Mutex<HashMap<String, PushSubscriber>>>,
}

/// ServerState 内部可变部分，加锁存放 gateway 地址 + 懒加载的 [`FutuClient`]。
struct Inner {
    /// 网关 TCP 地址（如 `127.0.0.1:11111`）
    gateway: String,
    /// 懒加载的底层连接；首次调用 [`ServerState::client`] 时建立，后续复用
    client: Option<Arc<FutuClient>>,
}

impl ServerState {
    /// 创建默认 state：`enable_trading=false` / `allow_real_trading=false` /
    /// 空 [`KeyStore`] / 无 authed_key。使用 `with_*` 链式方法注入额外能力。
    pub fn new(gateway: String) -> Self {
        Self {
            inner: Arc::new(Mutex::new(Inner {
                gateway,
                client: None,
            })),
            enable_trading: false,
            allow_real_trading: false,
            key_store: Arc::new(KeyStore::empty()),
            authed_key: None,
            trade_pwd_account: None,
            counters: Arc::new(RuntimeCounters::new()),
            push_subscribers: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// 启用交易写工具（构造器式链式设置）
    pub fn with_trading(mut self, enable_trading: bool, allow_real_trading: bool) -> Self {
        self.enable_trading = enable_trading;
        self.allow_real_trading = allow_real_trading;
        self
    }

    /// 设置 KeyStore（新授权模式）
    pub fn with_key_store(mut self, store: Arc<KeyStore>) -> Self {
        self.key_store = store;
        self
    }

    /// 设置已通过验证的 API Key 记录
    pub fn with_authed_key(mut self, key: Option<Arc<KeyRecord>>) -> Self {
        self.authed_key = key;
        self
    }

    /// 设置交易密码所属登录账号（MCP 只连 gateway，本身无法可靠推断 daemon
    /// 的 login account；由 CLI/env/config 显式注入）。
    pub fn with_trade_pwd_account(mut self, account: Option<String>) -> Self {
        self.trade_pwd_account = account;
        self
    }

    /// 是否启用了 scope 授权模式
    pub fn is_scope_mode(&self) -> bool {
        self.key_store.is_configured()
    }

    /// 交易写工具开关（legacy mode）。
    pub fn enable_trading(&self) -> bool {
        self.enable_trading
    }

    /// real 环境交易写工具开关（legacy mode）。
    pub fn allow_real_trading(&self) -> bool {
        self.allow_real_trading
    }

    /// 当前 MCP API key store。返回共享引用，避免调用方替换 runtime storage。
    pub fn key_store(&self) -> &Arc<KeyStore> {
        &self.key_store
    }

    /// startup 阶段验证过的 key 快照；调用方需要 fresh record 时仍应按 id 回查 key store。
    pub fn authed_key(&self) -> Option<Arc<KeyRecord>> {
        self.authed_key.clone()
    }

    /// 交易密码所属登录账号。
    pub fn trade_pwd_account(&self) -> Option<&str> {
        self.trade_pwd_account.as_deref()
    }

    /// 限额运行时计数器。返回共享引用，避免调用方替换 runtime storage。
    pub fn counters(&self) -> &Arc<RuntimeCounters> {
        &self.counters
    }

    /// 当前配置的 gateway 地址。
    pub async fn gateway(&self) -> String {
        self.inner.lock().await.gateway.clone()
    }

    /// 获取（或懒加载）网关客户端
    pub async fn client(&self) -> Result<Arc<FutuClient>> {
        let gateway = {
            let guard = self.inner.lock().await;
            if let Some(c) = &guard.client {
                return Ok(c.clone());
            }
            guard.gateway.clone()
        };

        let config = ClientConfig {
            addr: gateway.clone(),
            client_ver: env!("CARGO_PKG_VERSION").to_string(),
            client_id: "futu-mcp".to_string(),
            recv_notify: false,
            rsa_key: None,
        };
        let policy =
            ReconnectPolicy::new(MCP_CONNECT_RETRY_DELAY, MCP_CONNECT_RETRY_DELAY, Some(1));
        let mut reconnector = ReconnectingClient::new(config).with_policy(policy);
        let connect_result =
            tokio::time::timeout(MCP_CONNECT_TOTAL_TIMEOUT, reconnector.connect()).await;
        let (client, mut push_rx, _info) = match connect_result {
            Ok(result) => {
                result.with_context(|| format!("connect to futu gateway at {gateway}"))?
            }
            Err(_) => {
                return Err(anyhow!(
                    "connect to futu gateway at {gateway} timed out after {}s",
                    MCP_CONNECT_TOTAL_TIMEOUT.as_secs()
                ));
            }
        };

        let arc = Arc::new(client);
        {
            let mut guard = self.inner.lock().await;
            if let Some(c) = &guard.client {
                return Ok(c.clone());
            }
            guard.client = Some(arc.clone());
        }

        // v1.4.38 Phase 5 (100%): 按 acc_ids 过滤的 push broadcast
        //
        // 流程：
        // 1. push_rx 收 daemon 转发的 push
        // 2. 对 TRD_UPDATE_ORDER (2208) / TRD_UPDATE_ORDER_FILL (2218) 解包
        //    提取 acc_id
        // 3. 遍历订阅者，**只推给 acc_ids 匹配的**（或订阅者 acc_ids 空 = 不
        //    过滤，所有 acc 都收）
        // 4. 行情 push（QOT_UPDATE_*）无 acc_id 语义，广播给所有订阅者
        //
        // Per-session 独立 spawn notify，避免一个慢 session 阻塞其他
        let subs_for_push = Arc::downgrade(&self.push_subscribers);
        // v1.4.105 F5 fix (codex review C4 USER_ACK B): MCP push filter 改用
        // FilterRegistry::should_drop_event 单一注册中心 (跟 4 surface 一致),
        // 替代之前 inline subscriber_should_receive_with_market. 防 sibling-route
        // bypass — 加新 push event filter 维度只在 install_defaults 注册一次,
        // MCP 自动覆盖.
        let filter_registry =
            std::sync::Arc::new(futu_auth_pipeline::FilterRegistry::with_defaults());
        tokio::spawn(async move {
            while let Some(push) = push_rx.recv().await {
                let Some(subs_for_push) = subs_for_push.upgrade() else {
                    break;
                };
                let subscribers = {
                    let subs = subs_for_push.lock().await;
                    if subs.is_empty() {
                        Vec::new()
                    } else {
                        subs.iter()
                            .map(|(session_id, sub)| (session_id.clone(), sub.clone()))
                            .collect::<Vec<_>>()
                    }
                };
                if subscribers.is_empty() {
                    continue; // fast path: no listeners, drop
                }
                // v1.4.105 T-C2 + v1.4.106 codex 0932 F6/F7: classify push by proto_id
                // (set membership), 不再靠 body decode 成功推断. trade body decode
                // 失败现在归 TradePushDecode::DecodeFailed (event_type="trade",
                // 无 acc/market gate 信息) — restricted key 应 drop, unrestricted
                // 透传带 decode_status="failed".
                let decode_result = classify_trade_push(push.proto_id, &push.body);
                let (push_acc_id, push_trd_market, decode_status, event_type) = match &decode_result
                {
                    TradePushDecode::NotTrade => (None, None, "ok", "quote"),
                    TradePushDecode::Decoded { acc_id, trd_market } => {
                        (Some(*acc_id), Some(*trd_market), "ok", "trade")
                    }
                    TradePushDecode::DecodeFailed => (None, None, "failed", "trade"),
                };
                let push_trd_market_str = push_trd_market.map(trd_market_int_to_str);
                // v1.4.106 codex 0932 F7 [P3]: payload 加 event_type / trd_market /
                // decode_status — 让客户端不需要按 proto_id 自己 derive (4 surface 一致).
                // body_base64 后向兼容保留.
                let payload = serde_json::json!({
                    "kind": "futu_push",
                    "proto_id": push.proto_id,
                    "acc_id": push_acc_id,
                    "event_type": event_type,
                    "trd_market": push_trd_market_str,
                    "decode_status": decode_status,
                    "body_base64": base64_encode_bytes(&push.body),
                });
                let deliveries = {
                    let mut deliveries = Vec::with_capacity(subscribers.len());
                    for (session_id, sub) in subscribers.iter() {
                        // v1.4.106 codex 0932 F6 [P2]: trade decode-failed + restricted
                        // key (allowed_acc_ids 非 None) → DROP. 不能让 restricted key
                        // 看到无 acc gate 信息的 trade body 透传 (绕过 ACL).
                        // unrestricted key (allowed_acc_ids None / 空) 仍透传带 decode_status="failed".
                        if matches!(decode_result, TradePushDecode::DecodeFailed) {
                            let restricted = sub
                                .allowed_acc_ids_snapshot
                                .as_ref()
                                .map(|s| !s.is_empty())
                                .unwrap_or(false);
                            if restricted {
                                let key_id = sub.owner_key_id.as_deref().unwrap_or("<none>");
                                // 复用 cross-surface metric — reason="trade_decode_failed"
                                futu_auth::metrics::bump_ws_filtered("trade_decode_failed", key_id);
                                tracing::warn!(
                                    proto_id = push.proto_id,
                                    key_id,
                                    "v1.4.106 audit 0932 F6: trade push body decode failed; \
                                     dropped for restricted key (allowed_acc_ids set, \
                                     cannot ACL-gate body without acc_id)"
                                );
                                continue;
                            }
                            // unrestricted: fall through, broadcast 带 decode_status="failed"
                        }
                        // v1.4.105 F5 fix: 改用 FilterRegistry::should_drop_event.
                        // 行为对齐 4 surface — sub.acc_ids (MCP 显式订阅 list) 喂给
                        // ctx.sub_state (REST sub-acc-push state 同语义); sub.allowed_*
                        // _snapshot 喂给 ctx.allowed_* (caller key 限额).
                        //
                        // **行为微差** (与老 inline fn 一致, 不破老行为):
                        // - sub.acc_ids 空 = MCP 老语义"无限制订阅" → 传 None
                        //   (避免 REST sub_state 空集 tombstone 语义触发 drop-all)
                        // - sub.acc_ids 非空 → 传 Some(&sub.acc_ids), 跟 REST 一致
                        let sub_state_for_ctx = if sub.acc_ids.is_empty() {
                            None
                        } else {
                            Some(&sub.acc_ids)
                        };
                        let ctx = futu_auth_pipeline::PushEventCtx {
                            event_type,
                            event_acc: push_acc_id,
                            allowed_acc_ids: sub.allowed_acc_ids_snapshot.as_ref(),
                            sub_state: sub_state_for_ctx,
                            event_trd_market: push_trd_market_str,
                            allowed_markets: sub.allowed_markets_snapshot.as_ref(),
                        };
                        if filter_registry.should_drop_event(&ctx) {
                            // v1.4.105 T-C2 + F3 (codex review C4): bump filtered metric.
                            // **统一 label 命名** 跟 4 surface 一致 — gRPC subscribe_push
                            // / push_trd_acc / 都用 "trade_market" 标 Layer 3 (allowed_markets)
                            // 拒. 老 "push.trade" 命名是 surface-specific (T-C2 sole), 改成
                            // canonical "trade_market" 让跨 surface metrics jq aggregate 一致.
                            let key_id = sub.owner_key_id.as_deref().unwrap_or("<none>");
                            futu_auth::metrics::bump_ws_filtered("trade_market", key_id);
                            continue;
                        }
                        deliveries.push(PushDelivery {
                            peer: sub.peer.clone(),
                            data: payload.clone(),
                            session_id: session_id.clone(),
                            owner_key_id: sub.owner_key_id.clone(),
                            proto_id: push.proto_id,
                        });
                    }
                    deliveries
                };
                for delivery in deliveries {
                    tokio::spawn(async move {
                        let params = rmcp::model::LoggingMessageNotificationParam {
                            level: rmcp::model::LoggingLevel::Info,
                            logger: Some("futu_push".to_string()),
                            data: delivery.data,
                        };
                        if let Err(err) = delivery.peer.notify_logging_message(params).await {
                            tracing::warn!(
                                proto_id = delivery.proto_id,
                                session_id = delivery.session_id,
                                owner_key_id = delivery.owner_key_id.as_deref().unwrap_or("<none>"),
                                error = %err,
                                "mcp push notification send failed"
                            );
                        }
                    });
                }
            }
        });

        // v1.4.39 Phase 5 stale cleanup: 5 分钟跑一次，移除 registered_at > 4h
        // 的订阅者。避免长跑 daemon 累积陈旧 subscriber（客户端断开 /  rmcp
        // session gone 但没显式 unregister 的情况）。
        let subs_for_purge = Arc::downgrade(&self.push_subscribers);
        tokio::spawn(async move {
            use std::time::Duration;
            const PURGE_INTERVAL: Duration = Duration::from_secs(5 * 60);
            const MAX_AGE: Duration = Duration::from_secs(4 * 3600);
            let mut ticker = tokio::time::interval(PURGE_INTERVAL);
            ticker.tick().await; // skip the immediate first tick
            loop {
                ticker.tick().await;
                let Some(subs_for_purge) = subs_for_purge.upgrade() else {
                    break;
                };
                let now = std::time::Instant::now();
                let mut subs = subs_for_purge.lock().await;
                let before = subs.len();
                subs.retain(|_, sub| {
                    now.checked_duration_since(sub.registered_at)
                        .map(|age| age < MAX_AGE)
                        .unwrap_or(true)
                });
                let purged = before - subs.len();
                if purged > 0 {
                    tracing::info!(
                        purged,
                        remaining = subs.len(),
                        max_age_secs = MAX_AGE.as_secs(),
                        "v1.4.39 Phase 5: purged stale push subscribers (> 4h registered)"
                    );
                }
            }
        });

        Ok(arc)
    }
}

// ========== symbol 解析 ==========

pub fn parse_symbol(s: &str) -> Result<Security> {
    futu_qot::symbol::parse_symbol(s)
}

/// 格式化 Security 为 "MARKET.CODE"
pub fn format_symbol(sec: &Security) -> String {
    futu_qot::symbol::format_symbol(sec)
}

/// v1.4.90 P2-C: audit log Option<T> 序列化助手。
///
/// **背景**：之前 audit log 把 `Option<f64>` 用 `?req.price`（tracing 的 Debug
/// shorthand）记录，渲染成 JSON 字符串 `"Some(400.0)"` / `"None"`，下游 jq /
/// DuckDB 数值聚合炸（aggregator 期望 `400.0` number 或 `null`）。
///
/// **修法**：用 NaN sentinel 把 `Option<f64>` flatten 成 `f64`，tracing-subscriber
/// 的 JSON formatter 内部走 `serde_json::Value::from(f64::NAN)` →
/// `Number::from_f64(NaN) = None` → `Value::Null`。
/// 整数 / 字符串同理（i32 → f64 NaN sentinel；&str → "" 哨兵）。
///
/// 验证依据：
/// - `tracing_subscriber::fmt::format::json` line 501 `record_f64` 直接调
///   `serde_json::Value::from(value)`
/// - `serde_json::Value::from(f64)` impl: `Number::from_f64(f).map_or(Value::Null, Value::Number)`
pub mod audit_fmt {
    /// `Option<f64>` → `f64`（None → NaN）。tracing JSON 渲染 NaN 为 `null`。
    #[inline]
    pub fn opt_f64(v: Option<f64>) -> f64 {
        v.unwrap_or(f64::NAN)
    }

    /// `Option<i32>` → `f64`（None → NaN，Some(n) → n as f64）。
    /// i32 ≤ 2^31 < 2^52 mantissa，无精度损失。
    #[inline]
    pub fn opt_i32(v: Option<i32>) -> f64 {
        v.map(f64::from).unwrap_or(f64::NAN)
    }

    /// `Option<&str>` → `&str`（None → ""）。"" 哨兵在 audit 上下文里足以区分
    /// 不传 vs 传空（因为 Symbol / owner 等业务字段不会是空字符串）。
    #[inline]
    pub fn opt_str(v: Option<&str>) -> &str {
        v.unwrap_or("")
    }
}