futucli/cmd/daemon/

doctor.rs

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::time::{SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result};
use futu_core::diagnostic_text::update_check_unavailable_action;
use futu_qot::types::market_state_label;
use serde::Serialize;
use serde_json::{Value, json};

use crate::cmd::static_diag::summarize_static_status;
use crate::cmd::version_check::{self, UpdateDecision, UpdateLevel};
use crate::output::OutputFormat;

/// 聚合 REST 诊断端点，输出一份可贴给排障方的一键体检单。
pub async fn run_doctor(
    rest_url: Option<&str>,
    api_key: Option<&str>,
    symbol: Option<&str>,
    bundle_dir: Option<&Path>,
    check_update: bool,
    update_url: Option<&str>,
    update_timeout_ms: u64,
    output: OutputFormat,
) -> Result<()> {
    let base = super::resolve_rest_url(rest_url);
    let client = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(5))
        .build()
        .context("build reqwest client")?;
    let mut endpoints = Vec::new();
    endpoints.push(
        fetch_doctor_endpoint(
            &client,
            &base,
            "readyz",
            reqwest::Method::GET,
            "/readyz",
            api_key,
            None,
        )
        .await,
    );
    endpoints.push(
        fetch_doctor_endpoint(
            &client,
            &base,
            "admin-status",
            reqwest::Method::GET,
            "/api/admin/status",
            api_key,
            None,
        )
        .await,
    );
    endpoints.push(
        fetch_doctor_endpoint(
            &client,
            &base,
            "push-subscriber-info",
            reqwest::Method::GET,
            "/api/push-subscriber-info",
            api_key,
            None,
        )
        .await,
    );
    endpoints.push(
        fetch_doctor_endpoint(
            &client,
            &base,
            "sub-info",
            reqwest::Method::GET,
            "/api/sub-info",
            api_key,
            None,
        )
        .await,
    );
    endpoints.push(
        fetch_doctor_endpoint(
            &client,
            &base,
            "used-quota",
            reqwest::Method::POST,
            "/api/used-quota",
            api_key,
            Some(json!({})),
        )
        .await,
    );
    endpoints.push(
        fetch_doctor_endpoint(
            &client,
            &base,
            "history-kl-quota",
            reqwest::Method::POST,
            "/api/history-kl-quota",
            api_key,
            Some(json!({"c2s": {"b_get_detail": false}})),
        )
        .await,
    );
    endpoints.push(
        fetch_doctor_endpoint(
            &client,
            &base,
            "metrics",
            reqwest::Method::GET,
            "/metrics",
            api_key,
            None,
        )
        .await,
    );
    if let Some(symbol) = symbol {
        endpoints.push(
            fetch_doctor_endpoint(
                &client,
                &base,
                "market-state",
                reqwest::Method::POST,
                "/api/market-state",
                api_key,
                Some(json!({"symbols": [symbol]})),
            )
            .await,
        );
        endpoints.push(
            fetch_doctor_endpoint(
                &client,
                &base,
                "quote-capability",
                reqwest::Method::GET,
                &format!("/api/quote-capability?symbol={symbol}"),
                api_key,
                None,
            )
            .await,
        );
    }

    let mut findings = analyze_doctor(&endpoints, symbol);
    if check_update {
        findings.extend(fetch_update_check_finding(update_url, update_timeout_ms).await);
    }
    let report = DoctorReport {
        rest_url: base,
        symbol: symbol.map(str::to_string),
        endpoints,
        findings,
    };
    if let Some(dir) = bundle_dir {
        write_doctor_bundle(&report, dir)?;
        eprintln!("# doctor bundle written to {}", dir.display());
    }
    print_doctor_report(&report, output)
}

async fn fetch_update_check_finding(url: Option<&str>, timeout_ms: u64) -> Option<DoctorFinding> {
    match version_check::check_for_update(env!("CARGO_PKG_VERSION"), url, timeout_ms).await {
        Ok(decision) => update_check_finding_from_decision(&decision),
        Err(err) => Some(DoctorFinding {
            level: "INFO",
            area: "update",
            message: "version update check unavailable".to_string(),
            action: format!(
                "{err}; {}",
                update_check_unavailable_action().text_with_code()
            ),
        }),
    }
}

#[derive(Debug, Serialize)]
pub(super) struct DoctorReport {
    rest_url: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    symbol: Option<String>,
    endpoints: Vec<DoctorEndpoint>,
    findings: Vec<DoctorFinding>,
}

#[derive(Debug, Serialize)]
pub(super) struct DoctorEndpoint {
    pub(super) name: String,
    pub(super) method: String,
    pub(super) path: String,
    pub(super) ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) http_status: Option<u16>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) json: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) text_preview: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) error: Option<String>,
}

#[derive(Debug, Serialize)]
pub(super) struct DoctorFinding {
    pub(super) level: &'static str,
    pub(super) area: &'static str,
    pub(super) message: String,
    pub(super) action: String,
}

async fn fetch_doctor_endpoint(
    client: &reqwest::Client,
    base: &str,
    name: &str,
    method: reqwest::Method,
    path: &str,
    api_key: Option<&str>,
    json_body: Option<Value>,
) -> DoctorEndpoint {
    let method_text = method.as_str().to_string();
    match super::request_raw(client, base, method, path, api_key, json_body).await {
        Ok((status, body)) => {
            let json = serde_json::from_str::<Value>(&body).ok();
            let text_preview = if path == "/metrics" {
                Some(select_metrics_preview(&body))
            } else if json.is_none() {
                Some(preview_text(&body, 1200))
            } else {
                None
            };
            DoctorEndpoint {
                name: name.to_string(),
                method: method_text,
                path: path.to_string(),
                ok: (200..300).contains(&status),
                http_status: Some(status),
                json,
                text_preview,
                error: None,
            }
        }
        Err(err) => DoctorEndpoint {
            name: name.to_string(),
            method: method_text,
            path: path.to_string(),
            ok: false,
            http_status: None,
            json: None,
            text_preview: None,
            error: Some(err.to_string()),
        },
    }
}

pub(super) fn analyze_doctor(
    endpoints: &[DoctorEndpoint],
    symbol: Option<&str>,
) -> Vec<DoctorFinding> {
    let mut findings = Vec::new();
    for ep in endpoints {
        if !ep.ok {
            findings.push(DoctorFinding {
                level: if matches!(
                    ep.name.as_str(),
                    "metrics" | "used-quota" | "history-kl-quota"
                ) {
                    "WARN"
                } else {
                    "ERROR"
                },
                area: "endpoint",
                message: format!(
                    "{} {} unavailable{}",
                    ep.method,
                    ep.path,
                    ep.http_status
                        .map(|s| format!(" (HTTP {s})"))
                        .unwrap_or_default()
                ),
                action: ep
                    .error
                    .clone()
                    .unwrap_or_else(|| "查看 daemon 日志与 API key scope".to_string()),
            });
        }
    }

    if endpoint(endpoints, "readyz").is_some_and(|ep| ep.ok) {
        findings.push(DoctorFinding {
            level: "OK",
            area: "readyz",
            message: "REST readiness endpoint is reachable".to_string(),
            action: "gateway dispatch path is ready enough for REST diagnostics".to_string(),
        });
    }

    let market_state = endpoint_json(endpoints, "market-state").and_then(first_market_state);
    let market_active = market_state.map(is_market_state_active);
    if let Some(state) = market_state {
        findings.push(DoctorFinding {
            level: "INFO",
            area: "market-state",
            message: format!(
                "{} market_state={} ({})",
                symbol.unwrap_or("<symbol>"),
                market_state_label(state),
                state
            ),
            action: "用该状态辅助判断订阅无新 push 是闭市正常还是开市断流".to_string(),
        });
    }

    if let Some(push_json) = endpoint_json(endpoints, "push-subscriber-info")
        && let Some(push_health) = push_json.get("push_health")
    {
        if push_health
            .get("push_stream_healthy")
            .and_then(Value::as_bool)
            == Some(false)
        {
            findings.push(DoctorFinding {
                level: "ERROR",
                area: "push",
                message: "push_stream_healthy=false".to_string(),
                action:
                    "先看 consecutive_parse_errors / circuit breaker；必要时抓 daemon debug log"
                        .to_string(),
            });
        }
        if push_health
            .get("backend_connected")
            .and_then(Value::as_bool)
            == Some(false)
        {
            findings.push(DoctorFinding {
                level: "ERROR",
                area: "push",
                message: "backend_connected=false".to_string(),
                action: "request path TCP 未连接；先确认 broker runtime 重连状态".to_string(),
            });
        }
        if push_health
            .get("last_push_received_at_ms")
            .and_then(Value::as_i64)
            == Some(0)
        {
            findings.push(DoctorFinding {
                level: "WARN",
                area: "push",
                message: "daemon has not observed any backend push yet".to_string(),
                action: "确认已订阅行情或交易推送；仅启动未订阅时可正常为 0".to_string(),
            });
        }
    }

    if let Some(admin_json) = endpoint_json(endpoints, "admin-status")
        && let Some(warning) = admin_json
            .get("login")
            .and_then(|login| login.get("credential"))
            .and_then(|credential| credential.get("expiry_warning"))
            .and_then(Value::as_str)
    {
        findings.push(DoctorFinding {
            level: "WARN",
            area: "auth",
            message: warning.to_string(),
            action: "优先执行 futucli daemon-reload；若已过期则用密码/SMS 重新认证".to_string(),
        });
    }

    if let Some(admin_json) = endpoint_json(endpoints, "admin-status")
        && admin_json
            .get("cache")
            .and_then(|cache| cache.get("static_data"))
            .is_some()
        && let Ok(report) = summarize_static_status(admin_json)
        && (!report.ready
            || report.stock_list_sync_failed_total > 0
            || report.stock_list_sync_recoverable_retry_total > 0)
    {
        let level = if report.ready { "INFO" } else { "WARN" };
        findings.push(DoctorFinding {
            level,
            area: "static-data",
            message: format!(
                "stock-list readiness={} ready={} reason={}",
                report.stock_list_readiness, report.ready, report.reason
            ),
            action: format!("{}; inspect with futucli static-status", report.action),
        });
    }

    if let Some(admin_json) = endpoint_json(endpoints, "admin-status")
        && let Some(auth_refresh) = admin_json
            .get("login")
            .and_then(|login| login.get("auth_refresh"))
    {
        let proactive = auth_refresh
            .get("client_sig_proactive_refresh_enabled")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        let reactive = auth_refresh
            .get("client_sig_reactive_refresh_enabled")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        if proactive || reactive {
            findings.push(DoctorFinding {
                level: "INFO",
                area: "auth",
                message: format!(
                    "client_sig refresh opt-in enabled: proactive={proactive}, reactive={reactive}"
                ),
                action: "确认这是有意开启的长跑恢复配置；默认关闭，建议先用非关键账户验证"
                    .to_string(),
            });
        }
    }

    let subscribed_sub_types = endpoint_json(endpoints, "sub-info")
        .map(collect_sub_types)
        .unwrap_or_default();
    let push_watermarks = endpoint_json(endpoints, "push-subscriber-info")
        .and_then(|v| v.get("push_health"))
        .map(collect_sub_type_watermarks)
        .unwrap_or_default();
    if subscribed_sub_types.is_empty() {
        findings.push(DoctorFinding {
            level: "INFO",
            area: "subscription",
            message: "no subscribed SubType found in /api/sub-info".to_string(),
            action: "若期待行情 push，先执行 futucli sub 或 REST /api/subscribe".to_string(),
        });
    } else {
        for sub_type in subscribed_sub_types {
            if !push_watermarks.contains_key(&sub_type) {
                findings.push(missing_sub_type_finding(sub_type, market_active, symbol));
            }
        }
    }

    if let Some(metrics) = endpoint(endpoints, "metrics")
        && metrics.ok
        && !metrics
            .text_preview
            .as_deref()
            .unwrap_or_default()
            .contains("futu_gateway_push_health_stream_healthy")
    {
        findings.push(DoctorFinding {
            level: "WARN",
            area: "metrics",
            message: "/metrics reachable but push health metrics were not found".to_string(),
            action: "确认 daemon 版本包含 v1.4.113 push-health Prometheus renderer".to_string(),
        });
    }

    if let Some(capability_json) = endpoint_json(endpoints, "quote-capability") {
        findings.extend(quote_capability_decision_findings(capability_json));
    }

    findings
}

pub(super) fn update_check_finding_from_decision(
    decision: &UpdateDecision,
) -> Option<DoctorFinding> {
    match decision.level {
        UpdateLevel::UpToDate => None,
        UpdateLevel::Info => Some(DoctorFinding {
            level: "INFO",
            area: "update",
            message: decision.message.clone(),
            action: decision.action.clone(),
        }),
        UpdateLevel::Recommended => Some(DoctorFinding {
            level: "WARN",
            area: "update",
            message: decision.message.clone(),
            action: decision.action.clone(),
        }),
        UpdateLevel::Critical => Some(DoctorFinding {
            level: "ERROR",
            area: "update",
            message: decision.message.clone(),
            action: decision.action.clone(),
        }),
    }
}

fn endpoint<'a>(endpoints: &'a [DoctorEndpoint], name: &str) -> Option<&'a DoctorEndpoint> {
    endpoints.iter().find(|ep| ep.name == name)
}

fn endpoint_json<'a>(endpoints: &'a [DoctorEndpoint], name: &str) -> Option<&'a Value> {
    endpoint(endpoints, name).and_then(|ep| ep.json.as_ref())
}

fn missing_sub_type_finding(
    sub_type: i64,
    market_active: Option<bool>,
    symbol: Option<&str>,
) -> DoctorFinding {
    match market_active {
        Some(true) => DoctorFinding {
            level: "WARN",
            area: "subscription",
            message: format!(
                "SubType {sub_type} is subscribed but has no regular push watermark while {} is active",
                symbol.unwrap_or("the reference symbol")
            ),
            action: "开市仍无该类型 push：确认订阅参数、backend push 通道；必要时 /api/admin/reload 后重订阅".to_string(),
        },
        Some(false) => DoctorFinding {
            level: "INFO",
            area: "subscription",
            message: format!("SubType {sub_type} is subscribed but has no regular push watermark"),
            action: "参考 symbol 当前非连续交易时段；闭市/午休/盘前结束时部分 SubType 无新数据通常正常".to_string(),
        },
        None => DoctorFinding {
            level: "INFO",
            area: "subscription",
            message: format!("SubType {sub_type} is subscribed but has no regular push watermark"),
            action: "未传 --symbol，无法区分闭市正常与开市断流；可加 --symbol HK.00700 或查 /api/market-state".to_string(),
        },
    }
}

fn collect_sub_types(value: &Value) -> BTreeSet<i64> {
    let mut out = BTreeSet::new();
    collect_i64_key(value, "sub_type", &mut out);
    out
}

fn collect_sub_type_watermarks(push_health: &Value) -> BTreeMap<i64, i64> {
    let mut out = BTreeMap::new();
    if let Some(items) = push_health
        .get("last_push_by_sub_type")
        .and_then(Value::as_array)
    {
        for item in items {
            if let (Some(sub_type), Some(at_ms)) = (
                item.get("sub_type").and_then(Value::as_i64),
                item.get("last_push_received_at_ms").and_then(Value::as_i64),
            ) {
                out.insert(sub_type, at_ms);
            }
        }
    }
    out
}

fn collect_i64_key(value: &Value, key: &str, out: &mut BTreeSet<i64>) {
    match value {
        Value::Object(map) => {
            if let Some(v) = map.get(key).and_then(Value::as_i64) {
                out.insert(v);
            }
            for child in map.values() {
                collect_i64_key(child, key, out);
            }
        }
        Value::Array(items) => {
            for child in items {
                collect_i64_key(child, key, out);
            }
        }
        _ => {}
    }
}

fn quote_capability_decision_findings(value: &Value) -> Vec<DoctorFinding> {
    let Some(decisions) = value.get("decisions").and_then(Value::as_array) else {
        return Vec::new();
    };
    decisions
        .iter()
        .filter_map(|decision| {
            let allowed = decision.get("allowed").and_then(Value::as_bool)?;
            if allowed {
                return None;
            }
            let topic = decision
                .get("topic")
                .and_then(Value::as_str)
                .unwrap_or("unknown");
            let reason = decision
                .get("reason")
                .and_then(Value::as_str)
                .unwrap_or("blocked by quote capability decision");
            let source = decision
                .get("source")
                .and_then(Value::as_str)
                .unwrap_or("unknown");
            let freshness = decision
                .get("freshness")
                .and_then(Value::as_str)
                .unwrap_or("unknown");
            let action = decision
                .get("action")
                .and_then(Value::as_str)
                .unwrap_or("inspect futucli quote-rights and quote-capability")
                .to_string();
            Some(DoctorFinding {
                level: "WARN",
                area: "quote-capability",
                message: format!(
                    "{topic} blocked: {reason} (source={source}, freshness={freshness})"
                ),
                action,
            })
        })
        .collect()
}

fn first_market_state(value: &Value) -> Option<i32> {
    let mut values = BTreeSet::new();
    collect_i64_key(value, "market_state", &mut values);
    values
        .into_iter()
        .next()
        .and_then(|v| i32::try_from(v).ok())
}

fn is_market_state_active(state: i32) -> bool {
    matches!(
        state,
        1 | 3 | 5 | 8 | 10 | 13 | 15 | 19 | 21 | 23 | 25 | 28 | 32 | 33 | 35 | 36 | 37
    )
}

fn print_doctor_report(report: &DoctorReport, output: OutputFormat) -> Result<()> {
    match output {
        OutputFormat::Json => {
            println!("{}", serde_json::to_string_pretty(report)?);
        }
        OutputFormat::Jsonl => {
            println!("{}", serde_json::to_string(report)?);
        }
        OutputFormat::Table | OutputFormat::Markdown => {
            println!("futucli doctor");
            println!("REST: {}", report.rest_url);
            if let Some(symbol) = &report.symbol {
                println!("Symbol: {symbol}");
            }
            println!();
            println!("Endpoints:");
            for ep in &report.endpoints {
                let status = ep
                    .http_status
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| "-".to_string());
                let mark = if ep.ok { "OK" } else { "FAIL" };
                println!("- [{mark}] {} {} HTTP {}", ep.method, ep.path, status);
                if let Some(error) = &ep.error {
                    println!("  error: {error}");
                }
            }
            println!();
            println!("Findings:");
            if report.findings.is_empty() {
                println!("- [OK] no actionable findings");
            } else {
                for finding in &report.findings {
                    println!(
                        "- [{}] {}: {}",
                        finding.level, finding.area, finding.message
                    );
                    println!("  next: {}", finding.action);
                }
            }
        }
    }
    Ok(())
}

fn select_metrics_preview(body: &str) -> String {
    let selected: Vec<&str> = body
        .lines()
        .filter(|line| {
            line.contains("futu_gateway_push_health")
                || line.contains("futu_gateway_broker_tcp")
                || line.contains("futu_metrics_registry")
        })
        .collect();
    if selected.is_empty() {
        preview_text(body, 1200)
    } else {
        selected.join("\n")
    }
}

fn preview_text(body: &str, max_chars: usize) -> String {
    body.chars().take(max_chars).collect()
}

fn write_doctor_bundle(report: &DoctorReport, dir: &Path) -> Result<()> {
    std::fs::create_dir_all(dir).with_context(|| format!("create bundle dir {}", dir.display()))?;
    let endpoints_dir = dir.join("endpoints");
    std::fs::create_dir_all(&endpoints_dir)
        .with_context(|| format!("create endpoints dir {}", endpoints_dir.display()))?;

    let redacted_report = redacted_json(report)?;
    write_json_file(&dir.join("doctor-report.json"), &redacted_report)?;

    let findings_path = dir.join("doctor-findings.jsonl");
    let mut findings = String::new();
    for finding in &report.findings {
        findings.push_str(&serde_json::to_string(finding)?);
        findings.push('\n');
    }
    std::fs::write(&findings_path, redact_sensitive_text_for_bundle(&findings))
        .with_context(|| format!("write {}", findings_path.display()))?;

    let mut files = vec![
        "doctor-report.json".to_string(),
        "doctor-findings.jsonl".to_string(),
    ];
    for endpoint in &report.endpoints {
        let base_name = bundle_file_stem(&endpoint.name);
        if let Some(json) = &endpoint.json {
            let mut value = json.clone();
            redact_sensitive_json_for_bundle(&mut value);
            let rel = format!("endpoints/{base_name}.json");
            write_json_file(&dir.join(&rel), &value)?;
            files.push(rel);
        }
        if let Some(text) = &endpoint.text_preview {
            let rel = format!("endpoints/{base_name}.txt");
            let path = dir.join(&rel);
            std::fs::write(&path, redact_sensitive_text_for_bundle(text))
                .with_context(|| format!("write {}", path.display()))?;
            files.push(rel);
        }
        if let Some(error) = &endpoint.error {
            let rel = format!("endpoints/{base_name}.error.txt");
            let path = dir.join(&rel);
            std::fs::write(&path, redact_sensitive_text_for_bundle(error))
                .with_context(|| format!("write {}", path.display()))?;
            files.push(rel);
        }
    }

    let manifest = json!({
        "schema_version": 1,
        "generated_unix_secs": SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(),
        "rest_url": report.rest_url,
        "symbol": report.symbol,
        "redaction": "sensitive keys and key=value tokens are replaced with <redacted>",
        "files": files,
    });
    write_json_file(&dir.join("MANIFEST.json"), &manifest)?;
    Ok(())
}

fn write_json_file(path: &Path, value: &Value) -> Result<()> {
    let text = serde_json::to_string_pretty(value)?;
    std::fs::write(path, text).with_context(|| format!("write {}", path.display()))
}

fn redacted_json<T: Serialize>(value: &T) -> Result<Value> {
    let mut json = serde_json::to_value(value)?;
    redact_sensitive_json_for_bundle(&mut json);
    Ok(json)
}

fn bundle_file_stem(name: &str) -> String {
    name.chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '-'
            }
        })
        .collect()
}

pub(super) fn redact_sensitive_json_for_bundle(value: &mut Value) {
    match value {
        Value::Object(map) => {
            for (key, child) in map.iter_mut() {
                if is_sensitive_key(key) {
                    *child = Value::String("<redacted>".to_string());
                } else {
                    redact_sensitive_json_for_bundle(child);
                }
            }
        }
        Value::Array(items) => {
            for item in items {
                redact_sensitive_json_for_bundle(item);
            }
        }
        _ => {}
    }
}

pub(super) fn redact_sensitive_text_for_bundle(text: &str) -> String {
    text.split_whitespace()
        .map(|token| {
            let Some((key, _value)) = token.split_once('=') else {
                return token.to_string();
            };
            let normalized = key.trim_matches(|c: char| !c.is_ascii_alphanumeric() && c != '_');
            if is_sensitive_key(normalized) {
                format!("{key}=<redacted>")
            } else {
                token.to_string()
            }
        })
        .collect::<Vec<_>>()
        .join(" ")
}

fn is_sensitive_key(key: &str) -> bool {
    let k = key.to_ascii_lowercase();
    if matches!(
        k.as_str(),
        "client_sig_proactive_refresh_enabled" | "client_sig_reactive_refresh_enabled"
    ) {
        return false;
    }
    k.contains("password")
        || k == "pwd"
        || k.contains("api_key")
        || k.contains("token")
        || k.contains("secret")
        || k.contains("tgtgt")
        || k.contains("device_sig")
        || k.contains("dvs")
        || k.contains("auth_code")
        || k.contains("session_key")
        || k.contains("client_key")
        || k.contains("client_sig")
        || k.contains("web_sig")
}