futu_grpc/

server.rs

//! gRPC 服务实现
//!
//! FutuOpenD 服务通过通用的 proto_id + body 方式，
//! 将所有请求转发到现有的 RequestRouter。
//! 支持流式推送：行情、交易、广播事件通过 SubscribePush 接口推送给客户端。
//!
//! ## v1.4.106 codex 0517 ζ25-redo F2: stateful QOT stable identity
//!
//! gRPC `request()` 与 `subscribe_push()` 共享同一身份派生函数
//! [`auth::derive_grpc_conn_id`]：从 `(Bearer token, optional grpc-session-id
//! metadata)` 派生 deterministic stable conn_id（在
//! [`auth::GRPC_STABLE_CONN_NAMESPACE`] 即 bit 62 namespace 内）。
//!
//! 这让同一 caller 的连续 RPC 命中同一 SubscriptionManager / cache 状态：
//! - subscribe → unsubscribe / get_sub_info / query_subscription 全对齐
//! - quote cache 命中（per-conn cache 不再每次 RPC miss）
//! - QOT push fanout per-conn filter 能找到 caller
//!
//! 不同 Bearer / 不同 session_id → 自然隔离（caller 之间互不影响）。
//! Legacy mode（KeyStore 未配置 / Bearer 缺失）→ 共享一个固定 conn_id（
//! 全 legacy caller 共享同 sub state，对齐"无鉴权配置 = 单租户"语义）。
//!
//! 历史：v1.4.105 之前用自增 `conn_id_counter`，每次 RPC 拿新 ID，
//! 等价 REST v1.4.90 P0-B 之前的 quota 永久泄漏 bug。F2 修复对齐
//! `REST_SHARED_CONN` 设计哲学，不同点是 gRPC 按 caller 隔离（REST 共享）。

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

use bytes::Bytes;
use prost::Message as _;
use tokio::sync::{broadcast, mpsc};
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};

use futu_auth::{KeyStore, RuntimeCounters, Scope};
use futu_codec::header::ProtoFmtType;
use futu_server::conn::IncomingRequest;
use futu_server::router::RequestRouter;

use crate::auth::{
    derive_grpc_conn_id, extract_grpc_idempotency_key, extract_grpc_session_id, extract_grpc_token,
    grpc_audit_context, grpc_status_for,
};
use crate::proto::futu_open_d_server::FutuOpenD;
use crate::proto::{FutuRequest, FutuResponse, PushEvent, SubscribePushRequest};
use futu_auth_pipeline::{
    AuthDecision, AuthEnvelope, Credential, Endpoint, FilterRegistry, PushEventCtx, SurfaceId,
    authenticate_request,
};

mod generated_exposure {
    include!(concat!(env!("OUT_DIR"), "/generated_grpc_exposure.rs"));
}

mod push;
mod startup;
pub use push::GrpcPushBroadcaster;
pub use startup::{
    build_service_with_auth, start, start_with_auth, start_with_auth_until_shutdown,
    start_with_auth_until_shutdown_with_listener_events,
};

/// Explicit gRPC message boundary.
///
/// Native FTAPI frames are capped at 12 MiB; keeping gRPC at the same ceiling
/// prevents relying on tonic defaults while preserving protocol-sized payloads.
pub const GRPC_MAX_MESSAGE_SIZE_BYTES: usize = 12 * 1024 * 1024;

/// gRPC 服务实现
pub struct FutuGrpcService {
    router: Arc<RequestRouter>,
    push_broadcaster: Arc<GrpcPushBroadcaster>,
    key_store: Arc<KeyStore>,
    counters: Arc<RuntimeCounters>,
    /// v1.4.104: response filter 注册中心 (proto 2001 = AccListFilter).
    /// 加新 filter 在 `FilterRegistry::install_defaults` 注册一次, 4 surface
    /// 自动生效.
    filter_registry: Arc<FilterRegistry>,
    /// v1.4.106 codex 0517 ζ25-redo F2: gRPC `conn_id_counter` 已删除 —
    /// 自增 conn_id 让同一 caller 连续 RPC 拿不到同 sub state, 等价 REST
    /// v1.4.90 P0-B 之前的 quota 永久泄漏 bug. 现在改用
    /// [`auth::derive_grpc_conn_id`] 从 `(bearer, session_id)` 派生
    /// deterministic stable conn_id, 同 caller 连续 RPC 命中同一 sub state.
    serial_counter: AtomicU32,
}

impl FutuGrpcService {
    pub fn new(router: Arc<RequestRouter>, push_broadcaster: Arc<GrpcPushBroadcaster>) -> Self {
        Self::with_auth(
            router,
            push_broadcaster,
            Arc::new(KeyStore::empty()),
            Arc::new(RuntimeCounters::new()),
        )
    }

    /// 完整构造：同时接 key_store + counters（v1.0 推荐入口）
    ///
    /// `counters` 应由 main 全进程共享：REST / gRPC / MCP 共用一个实例才能保证
    /// rate limit / 日累计跨接口一致
    pub fn with_auth(
        router: Arc<RequestRouter>,
        push_broadcaster: Arc<GrpcPushBroadcaster>,
        key_store: Arc<KeyStore>,
        counters: Arc<RuntimeCounters>,
    ) -> Self {
        Self {
            router,
            push_broadcaster,
            key_store,
            counters,
            filter_registry: Arc::new(FilterRegistry::with_defaults()),
            serial_counter: AtomicU32::new(1),
        }
    }

    fn next_serial(&self) -> u32 {
        self.serial_counter.fetch_add(1, Ordering::Relaxed)
    }
}

#[tonic::async_trait]
impl FutuOpenD for FutuGrpcService {
    /// 通用请求-响应
    async fn request(
        &self,
        request: Request<FutuRequest>,
    ) -> Result<Response<FutuResponse>, Status> {
        // v1.4.104: 走 futu_auth_pipeline::authenticate_request 单一函数,
        // 不再 inline authenticate / check_scope / rate gate / body-aware /
        // audit. surface adapter 极薄 — 仅 transport extract + RejectKind →
        // Status 翻译.

        let proto_id = request.get_ref().proto_id;
        let caller_is_loopback = request
            .remote_addr()
            .is_some_and(|addr| addr.ip().is_loopback());
        if proto_id == 0 {
            return Err(Status::invalid_argument("proto_id is required"));
        }
        // v1.4.106 codex 0532 F3 (P2): daemon-internal proto_id (高位
        // 0x8000_0000 bit) 绝不应从 gRPC 公开 surface 进入 — 仅 REST handler
        // 内部合成给 router. 显式 reject + audit, 防探测 daemon 内部 routing.
        if futu_auth::is_internal_proto_id(proto_id) {
            tracing::warn!(
                proto_id,
                "rejecting daemon-internal proto_id at gRPC public surface (audit 0532 F3)"
            );
            return Err(Status::permission_denied(
                "daemon-internal proto_id not allowed on public surface",
            ));
        }

        // 1. extract token + session (owned String 避免 borrow vs move 冲突).
        // v1.4.106 codex 0517 ζ25-redo F2: 同时拿 grpc-session-id metadata,
        // 派生 stateful QOT stable conn_id (derive_grpc_conn_id).
        let token = extract_grpc_token(&request);
        let session_id = extract_grpc_session_id(&request);
        let idempotency_key = extract_grpc_idempotency_key(&request);
        let stable_conn_id = derive_grpc_conn_id(token.as_deref(), session_id.as_deref());
        let audit_ctx = grpc_audit_context(&request, stable_conn_id);
        // The manifest-generated match is the only public gRPC exposure registry.
        // Unknown proto IDs cannot bypass EndpointSpec into auth/router.
        let spec = match generated_exposure::generated_grpc_exposure(proto_id) {
            Ok(spec) => spec,
            Err(generated_exposure::GeneratedGrpcReject::NotExposed { endpoint, reason }) => {
                tracing::warn!(
                    proto_id,
                    endpoint,
                    reason,
                    "rejecting gRPC request for endpoint not exposed to gRPC"
                );
                return Err(Status::permission_denied(reason));
            }
            Err(generated_exposure::GeneratedGrpcReject::Unknown) => {
                tracing::warn!(
                    proto_id,
                    "rejecting gRPC request for proto_id not declared in the surface manifest"
                );
                return Err(Status::invalid_argument(
                    "proto_id is not declared in the surface manifest",
                ));
            }
        };
        let endpoint_name = spec.canonical_name;
        tracing::debug!(
            proto_id,
            endpoint = endpoint_name,
            "gRPC request dispatch (Layer 2 spec lookup)"
        );
        let req_inner = request.into_inner();
        let request_body = normalize_grpc_history_query_body(proto_id, req_inner.body)?;
        let needed_scope = if spec.runtime.side_effects == futu_surface_spec::SideEffectKind::Write
            && matches!(spec.runtime.scope, Scope::TradeReal | Scope::TradeSimulate)
        {
            futu_auth_pipeline::body_aware::trade_write_scope_from_body(proto_id, &request_body)
                .or(Some(spec.runtime.scope))
        } else {
            Some(spec.runtime.scope)
        };

        // 2. 构 credential + envelope, 调 pipeline
        let credential = match token.as_deref() {
            Some(t) => Credential::Bearer(t),
            None => Credential::None,
        };
        let env = AuthEnvelope {
            surface: SurfaceId::Grpc,
            endpoint: Endpoint::Proto(proto_id),
            needed_scope,
            credential,
            proto_id: Some(proto_id),
            body: &request_body,
            explicit_acc_id: None,
            explicit_ctx: None,
            commit_rate: true, // gRPC middleware 层 commit rate (trade:real)
            audit_emit: true,
        };
        let (allowed_for_filter, caller_rec) =
            match futu_auth::audit::with_context(audit_ctx.clone(), || {
                authenticate_request(&self.key_store, &self.counters, env)
            }) {
                AuthDecision::Reject { kind, reason, .. } => {
                    return Err(grpc_status_for(kind, reason));
                }
                AuthDecision::Allow {
                    allowed_acc_ids,
                    rec,
                    ..
                } => (allowed_acc_ids, rec),
            };

        // 3. dispatch + response filter
        // v1.4.105 D2 T-A1 fix: caller_allowed_acc_ids 从 pipeline allow decision
        // 真填进 IncomingRequest, 让 dispatch handler (e.g. SubAccPushHandler)
        // 端 enforce per-acc whitelist defense-in-depth.
        // codex 0522 F1 v1.4.106: 同步填 caller_key_id 让 cross-surface handler
        // 能识别 gRPC caller.
        let caller_allowed = allowed_for_filter
            .as_ref()
            .map(|s| std::sync::Arc::new(s.clone()));
        let caller_key_id = caller_rec.as_ref().map(|r| r.id.clone());
        let caller_has_auth_setup_scope = caller_rec
            .as_ref()
            .is_some_and(|record| record.scopes.contains(&Scope::AuthSetup));
        let serial_no = self.next_serial();
        let body = futu_server::trade_packet_id::fill_omitted_trade_packet_id_bytes(
            proto_id,
            request_body,
            stable_conn_id,
            serial_no,
        )
        .map_err(|e| Status::internal(format!("failed to prepare trade PacketID: {e}")))?;
        let incoming = IncomingRequest::builder(
            stable_conn_id,
            proto_id,
            serial_no,
            ProtoFmtType::Protobuf,
            Bytes::from(body),
        )
        .with_idempotency_key(idempotency_key)
        .with_caller_scope(caller_allowed, caller_key_id)
        .with_auth_setup_admission(
            caller_has_auth_setup_scope,
            caller_is_loopback,
            !self.key_store.is_configured(),
        )
        .build();

        match self.router.dispatch(incoming.conn_id, &incoming).await {
            Some(resp_bytes) => {
                // v1.4.104: 用 FilterRegistry (单一注册表) 替代 inline
                // filter_acc_list_response. 加新 filter (e.g. cash-flow) 只
                // 在 registry 注册一次, 4 surface 自动生效.
                let filtered_body =
                    self.filter_registry
                        .apply(proto_id, resp_bytes, allowed_for_filter.as_ref());
                Ok(Response::new(FutuResponse {
                    ret_type: 0,
                    ret_msg: String::new(),
                    proto_id,
                    body: filtered_body,
                }))
            }
            None => Ok(Response::new(FutuResponse {
                ret_type: -1,
                ret_msg: "handler returned no response".to_string(),
                proto_id,
                body: Vec::new(),
            })),
        }
    }

    type SubscribePushStream = ReceiverStream<Result<PushEvent, Status>>;

    /// 流式推送：客户端建立连接后持续接收行情、交易、广播推送
    ///
    /// v1.1：按订阅 key 的 scope 过滤推送 —— `qot:read`-only 的 key 不会收到
    /// `trade` 类（账户交易回报），对齐 REST `/ws` v0.9.0 加的 push filter。
    ///
    /// ## v1.4.104 阶段 7-2: pipeline 委托
    ///
    /// **handshake**: pipeline 调一次 with `Endpoint::Event("subscribe_push")` +
    /// `needed_scope=Some(qot:read)`. Allow → 拿 rec snapshot. Reject (Bearer
    /// invalid / expired / no qot:read) → translate to gRPC Status.
    ///
    /// **per-event filter**: stream 内每 event 调一次 pipeline with
    /// `Credential::PreVerified(rec)` + `Endpoint::Event(event_type)` +
    /// `needed_scope=Some(scope_for_event(...))` + `explicit_acc_id` (trade event
    /// 给 event.acc_id, 其他不传) + `audit_emit=false` (避免每 event 一条 audit
    /// 把日志冲爆) + `commit_rate=false` (push 不计 rate). Reject → silent drop +
    /// `metrics::bump_ws_filtered`. Allow → forward.
    ///
    /// `SubscribePush` 是 quote-first mixed push stream: handshake 必须有
    /// qot:read；caller 若还持有 acc:read, stream 内 trade event 再由 per-event
    /// scope match + trade acc_id whitelist 放行。acc:read-only key 不能打开
    /// mixed stream。
    async fn subscribe_push(
        &self,
        request: Request<SubscribePushRequest>,
    ) -> Result<Response<Self::SubscribePushStream>, Status> {
        if !grpc_push_ready(&self.router.startup_readiness()) {
            return Err(Status::unavailable(
                "gateway authentication is not ready for push subscriptions",
            ));
        }
        // v1.4.106 codex 1125 F6 [P2]: 显式 notify subscription opt-in.
        // 对齐 C++ raw TCP `IsConnSubRecvNotify` 语义 (broadcast push 必须显式
        // sub 才下发).
        let notify_subscribe = request.get_ref().notify_subscribe;

        // ── handshake: pipeline 调一次拿 caller-key + audit 一次 ──────────────────
        //
        // Bearer 提取 / verify / expiry / qot:read scope 都走 pipeline
        // (单一 source)。trade push 不是单独的 acc-only stream；caller 需要
        // qot:read 打开 stream, 再靠 per-event acc:read 过滤接收 trade event.
        //
        // v1.4.106 codex 0517 ζ25-redo F2: 与 `request()` 共享 stable conn_id.
        // 这让 ops/log 能把 push stream 和同 caller 的 subscribe request 串起来:
        // grep `conn_id=<stable>` 看到 subscribe RPC + push event filter 同 id.
        let token = extract_grpc_token(&request);
        let session_id = extract_grpc_session_id(&request);
        let stable_conn_id = derive_grpc_conn_id(token.as_deref(), session_id.as_deref());
        let audit_ctx = grpc_audit_context(&request, stable_conn_id);
        let credential = match token.as_deref() {
            Some(t) => Credential::Bearer(t),
            None => Credential::None,
        };
        let env = AuthEnvelope {
            surface: SurfaceId::Grpc,
            endpoint: Endpoint::Event("subscribe_push"),
            needed_scope: Some(Scope::QotRead),
            credential,
            proto_id: None,
            body: &[],
            explicit_acc_id: None,
            explicit_ctx: None,
            commit_rate: false,
            audit_emit: true, // handshake 一次 audit
        };
        let rec_snapshot = match futu_auth::audit::with_context(audit_ctx.clone(), || {
            authenticate_request(&self.key_store, &self.counters, env)
        }) {
            AuthDecision::Reject { kind, reason, .. } => {
                return Err(grpc_status_for(kind, reason));
            }
            AuthDecision::Allow { rec, .. } => rec,
        };

        let (scopes, key_id, allowed_acc_ids) = match rec_snapshot.as_ref() {
            Some(rec) => (
                rec.scopes.clone(),
                rec.id.clone(),
                rec.allowed_acc_ids.clone(),
            ),
            None => (
                // legacy: 全 scope + 无 acc 限制
                [
                    Scope::QotRead,
                    Scope::AccRead,
                    Scope::TradeSimulate,
                    Scope::TradeReal,
                ]
                .into_iter()
                .collect::<std::collections::HashSet<Scope>>(),
                "<none>".to_string(),
                None,
            ),
        };

        let (tx, rx) = mpsc::channel(256);
        let mut push_rx = self.push_broadcaster.subscribe();

        tracing::info!(
            key_id = %key_id,
            // v1.4.106 codex 0517 ζ25-redo F2: stable conn_id 让 subscribe RPC
            // 和 push stream 在 log 里可关联.
            conn_id = stable_conn_id,
            scopes = ?scopes,
            allowed_acc_ids = ?allowed_acc_ids.as_ref().map(|s| s.len()),
            "gRPC client subscribed to push events",
        );

        // ── per-event filter: pipeline 二次调 (audit_emit=false 防日志冲爆) ───────
        //
        // 每 event 调一次 pipeline:
        //   - Credential::PreVerified(rec) 复用 handshake 已 verify 的 rec
        //   - Endpoint::Event(event_type) 给 audit label (虽然 audit_emit=false)
        //   - needed_scope = scope_for_event(event_type)
        //   - explicit_acc_id = trade event 给 event.acc_id, 其他 None
        //
        // legacy mode (rec_snapshot=None) 不调 pipeline (没必要, 全放行).
        let key_store_arc = self.key_store.clone();
        let counters_arc = self.counters.clone();
        // v1.4.105 D3 (Phase 4): clone filter_registry for per-event Layer 3
        // (allowed_markets) check + Layer 1/2 reuse.
        let filter_registry_arc = self.filter_registry.clone();
        let push_readiness = self.router.startup_readiness();
        tokio::spawn(async move {
            loop {
                match push_rx.recv().await {
                    Ok(event) => {
                        if !grpc_push_ready(&push_readiness) {
                            futu_auth::metrics::bump_ws_filtered("startup_not_ready", &key_id);
                            continue;
                        }
                        // legacy fast-path: rec=None → 全放行, scope check / acc_id
                        // whitelist 都不需要走 pipeline.
                        let allow_event = if let Some(rec) = rec_snapshot.as_ref() {
                            let needed = scope_for_event(&event.event_type);
                            let explicit_acc_id = if event.event_type == "trade" {
                                Some(event.acc_id)
                            } else {
                                None
                            };
                            let env = AuthEnvelope {
                                surface: SurfaceId::Grpc,
                                endpoint: Endpoint::Event(&event.event_type),
                                needed_scope: Some(needed),
                                credential: Credential::PreVerified(rec.clone()),
                                proto_id: None,
                                body: &[],
                                explicit_acc_id,
                                explicit_ctx: None,
                                commit_rate: false, // push 不计 rate
                                audit_emit: false,  // per-event 不 audit, 避免冲爆
                            };
                            matches!(
                                authenticate_request(&key_store_arc, &counters_arc, env),
                                AuthDecision::Allow { .. }
                            )
                        } else {
                            true
                        };

                        if !allow_event {
                            // metric label 区分 trade acc_id reject vs scope reject
                            // (近似旧行为: trade 类 reject 标 "trade_acc_id" 仅当
                            // acc_id whitelist 非空; 否则按 event_type)
                            let label = if event.event_type == "trade"
                                && rec_snapshot
                                    .as_ref()
                                    .and_then(|r| r.allowed_acc_ids.as_ref())
                                    .is_some_and(|s| !s.is_empty())
                            {
                                "trade_acc_id"
                            } else {
                                event.event_type.as_str()
                            };
                            futu_auth::metrics::bump_ws_filtered(label, &key_id);
                            continue;
                        }

                        // v1.4.105 D3 (Phase 4): FilterRegistry::should_drop_event
                        // Layer 3 (allowed_markets). Layer 1 (allowed_acc_ids) 已在
                        // 上面 pipeline body-aware 跑过, 此处串行 Layer 3 双重
                        // 防御 + 未来 Layer 扩展自动覆盖.
                        // ⚠️ UNVERIFIED — pending real-machine verify (HK+US 双
                        // 账户跨 market 推送). 这里是 backend-semantic 风险,
                        // 只能靠真机跨 market trade push 验证后再升 confidence.
                        //
                        // v1.4.105 T-B3: trd_market 改读 PushEvent.trd_market 字段
                        // (PushDispatcher 端一次 decode), 不再各 surface 独立
                        // decode body. 空字符串 = 非 trade event / decode 失败 /
                        // market 未知 → 转 None (Layer 3 不 trigger drop, 向后
                        // 兼容).
                        let event_trd_market =
                            if event.event_type == "trade" && !event.trd_market.is_empty() {
                                Some(event.trd_market.as_str())
                            } else {
                                None
                            };
                        let allowed_markets_for_filter = rec_snapshot
                            .as_ref()
                            .and_then(|r| r.allowed_markets.as_ref());
                        let push_ctx = PushEventCtx {
                            event_type: &event.event_type,
                            event_acc: if event.event_type == "trade" {
                                Some(event.acc_id)
                            } else {
                                None
                            },
                            // Layer 1 已 pipeline 跑, 此处传 None 防双重 reject (LE 之间 OR 短路)
                            allowed_acc_ids: None,
                            // gRPC 没 sub_state (REST 特有)
                            sub_state: None,
                            // Layer 3 — 新加
                            event_trd_market,
                            allowed_markets: allowed_markets_for_filter,
                        };
                        if filter_registry_arc.should_drop_event(&push_ctx) {
                            futu_auth::metrics::bump_ws_filtered("trade_market", &key_id);
                            continue;
                        }

                        // v1.4.106 codex 1125 F6 [P2]: notify_subscribe gate.
                        // 对齐 C++ `IsConnSubRecvNotify` (raw TCP). broadcast notify
                        // 类 push (e.g. price reminder) 必须 client 显式 sub 才下发.
                        if event.event_type == "notify" && !notify_subscribe {
                            futu_auth::metrics::bump_ws_filtered("notify_unsub", &key_id);
                            continue;
                        }

                        if tx.send(Ok(event)).await.is_err() {
                            break; // 客户端断开
                        }
                    }
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        tracing::warn!(skipped = n, "gRPC push client lagged, skipped events");
                        // 继续接收，不断开
                    }
                    Err(broadcast::error::RecvError::Closed) => {
                        break; // 广播器关闭
                    }
                }
            }
            tracing::info!("gRPC push stream ended");
        });

        Ok(Response::new(ReceiverStream::new(rx)))
    }
}

fn normalize_grpc_history_query_body(proto_id: u32, body: Vec<u8>) -> Result<Vec<u8>, Status> {
    match proto_id {
        futu_core::proto_id::TRD_GET_HISTORY_ORDER_LIST => {
            normalize_history_order_list_filter_market(body)
        }
        futu_core::proto_id::TRD_GET_HISTORY_ORDER_FILL_LIST => {
            normalize_history_order_fill_list_filter_market(body)
        }
        _ => Ok(body),
    }
}

fn normalize_history_order_list_filter_market(body: Vec<u8>) -> Result<Vec<u8>, Status> {
    let Ok(mut req) = futu_proto::trd_get_history_order_list::Request::decode(body.as_slice())
    else {
        return Ok(body);
    };
    if default_history_filter_market(
        req.c2s.header.trd_market,
        &mut req.c2s.filter_conditions.filter_market,
    ) {
        return Ok(req.encode_to_vec());
    }
    Ok(body)
}

fn normalize_history_order_fill_list_filter_market(body: Vec<u8>) -> Result<Vec<u8>, Status> {
    let Ok(mut req) = futu_proto::trd_get_history_order_fill_list::Request::decode(body.as_slice())
    else {
        return Ok(body);
    };
    if default_history_filter_market(
        req.c2s.header.trd_market,
        &mut req.c2s.filter_conditions.filter_market,
    ) {
        return Ok(req.encode_to_vec());
    }
    Ok(body)
}

fn default_history_filter_market(header_market: i32, filter_market: &mut Option<i32>) -> bool {
    // Ref: FutuOpenD/Src/APIServer/Business/Trade/APIServer_Trd_GetHistoryOrderList.cpp:48-55
    // and _APIServer_Trd_Comm.cpp:3233-3247. C++ uses filterMarket both for
    // time parsing and response market filtering; missing filterMarket means
    // Unknown and passes every market. CLI/MCP/REST high-level paths already
    // default it from header.trd_market, so generic gRPC must do the same.
    if filter_market.is_none() && header_market != 0 {
        *filter_market = Some(header_market);
        return true;
    }
    false
}

// v1.4.104: `grpc_handler_full_check` 已删除 — 功能搬到
// `futu_auth_pipeline::pipeline::authenticate_request` 内的 body-aware step.
// gRPC `request()` 调一次 pipeline 即覆盖所有 (caller-key + scope + body-aware
// + audit + rate). 4 surface 共用同一函数, 不再有 hand-copy diverge.

/// gRPC PushEvent 的 event_type → 客户端必须持有的 scope (与 REST `/ws` 对齐).
/// 用于 SubscribePush stream 内 per-event filter.
fn scope_for_event(event_type: &str) -> Scope {
    match event_type {
        "trade" => Scope::AccRead, // 账户交易回报
        _ => Scope::QotRead,       // quote / notify / 未知都按行情门槛
    }
}

fn grpc_push_ready(readiness: &futu_server::identity::StartupReadiness) -> bool {
    readiness.snapshot().state == futu_server::identity::StartupState::Ready
}

// v1.4.105 D3 (Phase 4) T-B3: 旧 `extract_trd_market_from_trade_event` 已搬到
// `futu_server::push::extract_trd_market_from_trade_body` + PushDispatcher 端
// 一次 decode 后透传给 PushEvent.trd_market 字段, 4 surface (gRPC / REST WS /
// MCP) 共用. gRPC subscribe_push 现读 event.trd_market 字段不再 decode body.
//
// 这里若想要静态 import 提示, 可保留:
//
//     use futu_server::push::extract_trd_market_from_trade_body;
//
// 但目前 server.rs 不再调用, 全凭 PushEvent 字段透传, 故不 import.

#[cfg(test)]
mod tests;