use runledger_core::jobs::WorkflowType;
use sqlx::types::Uuid;

use crate::{DbPool, DbTx, Result};

use super::super::errors::validate_pagination;
use super::super::row_decode::{
    parse_job_stage, parse_job_type_name, parse_step_key_name, parse_workflow_release_mode,
    parse_workflow_run_status, parse_workflow_step_execution_kind, parse_workflow_step_status,
    parse_workflow_type_name,
};
use super::super::workflow_types::{
    WorkflowRunCountFilter, WorkflowRunDbRecord, WorkflowRunListFilter, WorkflowStepDbRecord,
    WorkflowStepDependencyDbRecord,
};

struct WorkflowRunLookupRow {
    id: Uuid,
    workflow_type: String,
    organization_id: Option<Uuid>,
    status: String,
    idempotency_key: Option<String>,
    result_step_key: Option<String>,
    metadata: serde_json::Value,
    started_at: chrono::DateTime<chrono::Utc>,
    finished_at: Option<chrono::DateTime<chrono::Utc>>,
    created_at: chrono::DateTime<chrono::Utc>,
    updated_at: chrono::DateTime<chrono::Utc>,
}

#[derive(sqlx::FromRow)]
struct WorkflowStepLookupRow {
    id: Uuid,
    workflow_run_id: Uuid,
    step_key: String,
    execution_kind: String,
    job_type: Option<String>,
    organization_id: Option<Uuid>,
    payload: serde_json::Value,
    priority: Option<i32>,
    max_attempts: Option<i32>,
    timeout_seconds: Option<i32>,
    stage: Option<String>,
    status: String,
    job_id: Option<Uuid>,
    released_at: Option<chrono::DateTime<chrono::Utc>>,
    started_at: Option<chrono::DateTime<chrono::Utc>>,
    finished_at: Option<chrono::DateTime<chrono::Utc>>,
    dependency_count_total: i32,
    dependency_count_pending: i32,
    dependency_count_unsatisfied: i32,
    status_reason: Option<String>,
    last_error_code: Option<String>,
    last_error_message: Option<String>,
    output: Option<serde_json::Value>,
    created_at: chrono::DateTime<chrono::Utc>,
    updated_at: chrono::DateTime<chrono::Utc>,
}

#[derive(sqlx::FromRow)]
struct WorkflowStepDependencyLookupRow {
    workflow_run_id: Uuid,
    prerequisite_step_id: Uuid,
    dependent_step_id: Uuid,
    release_mode: String,
    created_at: chrono::DateTime<chrono::Utc>,
}

fn workflow_run_db_record_from_lookup_row(
    row: WorkflowRunLookupRow,
) -> Result<WorkflowRunDbRecord> {
    Ok(WorkflowRunDbRecord {
        id: row.id,
        workflow_type: parse_workflow_type_name(row.workflow_type)?,
        organization_id: row.organization_id,
        status: parse_workflow_run_status(row.status)?,
        idempotency_key: row.idempotency_key,
        result_step_key: row.result_step_key.map(parse_step_key_name).transpose()?,
        metadata: row.metadata,
        started_at: row.started_at,
        finished_at: row.finished_at,
        created_at: row.created_at,
        updated_at: row.updated_at,
    })
}

fn workflow_step_db_record_from_lookup_row(
    row: WorkflowStepLookupRow,
) -> Result<WorkflowStepDbRecord> {
    Ok(WorkflowStepDbRecord {
        id: row.id,
        workflow_run_id: row.workflow_run_id,
        step_key: parse_step_key_name(row.step_key)?,
        execution_kind: parse_workflow_step_execution_kind(row.execution_kind)?,
        job_type: row.job_type.map(parse_job_type_name).transpose()?,
        organization_id: row.organization_id,
        payload: row.payload,
        priority: row.priority,
        max_attempts: row.max_attempts,
        timeout_seconds: row.timeout_seconds,
        stage: row.stage.map(parse_job_stage).transpose()?,
        status: parse_workflow_step_status(row.status)?,
        job_id: row.job_id,
        released_at: row.released_at,
        started_at: row.started_at,
        finished_at: row.finished_at,
        dependency_count_total: row.dependency_count_total,
        dependency_count_pending: row.dependency_count_pending,
        dependency_count_unsatisfied: row.dependency_count_unsatisfied,
        status_reason: row.status_reason,
        last_error_code: row.last_error_code,
        last_error_message: row.last_error_message,
        output: row.output,
        created_at: row.created_at,
        updated_at: row.updated_at,
    })
}

fn workflow_step_dependency_db_record_from_lookup_row(
    row: WorkflowStepDependencyLookupRow,
) -> Result<WorkflowStepDependencyDbRecord> {
    Ok(WorkflowStepDependencyDbRecord {
        workflow_run_id: row.workflow_run_id,
        prerequisite_step_id: row.prerequisite_step_id,
        dependent_step_id: row.dependent_step_id,
        release_mode: parse_workflow_release_mode(row.release_mode)?,
        created_at: row.created_at,
    })
}

pub async fn get_workflow_run_by_id(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_run_id: Uuid,
) -> Result<Option<WorkflowRunDbRecord>> {
    let row = sqlx::query_as!(
        WorkflowRunLookupRow,
        "SELECT
            id,
            workflow_type,
            organization_id,
            status::text AS \"status!\",
            idempotency_key,
            result_step_key,
            metadata,
            started_at,
            finished_at,
            created_at,
            updated_at
         FROM workflow_runs
         WHERE id = $1
           AND ($2::uuid IS NULL OR organization_id = $2)
         LIMIT 1",
        workflow_run_id,
        organization_id,
    )
    .fetch_optional(pool)
    .await
    .map_err(|error| crate::Error::from_query_sqlx_with_context("get workflow run by id", error))?;

    row.map(workflow_run_db_record_from_lookup_row).transpose()
}

pub(in crate::jobs::workflows) async fn load_workflow_run_by_id_tx(
    tx: &mut DbTx<'_>,
    workflow_run_id: Uuid,
    context: &'static str,
) -> Result<WorkflowRunDbRecord> {
    let run_row = sqlx::query_as!(
        WorkflowRunLookupRow,
        "SELECT
            id,
            workflow_type,
            organization_id,
            status::text AS \"status!\",
            idempotency_key,
            result_step_key,
            metadata,
            started_at,
            finished_at,
            created_at,
            updated_at
         FROM workflow_runs
         WHERE id = $1",
        workflow_run_id,
    )
    .fetch_one(&mut **tx)
    .await
    .map_err(|error| crate::Error::from_query_sqlx_with_context(context, error))?;

    workflow_run_db_record_from_lookup_row(run_row)
}

pub async fn list_workflow_steps(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_run_id: Uuid,
) -> Result<Vec<WorkflowStepDbRecord>> {
    let rows = sqlx::query_as::<_, WorkflowStepLookupRow>(
        "SELECT
            ws.id,
            ws.workflow_run_id,
            ws.step_key,
            ws.execution_kind::text AS execution_kind,
            ws.job_type,
            ws.organization_id,
            ws.payload,
            ws.priority,
            ws.max_attempts,
            ws.timeout_seconds,
            ws.stage,
            ws.status::text AS status,
            ws.job_id,
            ws.released_at,
            ws.started_at,
            ws.finished_at,
            ws.dependency_count_total,
            ws.dependency_count_pending,
            ws.dependency_count_unsatisfied,
            ws.status_reason,
            ws.last_error_code,
            ws.last_error_message,
            ws.output,
            ws.created_at,
            ws.updated_at
         FROM workflow_steps ws
         JOIN workflow_runs wr ON wr.id = ws.workflow_run_id
         WHERE ws.workflow_run_id = $1
           AND ($2::uuid IS NULL OR wr.organization_id = $2)
         ORDER BY ws.created_at ASC, ws.id ASC",
    )
    .bind(workflow_run_id)
    .bind(organization_id)
    .fetch_all(pool)
    .await
    .map_err(|error| crate::Error::from_query_sqlx_with_context("list workflow steps", error))?;

    rows.into_iter()
        .map(workflow_step_db_record_from_lookup_row)
        .collect()
}

pub async fn list_workflow_steps_page(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_run_id: Uuid,
    limit: i64,
    offset: i64,
) -> Result<Vec<WorkflowStepDbRecord>> {
    validate_pagination(limit, offset)?;

    let rows = sqlx::query_as::<_, WorkflowStepLookupRow>(
        "SELECT
            ws.id,
            ws.workflow_run_id,
            ws.step_key,
            ws.execution_kind::text AS execution_kind,
            ws.job_type,
            ws.organization_id,
            ws.payload,
            ws.priority,
            ws.max_attempts,
            ws.timeout_seconds,
            ws.stage,
            ws.status::text AS status,
            ws.job_id,
            ws.released_at,
            ws.started_at,
            ws.finished_at,
            ws.dependency_count_total,
            ws.dependency_count_pending,
            ws.dependency_count_unsatisfied,
            ws.status_reason,
            ws.last_error_code,
            ws.last_error_message,
            ws.output,
            ws.created_at,
            ws.updated_at
         FROM workflow_steps ws
         JOIN workflow_runs wr ON wr.id = ws.workflow_run_id
         WHERE ws.workflow_run_id = $1
           AND ($2::uuid IS NULL OR wr.organization_id = $2)
         ORDER BY ws.created_at ASC, ws.id ASC
         LIMIT $3 OFFSET $4",
    )
    .bind(workflow_run_id)
    .bind(organization_id)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await
    .map_err(|error| {
        crate::Error::from_query_sqlx_with_context("list workflow steps page", error)
    })?;

    rows.into_iter()
        .map(workflow_step_db_record_from_lookup_row)
        .collect()
}

pub async fn count_workflow_steps(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_run_id: Uuid,
) -> Result<i64> {
    sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*)::bigint
         FROM workflow_steps ws
         JOIN workflow_runs wr ON wr.id = ws.workflow_run_id
         WHERE ws.workflow_run_id = $1
           AND ($2::uuid IS NULL OR wr.organization_id = $2)",
    )
    .bind(workflow_run_id)
    .bind(organization_id)
    .fetch_one(pool)
    .await
    .map_err(|error| crate::Error::from_query_sqlx_with_context("count workflow steps", error))
}

pub async fn list_workflow_runs(
    pool: &DbPool,
    filter: &WorkflowRunListFilter<'_>,
) -> Result<Vec<WorkflowRunDbRecord>> {
    validate_pagination(filter.limit, filter.offset)?;

    let status_text = filter.status.map(|status| status.as_db_value());

    let rows = sqlx::query_as!(
        WorkflowRunLookupRow,
        "SELECT
            id,
            workflow_type,
            organization_id,
            status::text AS \"status!\",
            idempotency_key,
            result_step_key,
            metadata,
            started_at,
            finished_at,
            created_at,
            updated_at
         FROM workflow_runs
         WHERE ($1::uuid IS NULL OR organization_id = $1)
           AND ($2::text IS NULL OR status = $2::text::workflow_run_status)
           AND ($3::text IS NULL OR workflow_type ILIKE '%' || $3 || '%')
         ORDER BY created_at DESC
         LIMIT $4 OFFSET $5",
        filter.organization_id,
        status_text,
        filter.workflow_type,
        filter.limit,
        filter.offset,
    )
    .fetch_all(pool)
    .await
    .map_err(|error| crate::Error::from_query_sqlx_with_context("list workflow runs", error))?;

    rows.into_iter()
        .map(workflow_run_db_record_from_lookup_row)
        .collect()
}

pub async fn count_workflow_runs(
    pool: &DbPool,
    filter: &WorkflowRunCountFilter<'_>,
) -> Result<i64> {
    let status_text = filter.status.map(|status| status.as_db_value());
    sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*)::bigint
         FROM workflow_runs
         WHERE ($1::uuid IS NULL OR organization_id = $1)
           AND ($2::text IS NULL OR status = $2::text::workflow_run_status)
           AND ($3::text IS NULL OR workflow_type ILIKE '%' || $3 || '%')",
    )
    .bind(filter.organization_id)
    .bind(status_text)
    .bind(filter.workflow_type)
    .fetch_one(pool)
    .await
    .map_err(|error| crate::Error::from_query_sqlx_with_context("count workflow runs", error))
}

pub async fn get_latest_workflow_run_by_type(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_type: WorkflowType<'_>,
) -> Result<Option<WorkflowRunDbRecord>> {
    let row = sqlx::query_as!(
        WorkflowRunLookupRow,
        "SELECT
            id,
            workflow_type,
            organization_id,
            status::text AS \"status!\",
            idempotency_key,
            result_step_key,
            metadata,
            started_at,
            finished_at,
            created_at,
            updated_at
         FROM workflow_runs
         WHERE ($1::uuid IS NULL OR organization_id = $1)
           AND workflow_type = $2
         ORDER BY created_at DESC
         LIMIT 1",
        organization_id,
        workflow_type as _,
    )
    .fetch_optional(pool)
    .await
    .map_err(|error| {
        crate::Error::from_query_sqlx_with_context("get latest workflow run by type", error)
    })?;

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

    Ok(Some(workflow_run_db_record_from_lookup_row(row)?))
}

pub async fn list_workflow_step_dependencies(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_run_id: Uuid,
) -> Result<Vec<WorkflowStepDependencyDbRecord>> {
    let rows = sqlx::query_as::<_, WorkflowStepDependencyLookupRow>(
        "SELECT
            wsd.workflow_run_id,
            wsd.prerequisite_step_id,
            wsd.dependent_step_id,
            wsd.release_mode::text AS release_mode,
            wsd.created_at
         FROM workflow_step_dependencies wsd
         JOIN workflow_runs wr ON wr.id = wsd.workflow_run_id
         WHERE wsd.workflow_run_id = $1
           AND ($2::uuid IS NULL OR wr.organization_id = $2)
         ORDER BY
           wsd.prerequisite_step_id ASC,
           wsd.dependent_step_id ASC",
    )
    .bind(workflow_run_id)
    .bind(organization_id)
    .fetch_all(pool)
    .await
    .map_err(|error| {
        crate::Error::from_query_sqlx_with_context("list workflow step dependencies", error)
    })?;

    rows.into_iter()
        .map(workflow_step_dependency_db_record_from_lookup_row)
        .collect()
}

pub async fn list_workflow_step_dependencies_page(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_run_id: Uuid,
    limit: i64,
    offset: i64,
) -> Result<Vec<WorkflowStepDependencyDbRecord>> {
    validate_pagination(limit, offset)?;

    let rows = sqlx::query_as::<_, WorkflowStepDependencyLookupRow>(
        "SELECT
            wsd.workflow_run_id,
            wsd.prerequisite_step_id,
            wsd.dependent_step_id,
            wsd.release_mode::text AS release_mode,
            wsd.created_at
         FROM workflow_step_dependencies wsd
         JOIN workflow_runs wr ON wr.id = wsd.workflow_run_id
         WHERE wsd.workflow_run_id = $1
           AND ($2::uuid IS NULL OR wr.organization_id = $2)
         ORDER BY
           wsd.prerequisite_step_id ASC,
           wsd.dependent_step_id ASC
         LIMIT $3 OFFSET $4",
    )
    .bind(workflow_run_id)
    .bind(organization_id)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await
    .map_err(|error| {
        crate::Error::from_query_sqlx_with_context("list workflow step dependencies page", error)
    })?;

    rows.into_iter()
        .map(workflow_step_dependency_db_record_from_lookup_row)
        .collect()
}

pub async fn count_workflow_step_dependencies(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_run_id: Uuid,
) -> Result<i64> {
    sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*)::bigint
         FROM workflow_step_dependencies wsd
         JOIN workflow_runs wr ON wr.id = wsd.workflow_run_id
         WHERE wsd.workflow_run_id = $1
           AND ($2::uuid IS NULL OR wr.organization_id = $2)",
    )
    .bind(workflow_run_id)
    .bind(organization_id)
    .fetch_one(pool)
    .await
    .map_err(|error| {
        crate::Error::from_query_sqlx_with_context("count workflow step dependencies", error)
    })
}

pub async fn get_workflow_run_id_for_job(pool: &DbPool, job_id: Uuid) -> Result<Option<Uuid>> {
    sqlx::query_scalar!(
        "SELECT ws.workflow_run_id FROM workflow_steps ws WHERE ws.job_id = $1",
        job_id,
    )
    .fetch_optional(pool)
    .await
    .map_err(|error| {
        crate::Error::from_query_sqlx_with_context("get workflow run id for job", error)
    })
}

pub async fn get_workflow_run_by_type_and_idempotency_key(
    pool: &DbPool,
    organization_id: Option<Uuid>,
    workflow_type: WorkflowType<'_>,
    idempotency_key: &str,
) -> Result<Option<WorkflowRunDbRecord>> {
    let mut tx = pool
        .begin()
        .await
        .map_err(|error| crate::Error::ConnectionError(error.to_string()))?;
    let run = get_workflow_run_by_type_and_idempotency_key_tx(
        &mut tx,
        organization_id,
        workflow_type,
        idempotency_key,
    )
    .await?;
    tx.commit()
        .await
        .map_err(|error| crate::Error::ConnectionError(error.to_string()))?;
    Ok(run)
}

pub async fn get_workflow_run_by_type_and_idempotency_key_tx(
    tx: &mut DbTx<'_>,
    organization_id: Option<Uuid>,
    workflow_type: WorkflowType<'_>,
    idempotency_key: &str,
) -> Result<Option<WorkflowRunDbRecord>> {
    let row = if let Some(organization_id) = organization_id {
        sqlx::query_as!(
            WorkflowRunLookupRow,
            "SELECT
                id,
                workflow_type,
                organization_id,
                status::text AS \"status!\",
                idempotency_key,
                result_step_key,
                metadata,
                started_at,
                finished_at,
                created_at,
                updated_at
             FROM workflow_runs
             WHERE workflow_type = $1
               AND idempotency_key = $2
               AND organization_id = $3
             LIMIT 1",
            workflow_type as _,
            idempotency_key,
            organization_id,
        )
        .fetch_optional(&mut **tx)
        .await
    } else {
        sqlx::query_as!(
            WorkflowRunLookupRow,
            "SELECT
                id,
                workflow_type,
                organization_id,
                status::text AS \"status!\",
                idempotency_key,
                result_step_key,
                metadata,
                started_at,
                finished_at,
                created_at,
                updated_at
             FROM workflow_runs
             WHERE workflow_type = $1
               AND idempotency_key = $2
               AND organization_id IS NULL
             LIMIT 1",
            workflow_type as _,
            idempotency_key,
        )
        .fetch_optional(&mut **tx)
        .await
    }
    .map_err(|error| {
        crate::Error::from_query_sqlx_with_context(
            "get workflow run by type and idempotency key",
            error,
        )
    })?;

    row.map(workflow_run_db_record_from_lookup_row).transpose()
}
