use std::path::Path;
use std::sync::Arc;

use anyhow::{Context, Result};
use hive_sh4re::priv_proto::InfraAction;
use hive_sh4re::{HostRequest, HostResponse, LifecycleScope};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};

use crate::actions;
use crate::coordinator::Coordinator;
use crate::lifecycle;

pub async fn serve(socket: &Path, coord: Arc<Coordinator>) -> Result<()> {
    // Prefer a socket passed by systemd socket-activation (LISTEN_FDS).
    // When running under a `.socket` unit, systemd has already created,
    // bound, and chmod-ed the socket for us — we just accept on it.
    // Fall back to the traditional bind path when not socket-activated
    // (direct invocation, dev, tests).
    let listener = {
        let mut listenfd = listenfd::ListenFd::from_env();
        if let Some(std_listener) = listenfd
            .take_unix_listener(0)
            .context("take socket-activated unix listener")?
        {
            std_listener.set_nonblocking(true)?;
            UnixListener::from_std(std_listener)
                .context("convert socket-activated listener to tokio")?
        } else {
            // Standalone: create parent dir, remove any stale socket, bind.
            if let Some(parent) = socket.parent() {
                std::fs::create_dir_all(parent)
                    .with_context(|| format!("create socket parent {}", parent.display()))?;
            }
            if socket.exists() {
                std::fs::remove_file(socket).context("remove stale socket")?;
            }
            UnixListener::bind(socket)
                .with_context(|| format!("bind admin socket {}", socket.display()))?
        }
    };
    tracing::info!(socket = %socket.display(), hyperhive_flake = %coord.hyperhive_flake, "hive-c0re admin listening");

    loop {
        let (stream, _) = listener.accept().await.context("accept connection")?;
        let coord = coord.clone();
        tokio::spawn(async move {
            if let Err(e) = handle(stream, coord).await {
                tracing::warn!(error = ?e, "connection failed");
            }
        });
    }
}

async fn handle(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
    let (read, mut write) = stream.into_split();
    let mut reader = BufReader::new(read);
    let mut line = String::new();

    loop {
        line.clear();
        let n = reader.read_line(&mut line).await?;
        if n == 0 {
            return Ok(());
        }
        let resp = match serde_json::from_str::<HostRequest>(line.trim()) {
            Ok(req) => dispatch(&req, coord.clone()).await,
            Err(e) => HostResponse::error(format!("parse error: {e}")),
        };
        let mut payload = serde_json::to_string(&resp)?;
        payload.push('\n');
        write.write_all(payload.as_bytes()).await?;
        write.flush().await?;
    }
}

async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
    let result: anyhow::Result<HostResponse> = async {
        Ok(match req {
            HostRequest::Spawn { name } => handle_spawn(&coord, name).await?,
            HostRequest::RequestSpawn { name } => {
                tracing::info!(%name, "request_spawn");
                let id =
                    coord
                        .approvals
                        .submit_kind(name, hive_sh4re::ApprovalKind::Spawn, "", None)?;
                tracing::info!(%id, %name, "spawn approval queued");
                HostResponse::success()
            }
            HostRequest::Kill { name } => handle_kill(&coord, name).await?,
            HostRequest::Restart { name } => {
                tracing::info!(%name, "restart");
                lifecycle::restart(name).await?;
                HostResponse::success()
            }
            HostRequest::RestartAll => handle_restart_all().await?,
            HostRequest::Stop { scope, graceful } => {
                // Resolve the scope to explicit container names at the entry
                // point, then operate on names — never pass the bare "all
                // agents" flag deeper (it'd force every consumer, incl. the
                // graceful-stop queue, to re-expand it).
                let agents = scoped_agents(scope).await?;
                let infra = scoped_infra(scope);
                handle_stop(&coord, &agents, &infra, *graceful).await?
            }
            HostRequest::Start { scope } => {
                let agents = scoped_agents(scope).await?;
                let infra = scoped_infra(scope);
                handle_start(&agents, &infra).await?
            }
            HostRequest::Destroy { name, purge } => {
                actions::destroy(&coord, name, *purge).await?;
                HostResponse::success()
            }
            HostRequest::Rebuild { name } => handle_rebuild(&coord, name).await?,
            HostRequest::List => HostResponse::list(lifecycle::list().await?),
            HostRequest::Pending => HostResponse::pending(coord.approvals.pending()?),
            HostRequest::Approve { id } => {
                actions::approve(coord.clone(), *id).await?;
                HostResponse::success()
            }
            HostRequest::Deny { id } => {
                actions::deny(&coord, *id, None).await?;
                HostResponse::success()
            }
            HostRequest::SetParent { child, new_parent } => {
                tracing::info!(%child, ?new_parent, "set_parent");
                // `reparent_with_notify` wraps `topology::set_parent`
                // with the three notification messages + the
                // ContainerView rescan. Idempotent same-parent calls
                // skip both the messages and the disk write per the
                // topology fast-path.
                coord
                    .reparent_with_notify(child, new_parent.as_deref())
                    .await
                    .map_err(anyhow::Error::msg)?;
                HostResponse::success()
            }
        })
    }
    .await;
    match result {
        Ok(r) => r,
        Err(e) => HostResponse::error(format!("{e:#}")),
    }
}

/// Create + start the container for `name`, rolling back socket
/// registration and notifying the manager on failure.
async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
    tracing::info!(%name, "spawn");
    let agent_dir = coord.ensure_runtime(name)?;
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    match lifecycle::spawn(name, &hive, &paths).await {
        Ok(()) => {
            coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
                agent: name.to_owned(),
                ok: true,
                note: None,
                sha: None,
            });
        }
        Err(e) => {
            // Roll back socket registration if container creation failed.
            coord.unregister_agent(name);
            coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
                agent: name.to_owned(),
                ok: false,
                note: Some(format!("{e:#}")),
                sha: None,
            });
            return Err(e);
        }
    }
    Ok(HostResponse::success())
}

/// Kill `name`'s container, unregister its socket, notify the manager.
async fn handle_kill(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
    tracing::info!(%name, "kill");
    lifecycle::kill(name).await?;
    coord.unregister_agent(name);
    coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
        agent: name.to_owned(),
    });
    Ok(HostResponse::success())
}

/// Restart every container, aggregating per-agent failures into one
/// response rather than aborting on the first error.
async fn handle_restart_all() -> Result<HostResponse> {
    tracing::info!("restart-all");
    let agents = lifecycle::list().await?;
    let mut ok_agents: Vec<String> = Vec::new();
    let mut errors: Vec<String> = Vec::new();
    for agent in &agents {
        if let Err(e) = lifecycle::restart(agent).await {
            tracing::warn!(%agent, error = ?e, "restart-all: failed to restart agent");
            errors.push(format!("{agent}: {e:#}"));
        } else {
            ok_agents.push(agent.clone());
        }
    }
    if errors.is_empty() {
        Ok(HostResponse::list(ok_agents))
    } else {
        Ok(HostResponse {
            ok: false,
            error: Some(errors.join("; ")),
            agents: Some(ok_agents),
            approvals: None,
        })
    }
}

/// Stop the given `agents` (resolved logical names) then `infra` containers
/// (`hivectl stop`). Agents go down before infra so they're not mid-request
/// against a forge/matrix that's already gone. Per-target failures are
/// aggregated rather than aborting on the first error, mirroring
/// `handle_restart_all`. Callers resolve the [`LifecycleScope`] to these
/// explicit name lists up front — this never sees the "all" flag.
///
/// A `graceful` stop enqueues a `QueueKind::GracefulStop` per agent (signal the
/// harness, run one stop-checkpoint turn, drain, then container stop, with a
/// timeout fallback to a hard stop), mirroring the dashboard `?graceful=1`
/// path. `graceful` applies to agents only - infra containers have no harness
/// turn loop, so they're always hard-stopped.
async fn handle_stop(
    coord: &Arc<Coordinator>,
    agents: &[String],
    infra: &[&str],
    graceful: bool,
) -> Result<HostResponse> {
    tracing::info!(?agents, ?infra, graceful, "stop");
    let mut ok_items: Vec<String> = Vec::new();
    let mut errors: Vec<String> = Vec::new();
    let mut enqueued_graceful = false;

    for agent in agents {
        if graceful {
            // Graceful stop: enqueue the quiesce orchestration rather than a
            // hard kill. Serialised through the rebuild queue so it can't race
            // an in-flight rebuild for the same agent, and its per-step
            // progress surfaces on the queue snapshot + build log.
            coord.rebuild_queue.enqueue(
                crate::rebuild_queue::QueueKind::GracefulStop,
                agent.clone(),
                crate::rebuild_queue::QueueSource::Manual,
                "manual via hivectl graceful stop".to_owned(),
                None,
            );
            ok_items.push(agent.clone());
            enqueued_graceful = true;
            continue;
        }
        match lifecycle::kill(agent).await {
            Ok(()) => ok_items.push(agent.clone()),
            Err(e) => {
                tracing::warn!(%agent, error = ?e, "stop: agent kill failed");
                errors.push(format!("{agent}: {e:#}"));
            }
        }
    }
    if enqueued_graceful {
        coord.emit_rebuild_queue_snapshot();
    }

    for &container in infra {
        match crate::priv_client::control_infra_container(container, InfraAction::Stop).await {
            Ok(()) => ok_items.push(container.to_owned()),
            Err(e) => {
                tracing::warn!(%container, error = ?e, "stop: infra stop failed");
                errors.push(format!("{container}: {e:#}"));
            }
        }
    }

    Ok(finish_lifecycle(ok_items, &errors))
}

/// Start the given `infra` containers then `agents` (`hivectl start`) — the
/// inverse of [`handle_stop`]. Infra comes up before agents so the agents
/// find forge/matrix/gateway ready. Per-target failures aggregated. Callers
/// resolve the [`LifecycleScope`] to these explicit name lists up front.
async fn handle_start(agents: &[String], infra: &[&str]) -> Result<HostResponse> {
    tracing::info!(?agents, ?infra, "start");
    let mut ok_items: Vec<String> = Vec::new();
    let mut errors: Vec<String> = Vec::new();

    for &container in infra {
        match crate::priv_client::control_infra_container(container, InfraAction::Start).await {
            Ok(()) => ok_items.push(container.to_owned()),
            Err(e) => {
                tracing::warn!(%container, error = ?e, "start: infra start failed");
                errors.push(format!("{container}: {e:#}"));
            }
        }
    }

    for agent in agents {
        match lifecycle::start(agent).await {
            Ok(()) => ok_items.push(agent.clone()),
            Err(e) => {
                tracing::warn!(%agent, error = ?e, "start: agent start failed");
                errors.push(format!("{agent}: {e:#}"));
            }
        }
    }

    Ok(finish_lifecycle(ok_items, &errors))
}

/// Resolve which sub-agent logical names a scope targets: every live
/// container (from `lifecycle::list`) when `agents` is set or the scope is
/// "everything", plus any explicit `agent_names`. Returns de-duplicated
/// logical names with the `h-` container prefix stripped.
async fn scoped_agents(scope: &LifecycleScope) -> Result<Vec<String>> {
    use std::collections::BTreeSet;
    let mut set: BTreeSet<String> = BTreeSet::new();
    if scope.agents || scope.is_everything() {
        for c in lifecycle::list().await? {
            let logical = c
                .strip_prefix(lifecycle::AGENT_PREFIX)
                .unwrap_or(&c)
                .to_owned();
            set.insert(logical);
        }
    }
    for n in &scope.agent_names {
        set.insert(n.clone());
    }
    Ok(set.into_iter().collect())
}

/// Resolve which infra container names a scope targets. An "everything"
/// scope (no flags set) selects all controllable infra; otherwise each set
/// flag maps to its container. Fixed order for deterministic output.
fn scoped_infra(scope: &LifecycleScope) -> Vec<&'static str> {
    let everything = scope.is_everything();
    let mut out = Vec::new();
    if everything || scope.ci {
        out.push("hive-ci");
    }
    if everything || scope.forge {
        out.push("hive-forge");
    }
    if everything || scope.gateway {
        out.push("hive-gateway");
    }
    if everything || scope.matrix {
        out.push("hive-matrix");
    }
    out
}

/// Build the aggregated lifecycle response: `ok` with the touched names when
/// every target succeeded, otherwise `ok: false` with the joined errors and
/// the partial success list (matches `handle_restart_all`).
fn finish_lifecycle(ok_items: Vec<String>, errors: &[String]) -> HostResponse {
    if errors.is_empty() {
        HostResponse::list(ok_items)
    } else {
        HostResponse {
            ok: false,
            error: Some(errors.join("; ")),
            agents: Some(ok_items),
            approvals: None,
        }
    }
}

/// Rebuild `name`'s container, notifying the manager of the outcome
/// (success or failure) and kicking the agent's next turn on success.
async fn handle_rebuild(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
    tracing::info!(%name, "rebuild");
    let agent_dir = coord.ensure_runtime(name)?;
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    let result = lifecycle::rebuild(name, &hive, &paths, &|_| (), &|_| ()).await;
    // Mirror auto_update::rebuild_agent — the manager wants to know
    // about every rebuild attempt regardless of which surface triggered
    // it, especially failures (build error → manager can adjust the
    // agent's agent.nix). Without this the admin-socket CLI was a
    // notify-gap.
    match &result {
        Ok(()) => {
            coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
                agent: name.to_owned(),
                ok: true,
                note: None,
                sha: None,
                tag: None,
            });
            // Wake the agent's next turn with the "you were rebuilt"
            // hint. Same pattern as auto_update::rebuild_agent and the
            // dashboard rebuild path — this is the CLI's equivalent.
            coord.kick_agent(name, "container rebuilt");
        }
        Err(e) => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
            agent: name.to_owned(),
            ok: false,
            note: Some(format!("{e:#}")),
            sha: None,
            tag: None,
        }),
    }
    result?;
    Ok(HostResponse::success())
}