//! Handler for `mcptest drift` (schema + behavioral drift watch).
//!
//! `capture` connects to the server in a spec, snapshots its `tools/list`
//! catalog, runs the golden canary calls, and writes a drift-state file.
//! `check` compares a current capture against a baseline and fails the run on
//! a breaking schema change or any behavioral drift. The comparison is
//! deterministic; only the live capture touches the network.

use std::collections::BTreeMap;
use std::path::Path;

use anyhow::{Context, Result};
use serde::Deserialize;
use serde_json::{json, Value};

use mcptest_core::connector::connect_server;
use mcptest_core::drift::{compare, report_envelope, CanarySample, DriftReport, DriftState};
use mcptest_core::suite::ServerSpec;

use crate::cli::args::{DriftArgs, DriftCaptureArgs, DriftCheckArgs, DriftCommand};

/// The drift spec file: which server to watch and which canary calls to run.
#[derive(Debug, Deserialize)]
struct DriftSpec {
    server: ServerBlock,
    #[serde(default)]
    canaries: Vec<CanaryCall>,
}

/// The watched server's connection details.
#[derive(Debug, Deserialize)]
struct ServerBlock {
    #[serde(default)]
    url: Option<String>,
    #[serde(default)]
    command: Vec<String>,
    #[serde(default)]
    headers: BTreeMap<String, String>,
}

/// One canary tool call.
#[derive(Debug, Deserialize)]
struct CanaryCall {
    tool: String,
    #[serde(default)]
    args: Value,
}

/// Dispatch the `mcptest drift` subcommand.
pub(crate) fn drift_command(args: DriftArgs) -> Result<i32> {
    match args.command {
        DriftCommand::Capture(a) => capture(a),
        DriftCommand::Check(a) => check(a),
    }
}

/// Capture a drift state live and write it to disk.
fn capture(args: DriftCaptureArgs) -> Result<i32> {
    let spec = load_spec(&args.spec)?;
    let state = run_capture(&spec)?;
    let json = serde_json::to_string_pretty(&state)?;
    super::overwrite::ensure_overwritable(&args.output, args.force)?;
    std::fs::write(&args.output, json)
        .with_context(|| format!("could not write {}", args.output.display()))?;
    println!(
        "drift capture: wrote {} ({} tool(s), {} canary call(s))",
        args.output.display(),
        state
            .schema
            .get("tools")
            .and_then(Value::as_array)
            .map_or(0, Vec::len),
        state.canaries.len(),
    );
    Ok(0)
}

/// Compare a current capture against a baseline and gate.
fn check(args: DriftCheckArgs) -> Result<i32> {
    if args.current.is_some() == args.spec.is_some() {
        eprintln!("mcptest drift check: pass exactly one of --current or --spec");
        return Ok(2);
    }
    let baseline = load_state(&args.baseline)?;
    let current = match (&args.current, &args.spec) {
        (Some(path), _) => load_state(path)?,
        (_, Some(spec_path)) => run_capture(&load_spec(spec_path)?)?,
        _ => unreachable!("validated above"),
    };
    let report = compare(&baseline, &current).map_err(|e| anyhow::anyhow!("{e}"))?;
    if args.json {
        println!(
            "{}",
            serde_json::to_string_pretty(&report_envelope(&report))?
        );
    } else {
        print_human(&report);
    }
    Ok(i32::from(report.should_fail()))
}

/// Connect to the spec's server, snapshot the catalog, and run the canaries.
fn run_capture(spec: &DriftSpec) -> Result<DriftState> {
    let server_spec = build_server_spec(&spec.server)?;
    let label = spec
        .server
        .url
        .clone()
        .unwrap_or_else(|| spec.server.command.join(" "));
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .context("failed to build tokio runtime")?;
    runtime.block_on(capture_async(&server_spec, &spec.canaries, label))
}

/// The async half of [`run_capture`]: connect, list tools, run canaries.
async fn capture_async(
    server_spec: &ServerSpec,
    canaries: &[CanaryCall],
    label: String,
) -> Result<DriftState> {
    let live = connect_server(
        "drift",
        server_spec,
        "mcptest",
        mcptest_core::version::semver(),
        &mcptest_core::network::ProxyConfig::default(),
        &[],
    )
    .await
    .context("connect to the watched server failed")?;

    let schema = live
        .client
        .request("tools/list", json!({}))
        .await
        .map_err(|e| anyhow::anyhow!("tools/list failed: {e}"))?;

    let mut samples = Vec::with_capacity(canaries.len());
    for canary in canaries {
        let params = json!({ "name": canary.tool, "arguments": canary.args });
        let (response, error) = match live.client.request("tools/call", params).await {
            Ok(result) => {
                let is_error = result
                    .get("isError")
                    .and_then(Value::as_bool)
                    .unwrap_or(false);
                (result, is_error)
            }
            Err(e) => (json!({ "error": e.to_string() }), true),
        };
        samples.push(CanarySample {
            tool: canary.tool.clone(),
            args: canary.args.clone(),
            response,
            error,
        });
    }
    let _ = live.shutdown.shutdown().await;

    Ok(DriftState {
        version: "0.1".into(),
        generated_at: now_iso(),
        server: label,
        schema,
        canaries: samples,
    })
}

/// Build a connector [`ServerSpec`] from the spec's server block.
fn build_server_spec(block: &ServerBlock) -> Result<ServerSpec> {
    match (&block.url, block.command.split_first()) {
        (Some(_), Some(_)) => {
            anyhow::bail!("drift spec server sets both `url` and `command`; pick one")
        }
        (Some(url), None) => Ok(ServerSpec::StreamableHttp {
            url: url.clone(),
            headers: block.headers.clone(),
            oauth: None,
            stateless: false,
        }),
        (None, Some((command, args))) => Ok(ServerSpec::Stdio {
            command: command.clone(),
            args: args.to_vec(),
            env: BTreeMap::new(),
        }),
        (None, None) => anyhow::bail!("drift spec server must set `url` or `command`"),
    }
}

/// Render the drift report as a human-readable block.
fn print_human(report: &DriftReport) {
    if report.is_clean() {
        println!("drift check: clean (no schema or behavioral drift)");
        return;
    }
    if !report.schema_changes.is_empty() {
        println!("schema drift:");
        for change in &report.schema_changes {
            let tag = if change.breaking { "BREAKING" } else { "ok" };
            println!("  [{tag}] {} ({})", change.description, change.tool);
        }
    }
    if !report.behavioral_changes.is_empty() {
        println!("behavioral drift (golden canaries):");
        for change in &report.behavioral_changes {
            println!("  [BREAKING] {} ({})", change.description, change.tool);
        }
    }
    if report.should_fail() {
        eprintln!("drift check: FAIL (breaking schema change or behavioral drift)");
    } else {
        println!("drift check: changes are non-breaking");
    }
}

fn load_spec(path: &Path) -> Result<DriftSpec> {
    let text = std::fs::read_to_string(path)
        .with_context(|| format!("could not read {}", path.display()))?;
    serde_yaml::from_str(&text)
        .with_context(|| format!("could not parse drift spec {}", path.display()))
}

fn load_state(path: &Path) -> Result<DriftState> {
    let text = std::fs::read_to_string(path)
        .with_context(|| format!("could not read {}", path.display()))?;
    serde_json::from_str(&text)
        .with_context(|| format!("could not parse drift state {}", path.display()))
}

/// A best-effort UTC timestamp (seconds since the Unix epoch, RFC-3339-ish is
/// not needed here since the field is diagnostic only).
fn now_iso() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let secs = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    format!("unix:{secs}")
}
