use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context as _, Result, bail};
use clap::{Parser, Subcommand};
use hive_sh4re::{HostRequest, HostResponse};

// Every module hangs off the `hive_c0re` library (see `src/lib.rs`).
// The daemon and the `hivectl` sibling binary share the same module
// tree — no per-binary duplication. Enumerated rather than wildcard
// so clippy stays happy + the lib surface this bin consumes is
// explicit (any new daemon entry point reads off the next add).
use hive_c0re::coordinator::Coordinator;
use hive_c0re::{
    agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard,
    dashboard_events, events_vacuum, forge, manager_server, matrix, migrate, rebuild_queue,
    reminder_scheduler, scheduled_prompts_worker, server, stats_vacuum,
};

#[derive(Parser)]
#[command(name = "hive-c0re", about = "hyperhive coordinator daemon and CLI")]
struct Cli {
    /// Path to the host admin socket.
    #[arg(long, global = true, default_value = "/run/hyperhive/host.sock")]
    socket: PathBuf,

    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum Cmd {
    /// Run the coordinator daemon.
    Serve {
        /// URL of the hyperhive flake. Inlined into each per-agent
        /// `flake.nix` as the `hyperhive` input.
        #[arg(long, default_value = "/etc/hyperhive")]
        hyperhive_flake: String,
        /// Store-path URL of the nixpkgs to wire into the meta flake as
        /// `inputs.nixpkgs.url`. Set by the NixOS module to
        /// `"path:${pkgs.path}"` so the meta flake tracks exactly the
        /// nixpkgs the host was evaluated with (the host's own nixpkgs
        /// when `inputs.hyperhive.inputs.nixpkgs.follows = "nixpkgs"` is
        /// set, otherwise hyperhive's pin). Empty = legacy
        /// `follows = "hyperhive/nixpkgs"` fallback.
        #[arg(long, default_value = "")]
        nixpkgs_flake: String,
        /// Store-path URL of the nixpkgs-unstable to wire into the meta
        /// flake as `inputs.nixpkgs-unstable.url`. Hyperhive's
        /// `inputs.nixpkgs-unstable` then follows this top-level input.
        /// Set by the NixOS module; defaults to the hyperhive flake's own
        /// nixpkgs-unstable store path. Empty = legacy
        /// `follows = "hyperhive/nixpkgs-unstable"` fallback.
        #[arg(long, default_value = "")]
        nixpkgs_unstable_flake: String,
        /// Path to the sqlite message store.
        #[arg(long, default_value = "/var/lib/hyperhive/broker.sqlite")]
        db: PathBuf,
        /// Dashboard HTTP port.
        #[arg(long, default_value_t = 7000)]
        dashboard_port: u16,
        /// Operator pronouns (free text). Threaded into each
        /// container's harness via `HIVE_OPERATOR_PRONOUNS` so the
        /// system prompt can mention them. Default: `she/her`.
        #[arg(long, default_value = "she/her")]
        operator_pronouns: String,
        /// Per-model context-window sizes, as JSON object mapping model-family
        /// short name to token count. Threaded into each container as
        /// `HIVE_CONTEXT_WINDOW_TOKENS_<KEY_UPPER>` env vars. Set via the
        /// `services.hive-c0re.contextWindowTokens` NixOS option.
        #[arg(
            long,
            default_value = r#"{"haiku":200000,"sonnet":1000000,"opus":1000000}"#
        )]
        context_window_tokens: String,
    },
    /// Spawn a new agent container directly (`hive-agent-<name>`). Bypasses
    /// the approval queue — use only as an operator on the host. For
    /// approval-gated spawns, use `request-spawn` instead.
    Spawn { name: String },
    /// Queue a spawn request as an approval. The container is created on
    /// `approve <id>` (CLI) or the dashboard's APPR0VE button.
    RequestSpawn { name: String },
    /// Stop a managed container (graceful).
    Kill { name: String },
    /// Tear down a sub-agent container. Container is removed; persistent
    /// state (config repos + Claude credentials) is kept by default. Pass
    /// `--purge` to also wipe the agent's state dirs (config + creds +
    /// notes). No undo.
    Destroy {
        name: String,
        #[arg(long)]
        purge: bool,
    },
    /// Apply pending config to a managed container.
    Rebuild { name: String },
    /// List managed containers.
    List,
    /// List pending approval requests submitted by the manager.
    Pending,
    /// Approve a pending request by id; the action runs immediately.
    Approve { id: i64 },
    /// Deny a pending request by id.
    Deny { id: i64 },
    /// Move an agent in the topology tree. Set `--parent` to a new
    /// parent agent name; pass `--root` to promote the agent to root
    /// (no parent). Refuses cycles and unknown agents. The manager
    /// is reparentable like any other agent — its privileges come
    /// from the privileged MCP socket, not its tree position.
    SetParent {
        child: String,
        /// New parent agent name. Mutually exclusive with `--root`.
        /// Exactly one of `--parent` / `--root` is required — clap
        /// rejects both-absent calls so a fat-fingered
        /// `hive-c0re set-parent alice` doesn't silently promote
        /// alice to root.
        #[arg(long, conflicts_with = "root", required_unless_present = "root")]
        parent: Option<String>,
        /// Promote `child` to root (no parent).
        #[arg(long)]
        root: bool,
    },
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    let cli = Cli::parse();
    match cli.cmd {
        Cmd::Serve {
            hyperhive_flake,
            nixpkgs_flake,
            nixpkgs_unstable_flake,
            db,
            dashboard_port,
            operator_pronouns,
            context_window_tokens,
        } => {
            cmd_serve(
                hyperhive_flake,
                nixpkgs_flake,
                nixpkgs_unstable_flake,
                db,
                dashboard_port,
                operator_pronouns,
                context_window_tokens,
                &cli.socket,
            )
            .await
        }
        Cmd::Spawn { name } => {
            render(client::request(&cli.socket, HostRequest::Spawn { name }).await?)
        }
        Cmd::RequestSpawn { name } => {
            render(client::request(&cli.socket, HostRequest::RequestSpawn { name }).await?)
        }
        Cmd::Kill { name } => {
            render(client::request(&cli.socket, HostRequest::Kill { name }).await?)
        }
        Cmd::Destroy { name, purge } => {
            render(client::request(&cli.socket, HostRequest::Destroy { name, purge }).await?)
        }
        Cmd::Rebuild { name } => {
            render(client::request(&cli.socket, HostRequest::Rebuild { name }).await?)
        }
        Cmd::List => render(client::request(&cli.socket, HostRequest::List).await?),
        Cmd::Pending => render(client::request(&cli.socket, HostRequest::Pending).await?),
        Cmd::Approve { id } => {
            render(client::request(&cli.socket, HostRequest::Approve { id }).await?)
        }
        Cmd::Deny { id } => render(client::request(&cli.socket, HostRequest::Deny { id }).await?),
        Cmd::SetParent {
            child,
            parent,
            root,
        } => {
            let new_parent = if root { None } else { parent };
            render(
                client::request(&cli.socket, HostRequest::SetParent { child, new_parent }).await?,
            )
        }
    }
}

/// Start the coordinator daemon: open the broker, run migrations, spawn
/// background tasks (auto-update, vacuums, crash-watcher, reminder-scheduler,
/// dashboard), then serve the admin socket until a signal arrives.
async fn cmd_serve(
    hyperhive_flake: String,
    nixpkgs_flake: String,
    nixpkgs_unstable_flake: String,
    db: std::path::PathBuf,
    dashboard_port: u16,
    operator_pronouns: String,
    context_window_tokens: String,
    socket: &std::path::Path,
) -> Result<()> {
    let cwt: std::collections::HashMap<String, u64> = serde_json::from_str(&context_window_tokens)
        .context("--context-window-tokens: invalid JSON")?;
    let coord = Arc::new(Coordinator::open(
        &db,
        hyperhive_flake,
        nixpkgs_flake,
        nixpkgs_unstable_flake,
        dashboard_port,
        operator_pronouns,
        cwt,
    )?);
    manager_server::start(coord.clone())?;
    // Idempotent pre-flight: rewrite pre-meta-layout applied
    // repos, ensure proposed repos carry the `applied`
    // remote, bootstrap the meta repo, repoint containers at
    // `meta#<name>` (one-shot, guarded by a marker file).
    // Runs before manager auto-spawn so the new manager is
    // built against meta from the first attempt.
    if let Err(e) = migrate::run(&coord).await {
        tracing::warn!(error = ?e, "startup migration failed");
    }
    // Auto-create the root agent container if it isn't there yet. Block
    // on this — without root the system has no manager harness.
    // Failures are logged but allowed: a broken auto-spawn shouldn't
    // make the dashboard unreachable for debugging.
    if let Err(e) = auto_update::ensure_root_agent(&coord).await {
        tracing::warn!(error = ?e, "auto-spawn root agent failed");
    }
    // Auto-update in the background — don't block service start.
    // Sub-agent rebuilds can take tens of seconds; we want the admin
    // socket up immediately.
    let update_coord = coord.clone();
    tokio::spawn(async move {
        if let Err(e) = auto_update::run(update_coord).await {
            tracing::warn!(error = ?e, "auto-update task failed");
        }
    });
    // Forge user sweep: ensure every existing container has a
    // forgejo user + access token. No-op when the hive-forge
    // container isn't running. Backgrounded — touches the
    // forge state dir via `nixos-container run` which is slow.
    tokio::spawn(async move {
        forge::ensure_all().await;
    });
    // Matrix user sweep: same shape — ensure every container has
    // an account on the local matrix-tuwunel homeserver with an
    // access_token persisted to `<state>/matrix-token`. No-op when
    // the hive-matrix container isn't running. Backgrounded because
    // UIAA is a two-roundtrip dance per agent.
    tokio::spawn(async move {
        matrix::ensure_all().await;
    });
    // Periodic broker vacuum: drop fully-acked messages older
    // than 30 days. Delivered-but-unacked rows (recoverable via
    // requeue_inflight) and undelivered rows are always kept.
    // Runs hourly; first sweep happens immediately.
    let vacuum_coord = coord.clone();
    let mut vacuum_shutdown = coord.shutdown_rx();
    tokio::spawn(async move {
        let interval = std::time::Duration::from_hours(1);
        let keep_secs: i64 = 30 * 24 * 3600;
        loop {
            match vacuum_coord.broker.vacuum_delivered(keep_secs) {
                Ok(0) => {}
                Ok(n) => tracing::info!(removed = n, "broker vacuum"),
                Err(e) => tracing::warn!(error = ?e, "broker vacuum failed"),
            }
            tokio::select! {
                () = tokio::time::sleep(interval) => {}
                _ = vacuum_shutdown.changed() => {
                    tracing::info!("broker vacuum: shutdown signal received");
                    break;
                }
            }
        }
    });
    // Per-agent events.sqlite vacuum: host-side so the harness
    // doesn't need any retention wiring of its own.
    events_vacuum::spawn(&coord);
    // Per-agent turn-stats.sqlite vacuum: same pattern, 90-day
    // retention so trend analysis has enough history.
    stats_vacuum::spawn(&coord);
    // Per-agent bash-tasks file vacuum: host-side so the harness
    // cannot disable it. Deletes terminal task trios older than 48h.
    bash_tasks_vacuum::spawn(&coord);
    // build_logs.sqlite vacuum: c0re-side (single db). Failures kept
    // 30d, successes 24h — see `build_logs::vacuum` for the rule.
    hive_c0re::build_logs::spawn_vacuum(&coord);
    // Container crash watcher: emits HelperEvent::ContainerCrash
    // when a previously-running container goes away without an
    // operator-initiated transient state.
    crash_watch::spawn(coord.clone());
    // Agent-sockets marker poll: re-fires `agent_sockets::write`
    // and `gateway_nginx::write` every 10s so the JSON and nginx
    // config pick up newly-bound `.bound` markers after a rebuild.
    // Also retries any pending gateway nginx reload that failed on
    // the previous tick. write() is idempotent so steady-state cost
    // is one stat per agent per tick.
    // See `docs/gateway.md::Per-agent unix-socket upstream`.
    agent_sockets::spawn_poll();
    // Reminder scheduler: drains due reminders + handles
    // file_path payload persistence. See reminder_scheduler.rs.
    reminder_scheduler::spawn(coord.clone());
    // Scheduled-prompts worker: drains due scheduled_prompts rows
    // and fans the body out to each active target's inbox. See
    // scheduled_prompts_worker.rs.
    scheduled_prompts_worker::spawn(coord.clone());
    // Rebuild-queue worker: drains the global rebuild/meta-update/
    // spawn queue FIFO so hive-c0re never runs two heavyweight
    // container ops concurrently. Existing rebuild call sites
    // (auto_update, dashboard, manager, approval handler) enqueue
    // here instead of awaiting `rebuild_agent` inline. See
    // `rebuild_queue.rs`.
    {
        let q_coord = coord.clone();
        tokio::spawn(async move {
            rebuild_queue::run_worker(q_coord).await;
        });
    }
    // Forward every broker event onto the unified dashboard
    // channel with a freshly-stamped seq, so the dashboard SSE
    // sees broker messages + future mutation events on one
    // stream with one monotonic seq. The broker's intra-process
    // channel (used by `recv_blocking_batch`) stays untouched.
    spawn_broker_to_dashboard_forwarder(coord.clone());
    let dash_coord = coord.clone();
    tokio::spawn(async move {
        if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {
            tracing::error!(error = ?e, "dashboard failed");
        }
    });
    // Run the admin socket until a signal arrives; then signal
    // all background tasks so they exit cleanly before the
    // process terminates.
    let coord_sig = coord.clone();
    tokio::select! {
        res = server::serve(socket, coord) => { res? }
        _ = tokio::signal::ctrl_c() => {
            tracing::info!("SIGINT received — requesting shutdown");
            coord_sig.request_shutdown();
        }
        () = async {
            let mut sig = tokio::signal::unix::signal(
                tokio::signal::unix::SignalKind::terminate()
            ).expect("failed to install SIGTERM handler");
            sig.recv().await;
        } => {
            tracing::info!("SIGTERM received — requesting shutdown");
            coord_sig.request_shutdown();
        }
    }
    Ok(())
}

/// Re-emit every broker `MessageEvent` onto the dashboard channel as
/// a `DashboardEvent::Sent` / `Delivered` with a freshly-stamped seq.
/// Background task; runs for the life of the process. On a lagged
/// broker subscription we just keep going — the dashboard channel is
/// best-effort presentation plumbing, the broker keeps its own sqlite
/// log for replay.
fn spawn_broker_to_dashboard_forwarder(coord: Arc<Coordinator>) {
    use broker::MessageEvent;
    use dashboard_events::DashboardEvent;
    let mut rx = coord.broker.subscribe();
    tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(MessageEvent::Sent {
                    id,
                    from,
                    to,
                    body,
                    at,
                    in_reply_to,
                }) => {
                    let file_refs = dashboard::scan_validated_paths(&body);
                    coord.emit_dashboard_event(DashboardEvent::Sent {
                        seq: coord.next_seq(),
                        id,
                        from,
                        to,
                        body,
                        at,
                        in_reply_to,
                        file_refs,
                    });
                }
                Ok(MessageEvent::Delivered {
                    id,
                    from,
                    to,
                    body,
                    at,
                    in_reply_to,
                }) => {
                    let file_refs = dashboard::scan_validated_paths(&body);
                    coord.emit_dashboard_event(DashboardEvent::Delivered {
                        seq: coord.next_seq(),
                        id,
                        from,
                        to,
                        body,
                        at,
                        in_reply_to,
                        file_refs,
                    });
                }
                // Transient pings are not persisted and not shown in the
                // dashboard message history — ignore silently.
                Ok(MessageEvent::Ping { .. }) => {}
                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                    tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged");
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    });
}

fn render(resp: HostResponse) -> Result<()> {
    println!("{}", serde_json::to_string_pretty(&resp)?);
    if !resp.ok {
        bail!(resp.error.unwrap_or_else(|| "request failed".to_owned()));
    }
    Ok(())
}