//! Test adapter around the real Trove package manager, service bus and UDS supervisor.
//! stdin/stdout carry SDK bridge messages only; all state lives in a temporary directory.
use anyhow::{anyhow, ensure, Context, Result};
use serde_json::{json, Value};
use std::{
    collections::HashMap,
    future::Future,
    path::{Path, PathBuf},
    sync::Arc,
    time::Instant,
};
use tokio::io::{AsyncBufReadExt, BufReader};
use trove::{
    package::{AppPaths, InstalledPackage, PackageManager, SourceConfig, SourceKind},
    protocol::ProtocolMessage,
    runtime::{BackendRegistration, BackendSupervisor},
    service::{
        builtin::{fs::FsService, runtime::RuntimeService, storage::Storage},
        Caller, CurioRegistration, ProviderRef, ServiceBus, ServiceDescriptor,
    },
    transport::{uds::UnixGateway, Outbound, Peer},
};

fn emit(value: Value) {
    println!("{}", value);
}
fn read_json(path: impl AsRef<Path>) -> Result<Value> {
    Ok(serde_json::from_slice(&std::fs::read(path)?)?)
}
fn source(id: &str, index: &Path) -> SourceConfig {
    SourceConfig {
        id: id.into(),
        name: id.into(),
        kind: SourceKind::Local,
        index: index.display().to_string(),
        enabled: true,
        priority: 0,
        refresh_interval_seconds: 3600,
    }
}
fn target() -> &'static str {
    if cfg!(target_arch = "aarch64") {
        "aarch64-apple-darwin"
    } else {
        "x86_64-apple-darwin"
    }
}
fn expected_error<T>(result: trove::error::Result<T>, code: &str) -> Result<Value> {
    match result {
        Err(error) => {
            ensure!(
                error.code == code,
                "Expected {code}, got {}: {}",
                error.code,
                error.message
            );
            Ok(json!({"code":error.code}))
        }
        Ok(_) => Err(anyhow!("Expected {code}, operation succeeded")),
    }
}
async fn check<F: Future<Output = Result<Value>>>(
    rows: &mut Vec<Value>,
    name: impl Into<String>,
    operation: F,
) {
    let name = name.into();
    let start = Instant::now();
    let row = match operation.await {
        Ok(value) => {
            json!({"name":name,"status":"pass","value":value,"durationMs":start.elapsed().as_millis()})
        }
        Err(error) => {
            json!({"name":name,"status":"fail","error":{"message":format!("{error:#}")},"durationMs":start.elapsed().as_millis()})
        }
    };
    eprintln!("{} {}", row["status"].as_str().unwrap(), name);
    rows.push(row);
}

async fn package_checks(
    dist: &Path,
    paths: &AppPaths,
    packages: &PackageManager,
) -> Result<Vec<Value>> {
    let suite = read_json(dist.join("suite.json"))?;
    let mut rows = Vec::new();
    for entry in suite.as_array().context("suite is an array")? {
        if entry["directory"] == "test-upgrade-v2" {
            continue;
        }
        check(
            &mut rows,
            format!("安装并读取 {}", entry["directory"].as_str().unwrap()),
            async {
                let package = packages
                    .install_local(Path::new(entry["file"].as_str().unwrap()), true)
                    .await?;
                ensure!(
                    package.manifest.id == entry["manifest"]["id"],
                    "Installed ID changed"
                );
                ensure!(
                    package.manifest.development.is_none()
                        && package.manifest.build.is_none()
                        && package.manifest.package.is_none(),
                    "Source-only config leaked into package"
                );
                if let Some(ui) = &package.manifest.ui {
                    let resolved = trove::runtime::assets::resolve_asset(
                        &package.root,
                        &package.manifest.id,
                        &ui.entry,
                        &format!("/{}/{}", package.manifest.id, ui.entry),
                    )?;
                    ensure!(
                        std::fs::read_to_string(resolved)?.contains("./main.js"),
                        "UI entry is not the test console"
                    );
                    let base = Path::new(&ui.entry).parent().unwrap();
                    for relative in [
                        "main.js",
                        "config.js",
                        "checks.js",
                        "style.css",
                        "sdk/index.js",
                        "sdk/bridge.js",
                    ] {
                        ensure!(
                            package.root.join(base).join(relative).is_file(),
                            "Missing UI resource {relative}"
                        );
                    }
                    expected_error(
                        trove::runtime::assets::resolve_asset(
                            &package.root,
                            &package.manifest.id,
                            &ui.entry,
                            &format!("/{}/curio.json", package.manifest.id),
                        ),
                        "asset.forbidden",
                    )?;
                }
                if let Some(backend) = &package.manifest.backend {
                    use std::os::unix::fs::PermissionsExt;
                    ensure!(
                        std::fs::metadata(package.root.join(&backend.entry))?
                            .permissions()
                            .mode()
                            & 0o111
                            != 0,
                        "Backend isn't executable after install"
                    );
                }
                Ok(json!({"id":package.manifest.id,"version":package.active_version}))
            },
        )
        .await;
    }
    check(
        &mut rows,
        "重复安装同版本被拒绝且保留当前状态",
        async {
            let before = std::fs::read(paths.registry_file())?;
            let value = expected_error(
                packages
                    .install_local(&dist.join("packages/dev.trove.qa.host-1.0.0.curio"), true)
                    .await,
                "package.version_exists",
            )?;
            ensure!(
                std::fs::read(paths.registry_file())? == before,
                "Duplicate install changed registry"
            );
            ensure!(
                packages.get("dev.trove.qa.host")?.active_version == "1.0.0",
                "Duplicate install lost active version"
            );
            Ok(value)
        },
    )
    .await;
    for case in read_json(dist.join("invalid/cases.json"))?
        .as_array()
        .unwrap()
    {
        check(
            &mut rows,
            format!("拒绝无效包 {}", case["name"].as_str().unwrap()),
            async {
                let before = std::fs::read(paths.registry_file())?;
                let result = packages
                    .install_local(Path::new(case["file"].as_str().unwrap()), true)
                    .await;
                let value = expected_error(result, case["expectedError"].as_str().unwrap())?;
                ensure!(
                    std::fs::read(paths.registry_file())? == before,
                    "Failed install changed registry"
                );
                ensure!(
                    std::fs::read_dir(paths.support.join(".staging"))?
                        .next()
                        .is_none(),
                    "Failed install leaked staging files"
                );
                Ok(value)
            },
        )
        .await;
    }
    for case in read_json(dist.join("invalid/source-cases.json"))?
        .as_array()
        .unwrap()
    {
        check(
            &mut rows,
            format!("拒绝异常软件源 {}", case["name"].as_str().unwrap()),
            async {
                let sandbox = tempfile::tempdir()?;
                let manager =
                    PackageManager::open(AppPaths::under(sandbox.path()), "0.1.0", target())?;
                let id = case["id"].as_str().unwrap();
                manager.add_source(source(id, Path::new(case["file"].as_str().unwrap())))?;
                manager.refresh_source(id).await?;
                let result = manager
                    .install_from_source(
                        id,
                        case["curioId"].as_str().unwrap_or("dev.trove.qa.host"),
                        "1.0.0",
                        true,
                    )
                    .await;
                let value = expected_error(result, case["expectedError"].as_str().unwrap())?;
                ensure!(
                    manager.list().is_empty(),
                    "Invalid source installed a package"
                );
                Ok(value)
            },
        )
        .await;
    }
    check(
        &mut rows,
        "本地软件源刷新、安装、缓存恢复和禁用",
        async {
            let sandbox = tempfile::tempdir()?;
            let source_paths = AppPaths::under(sandbox.path());
            let manager = PackageManager::open(source_paths.clone(), "0.1.0", target())?;
            manager.add_source(source("trove-qa", &dist.join("index.json")))?;
            manager.refresh_source("trove-qa").await?;
            let count = manager.catalog().len();
            ensure!(count == 8, "Expected 8 normal Curio IDs, got {count}");
            let installed = manager
                .install_from_source("trove-qa", "dev.trove.qa.host", "1.0.0", true)
                .await?;
            ensure!(
                installed.source_id.as_deref() == Some("trove-qa"),
                "Source origin was lost"
            );
            drop(manager);
            let manager = PackageManager::open(source_paths, "0.1.0", target())?;
            ensure!(
                manager.catalog().len() == count && manager.list().len() == 1,
                "Durable source cache or installation lost"
            );
            let mut config = source("trove-qa", &dist.join("index.json"));
            config.enabled = false;
            manager.add_source(config)?;
            expected_error(
                manager
                    .install_from_source("trove-qa", "dev.trove.qa.isolation", "1.0.0", true)
                    .await,
                "source.disabled",
            )?;
            Ok(json!({"catalogIds":count,"sourceCacheRestored":true}))
        },
    )
    .await;
    check(&mut rows, "HTTP 软件源刷新、下载和校验安装", async {
        let sandbox = tempfile::tempdir()?;
        let manager = PackageManager::open(AppPaths::under(sandbox.path()), "0.1.0", target())?;
        let mut config = source("trove-qa", &dist.join("index.json"));
        config.kind = SourceKind::Http;
        config.index = std::env::var("QA_HTTP_SOURCE_URL")?;
        manager.add_source(config)?;
        manager.refresh_source("trove-qa").await?;
        let package = manager
            .install_from_source("trove-qa", "dev.trove.qa.host", "1.0.0", true)
            .await?;
        ensure!(
            package.manifest.id == "dev.trove.qa.host",
            "HTTP install identity changed"
        );
        Ok(json!({"downloaded":true,"hashVerified":true,"version":package.active_version}))
    })
    .await;
    check(&mut rows, "升级暂存、激活、回滚、卸载与数据保留", async {
        let id = "dev.trove.qa.upgrade";
        let data = paths.data_dir(id);
        std::fs::create_dir_all(&data)?;
        let marker = data.join("qa-upgrade-marker.txt");
        std::fs::write(&marker, "baseline-1.0.0")?;
        let file = dist.join("packages/dev.trove.qa.upgrade-1.1.0.curio");
        let pending = packages.install_local(&file, false).await?;
        ensure!(pending.active_version == "1.0.0" && pending.pending_version.as_deref() == Some("1.1.0"), "Update wasn't staged");
        ensure!(packages.activate(id, "1.1.0")?.active_version == "1.1.0", "Activation failed");
        ensure!(packages.rollback(id)?.active_version == "1.0.0", "Rollback failed");
        ensure!(std::fs::read_to_string(&marker)? == "baseline-1.0.0", "Update changed data");
        packages.set_enabled(id, false)?;
        ensure!(!packages.get(id)?.enabled, "Disable not persisted");
        packages.set_enabled(id, true)?;
        packages.uninstall(id, false)?;
        ensure!(marker.is_file(), "Keep-data uninstall deleted data");
        packages.install_local(&dist.join("packages/dev.trove.qa.upgrade-1.0.0.curio"), true).await?;
        ensure!(std::fs::read_to_string(&marker)? == "baseline-1.0.0", "Reinstall changed data");
        packages.uninstall(id, true)?;
        ensure!(!data.exists(), "Remove-data uninstall retained data");
        expected_error(packages.get(id), "package.not_installed")?;
        Ok(json!({"staged":true,"activated":true,"rolledBack":true,"reinstalled":true,"dataRemoved":true}))
    }).await;
    Ok(rows)
}

fn register(
    package: &InstalledPackage,
    paths: &AppPaths,
    runtime: &Arc<BackendSupervisor>,
    bus: &Arc<ServiceBus>,
) -> Result<()> {
    let manifest = &package.manifest;
    bus.register_curio(CurioRegistration {
        curio_id: manifest.id.clone(),
        version: manifest.version.clone(),
        enabled: package.enabled,
        provides: package
            .contracts
            .iter()
            .map(|contract| ServiceDescriptor {
                id: contract.service.clone(),
                version: contract.version.clone(),
                provider: ProviderRef::CurioBackend {
                    curio_id: manifest.id.clone(),
                },
                contract: contract.clone(),
                default_timeout_ms: 5000,
            })
            .collect(),
        requires: manifest
            .services
            .requires
            .iter()
            .map(|r| (r.id.clone(), r.version.clone()))
            .collect(),
    })?;
    if let Some(definition) = &manifest.backend {
        runtime.register(BackendRegistration {
            curio_id: manifest.id.clone(),
            version: manifest.version.clone(),
            root: package.root.clone(),
            definition: definition.clone(),
            data_dir: paths.data_dir(&manifest.id),
            cache_dir: paths.cache_dir(&manifest.id),
            log_dir: paths.log_dir(&manifest.id),
            instance_id: None,
            enabled: package.enabled,
            command: None,
        })?;
    }
    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().collect();
    if args.get(1).map(String::as_str) == Some("--remote-source") {
        return remote_source_check(
            args.get(2).context("remote source URL required")?,
            args.get(3).context("remote source ID required")?,
        )
        .await;
    }
    let dist = PathBuf::from(std::env::args().nth(1).context("dist directory required")?)
        .canonicalize()?;
    let sandbox = tempfile::Builder::new()
        .prefix("trove-qa-")
        .tempdir_in("/tmp")?;
    let paths = AppPaths::under(sandbox.path());
    let packages = Arc::new(PackageManager::open(paths.clone(), "0.1.0", target())?);
    let package_results = package_checks(&dist, &paths, &packages).await?;
    let runtime = BackendSupervisor::new(sandbox.path().join("host.sock"));
    let bus = ServiceBus::new(runtime.clone());
    bus.register_builtin(Arc::new(RuntimeService {
        paths: paths.clone(),
        packages: packages.clone(),
    }))?;
    bus.register_builtin(Arc::new(Arc::new(Storage::new(paths.support.join("data")))))?;
    bus.register_builtin(Arc::new(FsService {
        paths: paths.clone(),
    }))?;
    for package in packages.list() {
        register(&package, &paths, &runtime, &bus)?;
    }
    let gateway =
        UnixGateway::bind(runtime.socket_path().into(), runtime.clone(), bus.clone()).await?;
    emit(json!({"ready":true,"packageResults":package_results,"temporaryRoot":sandbox.path()}));
    let mut peers: HashMap<String, Arc<Peer>> = HashMap::new();
    let mut lines = BufReader::new(tokio::io::stdin()).lines();
    while let Some(line) = lines.next_line().await? {
        let envelope: Value = serde_json::from_str(&line)?;
        let op = envelope["op"].as_str().unwrap_or("");
        if op == "shutdown" {
            break;
        }
        if op == "send" {
            let peer = peers
                .get(envelope["connection"].as_str().unwrap_or(""))
                .context("Unknown connection")?
                .clone();
            let message = ProtocolMessage::from_value(envelope["message"].clone())?;
            let bus = bus.clone();
            tokio::spawn(async move {
                if let Err(error) = bus.dispatch(peer, message).await {
                    emit(json!({"fatal":error}));
                }
            });
            continue;
        }
        let result: Result<Value> = async {
            let id = envelope["curioId"].as_str().unwrap_or("");
            match op {
                "connect" => {
                    let package = packages.get(id)?;
                    let connection = envelope["connection"].as_str().context("connection required")?.to_string();
                    let (outbound, mut receiver) = Outbound::channel(512, 16 * 1024 * 1024);
                    let peer = Arc::new(Peer { id: connection.clone(), caller: Caller { curio_id: Some(id.into()), instance_id: Some(connection.clone()), endpoint: "ui".into() }, outbound, closed: Default::default(), features: vec!["call".into(),"cancel".into(),"subscribe".into()] });
                    bus.attach_peer(peer.clone());
                    peers.insert(connection.clone(), peer);
                    let key = connection.clone();
                    tokio::spawn(async move { while let Some(outbound) = receiver.recv().await { emit(json!({"connection":key,"message":outbound.message})); } });
                    Ok(json!({"curioId":id,"curioVersion":package.active_version,"instanceId":connection,"apiVersion":1,"mode":"production"}))
                }
                "disconnect" => {
                    let key = envelope["connection"].as_str().context("connection required")?;
                    bus.disconnect(key).await; peers.remove(key); Ok(json!({"disconnected":true}))
                }
                "status" => Ok(runtime.status(id).await),
                "diagnostics" => Ok(bus.diagnostics()),
                "strictContracts" => {
                    let enabled = envelope["value"].as_bool().context("strictContracts boolean required")?;
                    bus.strict_contracts(enabled);
                    Ok(json!({"strictContracts":enabled}))
                }
                "start" => { runtime.ensure_connected(id).await?; Ok(runtime.status(id).await) }
                "retry" => { runtime.retry(id).await?; Ok(runtime.status(id).await) }
                "stop" => { runtime.shutdown(id, "qa-test").await?; Ok(runtime.status(id).await) }
                "enabled" => {
                    let enabled = envelope["value"].as_bool().context("enabled boolean required")?;
                    runtime.shutdown(id, "qa-toggle").await?;
                    let package = packages.set_enabled(id, enabled)?;
                    register(&package, &paths, &runtime, &bus)?;
                    Ok(json!({"enabled":enabled}))
                }
                _ => Err(anyhow!("Unknown operation {op}")),
            }
        }.await;
        match result {
            Ok(value) => emit(json!({"controlId":envelope["controlId"],"result":value})),
            Err(error) => emit(
                json!({"controlId":envelope["controlId"],"error":{"message":format!("{error:#}")}}),
            ),
        }
    }
    bus.shutdown().await;
    runtime.shutdown_all().await;
    gateway.shutdown().await;
    Ok(())
}

async fn remote_source_check(url: &str, source_id: &str) -> Result<()> {
    let sandbox = tempfile::Builder::new()
        .prefix("trove-remote-qa-")
        .tempdir_in("/tmp")?;
    let paths = AppPaths::under(sandbox.path());
    let packages = Arc::new(PackageManager::open(paths.clone(), "0.1.0", target())?);
    let runtime = BackendSupervisor::new(sandbox.path().join("host.sock"));
    let bus = ServiceBus::new(runtime.clone());
    bus.register_builtin(Arc::new(RuntimeService {
        paths: paths.clone(),
        packages: packages.clone(),
    }))?;
    bus.register_builtin(Arc::new(Arc::new(Storage::new(paths.support.join("data")))))?;
    let gateway =
        UnixGateway::bind(runtime.socket_path().into(), runtime.clone(), bus.clone()).await?;
    let mut rows = Vec::new();
    check(&mut rows, "从真实远程源读取索引", async {
        packages.add_source(SourceConfig {
            id: source_id.into(),
            name: source_id.into(),
            kind: SourceKind::Http,
            index: url.into(),
            enabled: true,
            priority: 0,
            refresh_interval_seconds: 3600,
        })?;
        let refreshed = packages.refresh_source(source_id).await?;
        ensure!(!refreshed.stale, "Remote source refresh used stale cache");
        ensure!(
            !packages.catalog().is_empty(),
            "Remote source has no installable plugins"
        );
        Ok(json!({"url":url,"sourceId":source_id,"curioCount":packages.catalog().len()}))
    })
    .await;
    if rows.iter().all(|row| row["status"] == "pass") {
        for entry in packages.catalog() {
            for version in entry.package.versions {
                check(&mut rows, format!("远程安装 {}@{}", entry.package.id, version.version), async {
                    let package = packages.install_from_source(source_id, &entry.package.id, &version.version, true).await?;
                    ensure!(package.source_id.as_deref() == Some(source_id), "Installation wasn't bound to the remote source");
                    if let Some(ui) = &package.manifest.ui {
                        let asset = trove::runtime::assets::resolve_asset(&package.root, &package.manifest.id, &ui.entry, &format!("/{}/{}", package.manifest.id, ui.entry))?;
                        ensure!(asset.is_file(), "Installed UI entry is missing");
                    }
                    Ok(json!({"id":package.manifest.id,"version":package.active_version,"sourceId":package.source_id}))
                }).await;
            }
        }
        for package in packages.list() {
            register(&package, &paths, &runtime, &bus)?;
        }
        for package in packages
            .list()
            .into_iter()
            .filter(|package| package.manifest.backend.is_some())
        {
            check(
                &mut rows,
                format!("运行远程安装后端 {}", package.manifest.id),
                async {
                    let context = trove::service::CallContext::new(
                        Caller {
                            curio_id: Some(package.manifest.id.clone()),
                            instance_id: None,
                            endpoint: "ui".into(),
                        },
                        "remote-check",
                    );
                    let result = bus
                        .call(context, "@self", "info", json!({}), Default::default())
                        .await?;
                    ensure!(
                        result["curioId"] == package.manifest.id
                            && result["version"] == package.active_version,
                        "Backend identity doesn't match installed remote version"
                    );
                    ensure!(
                        result["pid"].as_u64().unwrap_or(0) > 0,
                        "Backend didn't report a process"
                    );
                    Ok(result)
                },
            )
            .await;
        }
        if packages.get("dev.trove.qa.consumer").is_ok()
            && packages.get("dev.trove.qa.provider").is_ok()
        {
            check(&mut rows, "远程安装插件间的服务调用", async {
                let context = trove::service::CallContext::new(
                    Caller {
                        curio_id: Some("dev.trove.qa.consumer".into()),
                        instance_id: None,
                        endpoint: "ui".into(),
                    },
                    "remote-consumer",
                );
                let result = bus
                    .call(
                        context,
                        "dev.trove.qa.provider.probe",
                        "echo",
                        json!({"remote":true,"text":"远程安装 🧪"}),
                        Default::default(),
                    )
                    .await?;
                ensure!(
                    result == json!({"remote":true,"text":"远程安装 🧪"}),
                    "Cross-plugin response changed"
                );
                Ok(result)
            })
            .await;
        }
    }
    bus.shutdown().await;
    runtime.shutdown_all().await;
    gateway.shutdown().await;
    sandbox.close()?;
    let failed = rows.iter().filter(|row| row["status"] == "fail").count();
    emit(json!({
        "schemaVersion":1,"sourceUrl":url,"sourceId":source_id,"target":target(),
        "scope":"Remote HTTP index and packages via real PackageManager; installed native backends via real UDS. Desktop UI rendering is not included.",
        "passed":rows.len()-failed,"failed":failed,"temporaryDataRemoved":true,"results":rows,
    }));
    ensure!(failed == 0, "Remote source verification failed");
    Ok(())
}