//! `GET /api/world/live` — the Clawmates Event Taxonomy as a Server-Sent Events
//! feed for the World + Observe visualizations. Read-only and workspace-scoped.
//!
//! Phase 1 emits the *real* shape of the workspace: a `topology.update` of its
//! agents, an `agent.status` per agent (working when it holds a live container,
//! else idle), and a `telemetry` snapshot — polled, mirroring `run_events_sse`.
//!
//! The richer `world.touch` / `agent.tool.call` events (an agent converging on
//! the node it acts upon — the Gource centerpiece) come from normalizing the
//! durable runner's `run_events`; that is the extension seam (see `normalize`),
//! filled in as the runner emits node targets. Until then the client's synthetic
//! fallback supplies that motion.

use std::collections::HashSet;
use std::convert::Infallible;
use std::time::Duration;

use axum::extract::{Query, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::IntoResponse;
use axum::Json;
use cm_domain::WorkspaceId;
use serde::Deserialize;
use serde_json::{json, Value};
use sqlx::{PgPool, Row};

use crate::{ApiError, AppState, Authed};

fn sse(event: &str, data: Value) -> Result<Event, Infallible> {
    Ok(Event::default().event(event).data(data.to_string()))
}

/// Agents that currently hold a live container (any kind) → "working".
async fn working_agents(pool: &PgPool, ws: WorkspaceId) -> HashSet<String> {
    let rows = sqlx::query(
        "SELECT DISTINCT a.id::text AS id
         FROM agents a JOIN agent_containers ac ON ac.agent_id = a.id
         WHERE a.workspace_id = $1",
    )
    .bind(ws.as_uuid())
    .fetch_all(pool)
    .await
    .unwrap_or_default();
    rows.into_iter().map(|r| r.get::<String, _>("id")).collect()
}

/// Currently-running runs in the workspace as (run_id, agent_id) — each is a
/// real "this agent is converging on its active work" signal (Gource).
async fn active_runs(pool: &PgPool, ws: WorkspaceId) -> Vec<(String, String)> {
    let rows = sqlx::query(
        "SELECT ar.id::text AS run_id, s.agent_id::text AS agent_id
         FROM agent_runs ar JOIN sessions s ON s.id = ar.session_id
         WHERE s.workspace_id = $1 AND ar.state = 'running'",
    )
    .bind(ws.as_uuid())
    .fetch_all(pool)
    .await
    .unwrap_or_default();
    rows.into_iter()
        .map(|r| (r.get::<String, _>("run_id"), r.get::<String, _>("agent_id")))
        .collect()
}

/// A short human label for a tool's input (for the tool-call target).
fn summarize_input(input: &Value) -> String {
    for k in ["target", "path", "url", "query", "name", "file", "command"] {
        if let Some(s) = input.get(k).and_then(|v| v.as_str()) {
            return s.chars().take(48).collect();
        }
    }
    String::new()
}

/// Normalize one durable `run_events` row into taxonomy events — the Rust twin of
/// the handoff bridge's normalize(). The runner already journals these, so the
/// live view shows REAL reasoning, tool-convergence and doors with no runner edit.
fn normalize_run_event(agent_id: &str, event_type: &str, payload: &Value) -> Vec<(&'static str, Value)> {
    let mut out = Vec::new();
    match event_type {
        "text_delta" => {
            if let Some(delta) = payload.get("delta").and_then(|v| v.as_str()) {
                out.push(("agent.reasoning.delta", json!({ "agentId": agent_id, "text": delta })));
            }
        }
        "step_started" => {
            let tool = payload.get("tool").and_then(|v| v.as_str()).unwrap_or("tool");
            let node_id = format!("tool:{tool}");
            let target = payload.get("input").map(summarize_input).unwrap_or_default();
            out.push(("agent.tool.call", json!({ "agentId": agent_id, "tool": tool, "target": target })));
            out.push(("node.activity", json!({ "nodeId": node_id, "label": tool, "kind": "service", "heat": 0.9 })));
            // the agent converges on the tool it's using (the Gource beam)
            out.push(("world.touch", json!({ "agentId": agent_id, "nodeId": node_id, "kind": "service" })));
        }
        "approval_required" => {
            let action = payload.get("action_type").and_then(|v| v.as_str()).unwrap_or("action");
            let category = payload.get("category").and_then(|v| v.as_str()).unwrap_or("");
            let door_id = payload.get("approval_id").and_then(|v| v.as_str()).unwrap_or("");
            out.push((
                "door.request",
                json!({ "doorId": door_id, "agentId": agent_id, "action": action, "target": category, "summary": action }),
            ));
        }
        _ => {}
    }
    out
}

/// Count of doors (approvals) awaiting a decision in the workspace.
async fn doors_pending(pool: &PgPool, ws: WorkspaceId) -> i64 {
    sqlx::query_scalar::<_, i64>(
        "SELECT count(*) FROM approvals WHERE workspace_id = $1 AND status = 'pending'",
    )
    .bind(ws.as_uuid())
    .fetch_one(pool)
    .await
    .unwrap_or(0)
}

/// `GET /api/world/live` — the taxonomy SSE feed.
pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) -> impl IntoResponse {
    let pool = state.pool.clone();
    let ws = user.workspace_id;

    let stream = async_stream::stream! {
        let mut first = true;
        // Remember last status per agent so we only push deltas after the seed.
        let mut last: std::collections::HashMap<String, String> = std::collections::HashMap::new();
        // Per-run journal cursor so we stream only NEW run_events each poll.
        let mut cursors: std::collections::HashMap<String, i64> = std::collections::HashMap::new();
        loop {
            let roster = match cm_db::repo::agents::roster(&pool, ws).await {
                Ok(r) => r,
                Err(_) => break,
            };
            let mut working = working_agents(&pool, ws).await;
            let runs = active_runs(&pool, ws).await;
            for (_, agent_id) in &runs {
                working.insert(agent_id.clone());
            }

            if first {
                // Seed the world graph with the workspace's agents as nodes.
                let nodes: Vec<Value> = roster
                    .iter()
                    .map(|a| json!({ "id": a.id.to_string(), "tier": "agent", "label": a.name }))
                    .collect();
                yield sse("topology.update", json!({ "formation": "live", "nodes": nodes }));
            }

            for a in &roster {
                let id = a.id.to_string();
                let status = if working.contains(&id) { "working" } else { "idle" };
                if last.get(&id).map(|s| s != status).unwrap_or(true) {
                    last.insert(id.clone(), status.to_string());
                    yield sse(
                        "agent.status",
                        json!({ "agentId": id, "status": status, "role": a.job_title }),
                    );
                }
            }

            // Real convergence: each running agent beams toward its active-run node.
            for (run_id, agent_id) in &runs {
                let node_id = format!("run:{}", &run_id[..run_id.len().min(8)]);
                yield sse(
                    "node.activity",
                    json!({ "nodeId": node_id, "label": "active run", "kind": "event", "heat": 0.85 }),
                );
                yield sse("world.touch", json!({ "agentId": agent_id, "nodeId": node_id, "kind": "event" }));

                // Tail the run's journal for richer real events (reasoning, tool
                // convergence, doors). On first sight, jump the cursor to the
                // current max so we stream forward without replaying the backlog.
                if let Some(&after) = cursors.get(run_id) {
                    let rows = sqlx::query(
                        "SELECT seq, event_type, payload FROM run_events
                         WHERE run_id = $1::uuid AND seq > $2 ORDER BY seq ASC LIMIT 200",
                    )
                    .bind(run_id)
                    .bind(after)
                    .fetch_all(&pool)
                    .await
                    .unwrap_or_default();
                    let mut maxseq = after;
                    for row in &rows {
                        let seq: i64 = row.get("seq");
                        let et: String = row.get("event_type");
                        let payload: Value = row.get("payload");
                        for (t, d) in normalize_run_event(agent_id, &et, &payload) {
                            yield sse(t, d);
                        }
                        if seq > maxseq {
                            maxseq = seq;
                        }
                    }
                    cursors.insert(run_id.clone(), maxseq);
                } else {
                    let maxseq: i64 = sqlx::query_scalar(
                        "SELECT coalesce(max(seq), -1) FROM run_events WHERE run_id = $1::uuid",
                    )
                    .bind(run_id)
                    .fetch_one(&pool)
                    .await
                    .unwrap_or(-1);
                    cursors.insert(run_id.clone(), maxseq);
                }
            }

            yield sse(
                "telemetry",
                json!({ "doorsPending": doors_pending(&pool, ws).await, "loops": runs.len() }),
            );

            first = false;
            tokio::time::sleep(Duration::from_secs(2)).await;
        }
    };

    Sse::new(stream).keep_alive(KeepAlive::default())
}

#[derive(Deserialize)]
pub struct ReplayQuery {
    hours: Option<i64>,
}

/// `GET /api/world/replay?hours=24` — a Gource-style timeline reconstructed from
/// the workspace's run history: a sorted list of timestamped taxonomy events the
/// client's WorldClock plays back into the same engine (live + replay).
pub async fn world_replay(
    State(state): State<AppState>,
    Authed(user): Authed,
    Query(q): Query<ReplayQuery>,
) -> Result<Json<Value>, ApiError> {
    let hours = q.hours.unwrap_or(24).clamp(1, 720);
    let rows = sqlx::query(
        "SELECT ar.id::text AS run_id, s.agent_id::text AS agent_id,
                extract(epoch FROM ar.created_at)::float8 AS started
         FROM agent_runs ar JOIN sessions s ON s.id = ar.session_id
         WHERE s.workspace_id = $1 AND ar.created_at > now() - ($2 * interval '1 hour')
         ORDER BY ar.created_at ASC",
    )
    .bind(user.workspace_id.as_uuid())
    .bind(hours)
    .fetch_all(&state.pool)
    .await?;

    let mut events: Vec<Value> = Vec::new();
    for r in &rows {
        let run_id: String = r.get("run_id");
        let agent_id: String = r.get("agent_id");
        let started: f64 = r.get("started");
        let node_id = format!("run:{}", &run_id[..run_id.len().min(8)]);
        events.push(json!({ "t": started, "type": "agent.status", "data": { "agentId": agent_id, "status": "working" }}));
        events.push(json!({ "t": started, "type": "node.activity", "data": { "nodeId": node_id, "label": "run", "kind": "event", "heat": 0.85 }}));
        events.push(json!({ "t": started, "type": "world.touch", "data": { "agentId": agent_id, "nodeId": node_id, "kind": "event" }}));
    }
    Ok(Json(json!({ "events": events, "hours": hours, "count": rows.len() })))
}

// THE NORMALIZE SEAM (future) -------------------------------------------------
// Translate one durable `run_events` row into zero+ taxonomy events, the Rust
// twin of the handoff bridge's normalize(). Wire this into the poll loop above
// once the runner emits node targets:
//   turn.started   -> agent.status(working) [+ agent.task.update]
//   turn.token     -> agent.reasoning.delta
//   tool.invoked   -> agent.tool.call [+ world.touch if a nodeId is present]
//   door.requested -> door.request ; door.resolved -> door.resolve
//   agent.message  -> agent.message ; runner.telemetry -> telemetry
#[allow(dead_code)]
fn normalize(_event_type: &str, _payload: &Value) -> Vec<(&'static str, Value)> {
    Vec::new()
}