//! Loops — durable recurring topology executions. The row holds the
//! definition (graph + task_template + triggers + repeat_policy) and a small
//! amount of scheduler state (enabled, next_fire_at, last_run_id).
//! Each fire produces a normal `topology_runs` row with loop_id + iteration
//! + parent_run_id set, so the run driver picks it up like any other job.
//!
//! See 0031 migration header for the state semantics and missed-window rule.

use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::PgPool;
use time::OffsetDateTime;
use uuid::Uuid;

use crate::DbError;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Loop {
    pub id: Uuid,
    pub workspace_id: Uuid,
    pub title: String,
    pub description: String,
    pub graph: Value,
    pub task_template: String,
    pub triggers: Value,
    pub repeat_policy: Value,
    pub enabled: bool,
    pub next_fire_at: Option<OffsetDateTime>,
    pub last_run_id: Option<Uuid>,
    pub webhook_token: Option<String>,
    pub webhook_signing_key: Option<String>,
    pub created_by: Uuid,
    pub created_at: OffsetDateTime,
    pub updated_at: OffsetDateTime,
}

/// Minimal fields the scheduler needs when it wakes up.
#[derive(Debug, Clone)]
pub struct DueLoop {
    pub id: Uuid,
    pub workspace_id: Uuid,
    pub graph: Value,
    pub task_template: String,
    pub triggers: Value,
    pub repeat_policy: Value,
    pub last_run_id: Option<Uuid>,
}

pub struct NewLoop<'a> {
    pub workspace_id: Uuid,
    pub title: &'a str,
    pub description: &'a str,
    pub graph: &'a Value,
    pub task_template: &'a str,
    pub triggers: &'a Value,
    pub repeat_policy: &'a Value,
    pub enabled: bool,
    pub next_fire_at: Option<OffsetDateTime>,
    pub webhook_token: Option<&'a str>,
    pub webhook_signing_key: Option<&'a str>,
    pub created_by: Uuid,
}

pub async fn create(pool: &PgPool, input: NewLoop<'_>) -> Result<Uuid, DbError> {
    let id = Uuid::now_v7();
    sqlx::query!(
        "INSERT INTO loops
            (id, workspace_id, title, description, graph, task_template,
             triggers, repeat_policy, enabled, next_fire_at,
             webhook_token, webhook_signing_key, created_by)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)",
        id,
        input.workspace_id,
        input.title,
        input.description,
        input.graph,
        input.task_template,
        input.triggers,
        input.repeat_policy,
        input.enabled,
        input.next_fire_at,
        input.webhook_token,
        input.webhook_signing_key,
        input.created_by,
    )
    .execute(pool)
    .await?;
    Ok(id)
}

pub async fn list(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<Loop>, DbError> {
    let rows = sqlx::query_as!(
        Loop,
        "SELECT id, workspace_id, title, description, graph, task_template,
                triggers, repeat_policy, enabled, next_fire_at, last_run_id,
                webhook_token, webhook_signing_key, created_by, created_at, updated_at
           FROM loops
          WHERE workspace_id = $1
          ORDER BY updated_at DESC",
        workspace_id,
    )
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

pub async fn get(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<Option<Loop>, DbError> {
    let row = sqlx::query_as!(
        Loop,
        "SELECT id, workspace_id, title, description, graph, task_template,
                triggers, repeat_policy, enabled, next_fire_at, last_run_id,
                webhook_token, webhook_signing_key, created_by, created_at, updated_at
           FROM loops
          WHERE id = $1 AND workspace_id = $2",
        id,
        workspace_id,
    )
    .fetch_optional(pool)
    .await?;
    Ok(row)
}

/// Look up a loop by its webhook token — used only by the webhook receiver,
/// which has no session context. Returns the minimal shape needed to enqueue
/// an iteration and verify the HMAC signature.
pub async fn get_by_webhook_token(
    pool: &PgPool,
    token: &str,
) -> Result<Option<(Uuid, Uuid, String, DueLoop)>, DbError> {
    let row = sqlx::query!(
        "SELECT id, workspace_id, graph, task_template, triggers, repeat_policy,
                last_run_id, webhook_signing_key
           FROM loops
          WHERE webhook_token = $1 AND enabled",
        token,
    )
    .fetch_optional(pool)
    .await?;
    Ok(row.and_then(|r| {
        let key = r.webhook_signing_key?;
        Some((
            r.id,
            r.workspace_id,
            key,
            DueLoop {
                id: r.id,
                workspace_id: r.workspace_id,
                graph: r.graph,
                task_template: r.task_template,
                triggers: r.triggers,
                repeat_policy: r.repeat_policy,
                last_run_id: r.last_run_id,
            },
        ))
    }))
}

pub struct UpdateLoop<'a> {
    pub title: &'a str,
    pub description: &'a str,
    pub graph: &'a Value,
    pub task_template: &'a str,
    pub triggers: &'a Value,
    pub repeat_policy: &'a Value,
    pub next_fire_at: Option<OffsetDateTime>,
}

pub async fn update(
    pool: &PgPool,
    id: Uuid,
    workspace_id: Uuid,
    input: UpdateLoop<'_>,
) -> Result<(), DbError> {
    sqlx::query!(
        "UPDATE loops
            SET title = $3, description = $4, graph = $5, task_template = $6,
                triggers = $7, repeat_policy = $8, next_fire_at = $9,
                updated_at = now()
          WHERE id = $1 AND workspace_id = $2",
        id,
        workspace_id,
        input.title,
        input.description,
        input.graph,
        input.task_template,
        input.triggers,
        input.repeat_policy,
        input.next_fire_at,
    )
    .execute(pool)
    .await?;
    Ok(())
}

pub async fn set_enabled(
    pool: &PgPool,
    id: Uuid,
    workspace_id: Uuid,
    enabled: bool,
) -> Result<(), DbError> {
    sqlx::query!(
        "UPDATE loops SET enabled = $3, updated_at = now()
          WHERE id = $1 AND workspace_id = $2",
        id,
        workspace_id,
        enabled,
    )
    .execute(pool)
    .await?;
    Ok(())
}

pub async fn delete(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<(), DbError> {
    sqlx::query!(
        "DELETE FROM loops WHERE id = $1 AND workspace_id = $2",
        id,
        workspace_id,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Persist the per-loop team container name + gateway URL after a
/// successful `spawn_loop` (P2). Dynamic query so the new columns don't
/// need a fresh .sqlx offline cache entry.
pub async fn set_zeroclaw_container(
    pool: &PgPool,
    loop_id: Uuid,
    workspace_id: Uuid,
    container: &str,
    gateway_url: &str,
) -> Result<(), DbError> {
    sqlx::query(
        "UPDATE loops
            SET zeroclaw_container = $3,
                zeroclaw_gateway_url = $4,
                updated_at = now()
          WHERE id = $1 AND workspace_id = $2",
    )
    .bind(loop_id)
    .bind(workspace_id)
    .bind(container)
    .bind(gateway_url)
    .execute(pool)
    .await?;
    Ok(())
}

/// Append one reorder rationale event to the loop's reorder_events
/// jsonb array. Called from the topology_worker completion hook after
/// parsing REORDER: markers out of the run output. Each event carries
/// the iteration index, run_id, text, and now() timestamp so a
/// downstream mini-timeline can show WHEN the plan was adjusted and
/// WHY. Idempotent: appending a duplicate text/run_id combo is allowed
/// (rare — indicates the parser matched twice on the same line).
pub async fn append_reorder_event(
    pool: &PgPool,
    loop_id: Uuid,
    run_id: Uuid,
    iteration: i32,
    text: &str,
) -> Result<(), DbError> {
    // Build the event server-side so `ts` uses postgres now() (canonical
    // wall clock; avoids skew if callers had stale local clocks).
    sqlx::query(
        "UPDATE loops
            SET reorder_events = reorder_events || jsonb_build_object(
                    'run_id',    $2::text,
                    'iteration', $3::int,
                    'text',      $4::text,
                    'ts',        to_char(now() AT TIME ZONE 'UTC',
                                         'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"')
                ),
                updated_at = now()
          WHERE id = $1",
    )
    .bind(loop_id)
    .bind(run_id.to_string())
    .bind(iteration)
    .bind(text)
    .execute(pool)
    .await?;
    Ok(())
}

/// Atomically append `completed` INT-XX ids to the loop's
/// `consumed_int_ids` array and bump `current_int_index` by the count
/// of NEW ids landed. Existing ids are not re-appended (idempotent on
/// re-runs). Called from the topology_worker completion hook after
/// parsing "COMPLETED: INT-XX" markers out of the run's final output.
pub async fn advance_after_completion(
    pool: &PgPool,
    loop_id: Uuid,
    completed: &[String],
) -> Result<(), DbError> {
    if completed.is_empty() {
        return Ok(());
    }
    // Use array set semantics: append only ids not already present.
    // The subquery computes the new list; length delta feeds the index bump.
    sqlx::query(
        "UPDATE loops
            SET consumed_int_ids = (
                SELECT ARRAY(
                    SELECT DISTINCT unnest(consumed_int_ids || $2::TEXT[])
                )
            ),
                current_int_index = current_int_index + (
                    SELECT count(*) FROM unnest($2::TEXT[]) AS n(v)
                     WHERE NOT (consumed_int_ids @> ARRAY[v])
                ),
                updated_at = now()
          WHERE id = $1",
    )
    .bind(loop_id)
    .bind(completed)
    .execute(pool)
    .await?;
    Ok(())
}

/// Bind (or unbind) a loop's source research topic. When set, the loop's
/// enqueue path prepends the topic's latest artifact + a "focus on the
/// next unconsumed INT" instruction to the coordinator task (option b,
/// order-sequential iteration).
pub async fn set_source_research_topic(
    pool: &PgPool,
    loop_id: Uuid,
    workspace_id: Uuid,
    source: Option<Uuid>,
) -> Result<(), DbError> {
    sqlx::query(
        "UPDATE loops
            SET source_research_topic_id = $3, updated_at = now()
          WHERE id = $1 AND workspace_id = $2",
    )
    .bind(loop_id)
    .bind(workspace_id)
    .bind(source)
    .execute(pool)
    .await?;
    Ok(())
}

/// Read a loop's source research topic id + consumed INT ids +
/// current index. Used by the enqueue path when building the
/// coordinator task string. Missing rows / NULL columns return None
/// so the caller can fall back to the plain task_template.
pub async fn source_research_context(
    pool: &PgPool,
    loop_id: Uuid,
) -> Result<Option<(Uuid, Vec<String>, i32)>, DbError> {
    use sqlx::Row;
    let row = sqlx::query(
        "SELECT source_research_topic_id, consumed_int_ids, current_int_index
           FROM loops
          WHERE id = $1",
    )
    .bind(loop_id)
    .fetch_optional(pool)
    .await?;
    Ok(row.and_then(|r| {
        let topic = r
            .try_get::<Option<Uuid>, _>("source_research_topic_id")
            .ok()
            .flatten()?;
        let consumed = r
            .try_get::<Vec<String>, _>("consumed_int_ids")
            .unwrap_or_default();
        let idx = r.try_get::<i32, _>("current_int_index").unwrap_or(0);
        Some((topic, consumed, idx))
    }))
}

/// Read the per-loop gateway URL (or None if the loop hasn't spawned a
/// container yet). Used by `topology_worker` to prefer the isolated
/// daemon over the workspace-wide one.
pub async fn zeroclaw_gateway_url(pool: &PgPool, loop_id: Uuid) -> Result<Option<String>, DbError> {
    use sqlx::Row;
    let row = sqlx::query("SELECT zeroclaw_gateway_url FROM loops WHERE id = $1")
        .bind(loop_id)
        .fetch_optional(pool)
        .await?;
    Ok(row.and_then(|r| {
        r.try_get::<Option<String>, _>("zeroclaw_gateway_url")
            .ok()
            .flatten()
    }))
}

// --- Staffing ---------------------------------------------------------------
//
// Loops attach agents, teams, and/or orgs. The three join tables are
// parallel; a loop can mix modes (e.g. one team + a couple of specialist
// agents). Callers use the `set_*` replace-all shape so PATCH is a single
// transactional swap — simpler than diffing and cheap for the list sizes
// this UI generates.

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentSlot {
    pub agent_id: Uuid,
    pub role_slot: Option<String>,
}

pub async fn set_agents(pool: &PgPool, loop_id: Uuid, slots: &[AgentSlot]) -> Result<(), DbError> {
    let mut tx = pool.begin().await?;
    sqlx::query!("DELETE FROM loop_agents WHERE loop_id = $1", loop_id)
        .execute(&mut *tx)
        .await?;
    for s in slots {
        sqlx::query!(
            "INSERT INTO loop_agents (loop_id, agent_id, role_slot)
             VALUES ($1, $2, $3)
             ON CONFLICT (loop_id, agent_id) DO UPDATE
                 SET role_slot = EXCLUDED.role_slot",
            loop_id,
            s.agent_id,
            s.role_slot,
        )
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(())
}

pub async fn agents(pool: &PgPool, loop_id: Uuid) -> Result<Vec<AgentSlot>, DbError> {
    let rows = sqlx::query!(
        "SELECT agent_id, role_slot FROM loop_agents WHERE loop_id = $1",
        loop_id,
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|r| AgentSlot {
            agent_id: r.agent_id,
            role_slot: r.role_slot,
        })
        .collect())
}

pub async fn set_teams(pool: &PgPool, loop_id: Uuid, ids: &[Uuid]) -> Result<(), DbError> {
    let mut tx = pool.begin().await?;
    sqlx::query!("DELETE FROM loop_teams WHERE loop_id = $1", loop_id)
        .execute(&mut *tx)
        .await?;
    for id in ids {
        sqlx::query!(
            "INSERT INTO loop_teams (loop_id, team_id) VALUES ($1, $2)
             ON CONFLICT (loop_id, team_id) DO NOTHING",
            loop_id,
            id,
        )
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(())
}

pub async fn teams(pool: &PgPool, loop_id: Uuid) -> Result<Vec<Uuid>, DbError> {
    let rows = sqlx::query!("SELECT team_id FROM loop_teams WHERE loop_id = $1", loop_id,)
        .fetch_all(pool)
        .await?;
    Ok(rows.into_iter().map(|r| r.team_id).collect())
}

pub async fn set_orgs(pool: &PgPool, loop_id: Uuid, ids: &[Uuid]) -> Result<(), DbError> {
    let mut tx = pool.begin().await?;
    sqlx::query!("DELETE FROM loop_orgs WHERE loop_id = $1", loop_id)
        .execute(&mut *tx)
        .await?;
    for id in ids {
        sqlx::query!(
            "INSERT INTO loop_orgs (loop_id, org_id) VALUES ($1, $2)
             ON CONFLICT (loop_id, org_id) DO NOTHING",
            loop_id,
            id,
        )
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(())
}

pub async fn orgs(pool: &PgPool, loop_id: Uuid) -> Result<Vec<Uuid>, DbError> {
    let rows = sqlx::query!("SELECT org_id FROM loop_orgs WHERE loop_id = $1", loop_id,)
        .fetch_all(pool)
        .await?;
    Ok(rows.into_iter().map(|r| r.org_id).collect())
}

/// Loops the scheduler tick should fire NOW. Only reads what the enqueue
/// path needs, so the tick stays cheap even when the workspace has hundreds
/// of loops.
pub async fn due(pool: &PgPool) -> Result<Vec<DueLoop>, DbError> {
    let rows = sqlx::query!(
        "SELECT id, workspace_id, graph, task_template, triggers, repeat_policy,
                last_run_id
           FROM loops
          WHERE enabled AND next_fire_at IS NOT NULL AND next_fire_at <= now()",
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|r| DueLoop {
            id: r.id,
            workspace_id: r.workspace_id,
            graph: r.graph,
            task_template: r.task_template,
            triggers: r.triggers,
            repeat_policy: r.repeat_policy,
            last_run_id: r.last_run_id,
        })
        .collect())
}

/// Post-fire bookkeeping: bump last_run_id + advance next_fire_at (NULL when
/// the loop has no cron trigger). Called by the scheduler after a successful
/// enqueue_iteration.
pub async fn mark_fired(
    pool: &PgPool,
    id: Uuid,
    run_id: Uuid,
    next_fire_at: Option<OffsetDateTime>,
) -> Result<(), DbError> {
    sqlx::query!(
        "UPDATE loops
            SET last_run_id = $2, next_fire_at = $3, updated_at = now()
          WHERE id = $1",
        id,
        run_id,
        next_fire_at,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Next iteration number for a loop (1 if it has never fired).
pub async fn next_iteration(pool: &PgPool, loop_id: Uuid) -> Result<i32, DbError> {
    let n: Option<i32> = sqlx::query_scalar!(
        "SELECT MAX(iteration) FROM topology_runs WHERE loop_id = $1",
        loop_id,
    )
    .fetch_one(pool)
    .await?;
    Ok(n.unwrap_or(0) + 1)
}

/// Enqueue an iteration as a normal `topology_runs` row. The scheduler,
/// on-completion hook, and webhook receiver all funnel through here so the
/// invariants (loop_id + iteration + parent_run_id all set together) stay
/// in one place.
pub async fn enqueue_iteration(
    pool: &PgPool,
    loop_id: Uuid,
    workspace_id: Uuid,
    task: &str,
    graph: &Value,
    iteration: i32,
    parent_run_id: Option<Uuid>,
) -> Result<Uuid, DbError> {
    let run_id = Uuid::now_v7();
    sqlx::query!(
        "INSERT INTO topology_runs
            (id, workspace_id, task, kind, status, graph, tier,
             loop_id, iteration, parent_run_id)
         VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7)",
        run_id,
        workspace_id,
        task,
        graph,
        loop_id,
        iteration,
        parent_run_id,
    )
    .execute(pool)
    .await?;
    Ok(run_id)
}