//! `pic apply [--file picloud.toml]` — reconcile the live app to the
//! manifest's desired state in one server-side transaction. Creates and
//! updates are applied; `--prune` additionally deletes live scripts/routes/
//! triggers absent from the manifest (secrets are never pruned).

use std::io::{IsTerminal, Write};
use std::path::Path;

use anyhow::{Context, Result};

use crate::client::{Client, NodeKind, NodePlanDto, PlanDto};
use crate::cmds::plan::build_bundle;
use crate::config;
use crate::manifest::{Manifest, ManifestProject};
use crate::output::{KvBlock, OutputMode};

#[allow(clippy::too_many_arguments)]
pub async fn run(
    manifest_path: &Path,
    env: Option<&str>,
    prune: bool,
    yes: bool,
    force: bool,
    takeover: bool,
    approve: &[String],
    mode: OutputMode,
) -> Result<()> {
    let creds = config::resolve()?;
    let client = Client::from_creds(&creds)?;

    let manifest = Manifest::load_with_env(manifest_path, env)?;
    // §3 M3: the env-approval gate is a property of the PROJECT, which lives at
    // the repo root. A leaf manifest carries no `[project]`, so find the
    // governing one (own block, else the nearest ancestor's) before gating.
    let governing = manifest
        .project
        .clone()
        .or_else(|| find_governing_project(manifest_path));
    require_env_approval(governing.as_ref(), env, approve)?;
    let base_dir = manifest_path.parent().unwrap_or_else(|| Path::new("."));
    let bundle = build_bundle(&manifest, base_dir)?;
    let kind = if manifest.is_group() {
        NodeKind::Group
    } else {
        NodeKind::App
    };
    let slug = manifest.slug().to_string();

    if prune && !yes {
        confirm_prune(&slug)?;
    }

    // Bound-plan check: replay the token from the last `pic plan` (for this
    // node) so the server refuses if it changed since it was reviewed.
    // `--force` skips it; no recorded plan means no check (apply still works
    // standalone). The token is single-use — cleared after a successful apply.
    let expected_token = if force {
        None
    } else {
        crate::linkstate::read_plan(base_dir)
            .filter(|l| l.app == slug)
            .map(|l| l.state_token)
    };

    // First-run guard: with no reviewed `.picloud/` plan, `apply` would mutate
    // with zero preview. Fetch a plan, show the change + blast-radius summary,
    // and confirm (interactive) or refuse a large cross-repo blast radius in CI.
    // `--yes`/`--force` opt out.
    if expected_token.is_none() && !yes && !force {
        let plan = client
            .plan_node(kind, &slug, &bundle, governing.as_ref())
            .await?;
        let changes = plan_change_count(&plan);
        let blast = plan
            .ownership
            .as_ref()
            .map_or(0, |o| o.blast_radius.iter().map(|b| b.apps).sum());
        if changes > 0 || blast > 0 {
            confirm_first_apply(&slug, changes, blast, &plan.warnings)?;
        }
    }

    let report = client
        .apply_node(
            kind,
            &slug,
            &bundle,
            prune,
            // §3 M3: send the GOVERNING project (own block, else nearest ancestor's)
            // so the server receives the env-approval policy and can enforce the
            // gate authoritatively. It is also the correct ownership project for
            // this node (a leaf applying under its repo's project).
            governing.as_ref(),
            takeover,
            expected_token.as_deref(),
            env,
            approve,
        )
        .await?;
    crate::linkstate::clear_plan(base_dir, &slug);

    let label = if manifest.is_group() { "group" } else { "app" };
    let mut block = KvBlock::new();
    block
        .field(label, slug.clone())
        .field(
            "scripts",
            format!(
                "+{} ~{} -{}",
                report.scripts_created, report.scripts_updated, report.scripts_deleted
            ),
        )
        .field(
            "routes",
            format!(
                "+{} ~{} -{}",
                report.routes_created, report.routes_updated, report.routes_deleted
            ),
        )
        .field(
            "triggers",
            format!("+{} -{}", report.triggers_created, report.triggers_deleted),
        )
        .field(
            "vars",
            format!(
                "+{} ~{} -{}",
                report.vars_created, report.vars_updated, report.vars_deleted
            ),
        )
        .field(
            "extension_points",
            format!(
                "+{} -{}",
                report.extension_points_created, report.extension_points_deleted
            ),
        )
        .field(
            "collections",
            format!(
                "+{} -{}",
                report.collections_created, report.collections_deleted
            ),
        )
        .field(
            "suppressions",
            format!(
                "+{} -{}",
                report.suppressions_created, report.suppressions_deleted
            ),
        )
        .field(
            "workflows",
            format!(
                "+{} ~{} -{}",
                report.workflows_created, report.workflows_updated, report.workflows_deleted
            ),
        );
    for w in &report.warnings {
        block.field("warning", w.clone());
    }
    block.print(mode);
    Ok(())
}

/// `pic apply --dir <root>` (Phase 5): reconcile a whole project tree — every
/// `picloud.toml` under `root` — in ONE atomic server transaction.
#[allow(clippy::too_many_arguments)]
pub async fn run_tree(
    dir: &Path,
    prune: bool,
    yes: bool,
    force: bool,
    takeover: bool,
    structure_mode: &str,
    approve: &[String],
    env: Option<&str>,
    mode: OutputMode,
) -> Result<()> {
    let creds = config::resolve()?;
    let client = Client::from_creds(&creds)?;
    let (bundle, node_count, project) = crate::discover::build_tree(dir, env)?;
    require_env_approval(project.as_ref(), env, approve)?;

    if prune && !yes {
        confirm_prune(&format!("{node_count} node(s) under {}", dir.display()))?;
    }

    let token_key = crate::cmds::plan::TREE_TOKEN_KEY;
    let expected_token = if force {
        None
    } else {
        crate::linkstate::read_plan(dir)
            .filter(|l| l.app == token_key)
            .map(|l| l.state_token)
    };

    // First-run guard (see `run`): no reviewed plan → preview + confirm the whole
    // tree's change count and cross-repo blast radius before mutating.
    if expected_token.is_none() && !yes && !force {
        let plan = client.plan_tree(&bundle, project.as_ref()).await?;
        let changes: usize = plan.nodes.iter().map(node_change_count).sum();
        let blast: u32 = plan
            .nodes
            .iter()
            .filter_map(|n| n.ownership.as_ref())
            .flat_map(|o| o.blast_radius.iter())
            .map(|b| b.apps)
            .sum();
        let warnings: Vec<String> = plan.nodes.iter().flat_map(|n| n.warnings.clone()).collect();
        if changes > 0 || blast > 0 {
            let target = format!("{node_count} node(s) under {}", dir.display());
            confirm_first_apply(&target, changes, blast, &warnings)?;
        }
    }

    let report = client
        .apply_tree(
            &bundle,
            prune,
            project.as_ref(),
            takeover,
            expected_token.as_deref(),
            structure_mode,
            env,
            approve,
        )
        .await?;
    crate::linkstate::clear_plan(dir, token_key);

    let mut block = KvBlock::new();
    block
        .field("nodes", node_count.to_string())
        .field(
            "scripts",
            format!(
                "+{} ~{} -{}",
                report.scripts_created, report.scripts_updated, report.scripts_deleted
            ),
        )
        .field(
            "routes",
            format!(
                "+{} ~{} -{}",
                report.routes_created, report.routes_updated, report.routes_deleted
            ),
        )
        .field(
            "triggers",
            format!("+{} -{}", report.triggers_created, report.triggers_deleted),
        )
        .field(
            "vars",
            format!(
                "+{} ~{} -{}",
                report.vars_created, report.vars_updated, report.vars_deleted
            ),
        )
        .field(
            "extension_points",
            format!(
                "+{} -{}",
                report.extension_points_created, report.extension_points_deleted
            ),
        )
        .field(
            "collections",
            format!(
                "+{} -{}",
                report.collections_created, report.collections_deleted
            ),
        )
        .field(
            "suppressions",
            format!(
                "+{} -{}",
                report.suppressions_created, report.suppressions_deleted
            ),
        )
        .field(
            "workflows",
            format!(
                "+{} ~{} -{}",
                report.workflows_created, report.workflows_updated, report.workflows_deleted
            ),
        );
    for w in &report.warnings {
        block.field("warning", w.clone());
    }
    block.print(mode);
    Ok(())
}

/// `--prune` deletes live scripts/routes/triggers absent from the manifest —
/// irreversible. Require an explicit go-ahead: an interactive `y`, or `--yes`
/// for non-interactive/CI use. Refuse a non-interactive prune without `--yes`
/// rather than silently deleting (review the deletions first with `pic plan`).
/// §3 M3: refuse `pic apply --env <e>` to a confirm-required env
/// (`[project.environments]` with `confirm = true`) unless it was explicitly
/// `--approve <e>`d. A blanket `--yes` does NOT cover a gated env (§4.2 "CI must
/// opt in per environment"). An unlisted or `confirm = false` env applies freely.
/// §3 M3: find the `[project]` that governs a single-file apply. The applied
/// manifest's own block wins (the caller checks that first); otherwise walk up
/// from its directory to the nearest ancestor `picloud.toml` that declares one
/// (the repo root), so `pic apply --file services/api/picloud.toml` still
/// honors the repo's env-approval gate. Loaded WITHOUT the env overlay — only
/// the `[project]` block matters here, and `load_with_env` would require an
/// overlay next to every ancestor and spuriously skip the governing manifest.
pub(crate) fn find_governing_project(manifest_path: &Path) -> Option<ManifestProject> {
    let dir = manifest_path.parent()?;
    dir.ancestors().skip(1).find_map(|anc| {
        let candidate = anc.join(crate::manifest::MANIFEST_FILE);
        Manifest::load(&candidate).ok().and_then(|m| m.project)
    })
}

fn require_env_approval(
    project: Option<&ManifestProject>,
    env: Option<&str>,
    approve: &[String],
) -> Result<()> {
    let (Some(env), Some(project)) = (env, project) else {
        return Ok(());
    };
    let Some(policy) = project.environments.get(env) else {
        return Ok(());
    };
    if policy.confirm && !approve.iter().any(|a| a == env) {
        anyhow::bail!(
            "environment `{env}` is confirm-required (`[project.environments]`); \
             re-run with `--approve {env}` to apply to it — a blanket `--yes` does \
             not cover a gated environment"
        );
    }
    Ok(())
}

/// A cross-repo blast radius at or above this many other-project apps is refused
/// non-interactively without `--yes` (§4.2 "this changes N apps — confirm").
const BLAST_RADIUS_CONFIRM_THRESHOLD: u32 = 10;

fn plan_change_count(plan: &PlanDto) -> usize {
    plan.scripts.len()
        + plan.routes.len()
        + plan.triggers.len()
        + plan.secrets.len()
        + plan.vars.len()
        + plan.extension_points.len()
        + plan.collections.len()
        + plan.suppressions.len()
}

fn node_change_count(n: &NodePlanDto) -> usize {
    n.scripts.len()
        + n.routes.len()
        + n.triggers.len()
        + n.secrets.len()
        + n.vars.len()
        + n.extension_points.len()
        + n.collections.len()
        + n.suppressions.len()
}

/// First-run guard: `pic apply` with no reviewed `.picloud/` plan mutates with
/// zero preview. When none was recorded (and not `--yes`/`--force`), the caller
/// fetches a plan and passes its change count + cross-repo blast radius here. On
/// a TTY we show the summary and prompt; non-interactively we allow a small
/// change but REFUSE a large blast radius without an explicit `--yes` (so a
/// wide-reaching change can't slip through CI unreviewed).
fn confirm_first_apply(
    target: &str,
    changes: usize,
    blast_apps: u32,
    warnings: &[String],
) -> Result<()> {
    if !std::io::stdin().is_terminal() {
        if blast_apps >= BLAST_RADIUS_CONFIRM_THRESHOLD {
            anyhow::bail!(
                "refusing to apply `{target}` non-interactively: no reviewed plan and this \
                 change fans out to {blast_apps} app(s) across other projects. Review with \
                 `pic plan`, then re-run with `--yes`."
            );
        }
        return Ok(());
    }
    eprintln!("no reviewed plan for `{target}`; this apply will make {changes} change(s).");
    if blast_apps > 0 {
        eprintln!("  cross-repo blast radius: {blast_apps} app(s) in other project(s).");
    }
    for w in warnings {
        eprintln!("  warning: {w}");
    }
    eprint!("Continue? [y/N] ");
    std::io::stderr().flush().ok();
    let mut answer = String::new();
    std::io::stdin()
        .read_line(&mut answer)
        .context("read confirmation")?;
    if !matches!(answer.trim(), "y" | "Y" | "yes" | "Yes") {
        anyhow::bail!("aborted");
    }
    Ok(())
}

fn confirm_prune(slug: &str) -> Result<()> {
    if !std::io::stdin().is_terminal() {
        anyhow::bail!(
            "refusing to `apply --prune` non-interactively without `--yes`: prune \
             deletes resources absent from the manifest and cannot be undone. \
             Review with `pic plan`, then re-run with `--yes`."
        );
    }
    eprint!(
        "apply --prune will DELETE live scripts/routes/triggers on `{slug}` that are \
         absent from the manifest. This cannot be undone. Continue? [y/N] "
    );
    std::io::stderr().flush().ok();
    let mut answer = String::new();
    std::io::stdin()
        .read_line(&mut answer)
        .context("read confirmation")?;
    if !matches!(answer.trim(), "y" | "Y" | "yes" | "Yes") {
        anyhow::bail!("aborted");
    }
    Ok(())
}