ek-ctp 0.8.0

//! `CtpTransport` 的官方实现:把驱动动作落到真实 CTP 会话,把 SPI 回调送回驱动。
//!
//! ## 线程模型(D-NATIVE-THREAD)
//!
//! SPI 回调跑在 **CTP 自己的线程**上。回调不做任何交接,直接 `on_md_event` /
//! `on_td_event` 进插件 —— 插件的驱动状态本来就在 `Mutex` 里自锁,`PluginContext`
//! 又是 `Arc<EventEngine>`(可跨线程 `put`),所以**不需要 ingress 线程**:
//! 回调线程 = 原稿的回调线程,`put` 语义与原稿"回调里直接 `gateway.on_tick`"同构。
//!
//! ## 双相接线(断环)
//!
//! 插件要先有传输层才构造得出来,而传输层要先有插件才投得回事件 —— 于是:
//! 传输层持 `Arc<Ingress>`,`Ingress` 里放 `OnceLock<Weak<CtpPlugin>>`,
//! 由 [`crate::plugin::CtpPlugin::with_native`] 在插件建好后回填。用 `Weak` 断环:
//! 否则 插件 → 传输层 → Ingress → 插件 会成环,两者永不析构。
//!
//! ## 会话释放顺序
//!
//! 先释放 CTP 会话(`Release()`),再放掉 `Ingress`:字段声明顺序即析构顺序
//! (`md`/`td` 在 `ingress` 之前声明),`shutdown()` 亦按此顺序显式执行。

use super::md::{MdApi, MdSpiCallbacks};
use super::pod::{
    EkDepthMarketData, EkErrorField, EkInputOrder, EkInstrumentCommissionRate, EkInstrumentField,
    EkInstrumentMarginRate, EkInstrumentStatus, EkInvestorPositionField, EkOrderField,
    EkTradeField, EkTradingAccountField,
};
use super::td::{TdApi, TdSpiCallbacks};
use crate::driver::{MdEvent, TdEvent};
use crate::plugin::{CtpPlugin, CtpTransport};
use crate::wire::{
    ErrorField, InputOrder, InputOrderAction, QryInstrumentCommissionRate, QryInstrumentMarginRate,
};
use std::ffi::c_void;
use std::path::PathBuf;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::Duration;

/// 回调侧入口:SPI 回调借它把事件交回插件。
///
/// 见模块文档的"双相接线":`plugin` 由 [`CtpPlugin::with_native`] 在构造完成后回填。
pub(crate) struct Ingress {
    plugin: OnceLock<Weak<CtpPlugin>>,
}

impl Ingress {
    fn new() -> Self {
        Self {
            plugin: OnceLock::new(),
        }
    }

    /// 回填插件弱引用(只允许一次;重复调用被 `OnceLock` 静默忽略)。
    fn attach(&self, plugin: &Arc<CtpPlugin>) {
        let _ = self.plugin.set(Arc::downgrade(plugin));
    }

    /// 取插件(已析构 / 尚未回填 → `None`,回调直接放弃)。
    fn plugin(&self) -> Option<Arc<CtpPlugin>> {
        self.plugin.get().and_then(Weak::upgrade)
    }
}

/// 在 `ctx` 指向的 [`Ingress`] 上执行 `f`;空指针直接跳过。
///
/// # Safety
///
/// `ctx` 必须是 [`NativeTransport`] 经 `register_spi` 注册的 `Arc<Ingress>` 裸指针,
/// 且该 `Ingress` 在本次调用期间仍然存活(由传输层持有 `Arc` 保证)。
unsafe fn with_ingress(ctx: *mut c_void, f: impl FnOnce(&Ingress)) {
    if ctx.is_null() {
        return;
    }
    // SAFETY: 见本函数的 Safety 契约。
    let ing = unsafe { &*(ctx as *const Ingress) };
    f(ing);
}

/// `error_id != 0` 才算失败 —— 原稿 `if not error["ErrorID"]` 的判定面
/// (认证、登录、行情订阅、结算确认这些"会判错"的回调)。
fn error_checked(p: *const EkErrorField) -> Option<ErrorField> {
    if p.is_null() {
        return None;
    }
    // SAFETY: shim 在回调期间保证指针有效且完整。
    let e = unsafe { (*p).to_wire() };
    if e.error_id == 0 { None } else { Some(e) }
}

/// 原样搬运:SDK 给了 info 结构就带上,**不看 `error_id`**。
///
/// 用在原稿"无条件处理回报"的两处(`onRspOrderInsert` 一律重造 REJECTED 单、
/// `onRspOrderAction` 一律记撤单失败)——那里 `error_id == 0` 也必须走失败面,
/// 否则会丢掉原稿行为(驱动侧 `RspOrderInsert` 要求 `Some(e)` 才重造单)。
fn error_raw(p: *const EkErrorField) -> Option<ErrorField> {
    if p.is_null() {
        return None;
    }
    // SAFETY: 同 `error_checked`。
    Some(unsafe { (*p).to_wire() })
}

/* ==================== MD 回调 ==================== */

unsafe extern "C" fn md_on_front_connected(ctx: *mut c_void) {
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_md_event(MdEvent::FrontConnected);
            }
        })
    }
}

unsafe extern "C" fn md_on_front_disconnected(ctx: *mut c_void, reason: std::ffi::c_int) {
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_md_event(MdEvent::FrontDisconnected(reason as i64));
            }
        })
    }
}

unsafe extern "C" fn md_on_rsp_user_login(ctx: *mut c_void, error: *const EkErrorField) {
    let error = error_checked(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_md_event(MdEvent::RspUserLogin { error });
            }
        })
    }
}

unsafe extern "C" fn md_on_rsp_error(ctx: *mut c_void, error: *const EkErrorField) {
    // 原稿 288-290 无条件记「行情接口报错」,故这里不看 error_id(RspError 也不是 Option)。
    if error.is_null() {
        return;
    }
    // SAFETY: shim 在回调期间保证指针有效。
    let e = unsafe { (*error).to_wire() };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_md_event(MdEvent::RspError(e));
            }
        })
    }
}

unsafe extern "C" fn md_on_rsp_sub_market_data(ctx: *mut c_void, error: *const EkErrorField) {
    let error = error_checked(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_md_event(MdEvent::RspSubMarketData { error });
            }
        })
    }
}

unsafe extern "C" fn md_on_rtn_depth_market_data(
    ctx: *mut c_void,
    depth: *const EkDepthMarketData,
) {
    if depth.is_null() {
        return;
    }
    // SAFETY: shim 在回调期间保证 POD 有效;`to_wire` 立刻把所有字段拷成拥有型数据。
    let tick = unsafe { (*depth).to_wire() };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_md_event(MdEvent::RtnDepthMarketData(tick));
            }
        })
    }
}

static MD_CALLBACKS: MdSpiCallbacks = MdSpiCallbacks {
    on_front_connected: Some(md_on_front_connected),
    on_front_disconnected: Some(md_on_front_disconnected),
    on_rsp_user_login: Some(md_on_rsp_user_login),
    on_rsp_error: Some(md_on_rsp_error),
    on_rsp_sub_market_data: Some(md_on_rsp_sub_market_data),
    on_rtn_depth_market_data: Some(md_on_rtn_depth_market_data),
};

/* ==================== TD 回调 ==================== */

unsafe extern "C" fn td_on_front_connected(ctx: *mut c_void) {
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::FrontConnected);
            }
        })
    }
}

unsafe extern "C" fn td_on_front_disconnected(ctx: *mut c_void, reason: std::ffi::c_int) {
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::FrontDisconnected(reason as i64));
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_authenticate(ctx: *mut c_void, error: *const EkErrorField) {
    let error = error_checked(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspAuthenticate { error });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_user_login(
    ctx: *mut c_void,
    front_id: std::ffi::c_int,
    session_id: std::ffi::c_int,
    error: *const EkErrorField,
) {
    let error = error_checked(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspUserLogin {
                    error,
                    front_id: front_id as i64,
                    session_id: session_id as i64,
                });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_settlement_info_confirm(
    ctx: *mut c_void,
    error: *const EkErrorField,
) {
    // 驱动忽略该 error(原稿 523 无条件写 log),这里照 `error_checked` 搬运即可。
    let error = error_checked(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspSettlementInfoConfirm { error });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_order_insert(
    ctx: *mut c_void,
    order: *const EkInputOrder,
    error: *const EkErrorField,
) {
    // 原稿无条件重造 REJECTED 单 → 需要 order 与 error 同时在场(驱动同判)。
    let error = error_raw(error);
    let req = if order.is_null() {
        None
    } else {
        // SAFETY: shim 在回调期间保证 POD 有效。
        Some(unsafe { (*order).to_wire() })
    };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspOrderInsert { error, req });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_order_action(ctx: *mut c_void, error: *const EkErrorField) {
    // 原稿 515-517 无条件记「交易撤单失败」→ 原样搬运(不看 error_id)。
    let error = error_raw(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspOrderAction { error });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_qry_instrument(
    ctx: *mut c_void,
    row: *const EkInstrumentField,
    is_last: std::ffi::c_int,
) {
    let row = if row.is_null() {
        None
    } else {
        // SAFETY: shim 在回调期间保证 POD 有效。
        Some(unsafe { (*row).to_wire() })
    };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspQryInstrument {
                    last: is_last != 0,
                    row,
                });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_qry_investor_position(
    ctx: *mut c_void,
    row: *const EkInvestorPositionField,
    is_last: std::ffi::c_int,
) {
    let row = if row.is_null() {
        None
    } else {
        // SAFETY: shim 在回调期间保证 POD 有效。
        Some(unsafe { (*row).to_wire() })
    };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspQryInvestorPosition {
                    last: is_last != 0,
                    row,
                });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_qry_trading_account(
    ctx: *mut c_void,
    row: *const EkTradingAccountField,
    is_last: std::ffi::c_int,
) {
    let row = if row.is_null() {
        None
    } else {
        // SAFETY: shim 在回调期间保证 POD 有效。
        Some(unsafe { (*row).to_wire() })
    };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspQryTradingAccount {
                    last: is_last != 0,
                    row,
                });
            }
        })
    }
}

unsafe extern "C" fn td_on_rtn_order(ctx: *mut c_void, order: *const EkOrderField) {
    if order.is_null() {
        return;
    }
    // SAFETY: shim 在回调期间保证 POD 有效。
    let of = unsafe { (*order).to_wire() };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RtnOrder(of));
            }
        })
    }
}

unsafe extern "C" fn td_on_rtn_trade(ctx: *mut c_void, trade: *const EkTradeField) {
    if trade.is_null() {
        return;
    }
    // SAFETY: shim 在回调期间保证 POD 有效。
    let tf = unsafe { (*trade).to_wire() };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RtnTrade(tf));
            }
        })
    }
}

/* ---- 附加接口面(超出原稿):三路均携带错误面 ---- */

unsafe extern "C" fn td_on_rsp_qry_instrument_margin_rate(
    ctx: *mut c_void,
    row: *const EkInstrumentMarginRate,
    error: *const EkErrorField,
    is_last: std::ffi::c_int,
) {
    let row = if row.is_null() {
        None
    } else {
        // SAFETY: shim 在回调期间保证 POD 有效。
        Some(unsafe { (*row).to_wire() })
    };
    // 与查询回报同判:`error_id != 0` 才算失败(见 error_checked)。
    let error = error_checked(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspQryInstrumentMarginRate {
                    row,
                    error,
                    last: is_last != 0,
                });
            }
        })
    }
}

unsafe extern "C" fn td_on_rsp_qry_instrument_commission_rate(
    ctx: *mut c_void,
    row: *const EkInstrumentCommissionRate,
    error: *const EkErrorField,
    is_last: std::ffi::c_int,
) {
    let row = if row.is_null() {
        None
    } else {
        // SAFETY: shim 在回调期间保证 POD 有效。
        Some(unsafe { (*row).to_wire() })
    };
    let error = error_checked(error);
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RspQryInstrumentCommissionRate {
                    row,
                    error,
                    last: is_last != 0,
                });
            }
        })
    }
}

unsafe extern "C" fn td_on_rtn_instrument_status(
    ctx: *mut c_void,
    status: *const EkInstrumentStatus,
) {
    // 原稿无条件记「状态推送」面(不是查询错误) → 传 null 直接放弃。
    if status.is_null() {
        return;
    }
    // SAFETY: shim 在回调期间保证 POD 有效。
    let s = unsafe { (*status).to_wire() };
    unsafe {
        with_ingress(ctx, |ing| {
            if let Some(p) = ing.plugin() {
                p.on_td_event(TdEvent::RtnInstrumentStatus(s));
            }
        })
    }
}

static TD_CALLBACKS: TdSpiCallbacks = TdSpiCallbacks {
    on_front_connected: Some(td_on_front_connected),
    on_front_disconnected: Some(td_on_front_disconnected),
    on_rsp_authenticate: Some(td_on_rsp_authenticate),
    on_rsp_user_login: Some(td_on_rsp_user_login),
    on_rsp_settlement_info_confirm: Some(td_on_rsp_settlement_info_confirm),
    on_rsp_order_insert: Some(td_on_rsp_order_insert),
    on_rsp_order_action: Some(td_on_rsp_order_action),
    on_rsp_qry_instrument: Some(td_on_rsp_qry_instrument),
    on_rsp_qry_investor_position: Some(td_on_rsp_qry_investor_position),
    on_rsp_qry_trading_account: Some(td_on_rsp_qry_trading_account),
    on_rtn_order: Some(td_on_rtn_order),
    on_rtn_trade: Some(td_on_rtn_trade),
    on_rsp_qry_instrument_margin_rate: Some(td_on_rsp_qry_instrument_margin_rate),
    on_rsp_qry_instrument_commission_rate: Some(td_on_rsp_qry_instrument_commission_rate),
    on_rtn_instrument_status: Some(td_on_rtn_instrument_status),
};

/* ==================== 传输层 ==================== */

/// 官方 SDK 实现的 [`CtpTransport`]。
///
/// 字段声明顺序即析构顺序:会话句柄(`md`/`td`)必须先于 `ingress` 释放,
/// 以免 `Release()` 期间 SDK 仍在回调里读已释放的 `Ingress`。
pub struct NativeTransport {
    md: Mutex<Option<MdApi>>,
    td: Mutex<Option<TdApi>>,
    /// MD / TD 各自的 reqid 计数器(原稿两个独立计数器,首个 reqid = 1)。
    md_reqid: AtomicI64,
    td_reqid: AtomicI64,
    /// CTP 流文件落盘根目录(D-FLOWPATH)。
    flow_root: PathBuf,
    ingress: Arc<Ingress>,
}

impl NativeTransport {
    /// 新建传输层。`ingress` 由 [`CtpPlugin::with_native`] 在插件构造后回填。
    fn new(ingress: Arc<Ingress>, flow_root: impl Into<PathBuf>) -> Self {
        Self {
            md: Mutex::new(None),
            td: Mutex::new(None),
            md_reqid: AtomicI64::new(0),
            td_reqid: AtomicI64::new(0),
            flow_root: flow_root.into(),
            ingress,
        }
    }

    fn ingress_ptr(&self) -> *mut c_void {
        // 只用指针做回调身份,CTP 原样回传;`Arc` 由 self 持有,生命周期覆盖会话。
        Arc::as_ptr(&self.ingress) as *mut c_void
    }

    fn plugin(&self) -> Option<Arc<CtpPlugin>> {
        self.ingress.plugin()
    }

    /// 传输层级失败上报(封装块创建失败等):经插件投一条业务日志。
    /// 回调面之外没有别的出口,且这类失败在原生层无法自愈。
    fn report(&self, what: &str, detail: &str) {
        if let Some(p) = self.plugin() {
            p.transport_log(format!("{what}失败:{detail}"));
        }
    }

    fn next_md_reqid(&self) -> i32 {
        self.md_reqid.fetch_add(1, Ordering::Relaxed) as i32 + 1
    }

    fn next_td_reqid(&self) -> i32 {
        self.td_reqid.fetch_add(1, Ordering::Relaxed) as i32 + 1
    }

    /// 落盘目录 = `<flow_root>/<网关名小写>/<Md|Td>`,对齐原稿
    /// `get_folder_path(gateway_name.lower()) + "\\Md"`(D-FLOWPATH)。
    ///
    /// 驱动给的 `flow_path` 形如 `"<gw>\\Md"`,这里把网关名段小写后拼到根目录下,
    /// 并确保目录存在(CTP 自己要往里写流文件,目录不存在会创建失败)。
    fn prepare_flow_dir(&self, flow_path: &str) -> PathBuf {
        let mut parts = flow_path.splitn(2, '\\');
        let gw = parts.next().unwrap_or("ctp").to_lowercase();
        let dir = match parts.next() {
            Some(tail) if !tail.is_empty() => self.flow_root.join(gw).join(tail),
            _ => self.flow_root.join(gw),
        };
        if let Err(e) = std::fs::create_dir_all(&dir) {
            self.report("创建 CTP 流目录", &format!("{}({})", dir.display(), e));
        }
        dir
    }

    /// 释放原生会话(幂等)。先关会话,再让调用方放掉 `ingress`。
    fn shutdown(&self) {
        // take() 后在锁外析构:MdApi/TdApi 的 Drop 会调 Release(),可能阻塞。
        let md = self.md.lock().unwrap().take();
        let td = self.td.lock().unwrap().take();
        drop(md);
        drop(td);
    }
}

impl Drop for NativeTransport {
    fn drop(&mut self) {
        self.shutdown();
    }
}

impl CtpTransport for NativeTransport {
    /// 单侧断开(configure 强制重连面,0.5.0):同款 take-锁外-Drop。
    /// Release 返回后 SDK 不再回调该会话;`ingress` 仍由另一侧持有,安全。
    fn td_release(&self) {
        let td = self.td.lock().unwrap().take();
        drop(td);
    }

    fn md_release(&self) {
        let md = self.md.lock().unwrap().take();
        drop(md);
    }

    fn md_connect(&self, addr: &str, flow_path: &str) {
        let mut slot = self.md.lock().unwrap();
        if slot.is_some() {
            // 驱动已守卫(原稿 MD connect 重复调用静默),这里只是幂等兜底。
            return;
        }
        let dir = self.prepare_flow_dir(flow_path);
        match MdApi::new(&dir) {
            Ok(mut api) => {
                api.register_front(addr);
                // SAFETY: ctx 是 self.ingress 的裸指针;self 持有该 Arc 直到会话释放之后。
                unsafe { api.register_spi(&MD_CALLBACKS, self.ingress_ptr()) };
                api.init();
                *slot = Some(api);
            }
            Err(e) => self.report("创建行情 API", &e),
        }
    }

    fn md_login(&self, user: &str, password: &str, broker: &str) {
        let mut slot = self.md.lock().unwrap();
        if let Some(api) = slot.as_mut() {
            // 原稿丢弃 reqUserLogin 的返回码(ctp_gateway.py 393),这里不额外记日志。
            let _ = api.req_user_login(broker, user, password, self.next_md_reqid());
        }
    }

    fn md_subscribe(&self, symbols: &[String]) {
        let mut slot = self.md.lock().unwrap();
        let Some(api) = slot.as_mut() else { return };
        // 原稿逐条 `subscribeMarketData(symbol)`(ctp_gateway.py 398):这里保持逐条,
        // 不合并成一次数组调用 —— 调用形态变了,CTP 侧的行为也就未必是同一条路径。
        for s in symbols {
            // 原稿同样丢弃返回码(398)。
            let one = [s.as_str()];
            let _ = api.subscribe(&one);
        }
    }

    fn td_connect(&self, addr: &str, flow_path: &str) {
        let mut slot = self.td.lock().unwrap();
        if slot.is_some() {
            // 原稿重复 connect 走 else → authenticate();该分支由驱动的 connect()
            // 守卫负责(它第二次返回 ReqAuthenticate),传输层只做幂等兜底。
            return;
        }
        let dir = self.prepare_flow_dir(flow_path);
        match TdApi::new(&dir) {
            Ok(mut api) => {
                // 原稿次序(749-777):createFtdcTraderApi → subscribePrivateTopic(0)
                // → subscribePublicTopic(0) → registerFront → init。
                api.subscribe_private_topic(super::td::TE_RESUME_RESTART);
                api.subscribe_public_topic(super::td::TE_RESUME_RESTART);
                api.register_front(addr);
                // SAFETY: 同 md_connect。
                unsafe { api.register_spi(&TD_CALLBACKS, self.ingress_ptr()) };
                api.init();
                *slot = Some(api);
            }
            Err(e) => self.report("创建交易 API", &e),
        }
    }

    fn td_auth(&self, user: &str, broker: &str, auth_code: &str, app_id: &str) {
        let mut slot = self.td.lock().unwrap();
        if let Some(api) = slot.as_mut() {
            // 原稿丢弃 reqAuthenticate 的返回码(792)。
            let _ = api.req_authenticate(broker, user, auth_code, app_id, self.next_td_reqid());
        }
    }

    fn td_login(&self, user: &str, password: &str, broker: &str) {
        let mut slot = self.td.lock().unwrap();
        if let Some(api) = slot.as_mut() {
            // 原稿丢弃 reqUserLogin 的返回码(806)。
            let _ = api.req_user_login(broker, user, password, self.next_td_reqid());
        }
    }

    fn td_settlement_confirm(&self, broker: &str, user: &str) {
        let mut slot = self.td.lock().unwrap();
        if let Some(api) = slot.as_mut() {
            // 原稿 InvestorID 用的是 self.userid(483),返回码同样丢弃(485)。
            let _ = api.req_settlement_confirm(broker, user, self.next_td_reqid());
        }
    }

    /// 合约查询 —— **含原稿的流控重发**(D-FLCN)。
    ///
    /// 原稿 519-533:结算确认回报里 `while True: n = reqQryInstrument({}); reqid += 1;
    /// if n == 0: break; sleep(1)` —— 无次数上限、无日志、重试也消耗 reqid。
    /// 本函数在 native 下同样跑在 CTP 回调线程上(`emit_td` 由 `on_td_event` 驱动),
    /// 因此"阻塞重发直到发出"与原稿结构同构。
    ///
    /// ⚠ 已知风险(与原稿同):若返回码永远非 0,这里会阻塞 CTP 的回调线程不放,
    /// 后续回报都进不来。原件就是这个行为,故此处不擅自加次数上限。
    fn td_qry_instrument(&self) {
        loop {
            let rc = {
                let mut slot = self.td.lock().unwrap();
                match slot.as_mut() {
                    Some(api) => api.req_qry_instrument(self.next_td_reqid()),
                    None => return,
                }
            };
            if rc == 0 {
                return;
            }
            std::thread::sleep(Duration::from_secs(1));
        }
    }

    fn td_qry_account(&self, broker: &str, user: &str) {
        let mut slot = self.td.lock().unwrap();
        if let Some(api) = slot.as_mut() {
            // 原稿丢弃返回码(876)。
            let _ = api.req_qry_trading_account(broker, user, self.next_td_reqid());
        }
    }

    fn td_qry_position(&self, broker: &str, user: &str) {
        let mut slot = self.td.lock().unwrap();
        if let Some(api) = slot.as_mut() {
            // 原稿丢弃返回码(886)。
            let _ = api.req_qry_investor_position(broker, user, self.next_td_reqid());
        }
    }

    fn td_insert(&self, order: &InputOrder) -> i32 {
        let (pod, _holders) = order.to_pod_cstr();
        let mut slot = self.td.lock().unwrap();
        match slot.as_mut() {
            Some(api) => api.req_order_insert(&pod, self.next_td_reqid()),
            None => -1,
        }
    }

    fn td_cancel(&self, action: &InputOrderAction) -> i32 {
        let (pod, _holders) = action.to_pod_cstr();
        let mut slot = self.td.lock().unwrap();
        match slot.as_mut() {
            Some(api) => api.req_order_action(&pod, self.next_td_reqid()),
            None => -1,
        }
    }

    /// 合约保证金率查询(附加接口面;返回码交由插件按原稿式失败面处理)。
    fn td_qry_margin_rate(&self, query: &QryInstrumentMarginRate) -> i32 {
        let (pod, _holders) = query.to_pod_cstr();
        let mut slot = self.td.lock().unwrap();
        match slot.as_mut() {
            Some(api) => api.req_qry_instrument_margin_rate(&pod, self.next_td_reqid()),
            None => -1,
        }
    }

    /// 合约手续费率查询(附加接口面)。
    fn td_qry_commission_rate(&self, query: &QryInstrumentCommissionRate) -> i32 {
        let (pod, _holders) = query.to_pod_cstr();
        let mut slot = self.td.lock().unwrap();
        match slot.as_mut() {
            Some(api) => api.req_qry_instrument_commission_rate(&pod, self.next_td_reqid()),
            None => -1,
        }
    }

    fn shutdown(&self) {
        NativeTransport::shutdown(self);
    }
}

/// 供 [`CtpPlugin::with_native`] 使用的构造三元组:传输层 + 回调入口 + 回填钩子。
///
/// 单独抽出来是为了让插件侧的构造顺序显式可读:先建 `ingress`,再建传输层,
/// 再建插件,最后回填(用 `Weak` 断环,见模块文档)。
pub(crate) fn native_parts(flow_root: impl Into<PathBuf>) -> (Box<dyn CtpTransport>, Arc<Ingress>) {
    let ingress = Arc::new(Ingress::new());
    let trans = NativeTransport::new(ingress.clone(), flow_root);
    (Box::new(trans), ingress)
}

/// 回填插件弱引用(第二相接线)。由 [`CtpPlugin::with_native`] 调用。
pub(crate) fn attach_plugin(ingress: &Ingress, plugin: &Arc<CtpPlugin>) {
    ingress.attach(plugin);
}

/// 仅供测试:确认传输层的会话槽位为空/已占(不触网)。
#[cfg(test)]
pub(crate) fn md_slot_filled(t: &NativeTransport) -> bool {
    t.md.lock().unwrap().is_some()
}

/// 仅供测试:构造一个不接插件的传输层。
#[cfg(test)]
pub(crate) fn test_transport(flow_root: impl Into<PathBuf>) -> NativeTransport {
    NativeTransport::new(Arc::new(Ingress::new()), flow_root)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// 落盘目录拼接对齐原稿:`gateway_name.lower()` + `"\\Md"`,挂在根目录下。
    #[test]
    fn flow_dir_follows_original_layout() {
        let t = test_transport("root");
        assert_eq!(
            t.prepare_flow_dir("Ctp\\Md"),
            PathBuf::from("root").join("ctp").join("Md")
        );
        assert_eq!(
            t.prepare_flow_dir("Ctp\\Td"),
            PathBuf::from("root").join("ctp").join("Td")
        );
        // 无尾段时退化为网关名目录本身
        assert_eq!(t.prepare_flow_dir("Ctp"), PathBuf::from("root").join("ctp"));
        assert!(!md_slot_filled(&t));
    }

    /// reqid 从 1 起、MD/TD 各自独立(原稿两个计数器)。
    #[test]
    fn reqid_starts_at_one_and_is_per_session() {
        let t = test_transport(std::env::temp_dir());
        assert_eq!(t.next_md_reqid(), 1);
        assert_eq!(t.next_md_reqid(), 2);
        assert_eq!(t.next_td_reqid(), 1);
    }

    /// 回调入口的两种退化输入都必须**安静放弃**,而不是崩在 C++ 帧里:
    /// ① C++ 侧传了空 `ctx`;② 插件尚未接线或已析构(`Weak` 升级失败)。
    #[test]
    fn callback_ingress_is_inert_without_ctx_or_plugin() {
        // SAFETY: 传空指针是本函数契约里显式允许的退化输入。
        unsafe { with_ingress(std::ptr::null_mut(), |_| panic!("空 ctx 不得执行回调体")) };

        let ingress = Arc::new(Ingress::new());
        assert!(ingress.plugin().is_none(), "未接线时不得有插件");
        // SAFETY: ingress 是本作用域内仍然存活的对象。
        unsafe {
            with_ingress(Arc::as_ptr(&ingress) as *mut c_void, |i| {
                assert!(i.plugin().is_none(), "未接线时回调体应看到 None 并放弃")
            })
        };
    }
}