futu_gateway_core/bridge/

mod.rs

// 业务桥接层：直连富途后端的完整网关
//
// 启动流程：HTTP 认证 → TCP 连接后端 → 登录 → 获取 session_key → 拉取账户列表 → 启动 API 服务
//
// # GatewayBridge boundary map
//
// `GatewayBridge` keeps facade/ownership state: shared backend, broker runtime,
// subscription runtime, client identity, idempotency/replay guards, and small
// control state.
// Cache-owned state lives under `CacheBundle`; push runtime state lives under
// `PushRuntime`; startup/reload/push/trade/QOT helpers are split into focused
// bridge submodules.
//
// Current boundary rules:
//
// 1. Handlers receive grouped dependency structs instead of cloning bridge
//    internals directly.
// 2. Cache access crosses the bridge boundary through `bridge.caches()` or
//    handler-owned `CacheBundle`.
// 3. Metrics, push health, and QOT login health cross the bridge boundary
//    through `bridge.push_runtime()`.
// 4. New bridge state must document its owner boundary before being exposed
//    outside this module.

use std::sync::Arc;

use arc_swap::ArcSwap;
use futu_backend::auth;
use futu_backend::conn::BackendConn;
use futu_core::error::Result;

// v1.4.110 P0-5 T15 wire: removed (handlers facade no longer exists)

// v1.4.60 Phase E1: bridge 模块化 split. 工具函数抽到子模块, 保持 pub(crate)
// 可见性让 sub-module + GatewayBridge 都能用.
mod account;
mod auth_runtime;
mod background_runtime;
mod broker_conn;
mod broker_runtime;
pub mod cache_bundle;
mod channel_runtime_shadow;
mod client_identity_runtime;
mod client_sig_reactive;
mod client_sig_timer;
pub mod clock;
mod dispatcher;
mod gateway_parts;
// v1.4.110 Layer 3 B: bridge::initialize() 拆 4 phases (zero behavior change).
mod init_phase1;
mod init_phase2;
mod init_phase3;
mod init_phase4_reconnect;
mod market_startup;
mod mkt_id_refresh;
mod platform_pool;
mod push_basic_qot;
mod push_broker_queue;
mod push_builder;
mod push_callback;
mod push_crypto_notify;
mod push_dedup;
pub mod push_delivery;
mod push_health;
mod push_kline;
mod push_msg_center;
mod push_orderbook;
pub mod push_parser;
mod push_rt;
mod push_runtime;
mod push_ticker;
mod push_trade_notify;
mod qot_login_health;
mod qot_right_push;
mod qot_right_refresh;
mod qot_runtime;
mod reference_data_download;
mod reference_runtime;
mod reload;
mod stock_list_sync;
mod subscription_runtime;
mod trade_dispatch;
mod trade_write_runtime;
mod types;
pub mod utils;
pub use types::*;

#[cfg(test)]
mod runtime_construction_tests;

/// 从 SharedBackend 加载当前后端连接（原子读取，支持重连后自动更新）
///
/// v1.4.110 P0-5 prep: 从 `handlers/mod.rs::load_backend` 提到 `bridge` 顶层,
/// 因 dispatcher.rs 3 处用 + handlers 子模块 48 处用, 是 bridge / handlers 共
/// 用 helper. 拆 3 crate 后位于 gateway-core (bridge 与 dispatcher 同 crate).
/// 原 `handlers::load_backend` 保留 `pub use` re-export 作 facade.
pub fn load_backend(
    shared: &SharedBackend,
) -> Option<std::sync::Arc<futu_backend::conn::BackendConn>> {
    let guard = shared.load();
    guard.as_ref().clone()
}

// v1.4.83 §9 Phase 2: Push channel health tracking (F2-F6)
pub use auth_runtime::AuthRuntime;
pub use background_runtime::{BackgroundRuntime, BackgroundTaskTracker};
use broker_conn::{
    BrokerChannelSetup, broker_tcp_login, establish_single_broker, keep_subscribe_quotes,
    resubscribe_quotes, sync_current_quote_desired_set,
};
pub use broker_runtime::{BROKER_RECONNECT_MAX_CONSECUTIVE_FAILURES, BrokerRuntime};
pub use cache_bundle::CacheBundle;
pub use client_identity_runtime::ClientIdentityRuntime;
pub use clock::ServerClockRuntime;
use gateway_parts::GatewayBridgeParts;
use push_builder::{build_order_fill_update_push, build_order_update_push, resolve_push_header};
pub use push_health::{
    PushHealth, PushHealthSnapshot, QotLoginHealth, QotLoginHealthSnapshot, SharedPushHealth,
    SharedQotLoginHealth,
};
pub use push_runtime::PushRuntime;
pub use qot_runtime::QotRuntime;
pub use reference_runtime::ReferenceRuntime;
pub use subscription_runtime::SubscriptionRuntime;
use trade_dispatch::{dispatch_trade_data_queries, retry_with_exp_backoff};
pub use trade_write_runtime::TradeWriteRuntime;
use utils::split_host_port;
// v1.4.93 P1-3 (BUG-5318-003): cross-module re-export 让 handlers/qot/mod.rs
// 能调 derive_exch_type_with_fallback. utils 模块自身保持 private.
pub use utils::{
    derive_exch_type_with_fallback, listing_date_to_str, market_code_to_exch_type_with_security,
    market_code_to_qot_market_with_security,
};

/// 共享的后端连接引用，支持重连后原子替换。
/// 所有 handler 持有同一个 Arc，重连时通过 ArcSwap::store 替换内部值，
/// handler 下次 load() 即可获取新连接。
pub type SharedBackend = Arc<ArcSwap<Option<Arc<BackendConn>>>>;

/// 网关桥接器
pub struct GatewayBridge {
    /// Cache ownership boundary for quote, trade, login, profile, and related
    /// shared cache state.
    caches: CacheBundle,
    /// Push-runtime boundary for metrics and push health state.
    push_runtime: PushRuntime,
    /// Broker/backend connection boundary for Platform and broker channels.
    broker_runtime: BrokerRuntime,
    /// QOT-specific mutable runtime state boundary.
    qot_runtime: QotRuntime,
    /// Subscription runtime boundary for QOT/TRD subscriptions and server push.
    subscription_runtime: SubscriptionRuntime,
    /// Auth/reload lifecycle runtime boundary.
    auth_runtime: AuthRuntime,
    /// Background task lifecycle boundary. Owns startup worker JoinHandles and
    /// aborts them when the bridge/runtime is dropped.
    background_runtime: BackgroundRuntime,
    /// Server-clock runtime boundary. Owns login-derived offset and injectable
    /// `ServerClock` provider.
    server_clock: ServerClockRuntime,
    /// Reference-data runtime boundary for HTTP-backed suspend/code-change caches.
    reference_runtime: ReferenceRuntime,
    /// Client identity boundary: app language + HTTP/API client type.
    client_identity: ClientIdentityRuntime,
    /// Trade-write runtime boundary for idempotency and PacketID replay guard.
    trade_write_runtime: TradeWriteRuntime,
}

impl GatewayBridge {
    pub fn new() -> Self {
        let parts = GatewayBridgeParts::new_default();
        Self {
            caches: parts.caches,
            push_runtime: parts.push_runtime,
            broker_runtime: parts.broker_runtime,
            qot_runtime: parts.qot_runtime,
            subscription_runtime: parts.subscription_runtime,
            auth_runtime: parts.auth_runtime,
            background_runtime: parts.background_runtime,
            server_clock: parts.server_clock,
            reference_runtime: parts.reference_runtime,
            client_identity: parts.client_identity,
            trade_write_runtime: parts.trade_write_runtime,
        }
    }

    /// Cache ownership boundary for quote, trade, login, profile, and related
    /// shared cache state.
    pub fn caches(&self) -> &CacheBundle {
        &self.caches
    }

    /// Trade-write runtime boundary for idempotency and PacketID replay guard.
    pub fn trade_write_runtime(&self) -> &TradeWriteRuntime {
        &self.trade_write_runtime
    }

    /// Push-runtime boundary for metrics, push health, qot-login health, and
    /// push event sender state.
    pub fn push_runtime(&self) -> &PushRuntime {
        &self.push_runtime
    }

    /// Auth/reload lifecycle runtime boundary.
    pub fn auth_runtime(&self) -> &AuthRuntime {
        &self.auth_runtime
    }

    /// Subscription runtime boundary for QOT/TRD subscriptions and server push.
    pub fn subscription_runtime(&self) -> &SubscriptionRuntime {
        &self.subscription_runtime
    }

    /// Broker/backend connection boundary for Platform and broker channels.
    pub fn broker_runtime(&self) -> &BrokerRuntime {
        &self.broker_runtime
    }

    pub(super) fn auth_runtime_mut(&mut self) -> &mut AuthRuntime {
        &mut self.auth_runtime
    }

    /// Client identity boundary: app language + HTTP/API client type.
    pub fn client_identity(&self) -> &ClientIdentityRuntime {
        &self.client_identity
    }

    pub(super) fn client_identity_mut(&mut self) -> &mut ClientIdentityRuntime {
        &mut self.client_identity
    }

    /// Reference-data runtime boundary for HTTP-backed suspend/code-change caches.
    pub fn reference_runtime(&self) -> &ReferenceRuntime {
        &self.reference_runtime
    }

    /// QOT-specific mutable runtime state boundary.
    pub fn qot_runtime(&self) -> &QotRuntime {
        &self.qot_runtime
    }

    /// Server-clock runtime boundary. Owns login-derived offset and injectable
    /// `ServerClock` provider.
    pub fn server_clock(&self) -> &ServerClockRuntime {
        &self.server_clock
    }

    /// Background task lifecycle boundary. Owns startup worker JoinHandles and
    /// aborts them when the bridge/runtime is dropped.
    pub fn background_runtime(&self) -> &BackgroundRuntime {
        &self.background_runtime
    }

    /// 取所有**已建立**的 broker 连接。顺序按 broker_id 升序，
    /// `fetch_real_accounts` 会遍历这个 vec 对每个 broker 各发一次 CMD 2282，
    /// 对齐 C++ `GTWCmdAndPushReply::OMEvProc_RePullAccList`（`GTWCmdAndPushReply.cpp:937-945`）
    /// —— 每个 broker conn 独立发 `INNProto_Trd_AccList::QueryAccList(uid, ConnTypeToBrokerID(m_enConnType))`。
    pub fn all_broker_backends(&self) -> Vec<(u32, Arc<BackendConn>)> {
        self.broker_runtime().all_broker_backends()
    }

    /// v1.4.110 codex audit P1 #3: 用户唯一已开户 crypto account 的 broker_id.
    ///
    /// 对齐 C++ `INNData_Trd_MainBrokerage::GetCryptoSupportedDefaultMainBroker`
    /// (line 70-123) 优先级 1:
    /// > 如果当前只开了一个 crypto account, 直接取该 account 的 broker.
    ///
    /// 返:
    /// - `Some(broker_id)`: trd_cache 恰好 1 个 `is_crypto_account()`, 取其 broker
    /// - `None`: 0 个或 ≥ 2 个 crypto account, 走 9419 crypto_brokers / main_brokers / fallback 路径
    ///
    /// 注入到 `resolve_qot_broker_for_request` / `resolve_or_reject_broker`
    /// 第 5 参数. QOT handler `securityFirm=Unknown(0)` 时此值决定 default broker.
    pub fn single_crypto_account_broker(&self) -> Option<u32> {
        futu_backend::crypto_trade::single_crypto_account_broker(&self.caches().trd_cache)
    }

    /// 完整的启动流程: 认证 → 连接后端 → 登录 → 拉取数据
    ///
    /// `verify_cb`: 短信验证码回调(GUI 模式传入, CLI 模式传 None 使用 stdin)
    ///
    /// v1.4.110 Layer 3 B refactor: 拆 4 phase 编排器, 每 phase 抽到独立子模块
    /// (zero behavior change):
    /// - [`init_phase1`]: HTTP 认证 + `--setup-only` 早退出
    /// - [`init_phase2`]: TCP 连接后端 + 登录 + ConnIP + broker 通道
    /// - [`init_phase3`]: 拉取账户列表 + 启动后台 worker (CMD 3020 / 6024 /
    ///   mkt_id refresh / qot_login_health / G1 client_sig timer /
    ///   stock_list_sync / reference_data_download)
    /// - [`init_phase4_reconnect`]: 心跳 + 断连监控 tokio task
    pub async fn initialize(
        &mut self,
        config: &GatewayConfig,
        verify_cb: Option<auth::VerifyCodeCallback>,
    ) -> Result<tokio::sync::mpsc::Receiver<PushEvent>> {
        let proactive_env = std::env::var("FUTU_CLIENT_SIG_PROACTIVE_REFRESH").ok();
        let reactive_env = std::env::var("FUTU_CLIENT_SIG_REACTIVE_REFRESH").ok();
        self.auth_runtime_mut()
            .set_client_sig_refresh_status(ClientSigRefreshStatus {
                client_sig_proactive_refresh_enabled:
                    client_sig_timer::effective_proactive_refresh_enabled(
                        config.client_sig_proactive_refresh,
                        proactive_env.as_deref(),
                    ),
                client_sig_reactive_refresh_enabled:
                    client_sig_reactive::effective_reactive_refresh_enabled(
                        config.client_sig_reactive_refresh,
                        reactive_env.as_deref(),
                    ),
            });

        // Phase 1: HTTP 认证 (+ `--setup-only` 早退出)
        let phase1 = match self.init_phase1_http_auth(config, verify_cb).await? {
            init_phase1::Phase1Outcome::SetupOnlyDone(rx) => return Ok(rx),
            init_phase1::Phase1Outcome::Continue(p1) => p1,
        };

        // Phase 2: TCP 连接后端 + 登录 + ConnIP + broker 通道
        let phase2 = self
            .init_phase2_tcp_connect(config, phase1.auth_result)
            .await?;
        let init_phase2::Phase2Result {
            backend,
            login_result,
            push_rx,
            push_cb,
            auth_result_for_reconnect,
        } = phase2;

        // Phase 3: 启动后台 worker
        self.init_phase3_workers(&backend).await?;

        // Phase 4: 心跳 + 断连监控
        let heartbeat_interval =
            std::time::Duration::from_secs(login_result.keep_alive_interval as u64);
        tracing::info!(interval_secs = ?heartbeat_interval, "starting backend heartbeat");
        let heartbeat_handle =
            futu_backend::heartbeat::start_heartbeat(Arc::clone(&backend), heartbeat_interval);

        init_phase4_reconnect::spawn_heartbeat_and_reconnect_monitor(
            init_phase4_reconnect::ReconnectMonitorDeps {
                config: config.clone(),
                auth: auth_result_for_reconnect,
                caches: self.caches().clone(),
                broker_runtime: self.broker_runtime().clone(),
                subscriptions: self.subscription_runtime().manager(),
                push_runtime: self.push_runtime().clone(),
                main_broker_cache: self.qot_runtime().main_broker_cache(),
                push_cb,
                commconfig: self.auth_runtime().commconfig(),
                auth_refresher: self.auth_runtime().auth_refresher(),
                account: config.account.clone(),
                initial_heartbeat: heartbeat_handle,
                background_runtime: self.background_runtime().clone(),
            },
        );

        // v1.4.106 codex 1140 F7: broker_dict refresher (CMD 18008 8h 周期).
        //
        // v1.4.110 P0-5 T15 P4 wire: 原本 bridge::initialize() 直接调
        // `handlers::qot::spawn_broker_dict_refresher(...)` 是 bridge → qot 反向
        // 引用 (cross-crate cycle 后违法). 现 spawn 移到
        // `futu_gateway_qot::register_handlers` 内部 — register 时 broker_runtime.backend
        // 已 ready (initialize 在它之前已完成), 同一 daemon 启动节奏不变.

        tracing::info!("gateway initialization complete");
        Ok(push_rx)
    }
    // v1.4.110 P0-5 prep (T12 P1): `register_handlers(&self, server)` 已删 —
    // 原本反向调用 `handlers::{qot,trd,sys}::register_handlers(router, self)`
    // 是 bridge → handlers cyclic dep (3 crate split 后 bridge 在 gateway-core,
    // handlers 在 gateway-trd/-qot, cross-crate cycle 不可). 改由 futu-opend
    // `startup/mod.rs` 在 bridge 建好后**显式**调 3 个域的 register fn.
}

impl Default for GatewayBridge {
    fn default() -> Self {
        Self::new()
    }
}