futu_server/conn/

model.rs

/// 单个客户端连接
pub use futu_core::INTERNAL_UI_CLIENT_ID;

pub(crate) struct DecodedInitConnect {
    request: futu_proto::init_connect::Request,
}

pub(crate) struct PreparedInitConnect {
    response_body: Vec<u8>,
    recv_notify: bool,
    ai_type: i32,
    enable_aes: bool,
}

impl DecodedInitConnect {
    pub(crate) fn is_internal_ui(&self) -> bool {
        self.request.c2s.client_id == INTERNAL_UI_CLIENT_ID
    }
}

pub struct ClientConn {
    /// 随机连接 ID（对应 C++ `GetRand_MilliTimeAndU22`）
    pub conn_id: u64,
    /// Monotonic physical-socket generation allocated at accept time. Direct
    /// async push owners must match this value inside the final connection map
    /// lookup; a retired request may never recreate it.
    pub session_generation: u64,
    /// 连接状态：InitConnect 前 / 后 / 已断开
    pub state: ConnState,
    /// 随机 AES-128 key（InitConnect 响应里下发给客户端）
    pub aes_key: [u8; 16],
    /// AES 加解密已启用（InitConnect 完成且配置了 RSA 时为 true）
    pub aes_encrypt_enabled: bool,
    /// 该连接协商的 proto 格式（Protobuf / JSON）
    pub proto_fmt_type: ProtoFmtType,
    /// 上次收到 KeepAlive 的时间，用于超时检查
    pub last_keepalive: Instant,
    /// InitConnect.C2S.recvNotify：此连接是否接收市场状态 / 交易解锁等通知。
    ///
    /// C++ 在 `APIServer_InitConnect.cpp` 里把该字段写入 ConnInfo；
    /// `RegQotPush` / `Qot_Sub(isRegOrUnRegPush)` 不会修改这个开关。
    pub recv_notify: bool,
    /// InitConnect.C2S.aiType：AI 调用类型。C++ 10.7
    /// `APIServer_InitConnect.cpp` 缺省为 0，并写入连接状态。
    pub ai_type: i32,
    /// 已收到的 KeepAlive 计数（监控用）
    pub keepalive_count: AtomicU32,
    /// 发送帧到此连接
    pub tx: mpsc::Sender<FutuFrame>,

    // ---- v1.0 WS per-message scope 鉴权（raw TCP legacy 兼容：scopes 空集全放行）----
    /// 该连接绑定的 API key id；WS 握手时填，未配 keys.json 时为 None
    pub key_id: Option<String>,
    /// 该连接持有的 scope 集合；空集 = legacy 模式 / TCP 直连，scope 检查放行
    pub scopes: HashSet<Scope>,
    /// v1.4.105 D3 (Phase 4) T-B2: 该连接 caller key 的 `allowed_markets` 硬
    /// 限额 (大写字符串 set, e.g. {"HK","US"}). `None` = 无限制 (legacy 模式
    /// / TCP 直连默认全开 / 未配 allowed_markets); `Some(set)` 非空 → push 端
    /// 应过滤 trd_market 不在 set 中的 trade event.
    ///
    /// **触发**: WS handshake 时从 `KeyRecord.allowed_markets` 拷贝过来.
    /// `PushDispatcher::push_trd_acc` 端 Layer 3 filter 检查. 与
    /// `caller_allowed_acc_ids` (Layer 1, per-call snapshot in IncomingRequest)
    /// 区别: 本字段是 per-conn snapshot (handshake 时一次性), 不随 per-call
    /// 重读 — KeyRecord SIGHUP reload 后**仅新建连接生效**, 老连接保持 snapshot
    /// (与 `scopes` / `caller_allowed_acc_ids` 的 snapshot 语义一致).
    pub allowed_markets: Option<std::sync::Arc<HashSet<String>>>,
    /// codex round 1 F4 (P2) v1.4.105: 该连接 caller key 的 `allowed_acc_ids`
    /// 硬限额 (per-conn snapshot, handshake 时一次性). `None` / `Some(empty)` =
    /// 无限制 (legacy 模式 / TCP 直连默认全开 / 未配 allowed_acc_ids);
    /// `Some(non-empty set)` →
    /// `PushDispatcher::push_trd_acc` 端 push-time 硬过滤 acc_id 不在 set 中
    /// 的 trade event (Layer 1, 与 `allowed_markets` 的 Layer 3 互补).
    /// Deny-all 使用 sentinel `{0}`，不使用空集合。
    ///
    /// **触发**: codex F4 指出 raw TCP push 端只查 `acc:read` scope +
    /// `allowed_markets`, 不查 `allowed_acc_ids`. 即使 request-time
    /// `SubAccPushHandler` 已阻止越权订阅, stale subscription / KeyRecord
    /// reload 后窄化的 acc 范围 / 历史 bug 留下的 conn→acc 关系 仍可能让 push
    /// 漏 leak. 本字段提供第二层 push-time 兜底.
    ///
    /// 与 `caller_allowed_acc_ids` (IncomingRequest, per-call) 区别: 本字段
    /// 在 push-time 用 (无 IncomingRequest), per-conn snapshot 与 `scopes` /
    /// `allowed_markets` 的 snapshot 语义一致.
    pub allowed_acc_ids: Option<std::sync::Arc<HashSet<u64>>>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RequestTransport {
    RawTcp,
    RawWebSocket,
}

/// 从连接接收到的请求
#[derive(Debug)]
pub struct IncomingRequest {
    response_commit: Arc<ResponseCommitState>,
    pub transport: RequestTransport,
    /// Physical socket generation copied from `ClientConn` by the listener
    /// immediately before dispatch. Non-raw adapters retain zero.
    pub session_generation: u64,
    /// 发送请求的连接 ID（用于响应路由 + SubscriptionManager / cache 记账）
    ///
    /// **跨 surface 命名空间分配**（v1.4.106 codex 0517 ζ25-redo F2 沉淀）：
    /// - raw TCP listener: `ClientConn::generate_conn_id()` 派生（u32 范围）
    /// - REST: `crates/futu-rest/src/routes/qot.rs::REST_SHARED_CONN`
    ///   = `0xFFFF_FFFE`（u32 上限附近, 单值共享）
    /// - gRPC: `crates/futu-grpc/src/auth.rs::GRPC_STABLE_CONN_NAMESPACE`
    ///   = `0x4000_0000_0000_0000`（bit 62 namespace, 按 caller 派生）
    /// - WS / MCP: 通常派生自所属物理 TCP 连接的 conn_id
    ///
    /// 各 surface 不重叠. 加新 surface 时分配一个 namespace base, 不要与
    /// 上述 4 个段重合.
    pub conn_id: u64,
    /// 协议 ID（对齐 C++ `NN_ProtoCmd_*`）
    pub proto_id: u32,
    /// 序列号（和 Response 配对，供 client 端请求-响应匹配）
    pub serial_no: u32,
    /// 请求体 proto 格式（Protobuf / JSON）
    pub proto_fmt_type: ProtoFmtType,
    /// 请求 body（已解密后的明文）
    pub body: Bytes,
    /// v1.4.38 Phase 4: 订单幂等 key（由 REST `Idempotency-Key` header / gRPC
    /// metadata / WS envelope / MCP tool args 填入）。None 表示客户端未传，
    /// handler 走无幂等直通 path（backward-compat）。
    pub idempotency_key: Option<String>,
    /// v1.4.106 codex 0920 F1 (P1): caller key id 副本 (per-call snapshot,
    /// 由 surface adapter 层从 KeyRecord 读取后填入).
    ///
    /// **目标**: idempotency cache key namespace 必须含 caller key id, 否则
    /// 不同 caller 用同 Idempotency-Key 会跨 caller 命中老 response —— 严重
    /// 跨账户数据泄漏 + 重复下单 silent fail.
    ///
    /// `None` = 无 caller 标识 (legacy TCP / 未 auth) → namespace 用 `<no_key>`
    /// 占位符. `Some("alice")` = WS / MCP / REST 已 auth 的 caller —— namespace
    /// 用 `<caller_key_id="alice">`, 不与其他 caller 串.
    pub caller_key_id: Option<String>,
    /// v1.4.105 D2 contract-hardening 补丁: caller key 的 `allowed_acc_ids` 硬限额
    /// 副本 (per-call snapshot, 在 surface adapter 层从 KeyRecord 读取后填入).
    ///
    /// **目标**: 让 dispatch-time handlers (e.g. `SubAccPushHandler` 注册 acc_id
    /// 到 SubscriptionManager) 也能 enforce per-acc whitelist — 即使上游 pipeline
    /// body-aware step 已 enforce, 让 handler 自己 defense-in-depth 防 future
    /// regression (新 surface 加进来漏调 pipeline body-aware).
    ///
    /// `None` / `Some(empty)` = caller 无 acc_id 限制 (legacy mode 或 unrestricted
    /// key) → handler 不 filter; `Some(non-empty set)` → handler 应 reject 不在
    /// set 中的 acc_id. Deny-all 使用 sentinel `{0}`，不使用空集合。
    pub caller_allowed_acc_ids: Option<std::sync::Arc<std::collections::HashSet<u64>>>,
    /// Central pre-login Verification admission facts supplied by adapters.
    pub caller_has_auth_setup_scope: bool,
    pub caller_is_loopback: bool,
    pub caller_legacy_local_mode: bool,
}

impl IncomingRequest {
    #[must_use]
    pub fn response_commit_waiter(&self) -> ResponseCommitWaiter {
        ResponseCommitWaiter {
            state: Arc::clone(&self.response_commit),
        }
    }

    pub(crate) fn mark_response_committed(&self) {
        self.response_commit
            .committed
            .store(true, std::sync::atomic::Ordering::Release);
        self.response_commit.notify.notify_waiters();
    }

    /// codex 0522 F4 v1.4.106: cross-surface 单测 hook. 构 IncomingRequest
    /// 并填 caller scope (`caller_key_id` + `caller_allowed_acc_ids`) — 让
    /// REST / gRPC / WS / MCP 等 surface 的 adapter 都用同一构造路径, 防
    /// "某个 surface 漏填字段" silent regression.
    ///
    /// 之前 4 surface 各写一份 struct literal, 加新字段需逐个改, 漏一个就
    /// 出现 silent None — 与坑 #54 schema-only fix 同模式 (实装符号 vs 真
    /// 行为差距). 本 helper 是 single point, 加新字段 schema 自动 propagate.
    ///
    /// **注意**: 本 helper 不 take ownership of body — caller 已 own bytes.
    /// idempotency_key / caller_key_id 接 String 而非 &str 让 caller 决定
    /// 是 clone 还是 move.
    pub fn builder(
        conn_id: u64,
        proto_id: u32,
        serial_no: u32,
        proto_fmt_type: ProtoFmtType,
        body: Bytes,
    ) -> IncomingRequestBuilder {
        IncomingRequestBuilder {
            request: Self {
                response_commit: Arc::new(ResponseCommitState::default()),
                transport: RequestTransport::RawTcp,
                session_generation: 0,
                conn_id,
                proto_id,
                serial_no,
                proto_fmt_type,
                body,
                idempotency_key: None,
                caller_allowed_acc_ids: None,
                caller_key_id: None,
                caller_has_auth_setup_scope: false,
                caller_is_loopback: false,
                caller_legacy_local_mode: false,
            },
        }
    }
}

#[derive(Debug, Default)]
struct ResponseCommitState {
    committed: std::sync::atomic::AtomicBool,
    notify: tokio::sync::Notify,
}

#[derive(Clone, Debug)]
pub struct ResponseCommitWaiter {
    state: Arc<ResponseCommitState>,
}

impl ResponseCommitWaiter {
    pub async fn wait(self) {
        loop {
            let notified = self.state.notify.notified();
            if self
                .state
                .committed
                .load(std::sync::atomic::Ordering::Acquire)
            {
                return;
            }
            notified.await;
        }
    }
}

/// Thin builder for `IncomingRequest`.
///
/// The base request shape is the wire envelope; idempotency and caller scope are
/// optional per-surface decorations. Keeping those defaults here avoids every
/// REST / gRPC / raw WS / MCP adapter spelling out `None` independently.
#[derive(Debug)]
pub struct IncomingRequestBuilder {
    request: IncomingRequest,
}

impl IncomingRequestBuilder {
    pub(crate) fn with_session_generation(mut self, session_generation: u64) -> Self {
        self.request.session_generation = session_generation;
        self
    }

    /// Reuse the wire request's response-commit state when an adapter rebuilds
    /// the request to attach an authenticated caller snapshot.
    ///
    /// The WebSocket pipeline sends the response for the original wire request,
    /// while handlers receive the rebuilt request. Sharing this state is what
    /// lets an ACK-gated handler observe the actual socket send commit.
    pub(crate) fn with_response_commit_from(mut self, wire_request: &IncomingRequest) -> Self {
        self.request.response_commit = Arc::clone(&wire_request.response_commit);
        self
    }

    pub fn with_transport(mut self, transport: RequestTransport) -> Self {
        self.request.transport = transport;
        self
    }

    pub fn with_idempotency_key(mut self, idempotency_key: Option<String>) -> Self {
        self.request.idempotency_key = idempotency_key;
        self
    }

    pub fn with_caller_scope(
        mut self,
        caller_allowed_acc_ids: Option<std::sync::Arc<HashSet<u64>>>,
        caller_key_id: Option<String>,
    ) -> Self {
        self.request.caller_allowed_acc_ids = caller_allowed_acc_ids;
        self.request.caller_key_id = caller_key_id;
        self
    }

    pub fn with_auth_setup_admission(
        mut self,
        caller_has_auth_setup_scope: bool,
        caller_is_loopback: bool,
        caller_legacy_local_mode: bool,
    ) -> Self {
        self.request.caller_has_auth_setup_scope = caller_has_auth_setup_scope;
        self.request.caller_is_loopback = caller_is_loopback;
        self.request.caller_legacy_local_mode = caller_legacy_local_mode;
        self
    }

    pub fn build(self) -> IncomingRequest {
        self.request
    }
}

impl From<IncomingRequestBuilder> for IncomingRequest {
    fn from(builder: IncomingRequestBuilder) -> Self {
        builder.build()
    }
}

fn conn_id_epoch_elapsed_or_zero() -> Duration {
    match SystemTime::now().duration_since(UNIX_EPOCH) {
        Ok(elapsed) => elapsed,
        Err(err) => {
            tracing::warn!(
                error = %err,
                "system wall clock is before UNIX_EPOCH; using zero duration fallback for conn_id"
            );
            Duration::ZERO
        }
    }
}