futu_backend/

quote_sub.rs

// 行情订阅管理
//
// FTAPI SubType → 后端 SubscribeBit 映射
// CMD 6211: 发送订阅请求到后端 (set-state semantics, server 覆盖式)
// CMD 6212: 接收行情推送
//
// **v1.4.106 codex 1131 F1+F2** 重构:
// - submit_global_desired_set 公共 fn 发"全集"CMD6211，并返回 QotSubError。
//   backend reject / decode 错 / timeout 都 loud 返回，caller 据此决定
//   ack-then-commit (F1)；旧 silent legacy wrapper 已在 v1.4.109 删除。
// - backend 覆盖式
//   ("每次请求，客户端都需要提供当前需要订阅的所有股票的所有订阅位")
//   unsub 通过发**新的更小集**实现 — 与 sub 同 entry 复用 (F2)
// - SubscribeRequestParams 携带 session / orderbook_detail / broker_detail /
//   rehab_type / first_push 等 backend req fields (F6)

use std::sync::Arc;

use prost::Message;

use futu_core::error::FutuError;

use crate::command_runtime::{
    SubscriptionDispatchContext, execute_qot_subscription_set,
    execute_qot_subscription_set_with_dispatch,
};
use crate::conn::BackendConn;
use crate::proto_internal::ft_cmd_stock_quote_sub;
use crate::proto_internal::ft_cmd_stock_quote_sub_data;

mod sub_bits;
mod ticker;

pub use futu_command_spec::CMD_QOT_PULL_TICKER;
pub use futu_command_spec::CMD_QOT_PUSH_SUB as CMD_QOT_SUB;
pub use futu_domain_qot_subscription::{
    EmptyDesiredMarket, SecurityWithOpts, SubBitOptions, SubscribeSetCommandMode,
    SubscribeSetPlanError, SubscribeSetSecurityPlan, empty_desired_market_for_sub,
    ensure_subscribe_set_backend_success, ftapi_market_to_quote_mkt, is_depth_sub_type,
    plan_empty_subscribe_set_commands, plan_subscribe_set_commands,
};
pub use sub_bits::{
    SubscribeBitInfo, sub_type_to_bit_infos_with_options, sub_type_to_bits,
    sub_type_to_bits_with_options,
};
pub use ticker::{TICKER_PAGE_MAX_ITEMS, pull_latest_ticker, pull_ticker_page};
#[cfg(test)]
use ticker::{
    build_ticker_page_request, common_session_to_nn, nn_quote_session, tick_period_type,
    ticker_periods_for_nn_session,
};

/// 后端推送命令 ID
pub const CMD_QOT_PUSH: u16 = 6212;
/// FTAPI SubType 枚举值
pub mod sub_type {
    pub const BASIC: i32 = 1;
    pub const ORDER_BOOK: i32 = 2;
    pub const TICKER: i32 = 4;
    pub const RT: i32 = 5;
    pub const KL_DAY: i32 = 6;
    pub const KL_5MIN: i32 = 7;
    pub const KL_15MIN: i32 = 8;
    pub const KL_30MIN: i32 = 9;
    pub const KL_60MIN: i32 = 10;
    pub const KL_1MIN: i32 = 11;
    pub const KL_WEEK: i32 = 12;
    pub const KL_MONTH: i32 = 13;
    pub const BROKER: i32 = 14;
    pub const KL_QUARTER: i32 = 15;
    pub const KL_YEAR: i32 = 16;
    pub const KL_3MIN: i32 = 17;
    pub const KL_10MIN: i32 = 18;
    pub const KL_120MIN: i32 = 19;
    pub const KL_180MIN: i32 = 20;
    pub const KL_240MIN: i32 = 21;
    pub const ORDER_BOOK_ODD: i32 = 22;
}

/// 后端 SubscribeBit 值
pub mod sbit {
    pub const PRICE: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_PRICE;
    pub const STOCK_STATE: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_STOCK_STATE;
    pub const STOCK_TYPE_SPECIFIC: u32 =
        futu_domain_qot_subscription::SUBSCRIBE_BIT_STOCK_TYPE_SPECIFIC;
    pub const ORDER_BOOK: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_ORDER_BOOK;
    pub const DEAL_STATISTICS: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_DEAL_STATISTICS;
    pub const HK_BROKER_QUEUE: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_HK_BROKER_QUEUE;
    pub const US_PREMARKET_AFTERHOURS: u32 =
        futu_domain_qot_subscription::SUBSCRIBE_BIT_US_PREMARKET_AFTERHOURS;
    pub const US_LV2_ORDER: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_US_LV2_ORDER;
    pub const TIME_SHARING: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_TIME_SHARING;
    pub const KLINE_1MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_1MIN;
    pub const KLINE_3MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_3MIN;
    pub const KLINE_5MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_5MIN;
    pub const KLINE_15MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_15MIN;
    pub const KLINE_30MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_30MIN;
    pub const KLINE_60MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_60MIN;
    pub const KLINE_DAY: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_DAY;
    pub const KLINE_WEEK: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_WEEK;
    pub const KLINE_MONTH: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_MONTH;
    pub const KLINE_QUARTER: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_QUARTER;
    pub const KLINE_YEAR: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_YEAR;
    // Ref: FutuOpenD/Src/NNProtoFile/Server/PB/Quote/FTCmdStockQuoteSubData.proto:92-98.
    // These backend subscribe bits are non-contiguous; do not derive them from
    // public FTAPI SubType values.
    pub const KLINE_120MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_120MIN;
    pub const KLINE_240MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_240MIN;
    pub const TICK: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_TICK;
    pub const MEGER_LV2_ORDER: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_MEGER_LV2_ORDER;
    pub const KLINE_10MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_10MIN;
    pub const KLINE_180MIN: u32 = futu_domain_qot_subscription::SUBSCRIBE_BIT_KLINE_180MIN;
}

/// **v1.4.106 codex 1131 F1**: `submit_global_desired_set` 错误类型.
/// 让 caller (SubHandler) 区分 backend reject vs decode err vs timeout →
/// ack-then-commit pattern (F1 P1).
#[derive(Debug)]
pub enum QotSubError {
    /// backend 返了 SubscribeSetRsp.result != 0 (reject), 携带 result 数 + warning.
    BackendRejected { result: i32, warning: i32 },
    /// 响应 decode 失败 — 该批次状态未知, 不写 local state.
    DecodeFailed(String),
    /// 网络 / TCP 错误 (timeout / 连接断). 透传 inner.
    Transport(FutuError),
    /// **v1.4.106 codex 0631 F3 [P2]**: caller 传了一个 backend 不识别的
    /// `ftapi_market` (`ftapi_market_to_quote_mkt → 0`). 防御性 fail loud:
    /// 不发任何 CMD6211, 整批 reject. caller 应早 validate 后再调.
    /// 携带 offending list 让 caller 报清晰错给用户.
    UnsupportedMarket { offending: Vec<i32> },
    /// **v1.4.106 codex 0631 F3 [P2]**: 多 market 分批发送时, 部分市场
    /// backend 失败 (BackendRejected / DecodeFailed) 但其它 OK — 该批次
    /// 是部分应用 (split state). caller 不能当全成功, 需明示用户
    /// "succeeded markets 已生效, failed 需重发".
    PartialMarketFailure { succeeded: Vec<u8>, failed: Vec<u8> },
}

impl std::fmt::Display for QotSubError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            QotSubError::BackendRejected { result, warning } => {
                write!(
                    f,
                    "backend rejected CMD6211: result={result} warning={warning}"
                )
            }
            QotSubError::DecodeFailed(s) => write!(f, "CMD6211 response decode failed: {s}"),
            QotSubError::Transport(e) => write!(f, "CMD6211 transport error: {e}"),
            QotSubError::UnsupportedMarket { offending } => write!(
                f,
                "CMD6211 unsupported ftapi_market(s): {offending:?} \
                 (ftapi_market_to_quote_mkt returned 0). Caller must validate \
                 ftapi_market before submit_global_desired_set."
            ),
            QotSubError::PartialMarketFailure { succeeded, failed } => write!(
                f,
                "CMD6211 partial-market failure: succeeded={succeeded:?} \
                 failed={failed:?}. State is split: succeeded markets are \
                 applied, failed markets need re-submit."
            ),
        }
    }
}

impl std::error::Error for QotSubError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            QotSubError::Transport(e) => Some(e),
            _ => None,
        }
    }
}

impl From<FutuError> for QotSubError {
    fn from(e: FutuError) -> Self {
        QotSubError::Transport(e)
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct QotSubscriptionWriterAdmission {
    pub connection_generation: u64,
    pub serial_no: u32,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DispatchAccepted {
    pub connection_generation: u64,
    pub serial_no: u32,
    pub quote_market_type: u8,
    pub runtime_dispatch_generation: u64,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum NormalDispatchOutcome {
    Succeeded,
    Failed,
}

#[derive(Clone)]
pub struct NormalSubscriptionDispatchHooks {
    pub try_admit: Arc<dyn Fn(QotSubscriptionWriterAdmission) -> bool + Send + Sync>,
    pub on_accepted:
        Arc<dyn Fn(QotSubscriptionWriterAdmission, u8, Vec<u8>) -> DispatchAccepted + Send + Sync>,
    pub on_outcome: Arc<dyn Fn(DispatchAccepted, NormalDispatchOutcome) + Send + Sync>,
}

/// **v1.4.110 Phase 2 Slice 4**: per-security 输入 — `(stock_id, broker_id, sub_types_with_opts)`
/// 3-tuple. broker_id = `None` 走 no-broker 路径 (普通股, Phase 2 默认),
/// `Some(NonZeroU32)` 走 broker-aware 路径 (crypto multi-broker).
///
/// 抽出 type alias 防 clippy `type_complexity` warn.
pub type SecuritySubscribeInput = SubscribeSetSecurityPlan;

fn plan_error_to_qot_sub_error(err: SubscribeSetPlanError) -> QotSubError {
    match err {
        SubscribeSetPlanError::EmptyDesiredSetWithoutMarket => {
            QotSubError::UnsupportedMarket { offending: vec![0] }
        }
        SubscribeSetPlanError::UnsupportedMarket { offending } => {
            QotSubError::UnsupportedMarket { offending }
        }
    }
}

/// **v1.4.106 codex 1131 F6**: 构建带 SubBitOptions 的订阅请求 — 真传 session
/// / detail / extended_time 给 backend.
///
/// **v1.4.110 Phase 2 Slice 4**: 升 3-tuple `(stock_id, broker_id, sub_types)`
/// 让 SecuritySubscribe.broker_id 真写出去 (对齐 C++ `MktQotSub.cpp:454-463`):
/// ```cpp
/// SecuritySubscribe::set_security_id(stSecKey.nStockID);
/// if (stSecKey.HasBroker()) {
///     SecuritySubscribe::set_broker_id(stSecKey.GetBrokerID());
/// }
/// ```
/// - `broker_id = None` ⟺ C++ `m_hasBroker = false`, 不调 `set_broker_id`
/// - `broker_id = Some(N)` ⟺ C++ `m_hasBroker = true`, 写 `broker_id = N`
pub fn build_subscribe_req_with_options(
    securities: &[SecuritySubscribeInput],
) -> ft_cmd_stock_quote_sub::SubscribeSetReq {
    build_subscribe_req_with_options_inner(securities, None)
}

pub fn build_keep_subscribe_req_with_options(
    securities: &[SecuritySubscribeInput],
) -> ft_cmd_stock_quote_sub::SubscribeSetReq {
    // Ref: C++ `MktQotSub.cpp:532-535` stores the last normal SubscribeSetReq
    // and sets `timer_sub=1` only for periodic keep-sub replay.
    build_subscribe_req_with_options_inner(securities, Some(1))
}

// The current 2607 C++ source still writes legacy `BitInfo::prob2` on the
// compatibility branches at `MktQotSubInstance.cpp:151,164`, while newer
// branches use `prob2_v2`. Preserve both proto2 presence paths explicitly.
#[allow(deprecated)]
fn build_subscribe_req_with_options_inner(
    securities: &[SecuritySubscribeInput],
    timer_sub: Option<i32>,
) -> ft_cmd_stock_quote_sub::SubscribeSetReq {
    let mut security_list = Vec::new();

    for (stock_id, broker_id, sub_types_with_opts) in securities {
        let mut bit_info_list = Vec::new();
        for (st, opts) in sub_types_with_opts {
            for info in sub_type_to_bit_infos_with_options(*st, opts.clone()) {
                bit_info_list.push(ft_cmd_stock_quote_sub_data::BitInfo {
                    bit: Some(info.bit),
                    prob: info.prob,
                    prob2: info.prob2,
                    prob2_v2: info.prob2_v2,
                });
            }
        }
        security_list.push(ft_cmd_stock_quote_sub_data::SecuritySubscribe {
            security_id: Some(*stock_id),
            bit_info_list,
            // v1.4.110 Phase 2 Slice 4: broker-aware CMD6211 wire.
            // Some(NZ) → 写 i32 (crypto multi-broker); None → 不写 (普通股).
            broker_id: broker_id.map(|nz| nz.get() as i32),
        });
    }

    ft_cmd_stock_quote_sub::SubscribeSetReq {
        security_list,
        reserved: None,
        timer_sub,
    }
}

/// **v1.4.106 codex 1131 F1+F2**: 给 caller (SubHandler) 的 set-state 接口.
///
/// **语义**: 发送整组 desired (stock_id, ftapi_market, sub_type_with_opts) →
/// backend 返 SubscribeSetRsp. backend "覆盖式" — 所有不在新集合中的旧订阅
/// 自动取消 (per FTCmdStockQuoteSub.proto 设计 doc:
/// "server会覆盖此客户端之前的订阅，并主动推送一次新增加的股票订阅位数据").
///
/// **F1 ack-then-commit**: caller 必须仅在 Ok 后才写 SubscriptionManager state.
/// Err 时 → 不写 state, 返用户 ret_type=-1 + 错误原因.
///
/// **F2 unsub via fresh set**: 退订通过传"new desired set 不含 removed key"
/// 实现, 不调单独 unsub backend cmd. SubHandler 计算
/// `current global - removed` 后调本 fn.
///
/// **max_sub_count**: 响应中 backend 下发的 quota — caller 应调
/// `SubscriptionManager::set_total_quota_from_backend(max_sub_count as u32)`
/// 同步真值 (F5).
///
/// **return 值**: backend 下发的 max_sub_count (caller 据此 update SubscriptionManager
/// 总配额 — F5 P2 dynamic quota).
pub async fn submit_global_desired_set(
    backend: &BackendConn,
    securities: &[SecurityWithOpts],
    hooks: &NormalSubscriptionDispatchHooks,
) -> std::result::Result<i32, QotSubError> {
    submit_global_desired_set_inner(backend, securities, hooks).await
}

async fn submit_global_desired_set_inner(
    backend: &BackendConn,
    securities: &[SecurityWithOpts],
    hooks: &NormalSubscriptionDispatchHooks,
) -> std::result::Result<i32, QotSubError> {
    let plan = plan_subscribe_set_commands(securities, SubscribeSetCommandMode::Normal)
        .map_err(plan_error_to_qot_sub_error)?;

    // **v1.4.106 codex 0631 F3 [P2]**: 多 market 分批发送, 部分失败 → 不能当
    // 全成功. 收集 succeeded / failed market 列表, 返 PartialMarketFailure.
    //
    // 老代码 .await? 短路: 第一个失败市场退出, max_sub_count=0 — caller 看不到
    // 哪些市场已成功 (split state). 改为收集所有结果, 部分失败 loud Err.
    let mut max_sub_count = 0i32;
    let mut succeeded_markets: Vec<u8> = Vec::new();
    let mut failed_markets: Vec<u8> = Vec::new();
    let mut first_transport_err: Option<FutuError> = None;
    for command in &plan.commands {
        match submit_subscribe_with_market(
            backend,
            &command.securities,
            command.mkt_type,
            command.is_depth,
            command.is_unsub_all,
            hooks,
        )
        .await
        {
            Ok(count) => {
                if count > max_sub_count {
                    max_sub_count = count;
                }
                if !succeeded_markets.contains(&command.mkt_type) {
                    succeeded_markets.push(command.mkt_type);
                }
            }
            Err(QotSubError::Transport(e)) => {
                // Transport error: 网络断 / TCP 错 — 整批中断, 透传不分批.
                // 这种情况下 partial 没意义 (后续市场也会同样 Transport 错).
                first_transport_err = Some(e);
                break;
            }
            Err(_) => {
                if !failed_markets.contains(&command.mkt_type) {
                    failed_markets.push(command.mkt_type);
                }
            }
        }
    }
    if let Some(e) = first_transport_err {
        return Err(QotSubError::Transport(e));
    }
    if !failed_markets.is_empty() {
        succeeded_markets.sort_unstable();
        failed_markets.sort_unstable();
        tracing::warn!(
            succeeded = ?succeeded_markets,
            failed = ?failed_markets,
            "v1.4.106 audit 0631 F3: submit_global_desired_set partial failure"
        );
        return Err(QotSubError::PartialMarketFailure {
            succeeded: succeeded_markets,
            failed: failed_markets,
        });
    }

    Ok(max_sub_count)
}

pub async fn submit_empty_desired_set_for_markets(
    backend: &BackendConn,
    markets: &[EmptyDesiredMarket],
    hooks: &NormalSubscriptionDispatchHooks,
) -> std::result::Result<i32, QotSubError> {
    let plan = plan_empty_subscribe_set_commands(markets).map_err(plan_error_to_qot_sub_error)?;

    let mut max_sub_count = 0i32;
    let mut succeeded_markets: Vec<u8> = Vec::new();
    let mut failed_markets: Vec<u8> = Vec::new();
    let mut first_transport_err: Option<FutuError> = None;

    for command in &plan.commands {
        match submit_subscribe_with_market(
            backend,
            &command.securities,
            command.mkt_type,
            command.is_depth,
            command.is_unsub_all,
            hooks,
        )
        .await
        {
            Ok(count) => {
                max_sub_count = max_sub_count.max(count);
                if !succeeded_markets.contains(&command.mkt_type) {
                    succeeded_markets.push(command.mkt_type);
                }
            }
            Err(QotSubError::Transport(e)) => {
                first_transport_err = Some(e);
                break;
            }
            Err(_) => {
                if !failed_markets.contains(&command.mkt_type) {
                    failed_markets.push(command.mkt_type);
                }
            }
        }
    }

    if let Some(e) = first_transport_err {
        return Err(QotSubError::Transport(e));
    }
    if !failed_markets.is_empty() {
        succeeded_markets.sort_unstable();
        succeeded_markets.dedup();
        failed_markets.sort_unstable();
        failed_markets.dedup();
        return Err(QotSubError::PartialMarketFailure {
            succeeded: succeeded_markets,
            failed: failed_markets,
        });
    }

    Ok(max_sub_count)
}

/// 发送单个 (mkt_type, is_depth) 的 CMD6211 请求, 返 max_sub_count.
///
/// v1.4.110 Phase 2 Slice 4: secs 升 3-tuple `(stock_id, broker_id, sub_types)`
/// 让 build_subscribe_req_with_options 能写出 SecuritySubscribe.broker_id.
async fn submit_subscribe_with_market(
    backend: &BackendConn,
    secs: &[SecuritySubscribeInput],
    mkt_type: u8,
    is_depth: bool,
    is_unsub_all: bool,
    hooks: &NormalSubscriptionDispatchHooks,
) -> std::result::Result<i32, QotSubError> {
    let req = if is_unsub_all {
        // 全退场景 — security_list 空, reserved=1 让 body 非零 (Windows backend 兼容).
        ft_cmd_stock_quote_sub::SubscribeSetReq {
            security_list: vec![],
            reserved: Some(1),
            timer_sub: None,
        }
    } else {
        build_subscribe_req_with_options(secs)
    };
    let body = req.encode_to_vec();

    let mut reserved = [0u8; 10];
    reserved[0] = mkt_type;
    // v1.4.110 Phase 2 Slice 4: request_bits 是 trace 用, 不必含 broker_id;
    // SecuritySubscribe.broker_id 在 build_subscribe_req_with_options 内部写.
    let request_bits: Vec<(u64, Vec<(u32, i64)>)> = secs
        .iter()
        .map(|(stock_id, _broker_id, sub_types)| {
            let bits = sub_types
                .iter()
                .flat_map(|(sub_type, opts)| sub_type_to_bits_with_options(*sub_type, opts.clone()))
                .collect();
            (*stock_id, bits)
        })
        .collect();

    tracing::info!(
        mkt_type,
        is_depth,
        is_unsub_all,
        count = secs.len(),
        body_len = body.len(),
        request_bits = ?request_bits,
        "v1.4.106 audit 1131 F1: sending CMD6211 subscribe (set-state)"
    );

    let accepted = Arc::new(parking_lot::Mutex::new(None));
    let context = SubscriptionDispatchContext {
        quote_market_type: mkt_type,
        exact_normal_wire: body.clone(),
        hooks: hooks.clone(),
        accepted: Arc::clone(&accepted),
    };
    let resp =
        match execute_qot_subscription_set_with_dispatch(backend, body.into(), reserved, context)
            .await
        {
            Ok(resp) => resp,
            Err(error) => {
                notify_normal_dispatch_outcome(hooks, &accepted, NormalDispatchOutcome::Failed);
                return Err(QotSubError::Transport(error));
            }
        };

    let parsed: ft_cmd_stock_quote_sub::SubscribeSetRsp = match Message::decode(resp.body.as_ref())
    {
        Ok(parsed) => parsed,
        Err(error) => {
            notify_normal_dispatch_outcome(hooks, &accepted, NormalDispatchOutcome::Failed);
            return Err(QotSubError::DecodeFailed(format!("{error}")));
        }
    };

    let status = match ensure_subscribe_set_backend_success(
        parsed.result,
        parsed.warning_code,
        parsed.max_sub_count,
    ) {
        Ok(status) => status,
        Err(reject) => {
            // **F1 P1**: backend reject → 让 caller 知道 (Err), 不 silent-warn.
            tracing::warn!(
                mkt_type,
                is_depth,
                result = reject.result,
                warning = reject.warning_code,
                request_bits = ?request_bits,
                "v1.4.106 audit 1131 F1: CMD6211 backend rejected"
            );
            notify_normal_dispatch_outcome(hooks, &accepted, NormalDispatchOutcome::Failed);
            return Err(QotSubError::BackendRejected {
                result: reject.result,
                warning: reject.warning_code,
            });
        }
    };

    tracing::info!(
        mkt_type,
        is_depth,
        max_sub_count = status.max_sub_count,
        "v1.4.106 audit 1131 F1: CMD6211 ok"
    );
    notify_normal_dispatch_outcome(hooks, &accepted, NormalDispatchOutcome::Succeeded);
    Ok(status.max_sub_count)
}

fn notify_normal_dispatch_outcome(
    hooks: &NormalSubscriptionDispatchHooks,
    accepted: &parking_lot::Mutex<Option<DispatchAccepted>>,
    outcome: NormalDispatchOutcome,
) {
    let receipt = accepted.lock().take();
    if let Some(receipt) = receipt {
        (hooks.on_outcome)(receipt, outcome);
    }
}

/// Submit one immutable cached keep request.
///
/// The caller supplies the exact last normal wire with only top-level
/// `timer_sub=1` changed. This path deliberately has no normal-dispatch hooks,
/// so keep cannot publish a new receipt or mutate normal eligibility state.
pub async fn submit_cached_keep_wire(
    backend: &BackendConn,
    quote_market_type: u8,
    wire_with_timer_sub_1: Vec<u8>,
) -> std::result::Result<i32, QotSubError> {
    let mut reserved = [0u8; 10];
    reserved[0] = quote_market_type;
    let resp = execute_qot_subscription_set(backend, wire_with_timer_sub_1.into(), reserved)
        .await
        .map_err(QotSubError::Transport)?;
    let parsed: ft_cmd_stock_quote_sub::SubscribeSetRsp = Message::decode(resp.body.as_ref())
        .map_err(|error| QotSubError::DecodeFailed(format!("{error}")))?;
    let status = ensure_subscribe_set_backend_success(
        parsed.result,
        parsed.warning_code,
        parsed.max_sub_count,
    )
    .map_err(|reject| QotSubError::BackendRejected {
        result: reject.result,
        warning: reject.warning_code,
    })?;
    Ok(status.max_sub_count)
}

#[cfg(test)]
mod tests;