futu_opend/startup/phase4/

surfaces.rs

//! v1.4.110 Layer 3 A: startup Phase 4 — surface server spawn helpers.
//!
//! 纯搬移自 `phase4.rs::run_phase4` 的 §7 WS / §8 REST / §9 gRPC / §10 Telnet
//! server spawn 块与 §11 TCP fail-closed gate 警告块（v1.8.0 large-file split,
//! 零行为变化）。每个 helper 体即原 `run_phase4` 里的同名块; 调用顺序、log 文案、
//! key_store holder 语义 (v1.4.103 B10) 全部由 `run_phase4` 原位保持。

use std::sync::Arc;

use futu_gateway_core::bridge::GatewayBridge;
use futu_server::listener::{ApiServer, ServerConfig};
use futu_server::listener_status::ListenerBindEvent;
use futu_server::ws_listener::{WsServer, WsServerDeps};

use crate::config::RuntimeConfig;

use super::listener_readiness::wrap_grpc_server_result;
use super::shutdown::SurfaceJoinHandle;
use super::summary::{RestTransportKind, merge_push_health_snapshots_for_rest};

// 7. 启动 WebSocket 服务（可选）
#[allow(clippy::too_many_arguments)]
pub(super) fn spawn_ws_server(
    config: &RuntimeConfig,
    server: &ApiServer,
    server_config: &ServerConfig,
    bridge: &Arc<GatewayBridge>,
    ws_key_store: &Option<Arc<futu_auth::KeyStore>>,
    shared_counters: &Arc<futu_auth::RuntimeCounters>,
    shutdown_rx: &tokio::sync::watch::Receiver<bool>,
    listener_events_tx: &tokio::sync::mpsc::UnboundedSender<ListenerBindEvent>,
) -> (Option<SurfaceJoinHandle>, Option<Arc<futu_auth::KeyStore>>) {
    // v1.4.103 (B10): holder 由 caller (`run_phase4`) 持有, 跨 server 块共享给
    // expand_allowed_card_nums spawn task; 本 helper 只在启用时填充并返回.
    let mut ws_key_store_holder: Option<std::sync::Arc<futu_auth::KeyStore>> = None;
    let ws_handle = if let Some(ws_port) = config.websocket_port {
        let ws_addr = format!("{}:{}", config.ip, ws_port);
        // v1.0：WS 握手鉴权 —— 复用 REST 的 key store 设计（`--ws-keys-file` 独立
        // 指定，不指定时 legacy 放行）
        // v1.4.102 BUG-007 fix (P1, leaf v1.4.100 报告): keys-file load 失败
        // 必须 fail-closed (abort daemon), 不再 silent fallback to legacy mode.
        // 历史: 用户传 `--ws-keys-file` 表示明确意图启用 auth, 文件 typo / parse
        // 错时 daemon 只 log error 然后继续无 auth → 用户以为有门锁, 实际门锁
        // 没装上 (security misconfig 比纯 legacy 更危险).
        let ws_key_store = ws_key_store.clone();
        let ws_counters = std::sync::Arc::clone(shared_counters);
        // v1.4.103 (B10): clone Arc 给外层 holder 持有, 同时让 ws_server 仍接管原 Arc.
        ws_key_store_holder = ws_key_store.as_ref().map(std::sync::Arc::clone);
        let ws_server = WsServer::with_auth(
            ws_addr.clone(),
            server_config.clone(),
            WsServerDeps::new(
                std::sync::Arc::clone(server.connections()),
                std::sync::Arc::clone(server.router()),
                Some(bridge.subscription_runtime().manager()),
            ),
            ws_key_store,
            Some(ws_counters),
        )
        .with_server_time_store(bridge.server_clock().anchor_store());
        tracing::info!(addr = %ws_addr, "starting WebSocket server");
        let ws_shutdown_rx = shutdown_rx.clone();
        let ws_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            ws_server
                .run_until_shutdown_with_listener_events(ws_shutdown_rx, Some(ws_listener_events))
                .await
        }))
    } else {
        None
    };
    (ws_handle, ws_key_store_holder)
}

// 8. 启动 REST API 服务（可选，含 WebSocket 推送）
#[allow(clippy::too_many_arguments)]
pub(super) fn spawn_rest_server(
    config: &RuntimeConfig,
    server: &ApiServer,
    bridge: &Arc<GatewayBridge>,
    ws_broadcaster: &Arc<futu_rest::ws::WsBroadcaster>,
    rest_key_store: &Option<Arc<futu_auth::KeyStore>>,
    rest_transport: futu_rest::server::RestTransport,
    rest_transport_kind: RestTransportKind,
    shared_counters: &Arc<futu_auth::RuntimeCounters>,
    shutdown_tx: &tokio::sync::watch::Sender<bool>,
    shutdown_rx: &tokio::sync::watch::Receiver<bool>,
    listener_events_tx: &tokio::sync::mpsc::UnboundedSender<ListenerBindEvent>,
) -> (Option<SurfaceJoinHandle>, Option<Arc<futu_auth::KeyStore>>) {
    let mut rest_key_store_holder: Option<std::sync::Arc<futu_auth::KeyStore>> = None;
    let rest_handle = if let Some(rest_port) = config.rest_port {
        let rest_addr = format!("{}:{}", config.ip, rest_port);
        let router = std::sync::Arc::clone(server.router());
        let broadcaster = std::sync::Arc::clone(ws_broadcaster);
        // v1.4.102 BUG-007 fix: 同 WS 路径 (fail-closed when keys-file 显式指定).
        let rest_key_store = rest_key_store
            .clone()
            .unwrap_or_else(|| std::sync::Arc::new(futu_auth::KeyStore::empty()));
        let rest_transport_name = match rest_transport_kind {
            RestTransportKind::Tls => "https",
            RestTransportKind::Plaintext => "http",
            RestTransportKind::Disabled => unreachable!("REST port is configured"),
        };
        tracing::info!(
            addr = %rest_addr,
            transport = rest_transport_name,
            "starting REST API server (WebSocket: /ws)"
        );

        // v1.4.103 (B10): clone Arc 给外层 holder 持有 (跨 server 块共享给
        // expand_allowed_card_nums spawn task). 仅在 keys 实际配置时填充
        // (empty store 不需要 expand).
        if rest_key_store.is_configured() {
            rest_key_store_holder = Some(std::sync::Arc::clone(&rest_key_store));
        }

        // v1.4.103 codex F3.1 (P1) round 3: REST 单独的 SIGHUP reload listener
        // 已**移除** — 防与 card_num expand 间的 race (多 listener 并发顺序乱
        // 可能让 reload 覆盖 expand 写入的 sentinel, 受限 key 在 reload 窗口
        // 内 silent unrestricted). 现在 reload + expand 由文件末尾的 unified
        // SIGHUP handler 顺序操作 (调 card_num_reload_and_expand_fn(true)).

        let rest_counters = std::sync::Arc::clone(shared_counters);
        // v1.4.32+ admin snapshot provider：closure 捕获 `Arc<GatewayBridge>`
        // 每次被 admin_status handler 调用时返回实时 StatusSnapshot JSON。
        // bridge 已经在上面 Arc 化，这里只做一次 clone。
        let bridge_for_status = std::sync::Arc::clone(bridge);
        let admin_status_provider: futu_rest::adapter::AdminStatusProvider =
            std::sync::Arc::new(move || {
                serde_json::to_value(bridge_for_status.snapshot_status())
                    .unwrap_or_else(|_| serde_json::json!({"error": "snapshot serialize failed"}))
            });
        let rest_shutdown_tx = shutdown_tx.clone();
        let admin_shutdown_handler: futu_rest::adapter::AdminShutdownHandler =
            std::sync::Arc::new(move || {
                rest_shutdown_tx
                    .send(true)
                    .map_err(|e| format!("shutdown receiver dropped: {e}"))
            });
        // v1.4.32+ admin reload handler：closure 调 Bridge::reload() 清 cipher cache
        // v1.4.34: reload 升级为 async（内部刷 credentials 走网络 I/O），
        // handler 返 Future，axum admin_reload async handler await 之
        // v1.4.106 codex 0554 F3 [P2]: reload 拆两阶段后变 sync (sync clear +
        // tokio::spawn refresh). closure 仍返 Future 类型不变 (向后兼容
        // AdminReloadHandler API), 但内部不 await — sync 阶段已完成 + spawn
        // 已派发, ReloadReport 立即可用.
        let bridge_for_reload = std::sync::Arc::clone(bridge);
        let admin_reload_handler: futu_rest::adapter::AdminReloadHandler =
            std::sync::Arc::new(move || {
                let bridge = std::sync::Arc::clone(&bridge_for_reload);
                Box::pin(async move {
                    serde_json::to_value(bridge.reload())
                        .unwrap_or_else(|_| serde_json::json!({"error": "reload serialize failed"}))
                })
            });
        // v1.4.83 §9 Phase 2 F5: push health snapshot provider —— closure
        // 捕获 bridge.push_runtime().push_health Arc, 每次
        // `/api/push-subscriber-info` 调用时返当前真实 snapshot.
        // v1.4.91 P1-D wiring: closure 同时捕获 bridge.push_runtime().qot_login_health,
        // 在返 push_health 同时多带一个 qot_login_health 字段供 ops 看
        // qot_logined self-heal counter (修 P1-D non-deterministic gap).
        let bridge_for_push_health = std::sync::Arc::clone(bridge);
        let push_health_snapshot_provider: futu_rest::adapter::PushHealthSnapshotProvider =
            std::sync::Arc::new(move || {
                let push = serde_json::to_value(
                    bridge_for_push_health
                        .push_runtime()
                        .push_health()
                        .snapshot(),
                )
                .unwrap_or_else(
                    |_| serde_json::json!({"error": "push_health snapshot serialize failed"}),
                );
                let qot_login = serde_json::to_value(
                    bridge_for_push_health
                        .push_runtime()
                        .qot_login_health()
                        .snapshot(),
                )
                .unwrap_or_else(
                    |_| serde_json::json!({"error": "qot_login_health snapshot serialize failed"}),
                );
                let backend_connected =
                    bridge_for_push_health.broker_runtime().platform_connected();
                let stock_list_status = bridge_for_push_health
                    .caches()
                    .static_cache
                    .stock_list_sync_status();
                let stock_list_static_query_ready = matches!(
                    futu_domain_static_data::stock_list_static_query_gate_action_for_status(
                        &stock_list_status.domain_facts(),
                    ),
                    futu_domain_static_data::StockListStaticQueryGateAction::Allow
                );
                merge_push_health_snapshots_for_rest(
                    push,
                    qot_login,
                    backend_connected,
                    stock_list_static_query_ready,
                )
            });
        // v1.4.105 D12 (Phase 2): 注入 card_num resolver 让 REST trade
        // handler (place_order / modify_order / cancel_all_order) 能解析
        // user 传 `card_num` 字段 → acc_id (覆盖 c2s.header.acc_id).
        // closure 捕获 bridge.caches().trd_cache, 调
        // `find_acc_ids_by_card_num(input) -> Vec<u64>`.
        let bridge_for_card_num = std::sync::Arc::clone(bridge);
        let card_num_resolver: futu_rest::adapter::CardNumResolver =
            std::sync::Arc::new(move |cn: &str| {
                bridge_for_card_num
                    .caches()
                    .trd_cache
                    .find_acc_ids_by_card_num(cn)
            });
        let rest_shutdown_rx = shutdown_rx.clone();
        let rest_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            futu_rest::server::start_with_auth_full_admin_until_shutdown_with_transport_and_listener_events(
                &rest_addr,
                router,
                broadcaster,
                rest_key_store,
                rest_counters,
                futu_rest::server::RestAdminHooks {
                    admin_status_provider: Some(admin_status_provider),
                    admin_shutdown_handler: Some(admin_shutdown_handler),
                    admin_reload_handler: Some(admin_reload_handler),
                    push_health_snapshot_provider: Some(push_health_snapshot_provider),
                    card_num_resolver: Some(card_num_resolver),
                },
                rest_transport,
                rest_shutdown_rx,
                Some(rest_listener_events),
            )
            .await
            .map_err(anyhow::Error::from)
        }))
    } else {
        None
    };
    (rest_handle, rest_key_store_holder)
}

// 9. 启动 gRPC 服务（可选，含流式推送）
pub(super) fn spawn_grpc_server(
    config: &RuntimeConfig,
    server: &ApiServer,
    grpc_broadcaster: &Arc<futu_grpc::server::GrpcPushBroadcaster>,
    grpc_key_store: &Option<Arc<futu_auth::KeyStore>>,
    shared_counters: &Arc<futu_auth::RuntimeCounters>,
    shutdown_rx: &tokio::sync::watch::Receiver<bool>,
    listener_events_tx: &tokio::sync::mpsc::UnboundedSender<ListenerBindEvent>,
) -> (Option<SurfaceJoinHandle>, Option<Arc<futu_auth::KeyStore>>) {
    let mut grpc_key_store_holder: Option<std::sync::Arc<futu_auth::KeyStore>> = None;
    let grpc_handle = if let Some(grpc_port) = config.grpc_port {
        let grpc_addr = format!("{}:{}", config.ip, grpc_port);
        let router = std::sync::Arc::clone(server.router());
        let broadcaster = std::sync::Arc::clone(grpc_broadcaster);
        // v1.4.102 BUG-007 fix: 同 WS / REST 路径 (fail-closed when keys-file 显式指定).
        let grpc_key_store = grpc_key_store
            .clone()
            .unwrap_or_else(|| std::sync::Arc::new(futu_auth::KeyStore::empty()));
        tracing::info!(addr = %grpc_addr, "starting gRPC server (SubscribePush: streaming)");

        // v1.4.103 (B10): clone Arc 给外层 holder 持有.
        if grpc_key_store.is_configured() {
            grpc_key_store_holder = Some(std::sync::Arc::clone(&grpc_key_store));
        }

        // v1.4.103 codex F3.1 (P1) round 3: gRPC 单独的 SIGHUP reload listener
        // 已**移除** — 同 REST 块, 由文件末尾的 unified SIGHUP handler 顺序
        // reload + expand 防 race.

        let grpc_counters = std::sync::Arc::clone(shared_counters);
        let grpc_shutdown_rx = shutdown_rx.clone();
        let grpc_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            let result = futu_grpc::server::start_with_auth_until_shutdown_with_listener_events(
                &grpc_addr,
                router,
                broadcaster,
                grpc_key_store,
                grpc_counters,
                grpc_shutdown_rx,
                Some(grpc_listener_events),
            )
            .await;
            wrap_grpc_server_result(result)
        }))
    } else {
        None
    };
    (grpc_handle, grpc_key_store_holder)
}

// 10. 启动 Telnet 管理服务（可选）
pub(super) fn spawn_telnet_server(
    config: &RuntimeConfig,
    server: &ApiServer,
    bridge: &Arc<GatewayBridge>,
    shutdown_tx: &tokio::sync::watch::Sender<bool>,
    shutdown_rx: &tokio::sync::watch::Receiver<bool>,
    listener_events_tx: &tokio::sync::mpsc::UnboundedSender<ListenerBindEvent>,
) -> Option<SurfaceJoinHandle> {
    let telnet_addr = super::telnet_bind_addr(&config.telnet_ip, config.telnet_port);
    if let Some(telnet_addr) = telnet_addr {
        // v1.4.97 P1-D-F: relogin callback closes over bridge to clear
        // login_cache; next 30s P1-D health tick triggers relogin.
        // Aligns with C++ GTWCmd_ReLogin (FTGateway/FTGTW_Define_Key.h:5).
        let bridge_for_relogin = std::sync::Arc::clone(bridge);
        let relogin_fn: futu_server::telnet::ReloginFn = std::sync::Arc::new(move || {
            tracing::warn!(
                "v1.4.97 P1-D-F: telnet relogin clearing login_cache; \
                     next P1-D tick will trigger central auth-session refresh"
            );
            bridge_for_relogin.caches().login_cache.clear();
        });
        let telnet_server = futu_server::telnet::TelnetServer::new(
            telnet_addr.clone(),
            std::sync::Arc::clone(server.connections()),
            Some(bridge.subscription_runtime().manager()),
            Some(std::sync::Arc::clone(server.metrics())),
            shutdown_tx.clone(),
        )
        .with_relogin_fn(relogin_fn);
        tracing::info!(addr = %telnet_addr, "starting Telnet server");
        let telnet_shutdown_rx = shutdown_rx.clone();
        let telnet_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            telnet_server
                .run_until_shutdown_with_listener_events(
                    telnet_shutdown_rx,
                    Some(telnet_listener_events),
                )
                .await
        }))
    } else {
        None
    }
}

// 11. v1.4.104 external reviewer S-001 (P0): native TCP keystore guard.
//
// 配任一 keys file (rest / grpc / ws) → 用户**意图**启用 scope mode.
// 但 TCP FTAPI 协议 InitConnect 没有 Bearer 字段, 无法做 caller-specific
// scope check (S-001). 默认 fail-closed: 不启 TCP listener, 用户需要
// 通过 REST/gRPC/WS 与 daemon 交互 (这些 surface 都已加 pipeline auth).
//
// 显式 `--allow-tcp-unauthenticated` opt-in → 启 TCP, 但 daemon 启动
// loud warn 该端口完全无 auth, agent skill 等 local process 可任意调用.
// 用之前 captured 的 local 变量, args 已被 merge_config 消费
pub(super) fn warn_tcp_keystore_gate(
    config: &RuntimeConfig,
    listen_addr: &str,
    tcp_disabled: bool,
    any_keys_configured: bool,
    allow_tcp_unauthenticated: bool,
) {
    if tcp_disabled {
        tracing::warn!(
            listen_addr = %listen_addr,
            "v1.4.104 external report S-001 (P0) fix: TCP listener (port {}) NOT started — \
             keys file configured but --allow-tcp-unauthenticated not set. \
             native TCP FTAPI protocol has no Bearer field, cannot enforce \
             caller-specific scope check; defaulting to fail-closed (skip TCP). \
             Use REST/gRPC/WS endpoints for authenticated access. \
             To restore TCP (legacy Python SDK clients) add --allow-tcp-unauthenticated, \
             but be aware that port {} will accept ANY local connection without \
             scope check (跨账户 leak risk).",
            config.port,
            config.port,
        );
        eprintln!(
            "⚠️  TCP listener (port {}) DISABLED (v1.4.104 external report S-001 fix): \
             keys file configured + no --allow-tcp-unauthenticated. \
             Pass --allow-tcp-unauthenticated to restore (with security warning).",
            config.port,
        );
    } else if any_keys_configured && allow_tcp_unauthenticated {
        tracing::warn!(
            listen_addr = %listen_addr,
            "⚠️  v1.4.104: TCP listener running WITHOUT scope check despite keys configured \
             (--allow-tcp-unauthenticated set). Port {} accepts ANY local connection — \
             跨账户 leak risk. Use REST/gRPC/WS for authenticated clients; reserve \
             TCP only for legacy Python SDK / C++ OpenD where Bearer not feasible.",
            config.port,
        );
        eprintln!(
            "⚠️  TCP port {} ACCEPTS UNAUTHENTICATED connections (--allow-tcp-unauthenticated). \
             受限 keys 不在该 surface 强制. 推荐改用 REST/gRPC/WS.",
            config.port,
        );
    }
}