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 tokio::sync::{broadcast, mpsc, watch};
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::push::ExternalPushSink;
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, FutuOpenDServer};
use crate::proto::{FutuRequest, FutuResponse, PushEvent, SubscribePushRequest};
use futu_auth_pipeline::{
    AuthDecision, AuthEnvelope, Credential, Endpoint, FilterRegistry, PushEventCtx, RejectKind,
    SurfaceId, authenticate_request,
};

/// 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 推送广播器
///
/// 实现 `ExternalPushSink` trait，接收 PushDispatcher 的推送事件，
/// 通过 broadcast channel 分发给所有 SubscribePush 流式连接。
#[derive(Clone)]
pub struct GrpcPushBroadcaster {
    tx: broadcast::Sender<PushEvent>,
}

impl GrpcPushBroadcaster {
    pub fn new(capacity: usize) -> Self {
        let (tx, _) = broadcast::channel(capacity);
        Self { tx }
    }

    /// 创建接收端
    pub fn subscribe(&self) -> broadcast::Receiver<PushEvent> {
        self.tx.subscribe()
    }

    fn has_receivers(&self) -> bool {
        self.tx.receiver_count() > 0
    }

    fn send(&self, event: PushEvent) {
        if !self.has_receivers() {
            return;
        }
        let proto_id = event.proto_id;
        let event_type = event.event_type.clone();
        if self.tx.send(event).is_err() {
            tracing::debug!(
                proto_id,
                event_type,
                receiver_count = self.tx.receiver_count(),
                "grpc push broadcast send skipped"
            );
        }
    }
}

impl ExternalPushSink for GrpcPushBroadcaster {
    /// **v1.4.106 codex 1131 F4 [P1]**: `rehab_type` 透传到 gRPC PushEvent.
    /// gRPC 当前 broadcast 模型 (所有 qot:read subscriber 收到所有 quote
    /// event), per-conn (sec_key, sub_type, rehab_type) 三元 filter 是 raw TCP
    /// 专属. rehab_type 通过 PushEvent 让客户端可见.
    fn on_quote_push(
        &self,
        sec_key: &str,
        sub_type: i32,
        rehab_type: i32,
        proto_id: u32,
        body: &[u8],
    ) {
        if !self.has_receivers() {
            return;
        }
        self.send(PushEvent {
            proto_id,
            sec_key: sec_key.to_string(),
            sub_type,
            rehab_type,
            body: body.to_vec(),
            event_type: "quote".to_string(),
            acc_id: 0,
            trd_market: String::new(), // 行情推送无 trd_market
        });
    }

    fn on_broadcast_push(&self, proto_id: u32, body: &[u8]) {
        if !self.has_receivers() {
            return;
        }
        self.send(PushEvent {
            proto_id,
            sec_key: String::new(),
            sub_type: 0,
            rehab_type: 0,
            body: body.to_vec(),
            event_type: "notify".to_string(),
            acc_id: 0,
            trd_market: String::new(),
        });
    }

    /// v1.4.105 D3 (Phase 4) T-B3: trade push trd_market 由 PushDispatcher 一
    /// 次 decode 后透传, 不再各 sink 独立 decode body. 空 / unknown → 空
    /// 字符串 (PushEvent proto3 默认值, 老 client 解析兼容).
    fn on_trade_push(&self, acc_id: u64, proto_id: u32, body: &[u8], trd_market: Option<&str>) {
        if !self.has_receivers() {
            return;
        }
        self.send(PushEvent {
            proto_id,
            sec_key: String::new(),
            sub_type: 0,
            rehab_type: 0,
            body: body.to_vec(),
            event_type: "trade".to_string(),
            acc_id,
            trd_market: trd_market.unwrap_or("").to_string(),
        });
    }
}

/// 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;
        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);
        // v1.4.110 Surface Spec v2: gRPC generic proto path is spec-owned.
        // Unknown proto_id must not bypass EndpointSpec into auth/router.
        let Some(spec) = futu_surface_spec::lookup_endpoint_by_proto_id(proto_id) else {
            tracing::warn!(
                proto_id,
                "rejecting gRPC request for proto_id not declared in EndpointSpec"
            );
            return Err(Status::invalid_argument(
                "proto_id is not declared in endpoint spec",
            ));
        };
        match spec.grpc_exposure() {
            futu_surface_spec::SurfaceExposure::Exposed(
                futu_surface_spec::GrpcSurface::GenericProtoRequest,
            ) => {}
            futu_surface_spec::SurfaceExposure::NotExposed { reason } => {
                tracing::warn!(
                    proto_id,
                    endpoint = spec.canonical_name,
                    reason,
                    "rejecting gRPC request for endpoint not exposed to gRPC"
                );
                return Err(Status::permission_denied(reason));
            }
        }
        let needed_scope = Some(spec.runtime.scope);
        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();

        // 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: &req_inner.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 serial_no = self.next_serial();
        let body = futu_server::trade_packet_id::fill_omitted_trade_packet_id_bytes(
            proto_id,
            req_inner.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)
        .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")` +
    /// `needed_scope=None` (跳 scope check, 走 OR 语义手工 check). Allow → 拿
    /// rec snapshot. Reject (Bearer invalid / expired) → 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.
    ///
    /// 与 v1.4.103 行为 byte-identical: handshake qot:read OR acc:read, per-event
    /// scope match + trade acc_id whitelist (受限 key + acc_id=0 仍 drop, 由 pipeline
    /// `body_aware::CheckCtx { acc_id: Some(0) }` + `allowed_acc_ids` 非空 → reject 实现).
    async fn subscribe_push(
        &self,
        request: Request<SubscribePushRequest>,
    ) -> Result<Response<Self::SubscribePushStream>, Status> {
        // 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 一次 ──────────────────
        //
        // 用 `needed_scope=None` 跳 pipeline scope check (因为我们要 qot:read OR
        // acc:read 双 OR 语义, 不是单 scope match). Bearer 提取 / verify / expiry
        // 都走 pipeline (单一 source).
        //
        // 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: None, // OR 语义手工 check, pipeline 不参与
            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,
        };

        // OR 语义 scope check: legacy mode (rec=None) 全放行;
        // scope mode 必须持 qot:read OR acc:read 任一.
        let (scopes, key_id, allowed_acc_ids) = match rec_snapshot.as_ref() {
            Some(rec) => {
                let qot_ok = rec.scopes.contains(&Scope::QotRead);
                let acc_ok = rec.scopes.contains(&Scope::AccRead);
                if !qot_ok && !acc_ok {
                    futu_auth::audit::with_context(audit_ctx.clone(), || {
                        futu_auth::audit::reject(
                            "grpc",
                            "event=subscribe_push",
                            &rec.id,
                            "missing scope qot:read OR acc:read",
                        );
                    });
                    return Err(grpc_status_for(
                        RejectKind::Forbidden,
                        "missing scope qot:read OR acc:read".to_string(),
                    ));
                }
                (
                    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();
        tokio::spawn(async move {
            loop {
                match push_rx.recv().await {
                    Ok(event) => {
                        // 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)))
    }
}

// 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 / 未知都按行情门槛
    }
}

// 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;

/// 构建 gRPC 服务（供外部调用 tonic Server 使用）
pub fn build_service(
    router: Arc<RequestRouter>,
    push_broadcaster: Arc<GrpcPushBroadcaster>,
) -> FutuOpenDServer<FutuGrpcService> {
    apply_grpc_message_limits(FutuOpenDServer::new(FutuGrpcService::new(
        router,
        push_broadcaster,
    )))
}

/// 构建 gRPC 服务（带 KeyStore 鉴权 + 共享限额 counters）
pub fn build_service_with_auth(
    router: Arc<RequestRouter>,
    push_broadcaster: Arc<GrpcPushBroadcaster>,
    key_store: Arc<KeyStore>,
    counters: Arc<RuntimeCounters>,
) -> FutuOpenDServer<FutuGrpcService> {
    apply_grpc_message_limits(FutuOpenDServer::new(FutuGrpcService::with_auth(
        router,
        push_broadcaster,
        key_store,
        counters,
    )))
}

fn apply_grpc_message_limits(
    service: FutuOpenDServer<FutuGrpcService>,
) -> FutuOpenDServer<FutuGrpcService> {
    service
        .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE_BYTES)
        .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE_BYTES)
}

/// 启动 gRPC 服务
pub async fn start(
    listen_addr: &str,
    router: Arc<RequestRouter>,
    push_broadcaster: Arc<GrpcPushBroadcaster>,
) -> Result<(), Box<dyn std::error::Error>> {
    start_with_auth(
        listen_addr,
        router,
        push_broadcaster,
        Arc::new(KeyStore::empty()),
        Arc::new(RuntimeCounters::new()),
    )
    .await
}

/// 启动 gRPC 服务（带 KeyStore 鉴权 + 共享限额 counters）
pub async fn start_with_auth(
    listen_addr: &str,
    router: Arc<RequestRouter>,
    push_broadcaster: Arc<GrpcPushBroadcaster>,
    key_store: Arc<KeyStore>,
    counters: Arc<RuntimeCounters>,
) -> Result<(), Box<dyn std::error::Error>> {
    let addr = listen_addr
        .parse()
        .map_err(|e| format!("invalid addr: {e}"))?;
    if !key_store.is_configured() {
        tracing::warn!(
            "gRPC server running WITHOUT API key auth (legacy mode); \
             all RPCs are open. Pass --grpc-keys-file to enable scope-based auth."
        );
    }
    let service = build_service_with_auth(router, push_broadcaster, key_store, counters);
    tracing::info!(addr = %listen_addr, "gRPC 服务已启动");
    tonic::transport::Server::builder()
        .add_service(service)
        .serve(addr)
        .await?;
    Ok(())
}

/// 启动 gRPC 服务（带 KeyStore 鉴权 + 共享限额 counters），并支持 daemon
/// 统一 shutdown 信号。
pub async fn start_with_auth_until_shutdown(
    listen_addr: &str,
    router: Arc<RequestRouter>,
    push_broadcaster: Arc<GrpcPushBroadcaster>,
    key_store: Arc<KeyStore>,
    counters: Arc<RuntimeCounters>,
    shutdown_rx: watch::Receiver<bool>,
) -> Result<(), Box<dyn std::error::Error>> {
    let addr = listen_addr
        .parse()
        .map_err(|e| format!("invalid addr: {e}"))?;
    if !key_store.is_configured() {
        tracing::warn!(
            "gRPC server running WITHOUT API key auth (legacy mode); \
             all RPCs are open. Pass --grpc-keys-file to enable scope-based auth."
        );
    }
    let service = build_service_with_auth(router, push_broadcaster, key_store, counters);
    tracing::info!(addr = %listen_addr, "gRPC 服务已启动");
    tonic::transport::Server::builder()
        .add_service(service)
        .serve_with_shutdown(addr, grpc_shutdown_requested(shutdown_rx))
        .await?;
    Ok(())
}

async fn grpc_shutdown_requested(mut shutdown_rx: watch::Receiver<bool>) {
    loop {
        if *shutdown_rx.borrow() {
            tracing::info!("gRPC server stopped by shutdown signal");
            return;
        }
        if shutdown_rx.changed().await.is_err() {
            tracing::info!("gRPC server stopped after shutdown sender dropped");
            return;
        }
    }
}