//! Persistence for workflow definitions (v1.2 Workflows).
//!
//! The `workflows` table (0071) is owner-polymorphic like `scripts`; M1 authors
//! **app-owned** workflows only (the `group_id` column ships unused). The read
//! trait backs the admin/CLI surface; the `*_tx` free-fns are used by the
//! declarative `apply` reconcile engine (Create/Update/Delete by `lower(name)`,
//! all in one transaction) — mirroring `trigger_repo::insert_trigger_tx`.
//!
//! Run/step state (`workflow_runs`, `workflow_run_steps`) lands in M2: the
//! durable orchestrator seeds a run (`start_run`), claims a `ready` step with
//! the same `FOR UPDATE SKIP LOCKED` lease `queue_repo` uses, and — in one
//! token-gated transaction — writes the step outcome and advances the DAG
//! (`complete_step_and_advance`). The graph-advance decision is a pure,
//! DB-free function ([`compute_advance`]) so it is unit-testable.

use std::collections::BTreeMap;

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use sqlx::PgPool;
use thiserror::Error;
use uuid::Uuid;

use picloud_shared::workflow::{
    OnError, RunStatus, StepStatus, WorkflowDefinition, WorkflowStepDef,
};
use picloud_shared::{AppId, WorkflowId, WorkflowRunId, WorkflowRunStepId};

#[derive(Debug, Error)]
pub enum WorkflowRepoError {
    #[error(transparent)]
    Db(#[from] sqlx::Error),
    #[error("workflow definition (de)serialization failed: {0}")]
    Serde(String),
}

/// A stored workflow definition (app-owned in M1).
#[derive(Debug, Clone)]
pub struct Workflow {
    pub id: WorkflowId,
    pub app_id: AppId,
    pub name: String,
    pub definition: WorkflowDefinition,
    pub enabled: bool,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// Raw row shape (definition decoded separately).
#[derive(sqlx::FromRow)]
struct WorkflowRow {
    id: Uuid,
    app_id: Uuid,
    name: String,
    definition: serde_json::Value,
    enabled: bool,
    created_at: DateTime<Utc>,
    updated_at: DateTime<Utc>,
}

impl TryFrom<WorkflowRow> for Workflow {
    type Error = WorkflowRepoError;
    fn try_from(r: WorkflowRow) -> Result<Self, Self::Error> {
        Ok(Self {
            id: r.id.into(),
            app_id: r.app_id.into(),
            name: r.name,
            definition: serde_json::from_value(r.definition)
                .map_err(|e| WorkflowRepoError::Serde(e.to_string()))?,
            enabled: r.enabled,
            created_at: r.created_at,
            updated_at: r.updated_at,
        })
    }
}

const SELECT_COLS: &str = "id, app_id, name, definition, enabled, created_at, updated_at";

#[async_trait]
pub trait WorkflowRepo: Send + Sync {
    /// All workflows owned by `app_id`, ordered by name.
    async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Workflow>, WorkflowRepoError>;
    /// A single app-owned workflow by case-insensitive name.
    async fn get_by_name(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Workflow>, WorkflowRepoError>;
    /// A single workflow by id (app-scoped for isolation).
    async fn get_by_id(
        &self,
        app_id: AppId,
        id: WorkflowId,
    ) -> Result<Option<Workflow>, WorkflowRepoError>;
}

pub struct PostgresWorkflowRepo {
    pool: PgPool,
}

impl PostgresWorkflowRepo {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

#[async_trait]
impl WorkflowRepo for PostgresWorkflowRepo {
    async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Workflow>, WorkflowRepoError> {
        let rows: Vec<WorkflowRow> = sqlx::query_as(&format!(
            "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 ORDER BY LOWER(name)"
        ))
        .bind(app_id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        rows.into_iter().map(TryInto::try_into).collect()
    }

    async fn get_by_name(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Workflow>, WorkflowRepoError> {
        let row: Option<WorkflowRow> = sqlx::query_as(&format!(
            "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 AND LOWER(name) = LOWER($2)"
        ))
        .bind(app_id.into_inner())
        .bind(name)
        .fetch_optional(&self.pool)
        .await?;
        row.map(TryInto::try_into).transpose()
    }

    async fn get_by_id(
        &self,
        app_id: AppId,
        id: WorkflowId,
    ) -> Result<Option<Workflow>, WorkflowRepoError> {
        let row: Option<WorkflowRow> = sqlx::query_as(&format!(
            "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 AND id = $2"
        ))
        .bind(app_id.into_inner())
        .bind(id.into_inner())
        .fetch_optional(&self.pool)
        .await?;
        row.map(TryInto::try_into).transpose()
    }
}

/// Pool-based read of an app's workflows (used by `apply`'s `load_current`,
/// which reads several markers directly off the pool).
pub async fn list_workflows_for_app(
    pool: &PgPool,
    app_id: AppId,
) -> Result<Vec<Workflow>, WorkflowRepoError> {
    let rows: Vec<WorkflowRow> = sqlx::query_as(&format!(
        "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 ORDER BY LOWER(name)"
    ))
    .bind(app_id.into_inner())
    .fetch_all(pool)
    .await?;
    rows.into_iter().map(TryInto::try_into).collect()
}

// ---- reconcile tx free-fns (used by apply_service) ------------------------

/// Load an app's workflows inside the apply transaction (so the diff sees a
/// snapshot consistent with the writes).
pub async fn list_workflows_for_app_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    app_id: AppId,
) -> Result<Vec<Workflow>, WorkflowRepoError> {
    let rows: Vec<WorkflowRow> = sqlx::query_as(&format!(
        "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 ORDER BY LOWER(name)"
    ))
    .bind(app_id.into_inner())
    .fetch_all(&mut **tx)
    .await?;
    rows.into_iter().map(TryInto::try_into).collect()
}

/// Insert an app-owned workflow.
pub async fn insert_workflow_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    app_id: AppId,
    name: &str,
    definition: &WorkflowDefinition,
    enabled: bool,
) -> Result<WorkflowId, WorkflowRepoError> {
    let def =
        serde_json::to_value(definition).map_err(|e| WorkflowRepoError::Serde(e.to_string()))?;
    let (id,): (Uuid,) = sqlx::query_as(
        "INSERT INTO workflows (app_id, name, definition, enabled) \
         VALUES ($1, $2, $3, $4) RETURNING id",
    )
    .bind(app_id.into_inner())
    .bind(name)
    .bind(def)
    .bind(enabled)
    .fetch_one(&mut **tx)
    .await?;
    Ok(id.into())
}

/// Update a workflow's definition + enabled flag (name/identity unchanged).
pub async fn update_workflow_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    id: WorkflowId,
    definition: &WorkflowDefinition,
    enabled: bool,
) -> Result<(), WorkflowRepoError> {
    let def =
        serde_json::to_value(definition).map_err(|e| WorkflowRepoError::Serde(e.to_string()))?;
    sqlx::query(
        "UPDATE workflows SET definition = $2, enabled = $3, updated_at = NOW() WHERE id = $1",
    )
    .bind(id.into_inner())
    .bind(def)
    .bind(enabled)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// Delete a workflow (prune). RESTRICTs if runs reference it — a workflow with
/// history is kept; this is surfaced as an error to the reconcile caller.
pub async fn delete_workflow_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    id: WorkflowId,
) -> Result<(), WorkflowRepoError> {
    sqlx::query("DELETE FROM workflows WHERE id = $1")
        .bind(id.into_inner())
        .execute(&mut **tx)
        .await?;
    Ok(())
}

// ===========================================================================
// M2 — run/step state + the durable orchestrator's persistence surface.
// ===========================================================================

/// Everything needed to seed a fresh run (mirrors `NewQueueMessage`). The
/// orchestrator's M5 `workflow::start` and the M2 tests both build this.
#[derive(Debug, Clone)]
pub struct NewWorkflowRun {
    pub workflow_id: WorkflowId,
    pub app_id: AppId,
    pub input: serde_json::Value,
    /// Correlates every step execution in `execution_logs` under one id.
    pub root_execution_id: Uuid,
    /// 0 for a top-level run; a nested sub-workflow (M4) is parent + 1.
    pub workflow_depth: i32,
    pub parent_run_id: Option<WorkflowRunId>,
    pub parent_step_id: Option<WorkflowRunStepId>,
}

/// A run row, read for the admin/CLI surface + test assertions.
#[derive(Debug, Clone)]
pub struct WorkflowRun {
    pub id: WorkflowRunId,
    pub workflow_id: WorkflowId,
    pub app_id: AppId,
    pub status: RunStatus,
    pub input: serde_json::Value,
    pub output: Option<serde_json::Value>,
    pub error: Option<String>,
    pub root_execution_id: Uuid,
    pub workflow_depth: i32,
    pub parent_run_id: Option<WorkflowRunId>,
    pub parent_step_id: Option<WorkflowRunStepId>,
    pub started_at: Option<DateTime<Utc>>,
    pub finished_at: Option<DateTime<Utc>>,
    pub created_at: DateTime<Utc>,
}

/// A step row within a run (read for the admin/CLI surface + assertions).
#[derive(Debug, Clone)]
pub struct WorkflowRunStep {
    pub id: WorkflowRunStepId,
    pub run_id: WorkflowRunId,
    pub step_name: String,
    pub status: StepStatus,
    pub attempt: i32,
    pub max_attempts: i32,
    pub output: Option<serde_json::Value>,
    pub error: Option<String>,
    pub child_run_id: Option<WorkflowRunId>,
}

/// A claimed `ready` step, joined with its run + workflow definition so the
/// orchestrator has everything it needs to resolve, execute, and advance
/// without a second round trip.
#[derive(Debug, Clone)]
pub struct ClaimedStep {
    pub step_id: WorkflowRunStepId,
    pub run_id: WorkflowRunId,
    pub step_name: String,
    /// 1-indexed current attempt (the claim increments it).
    pub attempt: i32,
    pub max_attempts: i32,
    pub claim_token: Uuid,
    // ---- run context ----
    pub app_id: AppId,
    pub workflow_id: WorkflowId,
    pub root_execution_id: Uuid,
    pub workflow_depth: i32,
    pub run_input: serde_json::Value,
    pub definition: WorkflowDefinition,
}

impl ClaimedStep {
    /// The definition of the step being executed.
    #[must_use]
    pub fn step_def(&self) -> Option<&WorkflowStepDef> {
        self.definition
            .steps
            .iter()
            .find(|s| s.name == self.step_name)
    }
}

/// The outcome the orchestrator hands back after executing a step. On
/// `Failed`, the repo decides retry-vs-terminal from `attempt < max_attempts`;
/// the orchestrator supplies the delay to use if a retry remains.
#[derive(Debug, Clone)]
pub enum StepOutcome {
    Succeeded(serde_json::Value),
    Failed {
        error: String,
        retry_delay: chrono::Duration,
    },
    /// The step's `when` condition evaluated false — it is `skipped` (never
    /// executed) and counts as satisfied-but-empty for its dependents (M3).
    Skipped,
}

/// What `complete_step_and_advance` did — surfaced mostly for tests/logging.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AdvanceResult {
    /// The claim token no longer matched (a stale worker); nothing written.
    Stale,
    /// The step failed but had attempts left; it was re-armed as `ready`.
    Retried,
    /// The step reached a terminal status and the run graph was advanced.
    Advanced,
}

// ---- the pure graph-advance decision (DB-free, unit-tested) --------------

/// The write plan `complete_step_and_advance` derives after a step reaches a
/// terminal status: which `pending` steps become `ready`, and the run's new
/// status/output/error.
#[derive(Debug, Clone, PartialEq)]
pub struct AdvancePlan {
    pub promote: Vec<String>,
    pub run_status: RunStatus,
    pub run_error: Option<String>,
    pub run_output: Option<serde_json::Value>,
}

/// Is `dep` satisfied for the purpose of unblocking a dependent? Succeeded and
/// skipped always satisfy; a `failed` step satisfies **only** when its
/// definition says `on_error = continue` (the run limps on).
fn dep_satisfied(status: StepStatus, on_error: OnError) -> bool {
    match status {
        StepStatus::Succeeded | StepStatus::Skipped => true,
        StepStatus::Failed => matches!(on_error, OnError::Continue),
        _ => false,
    }
}

/// Pure DAG advance. Given the definition and the current per-step
/// status/output snapshot, decide the promotions + terminal run state.
///
/// - A `pending` step whose every `depends_on` is satisfied becomes `ready`.
/// - If any step `failed` with `on_error = fail`, the run fails immediately.
/// - Else if every step is terminal, the run succeeds; its output is the
///   object of `{ step_name: output }` over the succeeded steps.
/// - Otherwise the run is still `running`.
#[must_use]
pub fn compute_advance(
    def: &WorkflowDefinition,
    statuses: &BTreeMap<String, StepStatus>,
    outputs: &BTreeMap<String, serde_json::Value>,
) -> AdvancePlan {
    let by_name: BTreeMap<&str, &WorkflowStepDef> =
        def.steps.iter().map(|s| (s.name.as_str(), s)).collect();
    let status_of = |name: &str| statuses.get(name).copied().unwrap_or(StepStatus::Pending);

    // Promote every pending step whose deps are all satisfied.
    let mut promote = Vec::new();
    for step in &def.steps {
        if status_of(&step.name) != StepStatus::Pending {
            continue;
        }
        let ready = step.depends_on.iter().all(|dep| {
            let d_on_err = by_name
                .get(dep.as_str())
                .map_or(OnError::Fail, |s| s.on_error);
            dep_satisfied(status_of(dep), d_on_err)
        });
        if ready {
            promote.push(step.name.clone());
        }
    }

    // A hard failure (on_error = fail that exhausted) fails the whole run.
    if let Some(failed) = def
        .steps
        .iter()
        .find(|s| status_of(&s.name) == StepStatus::Failed && matches!(s.on_error, OnError::Fail))
    {
        return AdvancePlan {
            promote,
            run_status: RunStatus::Failed,
            run_error: Some(format!("step {:?} failed", failed.name)),
            run_output: None,
        };
    }

    // All steps terminal (accounting for the promotions we just queued — a
    // promoted step is pending → ready, i.e. not terminal, so the run stays
    // running whenever `promote` is non-empty).
    let all_terminal =
        promote.is_empty() && def.steps.iter().all(|s| status_of(&s.name).is_terminal());
    if all_terminal {
        let mut out = serde_json::Map::new();
        for step in &def.steps {
            if status_of(&step.name) == StepStatus::Succeeded {
                if let Some(v) = outputs.get(&step.name) {
                    out.insert(step.name.clone(), v.clone());
                }
            }
        }
        // A run reaches here as Succeeded even if some steps failed under
        // `on_error = continue` (a hard `fail` short-circuits above). Those
        // steps contribute no output, so surface them in the run's `error`
        // field — otherwise a partial failure is invisible at the run level.
        let failed: Vec<&str> = def
            .steps
            .iter()
            .filter(|s| status_of(&s.name) == StepStatus::Failed)
            .map(|s| s.name.as_str())
            .collect();
        let run_error = (!failed.is_empty()).then(|| {
            format!(
                "run succeeded with {} continued failure(s): {}",
                failed.len(),
                failed.join(", ")
            )
        });
        return AdvancePlan {
            promote,
            run_status: RunStatus::Succeeded,
            run_error,
            run_output: Some(serde_json::Value::Object(out)),
        };
    }

    AdvancePlan {
        promote,
        run_status: RunStatus::Running,
        run_error: None,
        run_output: None,
    }
}

// ---- run seeding ----------------------------------------------------------

/// Seed a run: insert the `workflow_runs` row + one `workflow_run_steps` row
/// per definition step (root steps — no `depends_on` — start `ready`, the rest
/// `pending`). All in one transaction so a partially-seeded run is never
/// visible to the orchestrator's claim scan.
pub async fn start_run(
    pool: &PgPool,
    new: NewWorkflowRun,
    definition: &WorkflowDefinition,
) -> Result<WorkflowRunId, WorkflowRepoError> {
    let mut tx = pool.begin().await?;
    let run_id = seed_run_tx(&mut tx, &new, definition).await?;
    tx.commit().await?;
    Ok(run_id)
}

/// Insert the `workflow_runs` row + its `workflow_run_steps` (roots `ready`,
/// the rest `pending`) inside an open transaction. Shared by `start_run` and
/// the nested-child start (`start_child_and_park`).
pub async fn seed_run_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    new: &NewWorkflowRun,
    definition: &WorkflowDefinition,
) -> Result<WorkflowRunId, WorkflowRepoError> {
    let (run_id,): (Uuid,) = sqlx::query_as(
        "INSERT INTO workflow_runs \
            (workflow_id, app_id, status, input, root_execution_id, \
             workflow_depth, parent_run_id, parent_step_id) \
         VALUES ($1, $2, 'pending', $3, $4, $5, $6, $7) RETURNING id",
    )
    .bind(new.workflow_id.into_inner())
    .bind(new.app_id.into_inner())
    .bind(&new.input)
    .bind(new.root_execution_id)
    .bind(new.workflow_depth)
    .bind(new.parent_run_id.map(WorkflowRunId::into_inner))
    .bind(new.parent_step_id.map(WorkflowRunStepId::into_inner))
    .fetch_one(&mut **tx)
    .await?;

    for step in &definition.steps {
        let status = if step.depends_on.is_empty() {
            "ready"
        } else {
            "pending"
        };
        let max_attempts = i32::try_from(step.max_attempts()).unwrap_or(1);
        sqlx::query(
            "INSERT INTO workflow_run_steps \
                (run_id, step_name, status, max_attempts) \
             VALUES ($1, $2, $3, $4)",
        )
        .bind(run_id)
        .bind(&step.name)
        .bind(status)
        .bind(max_attempts)
        .execute(&mut **tx)
        .await?;
    }

    Ok(run_id.into())
}

// ---- claim / complete / reclaim ------------------------------------------

#[derive(sqlx::FromRow)]
struct ClaimedRow {
    id: Uuid,
    run_id: Uuid,
    step_name: String,
    attempt: i32,
    max_attempts: i32,
    app_id: Uuid,
    workflow_id: Uuid,
    root_execution_id: Uuid,
    workflow_depth: i32,
    input: serde_json::Value,
    definition: serde_json::Value,
}

/// Atomically claim one `ready`, due step across all active runs — the queue
/// competing-consumer lease (`FOR UPDATE SKIP LOCKED`), so parallel workers
/// (and, cluster mode later, parallel nodes) never grab the same step. The
/// claimed step flips to `running` with a fresh `claim_token` in one atomic
/// statement. A *separate*, non-transactional follow-up statement then reflects
/// the run as `pending → running` — this is a promptness convenience, not part
/// of the claim's atomicity: a crash in between is harmless (the token-gated
/// advance, and `reclaim`, both re-derive run status). Returns `None` when
/// nothing is claimable.
pub async fn claim_ready_step(pool: &PgPool) -> Result<Option<ClaimedStep>, WorkflowRepoError> {
    let token = Uuid::new_v4();
    let row: Option<ClaimedRow> = sqlx::query_as(
        "WITH claimed AS ( \
            UPDATE workflow_run_steps \
            SET status = 'running', claim_token = $1, claimed_at = NOW(), \
                attempt = attempt + 1, updated_at = NOW() \
            WHERE id = ( \
                SELECT s.id FROM workflow_run_steps s \
                JOIN workflow_runs r ON r.id = s.run_id \
                WHERE s.status = 'ready' \
                  AND (s.next_attempt_at IS NULL OR s.next_attempt_at <= NOW()) \
                  AND r.status IN ('pending', 'running') \
                ORDER BY s.created_at \
                FOR UPDATE OF s SKIP LOCKED \
                LIMIT 1 \
            ) \
            RETURNING id, run_id, step_name, attempt, max_attempts \
        ) \
        SELECT c.id, c.run_id, c.step_name, c.attempt, c.max_attempts, \
               r.app_id, r.workflow_id, r.root_execution_id, r.workflow_depth, \
               r.input, w.definition \
        FROM claimed c \
        JOIN workflow_runs r ON r.id = c.run_id \
        JOIN workflows w ON w.id = r.workflow_id",
    )
    .bind(token)
    .fetch_optional(pool)
    .await?;

    let Some(r) = row else { return Ok(None) };

    // Reflect the run as running promptly (advance also does this, but a
    // long-running first step would otherwise leave the run 'pending').
    sqlx::query(
        "UPDATE workflow_runs SET status = 'running', started_at = COALESCE(started_at, NOW()) \
         WHERE id = $1 AND status = 'pending'",
    )
    .bind(r.run_id)
    .execute(pool)
    .await?;

    let definition = serde_json::from_value(r.definition)
        .map_err(|e| WorkflowRepoError::Serde(e.to_string()))?;
    Ok(Some(ClaimedStep {
        step_id: r.id.into(),
        run_id: r.run_id.into(),
        step_name: r.step_name,
        attempt: r.attempt,
        max_attempts: r.max_attempts,
        claim_token: token,
        app_id: r.app_id.into(),
        workflow_id: r.workflow_id.into(),
        root_execution_id: r.root_execution_id,
        workflow_depth: r.workflow_depth,
        run_input: r.input,
        definition,
    }))
}

#[derive(sqlx::FromRow)]
struct StepStateRow {
    step_name: String,
    status: String,
    output: Option<serde_json::Value>,
}

/// Write a claimed step's outcome and advance the run graph — all in one
/// token-gated transaction. A stale worker (claim reclaimed out from under it)
/// matches zero rows and writes nothing (`AdvanceResult::Stale`).
///
/// - `Succeeded` → the step is marked `succeeded` with its output, then the
///   graph advances (promotions + terminal run status).
/// - `Failed` with attempts left → the step is re-armed `ready` with a backoff
///   gate; the run keeps running (`Retried`, no advance).
/// - `Failed` exhausted → the step is marked `failed`, then the graph advances
///   (its `on_error` decides whether the run fails or limps on).
#[allow(clippy::too_many_lines)]
pub async fn complete_step_and_advance(
    pool: &PgPool,
    claimed: &ClaimedStep,
    outcome: StepOutcome,
) -> Result<AdvanceResult, WorkflowRepoError> {
    let mut tx = pool.begin().await?;

    // 1. Token-gated step write.
    match &outcome {
        StepOutcome::Succeeded(out) => {
            let res = sqlx::query(
                "UPDATE workflow_run_steps \
                 SET status = 'succeeded', output = $3, error = NULL, \
                     claim_token = NULL, claimed_at = NULL, updated_at = NOW() \
                 WHERE id = $1 AND claim_token = $2",
            )
            .bind(claimed.step_id.into_inner())
            .bind(claimed.claim_token)
            .bind(out)
            .execute(&mut *tx)
            .await?;
            if res.rows_affected() == 0 {
                tx.rollback().await?;
                return Ok(AdvanceResult::Stale);
            }
        }
        StepOutcome::Failed { error, retry_delay } => {
            if claimed.attempt < claimed.max_attempts {
                let next = Utc::now() + *retry_delay;
                let res = sqlx::query(
                    "UPDATE workflow_run_steps \
                     SET status = 'ready', error = $3, claim_token = NULL, \
                         claimed_at = NULL, next_attempt_at = $4, updated_at = NOW() \
                     WHERE id = $1 AND claim_token = $2",
                )
                .bind(claimed.step_id.into_inner())
                .bind(claimed.claim_token)
                .bind(error)
                .bind(next)
                .execute(&mut *tx)
                .await?;
                if res.rows_affected() == 0 {
                    tx.rollback().await?;
                    return Ok(AdvanceResult::Stale);
                }
                tx.commit().await?;
                return Ok(AdvanceResult::Retried);
            }
            let res = sqlx::query(
                "UPDATE workflow_run_steps \
                 SET status = 'failed', error = $3, claim_token = NULL, \
                     claimed_at = NULL, updated_at = NOW() \
                 WHERE id = $1 AND claim_token = $2",
            )
            .bind(claimed.step_id.into_inner())
            .bind(claimed.claim_token)
            .bind(error)
            .execute(&mut *tx)
            .await?;
            if res.rows_affected() == 0 {
                tx.rollback().await?;
                return Ok(AdvanceResult::Stale);
            }
        }
        StepOutcome::Skipped => {
            let res = sqlx::query(
                "UPDATE workflow_run_steps \
                 SET status = 'skipped', output = NULL, error = NULL, \
                     claim_token = NULL, claimed_at = NULL, updated_at = NOW() \
                 WHERE id = $1 AND claim_token = $2",
            )
            .bind(claimed.step_id.into_inner())
            .bind(claimed.claim_token)
            .execute(&mut *tx)
            .await?;
            if res.rows_affected() == 0 {
                tx.rollback().await?;
                return Ok(AdvanceResult::Stale);
            }
        }
    }

    // 2. Advance the run graph in the same tx.
    advance_run_tx(&mut tx, claimed.run_id, &claimed.definition).await?;
    tx.commit().await?;
    Ok(AdvanceResult::Advanced)
}

/// Advance one run's graph inside an open transaction: lock the run row (so
/// concurrent advances serialize — each does a full recompute, so this is
/// idempotent), promote pending steps whose deps are now satisfied, and write
/// the run's terminal status. A no-op if the run is already terminal. Does NOT
/// commit — the caller owns the transaction.
///
/// Shared by `complete_step_and_advance` (a function/skip step finished) and
/// `resume_finished_children` (a parked sub-workflow step finished).
pub async fn advance_run_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    run_id: WorkflowRunId,
    definition: &WorkflowDefinition,
) -> Result<(), WorkflowRepoError> {
    let run: Option<(String,)> =
        sqlx::query_as("SELECT status FROM workflow_runs WHERE id = $1 FOR UPDATE")
            .bind(run_id.into_inner())
            .fetch_optional(&mut **tx)
            .await?;
    let Some((run_status,)) = run else {
        return Ok(());
    };
    if RunStatus::from_str(&run_status).is_some_and(RunStatus::is_terminal) {
        // Another worker already finished the run; nothing to do.
        return Ok(());
    }

    let step_rows: Vec<StepStateRow> = sqlx::query_as(
        "SELECT step_name, status, output FROM workflow_run_steps WHERE run_id = $1",
    )
    .bind(run_id.into_inner())
    .fetch_all(&mut **tx)
    .await?;
    let mut statuses = BTreeMap::new();
    let mut outputs = BTreeMap::new();
    for r in step_rows {
        if let Some(st) = StepStatus::from_str(&r.status) {
            statuses.insert(r.step_name.clone(), st);
        }
        if let Some(o) = r.output {
            outputs.insert(r.step_name, o);
        }
    }

    let plan = compute_advance(definition, &statuses, &outputs);

    for name in &plan.promote {
        sqlx::query(
            "UPDATE workflow_run_steps \
             SET status = 'ready', next_attempt_at = NULL, updated_at = NOW() \
             WHERE run_id = $1 AND step_name = $2 AND status = 'pending'",
        )
        .bind(run_id.into_inner())
        .bind(name)
        .execute(&mut **tx)
        .await?;
    }

    match plan.run_status {
        RunStatus::Succeeded => {
            sqlx::query(
                "UPDATE workflow_runs \
                 SET status = 'succeeded', output = $2, error = $3, finished_at = NOW() \
                 WHERE id = $1",
            )
            .bind(run_id.into_inner())
            .bind(plan.run_output)
            .bind(plan.run_error)
            .execute(&mut **tx)
            .await?;
        }
        RunStatus::Failed => {
            sqlx::query(
                "UPDATE workflow_runs \
                 SET status = 'failed', error = $2, finished_at = NOW() WHERE id = $1",
            )
            .bind(run_id.into_inner())
            .bind(plan.run_error)
            .execute(&mut **tx)
            .await?;
        }
        _ => {
            sqlx::query(
                "UPDATE workflow_runs \
                 SET status = 'running', started_at = COALESCE(started_at, NOW()) WHERE id = $1",
            )
            .bind(run_id.into_inner())
            .execute(&mut **tx)
            .await?;
        }
    }

    Ok(())
}

/// Periodic safety net: a step leased by a crashed worker (still `running`
/// with a `claim_token` past the visibility timeout) is re-armed `ready` so
/// another worker retries it. Only touches steps of still-active runs.
///
/// The crashed attempt produced no outcome, so it must NOT consume the retry
/// budget: `claim_ready_step` incremented `attempt` up front, so reclaim
/// *decrements* it (floored at 0) — the re-claim will re-increment, leaving the
/// effective attempt count unchanged. `next_attempt_at` is cleared so the
/// reclaimed step is immediately eligible.
///
/// Returns the number of steps reclaimed.
pub async fn reclaim_stale_steps(
    pool: &PgPool,
    visibility_timeout_secs: u32,
) -> Result<u64, WorkflowRepoError> {
    let res = sqlx::query(
        "UPDATE workflow_run_steps s \
         SET status = 'ready', claim_token = NULL, claimed_at = NULL, \
             attempt = GREATEST(s.attempt - 1, 0), next_attempt_at = NULL, updated_at = NOW() \
         FROM workflow_runs r \
         WHERE s.run_id = r.id \
           AND s.claim_token IS NOT NULL \
           AND s.status = 'running' \
           AND s.claimed_at < NOW() - ($1 || ' seconds')::INTERVAL \
           AND r.status IN ('pending', 'running')",
    )
    .bind(i64::from(visibility_timeout_secs))
    .execute(pool)
    .await?;
    Ok(res.rows_affected())
}

// ---- run/step reads (admin surface + test assertions) --------------------

#[derive(sqlx::FromRow)]
struct RunRow {
    id: Uuid,
    workflow_id: Uuid,
    app_id: Uuid,
    status: String,
    input: serde_json::Value,
    output: Option<serde_json::Value>,
    error: Option<String>,
    root_execution_id: Uuid,
    workflow_depth: i32,
    parent_run_id: Option<Uuid>,
    parent_step_id: Option<Uuid>,
    started_at: Option<DateTime<Utc>>,
    finished_at: Option<DateTime<Utc>>,
    created_at: DateTime<Utc>,
}

impl TryFrom<RunRow> for WorkflowRun {
    type Error = WorkflowRepoError;
    fn try_from(r: RunRow) -> Result<Self, Self::Error> {
        Ok(Self {
            id: r.id.into(),
            workflow_id: r.workflow_id.into(),
            app_id: r.app_id.into(),
            status: RunStatus::from_str(&r.status).ok_or_else(|| {
                WorkflowRepoError::Serde(format!("bad run status {:?}", r.status))
            })?,
            input: r.input,
            output: r.output,
            error: r.error,
            root_execution_id: r.root_execution_id,
            workflow_depth: r.workflow_depth,
            parent_run_id: r.parent_run_id.map(Into::into),
            parent_step_id: r.parent_step_id.map(Into::into),
            started_at: r.started_at,
            finished_at: r.finished_at,
            created_at: r.created_at,
        })
    }
}

const RUN_COLS: &str = "id, workflow_id, app_id, status, input, output, error, \
    root_execution_id, workflow_depth, parent_run_id, parent_step_id, \
    started_at, finished_at, created_at";

/// Most-recent runs of a workflow (newest first), capped at `limit`. Backs the
/// admin/CLI run-history list.
pub async fn list_runs_for_workflow(
    pool: &PgPool,
    app_id: AppId,
    workflow_id: WorkflowId,
    limit: i64,
) -> Result<Vec<WorkflowRun>, WorkflowRepoError> {
    let rows: Vec<RunRow> = sqlx::query_as(&format!(
        "SELECT {RUN_COLS} FROM workflow_runs \
         WHERE app_id = $1 AND workflow_id = $2 ORDER BY created_at DESC LIMIT $3"
    ))
    .bind(app_id.into_inner())
    .bind(workflow_id.into_inner())
    .bind(limit)
    .fetch_all(pool)
    .await?;
    rows.into_iter().map(TryInto::try_into).collect()
}

/// Read a single run by id (app-scoped for isolation).
pub async fn get_run(
    pool: &PgPool,
    app_id: AppId,
    run_id: WorkflowRunId,
) -> Result<Option<WorkflowRun>, WorkflowRepoError> {
    let row: Option<RunRow> = sqlx::query_as(&format!(
        "SELECT {RUN_COLS} FROM workflow_runs WHERE id = $1 AND app_id = $2"
    ))
    .bind(run_id.into_inner())
    .bind(app_id.into_inner())
    .fetch_optional(pool)
    .await?;
    row.map(TryInto::try_into).transpose()
}

#[derive(sqlx::FromRow)]
struct StepRow {
    id: Uuid,
    run_id: Uuid,
    step_name: String,
    status: String,
    attempt: i32,
    max_attempts: i32,
    output: Option<serde_json::Value>,
    error: Option<String>,
    child_run_id: Option<Uuid>,
}

impl TryFrom<StepRow> for WorkflowRunStep {
    type Error = WorkflowRepoError;
    fn try_from(r: StepRow) -> Result<Self, Self::Error> {
        Ok(Self {
            id: r.id.into(),
            run_id: r.run_id.into(),
            step_name: r.step_name,
            status: StepStatus::from_str(&r.status).ok_or_else(|| {
                WorkflowRepoError::Serde(format!("bad step status {:?}", r.status))
            })?,
            attempt: r.attempt,
            max_attempts: r.max_attempts,
            output: r.output,
            error: r.error,
            child_run_id: r.child_run_id.map(Into::into),
        })
    }
}

/// All steps of a run, ordered by name (for the admin/CLI surface + tests).
/// App-scoped at the query (JOIN `workflow_runs`) so isolation is enforced here
/// rather than relying on the caller having pre-verified run ownership — a
/// run under another app returns no rows.
pub async fn list_run_steps(
    pool: &PgPool,
    app_id: AppId,
    run_id: WorkflowRunId,
) -> Result<Vec<WorkflowRunStep>, WorkflowRepoError> {
    let rows: Vec<StepRow> = sqlx::query_as(
        "SELECT s.id, s.run_id, s.step_name, s.status, s.attempt, s.max_attempts, s.output, \
                s.error, s.child_run_id \
         FROM workflow_run_steps s \
         JOIN workflow_runs r ON r.id = s.run_id \
         WHERE s.run_id = $1 AND r.app_id = $2 ORDER BY s.step_name",
    )
    .bind(run_id.into_inner())
    .bind(app_id.into_inner())
    .fetch_all(pool)
    .await?;
    rows.into_iter().map(TryInto::try_into).collect()
}

/// The `{ step_name: output }` map of a run's **succeeded** steps — the
/// accumulated context M3 resolves a step's `when` + `input` templates against
/// (see `workflow_template::RunContext`). Read fresh at execution time so a
/// step sees every prior step's output.
pub async fn load_run_step_outputs(
    pool: &PgPool,
    run_id: WorkflowRunId,
) -> Result<BTreeMap<String, serde_json::Value>, WorkflowRepoError> {
    let rows: Vec<(String, Option<serde_json::Value>)> = sqlx::query_as(
        "SELECT step_name, output FROM workflow_run_steps \
         WHERE run_id = $1 AND status = 'succeeded'",
    )
    .bind(run_id.into_inner())
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .filter_map(|(name, out)| out.map(|v| (name, v)))
        .collect())
}

// ---- nested sub-workflows (M4) --------------------------------------------

/// Resolve an app-owned workflow by case-insensitive name (a `workflow`-kind
/// step's target). Pool free-fn mirror of `WorkflowRepo::get_by_name`.
pub async fn get_workflow_by_name(
    pool: &PgPool,
    app_id: AppId,
    name: &str,
) -> Result<Option<Workflow>, WorkflowRepoError> {
    let row: Option<WorkflowRow> = sqlx::query_as(&format!(
        "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 AND LOWER(name) = LOWER($2)"
    ))
    .bind(app_id.into_inner())
    .bind(name)
    .fetch_optional(pool)
    .await?;
    row.map(TryInto::try_into).transpose()
}

/// Resolve an app-owned workflow by id (app-scoped). Backs the run-detail
/// API's DAG-edge lookup (`depends_on` lives on the definition, not the run).
pub async fn get_workflow_by_id(
    pool: &PgPool,
    app_id: AppId,
    id: WorkflowId,
) -> Result<Option<Workflow>, WorkflowRepoError> {
    let row: Option<WorkflowRow> = sqlx::query_as(&format!(
        "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 AND id = $2"
    ))
    .bind(app_id.into_inner())
    .bind(id.into_inner())
    .fetch_optional(pool)
    .await?;
    row.map(TryInto::try_into).transpose()
}

/// Start a nested sub-workflow run and **park** the parent step on it. In one
/// token-gated transaction: the parent step (still claimed by the caller) is
/// set `running` with `claim_token` cleared and `child_run_id` pointing at a
/// freshly-seeded child run (depth + 1, parent linkage). The parked step is
/// never re-claimed (only `ready` steps are) nor reclaimed (only leased
/// `running` steps — `claim_token` is NULL here); `resume_finished_children`
/// resolves it once the child terminates. A stale worker matches zero rows and
/// nothing (including the child) is written.
pub async fn start_child_and_park(
    pool: &PgPool,
    claimed: &ClaimedStep,
    child_workflow_id: WorkflowId,
    child_definition: &WorkflowDefinition,
    child_input: serde_json::Value,
) -> Result<AdvanceResult, WorkflowRepoError> {
    let mut tx = pool.begin().await?;

    // Token-gated park FIRST — so a stale claim short-circuits before we ever
    // insert an orphan child run (child_run_id is linked below).
    let res = sqlx::query(
        "UPDATE workflow_run_steps \
         SET status = 'running', claim_token = NULL, claimed_at = NULL, updated_at = NOW() \
         WHERE id = $1 AND claim_token = $2",
    )
    .bind(claimed.step_id.into_inner())
    .bind(claimed.claim_token)
    .execute(&mut *tx)
    .await?;
    if res.rows_affected() == 0 {
        tx.rollback().await?;
        return Ok(AdvanceResult::Stale);
    }

    let new = NewWorkflowRun {
        workflow_id: child_workflow_id,
        app_id: claimed.app_id,
        input: child_input,
        // Correlate the whole nested tree under the top-level run's root id.
        root_execution_id: claimed.root_execution_id,
        workflow_depth: claimed.workflow_depth + 1,
        parent_run_id: Some(claimed.run_id),
        parent_step_id: Some(claimed.step_id),
    };
    let child_run = seed_run_tx(&mut tx, &new, child_definition).await?;

    sqlx::query(
        "UPDATE workflow_run_steps SET child_run_id = $2, updated_at = NOW() WHERE id = $1",
    )
    .bind(claimed.step_id.into_inner())
    .bind(child_run.into_inner())
    .execute(&mut *tx)
    .await?;

    sqlx::query(
        "UPDATE workflow_runs SET status = 'running', started_at = COALESCE(started_at, NOW()) \
         WHERE id = $1 AND status = 'pending'",
    )
    .bind(claimed.run_id.into_inner())
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(AdvanceResult::Advanced)
}

#[derive(sqlx::FromRow)]
struct ParkedRow {
    parent_step_id: Uuid,
    parent_run_id: Uuid,
    child_status: String,
    child_output: Option<serde_json::Value>,
    child_error: Option<String>,
    parent_definition: serde_json::Value,
}

/// Resolve parent steps parked on a now-terminal child sub-workflow: write the
/// child's output (or error) onto the parent step and advance the parent run.
/// Idempotent — the conditional `status = 'running'` gate means a second pass
/// (or a second worker) matches zero rows. Returns the number resolved.
pub async fn resume_finished_children(pool: &PgPool) -> Result<u64, WorkflowRepoError> {
    let rows: Vec<ParkedRow> = sqlx::query_as(
        "SELECT ps.id AS parent_step_id, ps.run_id AS parent_run_id, \
                child.status AS child_status, child.output AS child_output, \
                child.error AS child_error, w.definition AS parent_definition \
         FROM workflow_run_steps ps \
         JOIN workflow_runs child ON child.id = ps.child_run_id \
         JOIN workflow_runs parent ON parent.id = ps.run_id \
         JOIN workflows w ON w.id = parent.workflow_id \
         WHERE ps.status = 'running' AND ps.child_run_id IS NOT NULL \
           AND ps.claim_token IS NULL \
           AND child.status IN ('succeeded', 'failed', 'canceled') \
         LIMIT 32",
    )
    .fetch_all(pool)
    .await?;

    let mut resolved = 0u64;
    for r in rows {
        let mut tx = pool.begin().await?;
        // A child terminates as succeeded / failed / canceled. There is no
        // distinct `canceled` step status (nor a run-cancel operation yet — the
        // `canceled` run status is reserved for a future admin cancel), so a
        // canceled child resolves the parent step to `failed` with a message
        // that names the cause rather than masquerading as a generic failure.
        let (new_status, out, err) = match r.child_status.as_str() {
            "succeeded" => ("succeeded", r.child_output.clone(), None),
            "canceled" => (
                "failed",
                None,
                Some("sub-workflow was canceled".to_string()),
            ),
            _ => (
                "failed",
                None,
                Some(
                    r.child_error
                        .clone()
                        .unwrap_or_else(|| "sub-workflow failed".to_string()),
                ),
            ),
        };
        let res = sqlx::query(
            "UPDATE workflow_run_steps \
             SET status = $2, output = $3, error = $4, updated_at = NOW() \
             WHERE id = $1 AND status = 'running' AND child_run_id IS NOT NULL",
        )
        .bind(r.parent_step_id)
        .bind(new_status)
        .bind(&out)
        .bind(&err)
        .execute(&mut *tx)
        .await?;
        if res.rows_affected() == 0 {
            tx.rollback().await?;
            continue; // already resolved by another pass
        }
        let parent_def: WorkflowDefinition = serde_json::from_value(r.parent_definition)
            .map_err(|e| WorkflowRepoError::Serde(e.to_string()))?;
        advance_run_tx(&mut tx, r.parent_run_id.into(), &parent_def).await?;
        tx.commit().await?;
        resolved += 1;
    }
    Ok(resolved)
}

#[cfg(test)]
mod tests {
    use super::*;
    use picloud_shared::workflow::WorkflowStepDef;
    use serde_json::json;

    fn step(name: &str, deps: &[&str]) -> WorkflowStepDef {
        WorkflowStepDef {
            name: name.into(),
            function: Some(format!("fn_{name}")),
            workflow: None,
            input: serde_json::Value::Null,
            depends_on: deps.iter().map(|s| (*s).to_string()).collect(),
            when: None,
            retry: None,
            on_error: OnError::Fail,
        }
    }

    fn def(steps: Vec<WorkflowStepDef>) -> WorkflowDefinition {
        WorkflowDefinition { steps }
    }

    fn statuses(pairs: &[(&str, StepStatus)]) -> BTreeMap<String, StepStatus> {
        pairs.iter().map(|(n, s)| ((*n).to_string(), *s)).collect()
    }

    #[test]
    fn promotes_dependent_when_dep_succeeds() {
        // a -> b. a succeeded, b still pending → b becomes ready.
        let d = def(vec![step("a", &[]), step("b", &["a"])]);
        let st = statuses(&[("a", StepStatus::Succeeded), ("b", StepStatus::Pending)]);
        let plan = compute_advance(&d, &st, &BTreeMap::new());
        assert_eq!(plan.promote, vec!["b".to_string()]);
        assert_eq!(plan.run_status, RunStatus::Running);
    }

    #[test]
    fn fan_in_waits_for_all_deps() {
        // c depends on [a, b]. Only a done → c must NOT be promoted.
        let d = def(vec![step("a", &[]), step("b", &[]), step("c", &["a", "b"])]);
        let st = statuses(&[
            ("a", StepStatus::Succeeded),
            ("b", StepStatus::Running),
            ("c", StepStatus::Pending),
        ]);
        let plan = compute_advance(&d, &st, &BTreeMap::new());
        assert!(plan.promote.is_empty());
        assert_eq!(plan.run_status, RunStatus::Running);

        // Now both a and b done → c promoted.
        let st = statuses(&[
            ("a", StepStatus::Succeeded),
            ("b", StepStatus::Succeeded),
            ("c", StepStatus::Pending),
        ]);
        let plan = compute_advance(&d, &st, &BTreeMap::new());
        assert_eq!(plan.promote, vec!["c".to_string()]);
    }

    #[test]
    fn all_terminal_succeeds_with_output_object() {
        let d = def(vec![step("a", &[]), step("b", &["a"])]);
        let st = statuses(&[("a", StepStatus::Succeeded), ("b", StepStatus::Succeeded)]);
        let mut out = BTreeMap::new();
        out.insert("a".to_string(), json!({ "x": 1 }));
        out.insert("b".to_string(), json!("done"));
        let plan = compute_advance(&d, &st, &out);
        assert!(plan.promote.is_empty());
        assert_eq!(plan.run_status, RunStatus::Succeeded);
        assert_eq!(
            plan.run_output,
            Some(json!({ "a": { "x": 1 }, "b": "done" }))
        );
    }

    #[test]
    fn on_error_fail_exhausted_fails_the_run() {
        let d = def(vec![step("a", &[]), step("b", &["a"])]);
        // a failed (on_error = fail) → run fails, b never promoted.
        let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Pending)]);
        let plan = compute_advance(&d, &st, &BTreeMap::new());
        assert!(plan.promote.is_empty());
        assert_eq!(plan.run_status, RunStatus::Failed);
        assert!(plan.run_error.is_some());
    }

    #[test]
    fn on_error_continue_lets_dependents_proceed_and_run_succeed() {
        let mut a = step("a", &[]);
        a.on_error = OnError::Continue;
        let d = def(vec![a, step("b", &["a"])]);
        // a failed but continue → b promoted.
        let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Pending)]);
        let plan = compute_advance(&d, &st, &BTreeMap::new());
        assert_eq!(plan.promote, vec!["b".to_string()]);
        assert_eq!(plan.run_status, RunStatus::Running);

        // a failed-continue + b succeeded → run succeeds (a contributes no output).
        let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Succeeded)]);
        let mut out = BTreeMap::new();
        out.insert("b".to_string(), json!(42));
        let plan = compute_advance(&d, &st, &out);
        assert_eq!(plan.run_status, RunStatus::Succeeded);
        assert_eq!(plan.run_output, Some(json!({ "b": 42 })));
    }

    #[test]
    fn skipped_dep_satisfies_dependents() {
        let d = def(vec![step("a", &[]), step("b", &["a"])]);
        let st = statuses(&[("a", StepStatus::Skipped), ("b", StepStatus::Pending)]);
        let plan = compute_advance(&d, &st, &BTreeMap::new());
        assert_eq!(plan.promote, vec!["b".to_string()]);
    }

    #[test]
    fn parallel_fan_out_promotes_all_ready_children() {
        // root -> b, root -> c (diamond top). After root succeeds, both promote.
        let d = def(vec![
            step("root", &[]),
            step("b", &["root"]),
            step("c", &["root"]),
        ]);
        let st = statuses(&[
            ("root", StepStatus::Succeeded),
            ("b", StepStatus::Pending),
            ("c", StepStatus::Pending),
        ]);
        let plan = compute_advance(&d, &st, &BTreeMap::new());
        assert_eq!(plan.promote, vec!["b".to_string(), "c".to_string()]);
        assert_eq!(plan.run_status, RunStatus::Running);
    }
}