ek-ctp 0.6.0

//! # CtpPlugin —— ek-core 宿主注册的 CTP 网关插件。
//!
//! 谁发谁收:
//! - **发**:驱动输出 → `ctx.put` 上总线(纯生产者:**裸线键** `eTick.`/`eContract.`/
//!   `eOrder.`/`eTrade.`/`ePosition.`/`eAccount.`/
//!   `ePositionDone.`/`eLog`;0.4.0 D-EMIT 起裸键,网关身份归载荷 `gateway_name`;
//!   线值与 ek-oms 的 wire 模块同表同值)。
//! - **收**:`eTimer`(ek-event 秒针,2s 轮转驱动 + 行情日期刷新)与自身寻址
//!   `eReq.<名>`(订阅/发单/撤单三项服务,对齐 oms 的 ask 面形制)。
//!
//! 传输层(`CtpTransport`)是 native 桥的接缝:默认构建无传输(可测纯逻辑与
//! 请求面),native 特性(0.2.0)以官方 SDK 二进制实现该接口。
//!
//! 约定:
//! - 原稿 `CtpGateway.count`(224-233 的 2 秒 gate)在此为插件内计数器;
//! - `md.update_date` 与查询轮转同 gate(共节拍);
//! - `EVENT_POSITION_QUERY_DONE` 落线值 `ePositionDone.<gw>`(与 ek-oms 相位门同表)。

use crate::driver::{self, MdDriver, MdEvent, MdOut, TdDriver, TdEvent, TdOut};
use crate::wire::InputOrder;
use ek_core::context::PluginContext;
use ek_core::envelope::{Envelope, ReplyErr, e_req_type};
use ek_core::plugin::Plugin;
use ek_core::service::ServiceDesc;
use ek_data::cancel_request::CancelRequest;
use ek_data::log::LogData;
// 只被 native 面的 `transport_log` 使用(默认构建没有它,故按特性引入)。
#[cfg(feature = "native")]
use ek_data::log::LEVEL_ERROR;
use ek_data::order_request::OrderRequest;
use ek_event::Event;
use std::borrow::Cow;
use std::sync::{Arc, Mutex, OnceLock};

// ================ 总线线值(注意:与 ek-oms 的 wire 模块同表同值) ================

const W_TICK: &str = "eTick.";
const W_ORDER: &str = "eOrder.";
const W_TRADE: &str = "eTrade.";
const W_POSITION: &str = "ePosition.";
const W_ACCOUNT: &str = "eAccount.";
const W_CONTRACT: &str = "eContract.";
/// 原稿 EVENT_POSITION_QUERY_DONE(event.py:17):**裸键,无 gw 后缀**,gw 放载荷
/// ——ek-oms 侧以该裸键精确订阅(CtpPlugin emit 面),此处保持一字面。
const W_POS_DONE: &str = "ePositionDone.";
/// 日志线单源(ek-data::event::EVENT_LOG = "eLog",不再本地复定义)。
use ek_data::event::EVENT_LOG as W_LOG;
/// ek-event 秒针广播键(引用共享常量,不再重复定义字面值)。
/// 由 [ek_event::EVENT_TIMER] 单源提供。
pub use ek_event::EVENT_TIMER as ROUTE_TIMER;

// ================ 服务请求/回包契约(ask 载荷类型) ================

/// 订阅行情请求。
#[derive(Debug, Clone)]
pub struct SubscribeReq {
    /// 合约代码symbols(逐笔)
    pub symbols: Vec<String>,
}

/// 订阅请求回包。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubscribeReply {
    /// 是否受理(true = 已登记/已发起)
    pub ok: bool,
    /// 说明
    pub msg: String,
}

/// 发单请求。
#[derive(Debug, Clone)]
pub struct SendOrderReq {
    /// 下单请求
    pub req: OrderRequest,
}

/// 发单回包。
#[derive(Debug, Clone, PartialEq)]
pub struct SendOrderReply {
    /// 是否受理(原稿 `""` 面 → false)
    pub ok: bool,
    /// 成功 = `"{gw}.{front}_{session}_{ref}"`(OrderData.en_orderid 同式)
    pub en_orderid: String,
    /// 拒绝时说明
    pub msg: String,
}

/// 撤单请求。
#[derive(Debug, Clone)]
pub struct CancelOrderReq {
    /// 撤单请求
    pub req: CancelRequest,
}

/// 撤单回包。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CancelOrderReply {
    /// 是否受理(true = 命令已交付)
    pub ok: bool,
    /// 说明
    pub msg: String,
}

/// 费率/保证金率查询请求(附加接口面,超出原稿)。
///
/// `kind`: "margin" = 保证金率(hedge 过滤参与查询);"commission" = 手续费率
/// (无 hedge 过滤)。查询结果经 TD 回调 → 驱动 eLog 落线(不新增事件实体)。
#[derive(Debug, Clone)]
pub struct QueryRateReq {
    /// 查询类型("margin" / "commission")
    pub kind: String,
    /// 合约代码
    pub instrument_id: String,
    /// 投机套保档(仅 margin 查询用;'1'投机)
    pub hedge_flag: String,
    /// 交易所代码(过滤;可空)
    pub exchange_id: String,
}

/// 费率查询回包(是否已交付 + RT 返回码文本)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueryRateReply {
    /// 是否受理(true = 命令已交付 native)
    pub ok: bool,
    /// 说明(交付失败时含错误代码)
    pub msg: String,
}

/// 网关配置请求(**插件自带配置面**,2026-10-17 定盘:配置归插件自身,
/// 界面/宿主手工输入经 ask 进来;插件永不读配置文件)。
///
/// 语义 = 更新账号/前置 + 复位连接守卫 + 立即发起连接(先 TD 后 MD,与
/// `connect` 同式)。`md_user`/`md_password` 留空回落 `td_user`/`td_password`
/// (SimNow 同账号常态,界面少填两格)。
#[derive(Debug, Clone, Default)]
pub struct ConfigureReq {
    /// 经纪公司代码
    pub broker_id: String,
    /// 交易前置(空 = 不连 TD;与 md_addr 至少一个非空)
    pub td_addr: String,
    /// 交易用户
    pub td_user: String,
    /// 交易密码
    pub td_password: String,
    /// 授权编码(空 = 直接登录)
    pub td_auth_code: String,
    /// 产品名称(AppID)
    pub td_app_id: String,
    /// 行情前置(空 = 不连 MD)
    pub md_addr: String,
    /// 行情用户(空回落 td_user)
    pub md_user: String,
    /// 行情密码(空回落 td_password)
    pub md_password: String,
    /// 收录的产品类白名单(CTP 单字符码,如 `["1"]`;空 = 回落默认「期货」)。
    /// 界面勾选面:期货 `'1'` / 期权 `'2'`(现货期权 `'6'` 可配但不在默认勾选里)。
    pub product_classes: Vec<String>,
}

/// 配置回包(ok = 已应用并发起连接;false = 校验未过,msg 报字段)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConfigureReply {
    /// 是否受理
    pub ok: bool,
    /// 说明
    pub msg: String,
    /// 实际生效的产品类白名单(已归一化;空请求回落默认「期货」`'1'`)。
    /// 校验未过时回带**当前仍在生效**的那一份 —— 不是空表:空表会让界面误清空勾选,
    /// 而网关其实还跑在旧勾选上。
    pub product_classes: Vec<String>,
}

// ================ 传输层接缝(native 桥实现/dev 以测试桩实现) ================

/// 传输出口:驱动动作落到真实 CTP 会话(0.2.0 native 才有官方实现)。
pub trait CtpTransport: Send + Sync {
    /// 创建行情 API 并 registerFront/init。
    fn md_connect(&self, addr: &str, flow_path: &str);
    /// 行情登录。
    fn md_login(&self, user: &str, password: &str, broker: &str);
    /// 行情订阅。
    fn md_subscribe(&self, symbols: &[String]);
    /// 创建交易 API 并 registerFront/init。
    fn td_connect(&self, addr: &str, flow_path: &str);
    /// 认证。
    fn td_auth(&self, user: &str, broker: &str, auth_code: &str, app_id: &str);
    /// 登录。
    fn td_login(&self, user: &str, password: &str, broker: &str);
    /// 结算单确认。
    fn td_settlement_confirm(&self, broker: &str, user: &str);
    /// 合约查询。
    fn td_qry_instrument(&self);
    /// 资金查询。
    fn td_qry_account(&self, broker: &str, user: &str);
    /// 持仓查询。
    fn td_qry_position(&self, broker: &str, user: &str);
    /// 插单;返回 `reqOrderInsert` 的 rt 返回码(0 = 受理;非 0 = 原稿 846-848 的
    /// 失败面:记「委托请求发送失败,错误代码:n」且**不回 SUBMITTING 单**)。
    fn td_insert(&self, order: &InputOrder) -> i32;
    /// 撤单;返回 `reqOrderAction` 的 rt 返回码(原稿不检,原样透传给 native 自定)。
    fn td_cancel(&self, action: &crate::wire::InputOrderAction) -> i32;
    /// 合约保证金率查询(附加接口面;返回 `ReqQryInstrumentMarginRate` rt 返回码)。
    fn td_qry_margin_rate(&self, query: &crate::wire::QryInstrumentMarginRate) -> i32;
    /// 合约手续费率查询(附加接口面;返回 `ReqQryInstrumentCommissionRate` rt 返回码)。
    fn td_qry_commission_rate(&self, query: &crate::wire::QryInstrumentCommissionRate) -> i32;

    // 交易状态没有查询面(SDK 6.7.7 无 ReqQryInstrumentStatus):该件纯被动,
    // 由公有流推送经 TD 回调直达驱动,传输层不提供主动入口。

    /// 关闭原生会话(宿主停插件时调用)。
    ///
    /// 默认空实现 = 无资源可放。官方实现里这一步是必需的:`Release()` 之后 SDK
    /// **不再回调**,回调入口(ctx)才允许被放下 —— 顺序反了就是 use-after-free。
    fn shutdown(&self) {}

    /// 断开交易会话(释放句柄;默认空 = 纯逻辑面无会话可断)。
    /// configure 强制重连面(0.5.0)用:旧登录态先断,新账号/密码才有显形机会。
    fn td_release(&self) {}
    /// 断开行情会话(释放句柄;默认空 = 纯逻辑面无会话可断)。
    fn md_release(&self) {}
}

// ================ 网关配置 ================

/// 网关配置(对齐原稿 connect setting 141-171;地址补前缀由使用方完成——
/// 原稿在 connect 里对缺 "tcp://"/"ssl://" 的地址自动补 "tcp://",
/// 本 crate 要求调用方先行规整,保持驱动数据纯净)。
#[derive(Debug, Clone, Default)]
pub struct CtpConfig {
    /// 网关/插件注册名(唯一;默认 "ctp")
    pub gateway_name: String,
    /// 经纪公司代码
    pub broker_id: String,
    /// 交易前置
    pub td_addr: String,
    /// 交易用户
    pub td_user: String,
    /// 交易密码
    pub td_password: String,
    /// 授权编码(空 = 直接登录,原稿 448-451)
    pub td_auth_code: String,
    /// 产品名称(AppID)
    pub td_app_id: String,
    /// 行情前置
    pub md_addr: String,
    /// 行情用户
    pub md_user: String,
    /// 行情密码
    pub md_password: String,
    /// 收录的产品类白名单(D-PC-FILTER;空 = 回落默认「期货」`'1'`)。
    /// 查询侧不过滤,收录侧按它严格丢弃(不进共享表、不上抛 `eContract`)。
    pub product_classes: crate::map::ProductClasses,
}

impl CtpConfig {
    /// 缺省名 "ctp" 的兜底。
    pub fn gateway_name_or_default(&self) -> String {
        if self.gateway_name.is_empty() {
            "ctp".to_string()
        } else {
            self.gateway_name.clone()
        }
    }
}

// ================ 插件 ================

/// CTP 网关插件(纯生产者 + 自服务)。
pub struct CtpPlugin {
    gw: String,
    /// 当前网关配置(connect 空地址守卫与费率查询的单一事实源)。
    cfg: Mutex<CtpConfig>,
    /// 经纪公司代码(费率查询的 broker_id;configure 可换)。
    broker: Mutex<String>,
    /// 投资者代码(费率查询的 investor_id;configure 可换)。
    user: Mutex<String>,
    md: Mutex<MdDriver>,
    td: Mutex<TdDriver>,
    /// 2s gate 计数器(原稿 self.count)
    timer_count: Mutex<u32>,
    /// 传输层(默认构建 None;native/测试注入)
    trans: Option<Box<dyn CtpTransport>>,
    ctx: OnceLock<PluginContext>,
    /// 「未启用传输」提示只发一次
    trans_warned: Mutex<bool>,
}

impl CtpPlugin {
    /// 纯逻辑构造(native 尚未接传输)。
    pub fn new(cfg: CtpConfig) -> Self {
        let gw = cfg.gateway_name_or_default();
        Self::with_parts(cfg, gw, None)
    }

    /// 注入传输层(native 桥或测试桩)。
    pub fn with_transport(cfg: CtpConfig, trans: Box<dyn CtpTransport>) -> Self {
        let gw = cfg.gateway_name_or_default();
        Self::with_parts(cfg, gw, Some(trans))
    }

    fn with_parts(cfg: CtpConfig, gw: String, trans: Option<Box<dyn CtpTransport>>) -> Self {
        let contracts = driver::contracts_empty();
        let cfg_broker = cfg.broker_id.clone();
        let cfg_user = cfg.td_user.clone();
        let md_cfg = driver::MdConfig {
            addr: cfg.md_addr.clone(),
            user: cfg.md_user.clone(),
            password: cfg.md_password.clone(),
            broker: cfg.broker_id.clone(),
        };
        let td_cfg = driver::TdConfig {
            addr: cfg.td_addr.clone(),
            user: cfg.td_user.clone(),
            password: cfg.td_password.clone(),
            broker: cfg.broker_id.clone(),
            auth_code: cfg.td_auth_code.clone(),
            app_id: cfg.td_app_id.clone(),
            product_classes: cfg.product_classes.clone(),
        };
        // 原稿 md.current_date 由 datetime.now() 起手;这里初值 = 启动时的上海日期
        let md = Mutex::new(MdDriver::new(
            &gw,
            md_cfg,
            contracts.clone(),
            crate::map::shanghai_now_date(),
        ));
        let td = Mutex::new(TdDriver::new(&gw, td_cfg, contracts.clone()));
        Self {
            gw,
            cfg: Mutex::new(cfg),
            broker: Mutex::new(cfg_broker),
            user: Mutex::new(cfg_user),
            md,
            td,
            timer_count: Mutex::new(0),
            trans,
            ctx: OnceLock::new(),
            trans_warned: Mutex::new(false),
        }
    }

    /// native 构造:装配官方 SDK 传输层并完成回调接线(见 `native::transport` 文档)。
    ///
    /// 两相接线:① 先建回调入口 + 传输层(传输层持有 `Arc<Ingress>`);② 插件建好后
    /// 回填 `Weak<CtpPlugin>` —— 必须用弱引用断环,否则 插件 ↔ 传输层 互相持有,
    /// 两者永不析构(回调也不会停)。
    ///
    /// `flow_root` = CTP 流文件落盘根目录(D-FLOWPATH)。构造本身**不碰磁盘**:
    /// 目录在真正连接时才按 `<flow_root>/<网关名小写>/{Md,Td}` 创建。
    #[cfg(feature = "native")]
    pub fn with_native(cfg: CtpConfig, flow_root: impl Into<std::path::PathBuf>) -> Arc<Self> {
        let (trans, ingress) = crate::native::transport::native_parts(flow_root);
        let plugin = Arc::new(Self::with_transport(cfg, trans));
        crate::native::transport::attach_plugin(&ingress, &plugin);
        plugin
    }

    /// 连接(原稿 `CtpGateway.connect` 162-190:先 TD 后 MD;`init_query` 只是挂秒针)。
    ///
    /// **连接触发点住在本插件里**(不放到核心/公共层,每个插件自持自己的连接):
    /// 宿主 `start()` 会经 [`Plugin::on_start`] 调它一次;要显式控制(先配后连、
    /// 断线后重连)也可以直接调。幂等由驱动守卫裁决 —— 原稿 MD 重复调用静默,
    /// TD 重复调用转认证(776-777),两者都在 `MdDriver::connect`/`TdDriver::connect`。
    ///
    /// 原稿末尾的 `init_query` 只是把查询轮转注册到 `EVENT_TIMER`(237-240);
    /// 本 crate 的对应面是宿主 eTimer → [`CtpPlugin::on_timer`],这里无需动作。
    pub fn connect(&self) {
        // 空地址守卫(0.3.0):td/md 前置都空 → 不瞎连(on_start 空配置形态),
        // 等 `{gw}.configure` 手工注入后再发起(插件自带配置面,定盘口径)。
        let (td_empty, md_empty) = {
            let c = self.cfg.lock().unwrap();
            (c.td_addr.is_empty(), c.md_addr.is_empty())
        };
        if td_empty && md_empty {
            self.put_log(LogData {
                gateway_name: self.gw.clone(),
                msg: "未配置前置地址,跳过连接(等待 configure)".to_string(),
                level: 20,
            });
            return;
        }
        // 先 TD 后 MD:与原稿 connect 的调用次序一致。
        // 锁在各自语句结束时释放 —— emit 可能落到会阻塞的原生调用(如合约查询的
        // 流控重发),绝不能在持驱动锁时做。
        let td_outs = if td_empty {
            Vec::new()
        } else {
            self.td.lock().unwrap().connect()
        };
        let md_outs = if md_empty {
            Vec::new()
        } else {
            self.md.lock().unwrap().connect()
        };
        for o in td_outs {
            self.emit_td(o);
        }
        for o in md_outs {
            self.emit_md(o);
        }
    }

    /// 传输层级失败上报:native 桥在回调面之外没有别的出口。
    ///
    /// 只用在本插件/原生层自己的故障上(建不出 API、建不出流目录)。原稿没有对应面
    /// ——那里失败会直接抛异常,进程级可见;Rust 侧不能抛,所以落一条 ERROR 日志。
    /// **请求返回码不在此列**:原稿除下单外一律丢弃返回码,那些日志会是原稿没有的噪音。
    #[cfg(feature = "native")]
    pub(crate) fn transport_log(&self, msg: String) {
        self.put_log(LogData {
            gateway_name: self.gw.clone(),
            msg,
            level: LEVEL_ERROR,
        });
    }

    /// 原生适配入口(native 桥的回调直接调这四个):MD 回报步进 + 总线上抛。
    pub fn on_md_event(&self, ev: MdEvent) {
        let outs = self.md.lock().unwrap().step(ev);
        for o in outs {
            self.emit_md(o);
        }
    }

    /// 原生适配入口(native 桥的回调直接调这四个):TD 回报步进 + 总线上抛。
    pub fn on_td_event(&self, ev: TdEvent) {
        let outs = self.td.lock().unwrap().step(ev);
        for o in outs {
            self.emit_td(o);
        }
    }

    /// eTimer 一拍(原稿 gateway.process_timer_event):2s gate → 轮转查询 +
    /// 行情日期刷新。这样独立出来服务测试与 native 适配。
    pub fn on_timer(&self) {
        let gate_pass = {
            let mut c = self.timer_count.lock().unwrap();
            *c += 1;
            if *c < 2 {
                false
            } else {
                *c = 0;
                true
            }
        };
        if !gate_pass {
            return;
        }
        // 原稿 231-235 次序:先 pop 查询函数执行,再 md_api.update_date
        let outs = self.td.lock().unwrap().on_timer();
        for o in outs {
            self.emit_td(o);
        }
        let date = crate::map::shanghai_now_date();
        self.md.lock().unwrap().update_date(date);
    }

    fn emit_md(&self, o: MdOut) {
        match o {
            MdOut::Log(l) => self.put_log(l),
            MdOut::Tick(t) => self.put_data(W_TICK, t),
            MdOut::ReqConnect { addr, flow_path } => {
                if let Some(t) = &self.trans {
                    t.md_connect(&addr, &flow_path);
                } else {
                    self.warn_no_transport();
                }
            }
            MdOut::ReqLogin {
                user,
                password,
                broker,
            } => {
                if let Some(t) = &self.trans {
                    t.md_login(&user, &password, &broker);
                } else {
                    self.warn_no_transport();
                }
            }
            MdOut::ReqSubscribe(symbols) => {
                if let Some(t) = &self.trans {
                    t.md_subscribe(&symbols);
                } else {
                    self.warn_no_transport();
                }
            }
        }
    }

    fn emit_td(&self, o: TdOut) {
        match o {
            TdOut::Log(l) => self.put_log(l),
            TdOut::Contract(c) => self.put_data(W_CONTRACT, c),
            TdOut::Order(o) => self.put_data(W_ORDER, o),
            TdOut::Trade(t) => self.put_data(W_TRADE, t),
            TdOut::Position(p) => self.put_data(W_POSITION, p),
            TdOut::Account(a) => self.put_data(W_ACCOUNT, a),
            TdOut::PositionQueryDone(gw) => {
                // 原稿 593:on_event(EVENT_POSITION_QUERY_DONE, self.gateway_name)
                // → 裸键 "ePositionDone." + 载荷 gw(ek-oms 以此精确订阅)
                if let Some(ctx) = self.ctx.get() {
                    let _ = ctx.put(Event::with_data(W_POS_DONE, Arc::new(gw)));
                }
            }
            // 原稿 EVENT_CONTRACT_FINISH = "eContractFinish":全局线无 gw 后缀;
            // 载荷放网关名(多网关辨源,D-CFIN)。
            TdOut::ContractFinish(gw) => {
                if let Some(ctx) = self.ctx.get() {
                    let _ = ctx.put(Event::with_data("eContractFinish", Arc::new(gw)));
                }
            }
            TdOut::ReqConnect { addr, flow_path } => {
                if let Some(t) = &self.trans {
                    t.td_connect(&addr, &flow_path);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqAuthenticate {
                user,
                broker,
                auth_code,
                app_id,
            } => {
                if let Some(t) = &self.trans {
                    t.td_auth(&user, &broker, &auth_code, &app_id);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqLogin {
                user,
                password,
                broker,
            } => {
                if let Some(t) = &self.trans {
                    t.td_login(&user, &password, &broker);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqSettlementConfirm { broker, user } => {
                if let Some(t) = &self.trans {
                    t.td_settlement_confirm(&broker, &user);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqQryInstrument => {
                if let Some(t) = &self.trans {
                    t.td_qry_instrument();
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqQryAccount { broker, user } => {
                if let Some(t) = &self.trans {
                    t.td_qry_account(&broker, &user);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqQryPosition { broker, user } => {
                if let Some(t) = &self.trans {
                    t.td_qry_position(&broker, &user);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::Insert(o) => {
                if let Some(t) = &self.trans {
                    t.td_insert(&o);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::Cancel(a) => {
                if let Some(t) = &self.trans {
                    t.td_cancel(&a);
                } else {
                    self.warn_no_transport();
                }
            }
        }
    }

    /// 业务日志统一出口(LogData 挂 "eLog" 固定键;ctx 文档同款)。
    fn put_log(&self, l: LogData) {
        if let Some(ctx) = self.ctx.get() {
            let _ = ctx.put(Event::with_data(W_LOG, Arc::new(l)));
        }
    }

    /// 数据上抛(**裸线键 `eXxx.` + 载荷**;0.4.0 集成修正)。
    ///
    /// 原稿 emit 面带 gw 后缀(`eOrder.ctp`),但 ek 总线派发是精确匹配
    /// (ek-event register 精确键,ek-core 照 routes 原样注册),而消费侧
    /// dm/oms 注册的是裸前缀(`eTick.`/`eOrder.`)——带后缀的行**没有任何
    /// 订阅者**,整条 ctp 数据流在装配里被静默丢弃(T5 实连揭出)。网关身份
    /// 本就在载荷 `gateway_name` 字段里,消费侧按它分流足够;sim 侧同款裸键
    /// 先例(ui_sim 全绿)钉死此契约。见 DESIGN D-EMIT。
    fn put_data<T: Send + Sync + 'static>(&self, prefix: &str, payload: T) {
        if let Some(ctx) = self.ctx.get() {
            let _ = ctx.put(Event::with_data(prefix, Arc::new(payload)));
        }
    }

    /// 无传输时的静默风告(只发一次;info 级防刷屏)。
    fn warn_no_transport(&self) {
        let mut warned = self.trans_warned.lock().unwrap();
        if !*warned {
            *warned = true;
            self.put_log(LogData {
                gateway_name: self.gw.clone(),
                msg: "传输层未接入(默认构建):连接/查询/发送动作已被吸收".to_string(),
                level: 20,
            });
        }
    }

    /// 服务面:`eReq.<gw>` 分派(arm 范式与 oms 同)。
    fn on_request(&self, event: &Event) {
        let Some(env) = event.data.downcast_ref::<Envelope>() else {
            return;
        };
        macro_rules! arm {
            ($ty:ty, $res:ty, $body:expr) => {
                if let Some(req) = env.payload.downcast_ref::<$ty>() {
                    let res: $res = $body(req);
                    let _ = env.reply.send(Box::new(Ok::<$res, ReplyErr>(res)));
                    return;
                }
            };
        }
        arm!(SubscribeReq, SubscribeReply, |r: &SubscribeReq| self
            .do_subscribe(r));
        arm!(SendOrderReq, SendOrderReply, |r: &SendOrderReq| self
            .do_send_order(r));
        arm!(CancelOrderReq, CancelOrderReply, |r: &CancelOrderReq| self
            .do_cancel_order(r));
        arm!(QueryRateReq, QueryRateReply, |r: &QueryRateReq| self
            .do_query_rate(r));
        arm!(ConfigureReq, ConfigureReply, |r: &ConfigureReq| self
            .do_configure(r));
        // 载荷不是任何已声明契约 → 协议错误(宿主既有语义)
        let _ = env
            .reply
            .send(Box::new(Err::<(), ReplyErr>(ReplyErr::InvalidPayload)));
    }

    fn do_subscribe(&self, r: &SubscribeReq) -> SubscribeReply {
        let mut outs = Vec::new();
        {
            let mut md = self.md.lock().unwrap();
            for s in &r.symbols {
                outs.extend(md.subscribe(s));
            }
        }
        for o in outs {
            self.emit_md(o);
        }
        SubscribeReply {
            ok: true,
            msg: String::new(),
        }
    }

    fn do_send_order(&self, r: &SendOrderReq) -> SendOrderReply {
        let outcome = {
            let mut td = self.td.lock().unwrap();
            td.send_order(&r.req)
        };
        if let Some(log) = &outcome.error {
            self.put_log(log.clone());
        }
        // 原稿 844-848:插单命令的 rt 返回码由传输层裁决;n != 0 →
        // 日志「委托请求发送失败,错误代码:n」且回 ""(不回 SUBMITTING 单)。
        // 其余命令面照常下发。
        let mut submit_order: Option<ek_data::order::OrderData> = None;
        let mut send_failed: Option<String> = None;
        for o in outcome.outs {
            match o {
                TdOut::Insert(input) => match &self.trans {
                    Some(t) => {
                        let n = t.td_insert(&input);
                        if n != 0 {
                            send_failed = Some(format!("委托请求发送失败,错误代码:{n}"));
                        }
                    }
                    None => self.warn_no_transport(),
                },
                TdOut::Order(od) => submit_order = Some(od),
                other => self.emit_td(other),
            }
        }
        if let Some(msg) = send_failed {
            let log = LogData {
                gateway_name: self.gw.clone(),
                msg: msg.clone(),
                level: 20,
            };
            self.put_log(log);
            return SendOrderReply {
                ok: false,
                en_orderid: String::new(),
                msg,
            };
        }
        if let Some(od) = submit_order {
            self.emit_td(TdOut::Order(od));
        }
        if outcome.en_orderid.is_empty() {
            let msg = outcome.error.map(|l| l.msg).unwrap_or_default();
            SendOrderReply {
                ok: false,
                en_orderid: String::new(),
                msg,
            }
        } else if self.trans.is_none() {
            self.warn_no_transport();
            SendOrderReply {
                ok: false,
                en_orderid: outcome.en_orderid,
                msg: "传输层未接入(默认构建)".to_string(),
            }
        } else {
            SendOrderReply {
                ok: true,
                en_orderid: outcome.en_orderid,
                msg: String::new(),
            }
        }
    }

    /// 费率查询实现(附加接口面)。margin → `td_qry_margin_rate`,
    /// commission → `td_qry_commission_rate`;status 无 Req,不在此线
    /// (纯被动,官文见 DESIGN)。RT 非零命中原稿式发送失败日志并回值。
    fn do_query_rate(&self, r: &QueryRateReq) -> QueryRateReply {
        let broker = self.broker.lock().unwrap().clone();
        let user = self.user.lock().unwrap().clone();
        let is_margin = r.kind == "margin";
        let is_commission = r.kind == "commission";
        if !is_margin && !is_commission {
            self.put_log(LogData {
                gateway_name: self.gw.clone(),
                msg: format!("费率查询类型未知:{}", r.kind),
                level: 20,
            });
            return QueryRateReply {
                ok: false,
                msg: format!("费率查询类型未知:{}", r.kind),
            };
        }
        if let Some(t) = &self.trans {
            let n = if is_margin {
                let query = crate::wire::QryInstrumentMarginRate {
                    broker_id: broker.clone(),
                    investor_id: user.clone(),
                    hedge_flag: r.hedge_flag.clone(),
                    exchange_id: r.exchange_id.clone(),
                    invest_unit_id: String::new(),
                    instrument_id: r.instrument_id.clone(),
                };
                t.td_qry_margin_rate(&query)
            } else {
                let query = crate::wire::QryInstrumentCommissionRate {
                    broker_id: broker.clone(),
                    investor_id: user.clone(),
                    exchange_id: r.exchange_id.clone(),
                    invest_unit_id: String::new(),
                    instrument_id: r.instrument_id.clone(),
                };
                t.td_qry_commission_rate(&query)
            };
            let label = if is_margin {
                "保证金率"
            } else {
                "手续费率"
            };
            if n != 0 {
                // 同会话查询节律 1 次/秒(eTimer 轮转同用每秒名额):RT != 0 时隔
                // 1.05s 重试一次(对齐示例第 11 步与 D-FLCN 的节拍纪律)。
                std::thread::sleep(std::time::Duration::from_millis(1050));
                let n2 = if is_margin {
                    let query = crate::wire::QryInstrumentMarginRate {
                        broker_id: broker.clone(),
                        investor_id: user.clone(),
                        hedge_flag: r.hedge_flag.clone(),
                        exchange_id: r.exchange_id.clone(),
                        invest_unit_id: String::new(),
                        instrument_id: r.instrument_id.clone(),
                    };
                    t.td_qry_margin_rate(&query)
                } else {
                    let query = crate::wire::QryInstrumentCommissionRate {
                        broker_id: broker.clone(),
                        investor_id: user.clone(),
                        exchange_id: r.exchange_id.clone(),
                        invest_unit_id: String::new(),
                        instrument_id: r.instrument_id.clone(),
                    };
                    t.td_qry_commission_rate(&query)
                };
                let n = n2;
                if n != 0 {
                    let msg = format!("{label}查询请求发送失败,错误代码:{n}(重发一次仍拒)");
                    self.put_log(LogData {
                        gateway_name: self.gw.clone(),
                        msg: msg.clone(),
                        level: 20,
                    });
                    return QueryRateReply { ok: false, msg };
                }
                QueryRateReply {
                    ok: true,
                    msg: format!("{label}查询首次被拒(错误代码另附),重发后受理"),
                }
            } else {
                QueryRateReply {
                    ok: true,
                    msg: String::new(),
                }
            }
        } else {
            self.warn_no_transport();
            QueryRateReply {
                ok: false,
                msg: "传输层未接入(默认构建)".to_string(),
            }
        }
    }

    fn do_cancel_order(&self, r: &CancelOrderReq) -> CancelOrderReply {
        let outs = {
            let mut td = self.td.lock().unwrap();
            td.cancel_order(&r.req)
        };
        let mut rejected_log: Option<LogData> = None;
        let outs: Vec<TdOut> = outs
            .into_iter()
            .filter(|o| match o {
                TdOut::Log(l) => {
                    rejected_log = Some(l.clone());
                    false
                }
                _ => true,
            })
            .collect();
        if let Some(l) = rejected_log {
            let msg = l.msg.clone();
            self.put_log(l);
            return CancelOrderReply { ok: false, msg };
        }
        for o in outs {
            self.emit_td(o);
        }
        if self.trans.is_none() {
            self.warn_no_transport();
            return CancelOrderReply {
                ok: false,
                msg: "传输层未接入(默认构建)".to_string(),
            };
        }
        CancelOrderReply {
            ok: true,
            msg: String::new(),
        }
    }

    /// 配置实现(插件自带配置面,0.3.0;定盘:界面/宿主手工输入经 ask,
    /// 插件不读配置文件)。
    ///
    /// 校验 → 应用(驱动 reconfigure 复位连接/认证/登录守卫 + 插件侧
    /// cfg/broker/user 换血)→ **强制重连**(0.5.0:先 `td_release`/`md_release`
    /// 断开旧会话——传输层幂等 connect 下,不断开就没有新登录显形,换账号/
    /// 改密码永不生效;Release 归 RAII Drop,可能阻塞到回调收尾)→ `connect()`
    /// (先 TD 后 MD)。纯逻辑面 release 默认空,行为不变。
    fn do_configure(&self, r: &ConfigureReq) -> ConfigureReply {
        self.configure(r)
    }

    /// 当前生效的产品类勾选(D-PC-FILTER)。
    ///
    /// 校验被拒时也回带它,而不是空表 —— 空表会让界面把勾选框清空,而网关其实还跑在
    /// 旧勾选上,那是假信息。
    fn effective_classes(&self) -> Vec<String> {
        self.td.lock().unwrap().product_classes()
    }

    /// 配置口(公开,与 `connect` 同级的显式控制面;ask 臂 `{gw}.configure`
    /// 走的也是它)。
    pub fn configure(&self, r: &ConfigureReq) -> ConfigureReply {
        if r.td_addr.is_empty() && r.md_addr.is_empty() {
            return ConfigureReply {
                ok: false,
                msg: "配置校验: td_addr/md_addr 至少一个非空".to_string(),
                product_classes: self.effective_classes(),
            };
        }
        if !r.td_addr.is_empty()
            && (r.broker_id.is_empty() || r.td_user.is_empty() || r.td_password.is_empty())
        {
            return ConfigureReply {
                ok: false,
                msg: "配置校验: TD 侧需 broker_id/td_user/td_password 齐".to_string(),
                product_classes: self.effective_classes(),
            };
        }
        let md_user = if !r.md_user.is_empty() {
            r.md_user.clone()
        } else {
            r.td_user.clone()
        };
        let md_password = if !r.md_password.is_empty() {
            r.md_password.clone()
        } else {
            r.td_password.clone()
        };
        if !r.md_addr.is_empty() && (md_user.is_empty() || md_password.is_empty()) {
            return ConfigureReply {
                ok: false,
                msg: "配置校验: MD 侧需 md_user/md_password(或 td 侧回落)".to_string(),
                product_classes: self.effective_classes(),
            };
        }
        // 产品类勾选校验(D-PC-FILTER):CTP 产品类码是单字符。非法项不静默接受,
        // 直接拒并列出;空串项视为「没填」,由归一化忽略(全空 → 回落默认期货)。
        let bad_codes: Vec<String> = r
            .product_classes
            .iter()
            .map(|c| c.trim().to_string())
            .filter(|c| !c.is_empty() && c.chars().count() != 1)
            .collect();
        if !bad_codes.is_empty() {
            return ConfigureReply {
                ok: false,
                msg: format!(
                    "配置校验: product_classes 需为单字符 CTP 码(非法项:{})",
                    bad_codes.join(",")
                ),
                product_classes: self.effective_classes(),
            };
        }
        let classes = crate::map::ProductClasses(r.product_classes.clone());
        let effective_classes = classes.normalize();
        {
            self.td.lock().unwrap().reconfigure(driver::TdConfig {
                addr: r.td_addr.clone(),
                user: r.td_user.clone(),
                password: r.td_password.clone(),
                broker: r.broker_id.clone(),
                auth_code: r.td_auth_code.clone(),
                app_id: r.td_app_id.clone(),
                product_classes: classes.clone(),
            });
        }
        {
            self.md.lock().unwrap().reconfigure(driver::MdConfig {
                addr: r.md_addr.clone(),
                user: md_user,
                password: md_password,
                broker: r.broker_id.clone(),
            });
        }
        *self.broker.lock().unwrap() = r.broker_id.clone();
        *self.user.lock().unwrap() = r.td_user.clone();
        *self.cfg.lock().unwrap() = CtpConfig {
            gateway_name: self.gw.clone(),
            broker_id: r.broker_id.clone(),
            td_addr: r.td_addr.clone(),
            td_user: r.td_user.clone(),
            td_password: r.td_password.clone(),
            td_auth_code: r.td_auth_code.clone(),
            td_app_id: r.td_app_id.clone(),
            md_addr: r.md_addr.clone(),
            md_user: r.md_user.clone(),
            md_password: r.md_password.clone(),
            product_classes: classes,
        };
        // 强制重连(0.5.0):旧会话先断,connect() 才会新建句柄走全链登录。
        if let Some(t) = &self.trans {
            t.td_release();
            t.md_release();
        }
        self.connect();
        ConfigureReply {
            ok: true,
            msg: format!(
                "已配置并发起连接(先 TD 后 MD);收录产品类:{}",
                effective_classes.join(" ")
            ),
            product_classes: effective_classes,
        }
    }
}

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

impl Plugin for CtpPlugin {
    fn name(&self) -> &str {
        &self.gw
    }

    fn routes(&self) -> Vec<Cow<'static, str>> {
        vec![Cow::Owned(e_req_type(&self.gw)), Cow::Borrowed(ROUTE_TIMER)]
    }

    fn version(&self) -> Cow<'static, str> {
        Cow::Borrowed(env!("CARGO_PKG_VERSION"))
    }

    fn services(&self) -> Vec<ServiceDesc> {
        vec![
            ServiceDesc::describe::<SubscribeReq, SubscribeReply>(
                format!("{}.subscribe", self.gw),
                "登记并(已登录时)下发行情订阅",
            ),
            ServiceDesc::describe::<SendOrderReq, SendOrderReply>(
                format!("{}.send_order", self.gw),
                "下单(InputOrder + SUBMITTING 单已上总线)",
            ),
            ServiceDesc::describe::<CancelOrderReq, CancelOrderReply>(
                format!("{}.cancel_order", self.gw),
                "撤单(三段订单号解析)",
            ),
            ServiceDesc::describe::<QueryRateReq, QueryRateReply>(
                format!("{}.query_rate", self.gw),
                "费率查询(margin=保证金率/commission=手续费率;状态为纯被动推送)",
            ),
            ServiceDesc::describe::<ConfigureReq, ConfigureReply>(
                format!("{}.configure", self.gw),
                "网关配置(插件自带配置面;手工输入账号/前置 → 复位守卫并发起连接)",
            ),
        ]
    }

    fn on_init(&self, ctx: &PluginContext) {
        let _ = self.ctx.set(ctx.clone());
    }

    fn on_start(&self) {
        // 传输层已接入 → 按配置发起连接(原稿由 app 显式调 CtpGateway.connect)。
        // 无传输层的纯逻辑构建保持原状:emit 走 warn_no_transport 提示一次。
        self.connect();
    }

    fn on_stop(&self) {
        // 先释放原生会话:之后 SDK 不再回调,回调入口才会随插件一起被放下。
        if let Some(t) = &self.trans {
            t.shutdown();
        }
    }

    fn on_mail(&self, event: &Event) {
        if event.event_type == ROUTE_TIMER {
            self.on_timer();
            return;
        }
        if event.event_type == e_req_type(&self.gw) {
            self.on_request(event);
        }
    }
}