ek-ctp 0.1.0

//! # CtpPlugin —— ek-core 宿主注册的 CTP 网关插件。
//!
//! 谁发谁收:
//! - **发**:驱动输出 → `ctx.put` 上总线(纯生产者:`eTick.<gw>`/`eContract.<gw>`/
//!   `eOrder.<gw>`/`eTrade.<gw>`/`ePosition.<gw>`/`eAccount.<gw>`/
//!   `ePositionDone.<gw>`/`eLog`;线值与 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::{e_req_type, Envelope, ReplyErr};
use ek_core::plugin::Plugin;
use ek_core::service::ServiceDesc;
use ek_data::cancel_request::CancelRequest;
use ek_data::log::LogData;
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,
}

// ================ 传输层接缝(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;
}

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

/// 网关配置(对齐原稿 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,
}

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,
    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 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(),
        };
        // 原稿 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,
            md,
            td,
            timer_count: Mutex::new(0),
            trans,
            ctx: OnceLock::new(),
            trans_warned: Mutex::new(false),
        }
    }

    /// 原生适配入口(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.<gateway_name>` + 载荷)。
    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(format!("{}{}", prefix, self.gw), 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));
        // 载荷不是任何已声明契约 → 协议错误(宿主既有语义)
        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() }
        }
    }

    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() }
    }
}

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),
                "撤单(三段订单号解析)",
            ),
        ]
    }

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

    fn on_start(&self) {
        // 纯生产者无需依赖;native 桥在 on_start 由宿主装配线接入传输层(v0.2.0)
    }

    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);
        }
    }
}