futucli/cmd/

repl.rs

//! `futucli repl` — 交互式 REPL
//!
//! 特性：
//! - 共享一条 FutuClient 长连接，避免每条命令重连网关
//! - 复用所有子命令（通过 `cli::dispatch`）
//! - rustyline 行编辑：↑↓ 历史、Ctrl-R 反向搜索、Ctrl-D 退出
//! - 历史持久化到 ~/.cache/futucli/history（或 $XDG_CACHE_HOME）
//! - 订阅推送实时打印且不打断 prompt（`ExternalPrinter`）
//! - REPL 专属命令：help / exit / reconnect / subs / unsub
//!
//! 实现要点：`rustyline::readline` 是阻塞式 API，放进 `tokio::task::spawn_blocking`
//! 以与 async 命令共存。

use std::collections::{BTreeMap, HashSet};
use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
use clap::Parser;
use futu_net::client::{FutuClient, PushMessage};
use futu_qot::push::{QuoteHandler, QuotePushDispatcher};
use futu_qot::types::{BasicQot, KLine, OrderBookData, Security, SubType};
use rustyline::error::ReadlineError;
use rustyline::{DefaultEditor, ExternalPrinter};
use tokio::sync::{Mutex, mpsc};

use crate::cli::{Cli, Command};
use crate::common::{connect_gateway, format_symbol, parse_sub_type, parse_symbol};
use crate::output::OutputFormat;

/// REPL 运行时状态。订阅状态按 `Security → 订阅的 SubType 集合` 记录，
/// 用 `BTreeMap` 保证 `subs` 命令输出稳定可读。
struct ReplState {
    gateway: String,
    output: OutputFormat,
    client: Arc<FutuClient>,
    subs: Mutex<BTreeMap<String, HashSet<SubType>>>,
}

/// 推送打印器：tty 下走 rustyline ExternalPrinter（不撞 prompt），
/// 非 tty（pipe / 重定向）下降级到 stderr，保证脚本 / 测试里 REPL 仍可用。
enum PushPrinter {
    External(Box<dyn ExternalPrinter + Send>),
    Stderr,
}

impl PushPrinter {
    fn print(&mut self, msg: String) {
        match self {
            PushPrinter::External(p) => {
                let _ = p.print(msg);
            }
            PushPrinter::Stderr => {
                eprint!("{msg}");
            }
        }
    }
}

/// 包一层 tokio Mutex，跨 task 共享。
type SharedPrinter = Arc<Mutex<PushPrinter>>;

pub async fn run(gateway: &str, output: OutputFormat) -> Result<()> {
    // 1. 建立长连接
    let (client, push_rx) = connect_gateway(gateway, "futucli-repl").await?;
    eprintln!("✓ connected to {gateway}");

    let state = Arc::new(ReplState {
        gateway: gateway.to_string(),
        output,
        client,
        subs: Mutex::new(BTreeMap::new()),
    });

    // 2. rustyline editor + 历史文件
    let history_path = history_file_path();
    let mut rl = DefaultEditor::new().context("init rustyline editor")?;
    if let Some(p) = &history_path {
        let _ = rl.load_history(p); // 首次运行无文件属正常
    }

    // 3. ExternalPrinter：供 push task 打印，不打断 prompt；
    //    非 tty 环境（pipe）直接降级到 stderr。
    let printer_inner = match rl.create_external_printer() {
        Ok(p) => PushPrinter::External(Box::new(p)),
        Err(e) => {
            eprintln!("note: external printer unavailable ({e}); push output will go to stderr");
            PushPrinter::Stderr
        }
    };
    let printer: SharedPrinter = Arc::new(Mutex::new(printer_inner));

    // 4. 启动 push 消费 task
    tokio::spawn(push_loop(push_rx, printer.clone()));

    // 5. 主循环：readline → 解析 → dispatch
    print_banner(gateway);
    loop {
        // rustyline 的 readline 是阻塞的，放到 blocking 线程里
        let prompt = format!("futu ({}) > ", state.gateway);
        let read_result = tokio::task::spawn_blocking(move || {
            let res = rl.readline(&prompt);
            (rl, res)
        })
        .await
        .context("readline task panicked")?;
        rl = read_result.0;

        let line = match read_result.1 {
            Ok(l) => l,
            Err(ReadlineError::Interrupted) => {
                eprintln!("(Ctrl-C) — type 'exit' or Ctrl-D to quit");
                continue;
            }
            Err(ReadlineError::Eof) => {
                eprintln!("bye");
                break;
            }
            Err(e) => {
                eprintln!("readline error: {e}");
                break;
            }
        };
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        let _ = rl.add_history_entry(trimmed);

        // 6. 派发：REPL 专属 > 子命令
        match handle_line(trimmed, &state, &printer).await {
            Ok(ShouldContinue::Continue) => {}
            Ok(ShouldContinue::Exit) => break,
            Err(e) => eprintln!("error: {e:#}"),
        }
    }

    // 退出前保存历史
    if let Some(p) = &history_path {
        let _ = std::fs::create_dir_all(p.parent().unwrap_or_else(|| std::path::Path::new(".")));
        let _ = rl.save_history(p);
    }
    Ok(())
}

enum ShouldContinue {
    Continue,
    Exit,
}

/// 路由一行输入。REPL 专属命令优先；否则走 clap 解析再转 `cli::dispatch`。
async fn handle_line(
    line: &str,
    state: &Arc<ReplState>,
    printer: &SharedPrinter,
) -> Result<ShouldContinue> {
    let tokens = shlex::split(line).ok_or_else(|| anyhow!("failed to tokenize input"))?;
    if tokens.is_empty() {
        return Ok(ShouldContinue::Continue);
    }

    match tokens[0].as_str() {
        "exit" | "quit" | ":q" => return Ok(ShouldContinue::Exit),
        "help" | "?" => {
            print_help();
            return Ok(ShouldContinue::Continue);
        }
        "reconnect" => {
            reconnect(state, printer).await?;
            return Ok(ShouldContinue::Continue);
        }
        "subs" => {
            list_subs(state).await;
            return Ok(ShouldContinue::Continue);
        }
        "sub" if tokens.len() >= 2 && looks_like_symbol(&tokens[1]) => {
            // 覆盖顶层 `sub` 子命令，改为 REPL 内部后台订阅
            return sub_inline(&tokens[1..], state, printer)
                .await
                .map(|_| ShouldContinue::Continue);
        }
        "unsub" => {
            return unsub_inline(&tokens[1..], state)
                .await
                .map(|_| ShouldContinue::Continue);
        }
        _ => {}
    }

    // clap 解析（注入一个伪 argv[0]）
    let argv = std::iter::once("futucli".to_string())
        .chain(tokens)
        .collect::<Vec<_>>();
    let parsed = match Cli::try_parse_from(&argv) {
        Ok(p) => p,
        Err(e) => {
            // clap 的帮助 / 错误消息自己会带换行；直接打出
            eprint!("{e}");
            return Ok(ShouldContinue::Continue);
        }
    };

    // REPL 里禁止再次进入 repl
    if matches!(parsed.command, Command::Repl) {
        bail!("already in REPL");
    }

    // 使用 REPL 的 gateway/output，忽略解析到的（避免 REPL 里切换网关造成长连接失效）
    crate::cli::dispatch(&state.gateway, state.output, parsed.command).await?;
    Ok(ShouldContinue::Continue)
}

// ========== REPL 专属命令实现 ==========

fn print_banner(gateway: &str) {
    eprintln!("futucli REPL — gateway {gateway}");
    eprintln!("type 'help' for commands, 'exit' or Ctrl-D to quit");
}

fn print_help() {
    let lines = [
        "REPL 内置命令:",
        "  help | ?              显示本帮助",
        "  exit | quit | :q      退出",
        "  reconnect             断开并重新连接网关",
        "  subs                  列出当前活跃订阅",
        "  sub <SYMBOL> [-t csv] 后台订阅（推送在 prompt 上方实时显示）",
        "  unsub <SYMBOL> [csv]  取消订阅（csv 省略则全部类型）",
        "",
        "子命令（和外层 futucli 一致）:",
        "  ping / quote / snapshot / kline / orderbook / ticker / rt / static /",
        "  broker / plate-list / plate-stocks / account / funds / position /",
        "  order / deal / unlock-trade",
        "",
        "示例:",
        "  quote HK.00700 US.AAPL",
        "  kline HK.00700 -t day -n 5",
        "  sub HK.00700 -t basic,orderbook",
    ];
    for l in lines {
        eprintln!("{l}");
    }
}

async fn reconnect(state: &Arc<ReplState>, printer: &SharedPrinter) -> Result<()> {
    let (new_client, new_rx) = connect_gateway(&state.gateway, "futucli-repl").await?;

    // 原子替换 client。FutuClient 存在 state 里的是 Arc，替换指向对象需要 Mutex；
    // 简化：reconnect 不热替换 state.client（会需要 Arc<Mutex<Arc<FutuClient>>>），
    // 而是让用户重启 REPL。如果有需要我们再做热替换。
    let _ = (new_client, new_rx); // 暂保留变量
    let _ = printer;
    eprintln!("reconnect not fully wired yet — please exit and re-enter REPL for a clean state");
    Ok(())
}

async fn list_subs(state: &Arc<ReplState>) {
    let guard = state.subs.lock().await;
    if guard.is_empty() {
        println!("(no active subscriptions)");
        return;
    }
    println!("Active subscriptions:");
    for (sym, types) in guard.iter() {
        let mut names: Vec<&'static str> = types.iter().copied().map(sub_type_label).collect();
        names.sort_unstable();
        println!("  {sym:<14} {}", names.join(","));
    }
}

fn looks_like_symbol(tok: &str) -> bool {
    tok.contains('.') && !tok.starts_with('-')
}

async fn sub_inline(
    args: &[String],
    state: &Arc<ReplState>,
    printer: &SharedPrinter,
) -> Result<()> {
    // 用法: sub <SYMBOL...> [-t csv]
    let (symbols, types_csv) = split_sub_args(args)?;
    let secs: Vec<Security> = symbols
        .iter()
        .map(|s| parse_symbol(s))
        .collect::<Result<_>>()?;
    let sub_types: Vec<SubType> = types_csv
        .split(',')
        .map(parse_sub_type)
        .collect::<Result<_>>()?;

    futu_qot::sub::subscribe(&state.client, &secs, &sub_types, true, true).await?;
    {
        let mut guard = state.subs.lock().await;
        for s in &symbols {
            let entry = guard.entry(s.clone()).or_default();
            for t in &sub_types {
                entry.insert(*t);
            }
        }
    }
    let _ = printer; // push 已在全局 task 里消费
    eprintln!("✓ subscribed symbols={symbols:?} types={types_csv}");
    Ok(())
}

async fn unsub_inline(args: &[String], state: &Arc<ReplState>) -> Result<()> {
    if args.is_empty() {
        bail!("usage: unsub <SYMBOL> [csv]");
    }
    let symbol = args[0].clone();
    let types_csv = args.get(1).cloned();

    let sec = parse_symbol(&symbol)?;
    let sub_types: Vec<SubType> = match types_csv {
        Some(csv) if !csv.is_empty() => {
            csv.split(',').map(parse_sub_type).collect::<Result<_>>()?
        }
        _ => {
            let guard = state.subs.lock().await;
            guard
                .get(&symbol)
                .map(|s| s.iter().copied().collect())
                .unwrap_or_default()
        }
    };
    if sub_types.is_empty() {
        bail!("no active subscription for {symbol}");
    }

    futu_qot::sub::unsubscribe(&state.client, std::slice::from_ref(&sec), &sub_types).await?;
    {
        let mut guard = state.subs.lock().await;
        if let Some(set) = guard.get_mut(&symbol) {
            for t in &sub_types {
                set.remove(t);
            }
            if set.is_empty() {
                guard.remove(&symbol);
            }
        }
    }
    eprintln!(
        "✓ unsubscribed {symbol} types=[{}]",
        sub_types
            .iter()
            .map(|t| sub_type_label(*t))
            .collect::<Vec<_>>()
            .join(",")
    );
    Ok(())
}

/// 拆 `sub HK.00700 US.AAPL -t basic,rt` → (["HK.00700","US.AAPL"], "basic,rt")
fn split_sub_args(args: &[String]) -> Result<(Vec<String>, String)> {
    let mut symbols = Vec::new();
    let mut types = "basic".to_string();
    let mut i = 0;
    while i < args.len() {
        let a = &args[i];
        if a == "-t" || a == "--type" {
            i += 1;
            types = args
                .get(i)
                .ok_or_else(|| anyhow!("-t needs a value"))?
                .clone();
        } else if let Some(rest) = a.strip_prefix("--type=") {
            types = rest.to_string();
        } else if looks_like_symbol(a) {
            symbols.push(a.clone());
        } else {
            bail!("unexpected token {a:?}");
        }
        i += 1;
    }
    if symbols.is_empty() {
        bail!("usage: sub <SYMBOL...> [-t csv]");
    }
    Ok((symbols, types))
}

// ========== push 异步打印 ==========

struct PrintHandler {
    printer: SharedPrinter,
}

impl PrintHandler {
    async fn emit(&self, s: String) {
        let mut g = self.printer.lock().await;
        g.print(format!("{s}\n"));
    }
}

#[async_trait]
impl QuoteHandler for PrintHandler {
    async fn on_basic_qot_update(&self, qot_list: Vec<BasicQot>) {
        for q in qot_list {
            let sym = format_symbol(&q.security);
            let change = q.cur_price - q.last_close_price;
            let pct = if q.last_close_price != 0.0 {
                change / q.last_close_price * 100.0
            } else {
                0.0
            };
            let sign = if change >= 0.0 { "+" } else { "" };
            self.emit(format!(
                "[{}] basic {sym:<12} px={:.3} {sign}{change:.3} ({sign}{pct:.2}%) vol={}",
                q.update_time, q.cur_price, q.volume
            ))
            .await;
        }
    }

    async fn on_kl_update(&self, security: Security, kl_list: Vec<KLine>) {
        let sym = format_symbol(&security);
        for k in kl_list {
            self.emit(format!(
                "[{}] kl    {sym:<12} O={:.3} H={:.3} L={:.3} C={:.3} V={}",
                k.time, k.open_price, k.high_price, k.low_price, k.close_price, k.volume
            ))
            .await;
        }
    }

    async fn on_order_book_update(&self, data: OrderBookData) {
        let sym = format_symbol(&data.security);
        let top_bid = data.bid_list.first();
        let top_ask = data.ask_list.first();
        let msg = match (top_bid, top_ask) {
            (Some(b), Some(a)) => format!(
                "              ob    {sym:<12} bid={:.3}x{} ask={:.3}x{} spr={:.3}",
                b.price,
                b.volume,
                a.price,
                a.volume,
                a.price - b.price
            ),
            _ => format!("              ob    {sym:<12} (empty)"),
        };
        self.emit(msg).await;
    }

    async fn on_ticker_update(
        &self,
        security: Security,
        ticker_list: Vec<futu_qot::ticker::Ticker>,
    ) {
        let sym = format_symbol(&security);
        for t in ticker_list {
            self.emit(format!(
                "              tick  {sym:<12} px={:.3} vol={} dir={}",
                t.price, t.volume, t.dir
            ))
            .await;
        }
    }

    async fn on_rt_update(&self, security: Security, rt_list: Vec<futu_qot::rt::TimeShare>) {
        let sym = format_symbol(&security);
        for r in rt_list {
            self.emit(format!(
                "[{}] rt    {sym:<12} px={:.3} avg={:.3} vol={}",
                r.time, r.price, r.avg_price, r.volume
            ))
            .await;
        }
    }
}

async fn push_loop(mut rx: mpsc::UnboundedReceiver<PushMessage>, printer: SharedPrinter) {
    let handler = PrintHandler { printer };
    while let Some(msg) = rx.recv().await {
        if let Err(e) = QuotePushDispatcher::dispatch(&handler, msg.proto_id, &msg.body).await {
            let mut g = handler.printer.lock().await;
            g.print(format!("push dispatch error: {e}\n"));
        }
    }
}

// ========== util ==========

fn sub_type_label(t: SubType) -> &'static str {
    match t {
        SubType::Basic => "basic",
        SubType::OrderBook => "orderbook",
        SubType::Ticker => "ticker",
        SubType::RT => "rt",
        SubType::KLDay => "kl_day",
        SubType::KL1Min => "kl_1min",
        SubType::KL3Min => "kl_3min",
        SubType::KL5Min => "kl_5min",
        SubType::KL15Min => "kl_15min",
        SubType::KL30Min => "kl_30min",
        SubType::KL60Min => "kl_60min",
        SubType::KLWeek => "kl_week",
        SubType::KLMonth => "kl_month",
        SubType::KLQuarter => "kl_quarter",
        SubType::KLYear => "kl_year",
        SubType::Broker => "broker",
        SubType::OrderDetail => "order_detail",
        _ => "other",
    }
}

fn history_file_path() -> Option<PathBuf> {
    let base = dirs::cache_dir()?;
    Some(base.join("futucli").join("history"))
}