/*
 * Author: Ruben Fiszel
 * Copyright: Windmill Labs, Inc 2022
 * This file and its contents are licensed under the AGPLv3 License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-AGPL for a copy of the license.
 */

use axum::http::HeaderValue;
use serde_json::value::RawValue;
use std::collections::HashMap;
use windmill_common::flow_status::RestartedFrom;

use crate::db::ApiAuthed;

use crate::{
    db::DB,
    users::{check_scopes, require_owner_of_path, OptAuthed},
    utils::require_super_admin,
    variables::get_workspace_key,
};
use anyhow::Context;
use axum::{
    extract::{FromRequest, Json, Path, Query},
    response::{IntoResponse, Response},
    routing::{get, post},
    Extension, Router,
};
use base64::Engine;
use chrono::Utc;
use hmac::Mac;
use hyper::{http, Request, StatusCode};
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use sql_builder::{prelude::*, quote, SqlBuilder};
use sqlx::types::JsonRawValue;
use sqlx::{query_scalar, types::Uuid, FromRow, Postgres, Transaction};
use tower_http::cors::{Any, CorsLayer};
use urlencoding::encode;
use windmill_audit::{audit_log, ActionKind};
use windmill_common::worker::{to_raw_value, CUSTOM_TAGS_PER_WORKSPACE, SERVER_CONFIG};
use windmill_common::{
    db::UserDB,
    error::{self, to_anyhow, Error},
    flow_status::{Approval, FlowStatus, FlowStatusModule},
    flows::FlowValue,
    jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode},
    oauth2::HmacSha256,
    scripts::{ScriptHash, ScriptLang},
    users::username_to_permissioned_as,
    utils::{not_found_if_none, now_from_db, paginate, require_admin, Pagination, StripPath},
};
use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL};
use windmill_queue::{
    add_completed_job_error, get_queued_job, get_result_by_id_from_running_flow, job_is_complete,
    push, CanceledBy, PushArgs, PushIsolationLevel,
};

pub fn workspaced_service() -> Router {
    let cors = CorsLayer::new()
        .allow_methods([http::Method::GET, http::Method::POST])
        .allow_headers([http::header::CONTENT_TYPE, http::header::AUTHORIZATION])
        .allow_origin(Any);

    Router::new()
        .route(
            "/run/f/*script_path",
            post(run_flow_by_path)
                .head(|| async { "" })
                .layer(cors.clone()),
        )
        .route(
            "/restart/f/:job_id/from/:step_id",
            post(restart_flow).head(|| async { "" }).layer(cors.clone()),
        )
        .route(
            "/restart/f/:job_id/from/:step_id/:branch_of_iteration_n",
            post(restart_flow).head(|| async { "" }).layer(cors.clone()),
        )
        .route(
            "/run/p/*script_path",
            post(run_job_by_path)
                .head(|| async { "" })
                .layer(cors.clone()),
        )
        .route(
            "/run_wait_result/p/*script_path",
            post(run_wait_result_script_by_path)
                .get(run_wait_result_job_by_path_get)
                .head(|| async { "" })
                .layer(cors.clone()),
        )
        .route(
            "/run_wait_result/h/:hash",
            post(run_wait_result_script_by_hash)
                .head(|| async { "" })
                .layer(cors.clone()),
        )
        .route(
            "/run_wait_result/f/*script_path",
            post(run_wait_result_flow_by_path)
                .get(run_wait_result_flow_by_path_get)
                .head(|| async { "" })
                .layer(cors.clone()),
        )
        .route(
            "/run/h/:hash",
            post(run_job_by_hash)
                .head(|| async { "" })
                .layer(cors.clone()),
        )
        .route("/run/preview", post(run_preview_job))
        .route("/add_batch_jobs/:n", post(add_batch_jobs))
        .route("/run/preview_flow", post(run_preview_flow_job))
        .route("/list", get(list_jobs))
        .route("/queue/list", get(list_queue_jobs))
        .route("/queue/count", get(count_queue_jobs))
        .route("/queue/cancel_all", post(cancel_all))
        .route("/completed/count", get(count_completed_jobs))
        .route(
            "/completed/list",
            get(list_completed_jobs).layer(cors.clone()),
        )
        .route(
            "/completed/get/:id",
            get(get_completed_job).layer(cors.clone()),
        )
        .route(
            "/completed/get_result/:id",
            get(get_completed_job_result).layer(cors.clone()),
        )
        .route(
            "/completed/get_result_maybe/:id",
            get(get_completed_job_result_maybe).layer(cors.clone()),
        )
        .route(
            "/completed/delete/:id",
            post(delete_completed_job).layer(cors.clone()),
        )
        .route(
            "/flow/resume/:id",
            post(resume_suspended_flow_as_owner).layer(cors.clone()),
        )
        .route(
            "/job_signature/:job_id/:resume_id",
            get(create_job_signature).layer(cors.clone()),
        )
        .route(
            "/resume_urls/:job_id/:resume_id",
            get(get_resume_urls).layer(cors.clone()),
        )
        .route(
            "/result_by_id/:job_id/:node_id",
            get(get_result_by_id).layer(cors.clone()),
        )
}

pub fn global_service() -> Router {
    Router::new()
        .route(
            "/resume/:job_id/:resume_id/:secret",
            get(resume_suspended_job),
        )
        .route(
            "/resume/:job_id/:resume_id/:secret",
            post(resume_suspended_job),
        )
        .route(
            "/cancel/:job_id/:resume_id/:secret",
            get(cancel_suspended_job),
        )
        .route(
            "/cancel/:job_id/:resume_id/:secret",
            post(cancel_suspended_job),
        )
        .route(
            "/get_flow/:job_id/:resume_id/:secret",
            get(get_suspended_job_flow),
        )
        .route("/get/:id", get(get_job))
        .route("/get_logs/:id", get(get_job_logs))
        .route("/completed/get/:id", get(get_completed_job))
        .route("/completed/get_result/:id", get(get_completed_job_result))
        .route(
            "/completed/get_result_maybe/:id",
            get(get_completed_job_result_maybe),
        )
        .route("/getupdate/:id", get(get_job_update))
        .route("/queue/cancel/:id", post(cancel_job_api))
        .route("/queue/force_cancel/:id", post(force_cancel))
}

pub fn global_root_service() -> Router {
    Router::new().route("/db_clock", get(get_db_clock))
}

#[derive(Deserialize)]
struct JsonPath {
    pub json_path: Option<String>,
}
async fn get_result_by_id(
    Extension(db): Extension<DB>,
    Path((w_id, flow_id, node_id)): Path<(String, Uuid, String)>,
    Query(JsonPath { json_path }): Query<JsonPath>,
) -> windmill_common::error::JsonResult<Box<JsonRawValue>> {
    let res = windmill_queue::get_result_by_id(db, w_id, flow_id, node_id, json_path).await?;
    Ok(Json(res))
}

async fn get_db_clock(Extension(db): Extension<DB>) -> windmill_common::error::JsonResult<i64> {
    Ok(Json(now_from_db(&db).await?.timestamp_millis()))
}

async fn cancel_job_api(
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    OptAuthed(opt_authed): OptAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
    Json(CancelJob { reason }): Json<CancelJob>,
) -> error::Result<String> {
    let tx = db.begin().await?;

    let username = match opt_authed {
        Some(authed) => authed.username,
        None => "anonymous".to_string(),
    };

    let (mut tx, job_option) =
        windmill_queue::cancel_job(&username, reason, id, &w_id, tx, &db, rsmq, false).await?;

    if let Some(id) = job_option {
        audit_log(
            &mut *tx,
            &username,
            "jobs.cancel",
            ActionKind::Delete,
            &w_id,
            Some(&id.to_string()),
            None,
        )
        .await?;
        tx.commit().await?;
        Ok(id.to_string())
    } else {
        tx.commit().await?;
        if job_is_complete(&db, id, &w_id).await.unwrap_or(false) {
            return Ok(format!("queued job id {} is already completed", id));
        } else {
            return Err(error::Error::NotFound(format!(
                "queued job id {} does not exist",
                id
            )));
        }
    }
}

async fn force_cancel(
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    OptAuthed(opt_authed): OptAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
    Json(CancelJob { reason }): Json<CancelJob>,
) -> error::Result<String> {
    let tx = db.begin().await?;

    let username = match opt_authed {
        Some(authed) => authed.username,
        None => "anonymous".to_string(),
    };

    let (mut tx, job_option) =
        windmill_queue::cancel_job(&username, reason, id, &w_id, tx, &db, rsmq, true).await?;

    if let Some(id) = job_option {
        audit_log(
            &mut *tx,
            &username,
            "jobs.force_cancel",
            ActionKind::Delete,
            &w_id,
            Some(&id.to_string()),
            None,
        )
        .await?;
        tx.commit().await?;
        Ok(id.to_string())
    } else {
        tx.commit().await?;
        if job_is_complete(&db, id, &w_id).await.unwrap_or(false) {
            return Ok(format!("queued job id {} is already completed", id));
        } else {
            return Err(error::Error::NotFound(format!(
                "queued job id {} does not exist",
                id
            )));
        }
    }
}

pub async fn get_path_for_hash<'c>(
    db: &mut Transaction<'c, Postgres>,
    w_id: &str,
    hash: i64,
) -> error::Result<String> {
    let path = sqlx::query_scalar!(
        "select path from script where hash = $1 AND workspace_id = $2",
        hash,
        w_id
    )
    .fetch_one(&mut **db)
    .await
    .map_err(|e| {
        Error::InternalErr(format!(
            "querying getting path for hash {hash} in {w_id}: {e}"
        ))
    })?;
    Ok(path)
}

pub async fn get_path_tag_limits_cache_for_hash(
    db: &DB,
    w_id: &str,
    hash: i64,
) -> error::Result<(
    String,
    Option<String>,
    Option<i32>,
    Option<i32>,
    Option<i32>,
    ScriptLang,
    Option<bool>,
    Option<i16>,
    Option<bool>,
    Option<i32>,
)> {
    let script = sqlx::query!(
        "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
        hash,
        w_id
    )
    .fetch_one(db)
    .await
    .map_err(|e| {
        Error::InternalErr(format!(
            "querying getting path for hash {hash} in {w_id}: {e}"
        ))
    })?;
    Ok((
        script.path,
        script.tag,
        script.concurrent_limit,
        script.concurrency_time_window_s,
        script.cache_ttl,
        script.language,
        script.dedicated_worker,
        script.priority,
        script.delete_after_use,
        script.timeout,
    ))
}

async fn get_job(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<Response> {
    let job = get_job_internal(&db, w_id.as_str(), id).await?;
    Ok(Json(job).into_response())
}

async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::Result<Job> {
    let cjob_maybe = sqlx::query_as::<_, CompletedJob>("SELECT 
        id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, 
        CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, right(logs, 20000000) as logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id,
        schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped,
        raw_lock, email, visible_to_owner, mem_peak, tag, priority
        FROM completed_job WHERE id = $1 AND workspace_id = $2")
            .bind(job_id)
            .bind(workspace_id)
            .fetch_optional(db)
            .await?
            .map(Job::CompletedJob);
    if let Some(cjob) = cjob_maybe {
        Ok(cjob)
    } else {
        let job_o = sqlx::query_as::<_, QueuedJob>(
            "SELECT  id, workspace_id, parent_job, created_by, created_at, started_at, scheduled_for, running,
                script_hash, script_path, CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, right(logs, 20000000) as logs, raw_code, canceled, canceled_by, canceled_reason, last_ping, 
                job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language,
                 suspend, suspend_until, same_worker, raw_lock, pre_run_error, email, visible_to_owner, mem_peak, 
                root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority
                FROM queue WHERE id = $1 AND workspace_id = $2",
        )
        .bind(job_id)
        .bind(workspace_id)
        .fetch_optional(db)
        .await?
        .map(Job::QueuedJob);
        let job: Job = not_found_if_none(job_o, "Job", job_id.to_string())?;
        Ok(job)
    }
}

async fn get_job_logs(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<String> {
    let text = sqlx::query_scalar!(
        "SELECT logs FROM completed_job WHERE id = $1 AND workspace_id = $2",
        id,
        w_id
    )
    .fetch_optional(&db)
    .await?
    .flatten();
    let text = not_found_if_none(text, "Job Logs", id.to_string())?;
    Ok(text)
}

#[derive(Debug, sqlx::FromRow, Serialize)]
pub struct ListableCompletedJob {
    pub r#type: String,
    pub workspace_id: String,
    pub id: Uuid,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent_job: Option<Uuid>,
    pub created_by: String,
    pub created_at: chrono::DateTime<chrono::Utc>,
    pub started_at: chrono::DateTime<chrono::Utc>,
    pub duration_ms: i64,
    pub success: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub script_hash: Option<ScriptHash>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub script_path: Option<String>,
    pub deleted: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub raw_code: Option<String>,
    pub canceled: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub canceled_by: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub canceled_reason: Option<String>,
    pub job_kind: JobKind,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub schedule_path: Option<String>,
    pub permissioned_as: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub flow_status: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub raw_flow: Option<serde_json::Value>,
    pub is_flow_step: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub language: Option<ScriptLang>,
    pub is_skipped: bool,
    pub email: String,
    pub visible_to_owner: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub mem_peak: Option<i32>,
    pub tag: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub priority: Option<i16>,
}

#[derive(Deserialize, Clone)]
pub struct RunJobQuery {
    scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
    scheduled_in_secs: Option<i64>,
    parent_job: Option<Uuid>,
    invisible_to_owner: Option<bool>,
    queue_limit: Option<i64>,
    payload: Option<String>,
    job_id: Option<Uuid>,
}

impl RunJobQuery {
    async fn get_scheduled_for<'c>(
        &self,
        db: &DB,
    ) -> error::Result<Option<chrono::DateTime<chrono::Utc>>> {
        if let Some(scheduled_for) = self.scheduled_for {
            Ok(Some(scheduled_for))
        } else if let Some(scheduled_in_secs) = self.scheduled_in_secs {
            let now = now_from_db(db).await?;
            Ok(Some(now + chrono::Duration::seconds(scheduled_in_secs)))
        } else {
            Ok(None)
        }
    }
}

#[derive(Deserialize)]
pub struct ListQueueQuery {
    pub script_path_start: Option<String>,
    pub script_path_exact: Option<String>,
    pub script_hash: Option<String>,
    pub created_by: Option<String>,
    pub started_before: Option<chrono::DateTime<chrono::Utc>>,
    pub started_after: Option<chrono::DateTime<chrono::Utc>>,
    pub created_before: Option<chrono::DateTime<chrono::Utc>>,
    pub created_after: Option<chrono::DateTime<chrono::Utc>>,
    pub created_or_started_before: Option<chrono::DateTime<chrono::Utc>>,
    pub created_or_started_after: Option<chrono::DateTime<chrono::Utc>>,
    pub running: Option<bool>,
    pub schedule_path: Option<String>,
    pub parent_job: Option<String>,
    pub order_desc: Option<bool>,
    pub job_kinds: Option<String>,
    pub suspended: Option<bool>,
    // filter by matching a subset of the args using base64 encoded json subset
    pub args: Option<String>,
    pub tag: Option<String>,
}

fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> SqlBuilder {
    let mut sqlb = SqlBuilder::select_from("queue")
        .fields(fields)
        .order_by("created_at", lq.order_desc.unwrap_or(true))
        .limit(1000)
        .and_where_eq("workspace_id", "?".bind(&w_id))
        .clone();

    if let Some(ps) = &lq.script_path_start {
        sqlb.and_where_like_left("script_path", "?".bind(ps));
    }
    if let Some(p) = &lq.script_path_exact {
        sqlb.and_where_eq("script_path", "?".bind(p));
    }
    if let Some(p) = &lq.schedule_path {
        sqlb.and_where_eq("schedule_path", "?".bind(p));
    }
    if let Some(h) = &lq.script_hash {
        sqlb.and_where_eq("script_hash", "?".bind(h));
    }
    if let Some(cb) = &lq.created_by {
        sqlb.and_where_eq("created_by", "?".bind(cb));
    }
    if let Some(t) = &lq.tag {
        sqlb.and_where_eq("tag", "?".bind(t));
    }
    if let Some(r) = &lq.running {
        sqlb.and_where_eq("running", &r);
    }
    if let Some(pj) = &lq.parent_job {
        sqlb.and_where_eq("parent_job", "?".bind(pj));
    }
    if let Some(dt) = &lq.started_before {
        sqlb.and_where_le("started_at", format!("to_timestamp({})", dt.timestamp()));
    }
    if let Some(dt) = &lq.started_after {
        sqlb.and_where_ge("started_at", format!("to_timestamp({})", dt.timestamp()));
    }

    if let Some(dt) = &lq.created_before {
        sqlb.and_where_le("created_at", format!("to_timestamp({})", dt.timestamp()));
    }
    if let Some(dt) = &lq.created_after {
        sqlb.and_where_ge("created_at", format!("to_timestamp({})", dt.timestamp()));
    }

    if let Some(dt) = &lq.created_or_started_after {
        let ts = dt.timestamp();
        sqlb.and_where(format!("(started_at IS NOT NULL AND started_at >= to_timestamp({})) OR (started_at IS NULL AND created_at >= to_timestamp({}))", ts, ts));
    }

    if let Some(dt) = &lq.created_or_started_before {
        let ts = dt.timestamp();
        sqlb.and_where(format!("(started_at IS NOT NULL AND started_at < to_timestamp({})) OR (started_at IS NULL AND created_at < to_timestamp({}))", ts, ts));
    }

    if let Some(s) = &lq.suspended {
        if *s {
            sqlb.and_where_gt("suspend", 0);
        } else {
            sqlb.and_where_eq("suspend", 0);
        }
    }

    if let Some(jk) = &lq.job_kinds {
        sqlb.and_where_in(
            "job_kind",
            &jk.split(',').into_iter().map(quote).collect::<Vec<_>>(),
        );
    }

    if let Some(args) = &lq.args {
        sqlb.and_where("args @> ?".bind(&args.replace("'", "''")));
    }

    sqlb
}

#[derive(Serialize, FromRow)]
struct ListableQueuedJob {
    pub id: Uuid,
    pub created_by: String,
    pub created_at: chrono::DateTime<chrono::Utc>,
    pub started_at: Option<chrono::DateTime<chrono::Utc>>,
    pub scheduled_for: chrono::DateTime<chrono::Utc>,
    pub script_hash: Option<ScriptHash>,
    pub script_path: Option<String>,
    pub args: Option<serde_json::Value>,
    pub job_kind: JobKind,
    pub schedule_path: Option<String>,
    pub is_flow_step: bool,
    pub language: Option<ScriptLang>,
    pub email: String,
    pub suspend: Option<i32>,
    pub tag: String,
    pub priority: Option<i16>,
}

async fn list_queue_jobs(
    Extension(db): Extension<DB>,
    Path(w_id): Path<String>,
    Query(lq): Query<ListQueueQuery>,
) -> error::JsonResult<Vec<ListableQueuedJob>> {
    let sql = list_queue_jobs_query(
        &w_id,
        &lq,
        &[
            "id",
            "created_by",
            "created_at",
            "started_at",
            "scheduled_for",
            "script_hash",
            "script_path",
            "null as args",
            "job_kind",
            "schedule_path",
            "permissioned_as",
            "is_flow_step",
            "language",
            "same_worker",
            "email",
            "suspend",
            "tag",
            "priority",
        ],
    )
    .sql()?;
    let jobs = sqlx::query_as::<_, ListableQueuedJob>(&sql)
        .fetch_all(&db)
        .await?;
    Ok(Json(jobs))
}

async fn cancel_all(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,

    Path(w_id): Path<String>,
) -> error::JsonResult<Vec<Uuid>> {
    require_admin(authed.is_admin, &authed.username)?;

    let mut jobs = sqlx::query!(
        "UPDATE queue SET canceled = true,  canceled_by = $2, scheduled_for = now(), suspend = 0 WHERE workspace_id = $1 AND schedule_path IS NULL RETURNING id, running",
        w_id,
        authed.username
    )
    .fetch_all(&db)
    .await?;

    let username = authed.username;
    for j in jobs.iter() {
        if !j.running {
            let e = serde_json::json!({"message": format!("Job canceled: cancel_all by {username}"), "name": "Canceled", "reason": "cancel_all", "canceler": username});
            let mut tx = db.begin().await?;
            let job_running = get_queued_job(j.id, &w_id, &mut tx).await?;
            tx.commit().await?;

            if let Some(job_running) = job_running {
                let add_job = add_completed_job_error(
                    &db,
                    &job_running,
                    format!("canceled by {username}: cancel_all"),
                    job_running.mem_peak.unwrap_or(0),
                    Some(CanceledBy {
                        username: Some(username.to_string()),
                        reason: Some("cancel_all".to_string()),
                    }),
                    e,
                    rsmq.clone(),
                    "server",
                )
                .await;
                if let Err(e) = add_job {
                    tracing::error!("Failed to add canceled job: {}", e);
                }
            }
        }
    }
    let uuids = jobs.iter_mut().map(|j| j.id).collect::<Vec<_>>();

    Ok(Json(uuids))
}

#[derive(Serialize, Debug, FromRow)]
struct QueueStats {
    database_length: i64,
}

async fn count_queue_jobs(
    Extension(db): Extension<DB>,
    Path(w_id): Path<String>,
) -> error::JsonResult<QueueStats> {
    Ok(Json(
        sqlx::query_as!(
            QueueStats,
            "SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM queue WHERE workspace_id = $1 AND scheduled_for <= now() AND running = false",
            w_id
        )
        .fetch_one(&db)
        .await?,
    ))
}

async fn count_completed_jobs(
    Extension(db): Extension<DB>,
    Path(w_id): Path<String>,
) -> error::JsonResult<QueueStats> {
    Ok(Json(
        sqlx::query_as!(
            QueueStats,
            "SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM completed_job WHERE workspace_id = $1",
            w_id
        )
        .fetch_one(&db)
        .await?,
    ))
}

async fn list_jobs(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Path(w_id): Path<String>,
    Query(pagination): Query<Pagination>,
    Query(lq): Query<ListCompletedQuery>,
) -> error::JsonResult<Vec<Job>> {
    check_scopes(&authed, || format!("listjobs"))?;

    let (per_page, offset) = paginate(pagination);
    let lqc = lq.clone();

    if lq.success.is_some() && lq.running.is_some_and(|x| x) {
        return Err(error::Error::BadRequest(
            "cannot specify both success and running".to_string(),
        ));
    }
    let sqlc = if lq.running.is_none() {
        Some(list_completed_jobs_query(
            &w_id,
            per_page + offset,
            0,
            &ListCompletedQuery { order_desc: Some(true), ..lqc },
            &[
                "'CompletedJob' as typ",
                "id",
                "workspace_id",
                "parent_job",
                "created_by",
                "created_at",
                "started_at",
                "null as scheduled_for",
                "null as running",
                "script_hash",
                "script_path",
                "null as args",
                "duration_ms",
                "success",
                "deleted",
                "canceled",
                "canceled_by",
                "job_kind",
                "schedule_path",
                "permissioned_as",
                "is_flow_step",
                "language",
                "is_skipped",
                "email",
                "visible_to_owner",
                "null as suspend",
                "mem_peak",
                "tag",
                "null as concurrent_limit",
                "null as concurrency_time_window_s",
                "priority",
            ],
        ))
    } else {
        None
    };

    let sql = if lq.success.is_none() {
        let sqlq = list_queue_jobs_query(
            &w_id,
            &ListQueueQuery {
                script_path_start: lq.script_path_start,
                script_path_exact: lq.script_path_exact,
                script_hash: lq.script_hash,
                created_by: lq.created_by,
                started_before: lq.started_before,
                started_after: lq.started_after,
                created_before: lq.created_before,
                created_after: lq.created_after,
                created_or_started_before: lq.created_or_started_before,
                created_or_started_after: lq.created_or_started_after,
                running: lq.running,
                parent_job: lq.parent_job,
                order_desc: Some(true),
                job_kinds: lq.job_kinds,
                suspended: lq.suspended,
                args: lq.args,
                tag: lq.tag,
                schedule_path: lq.schedule_path,
            },
            &[
                "'QueuedJob' as typ",
                "id",
                "workspace_id",
                "parent_job",
                "created_by",
                "created_at",
                "started_at",
                "scheduled_for",
                "running",
                "script_hash",
                "script_path",
                "null as args",
                "null as duration_ms",
                "null as success",
                "false as deleted",
                "canceled",
                "canceled_by",
                "job_kind",
                "schedule_path",
                "permissioned_as",
                "is_flow_step",
                "language",
                "false as is_skipped",
                "email",
                "visible_to_owner",
                "suspend",
                "mem_peak",
                "tag",
                "concurrent_limit",
                "concurrency_time_window_s",
                "priority",
            ],
        );

        if let Some(sqlc) = sqlc {
            format!(
                "{} UNION ALL {} LIMIT {} OFFSET {};",
                &sqlq.subquery()?,
                &sqlc.subquery()?,
                per_page,
                offset
            )
        } else {
            sqlq.query()?
        }
    } else {
        sqlc.unwrap().query()?
    };
    let mut tx = user_db.begin(&authed).await?;
    let jobs: Vec<UnifiedJob> = sqlx::query_as(&sql).fetch_all(&mut *tx).await?;
    tx.commit().await?;
    Ok(Json(jobs.into_iter().map(From::from).collect()))
}

pub async fn resume_suspended_flow_as_owner(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((_w_id, flow_id)): Path<(String, Uuid)>,
    QueryOrBody(value): QueryOrBody<serde_json::Value>,
) -> error::Result<StatusCode> {
    check_scopes(&authed, || format!("resumeflow"))?;
    let value = value.unwrap_or(serde_json::Value::Null);
    let mut tx = db.begin().await?;

    let (flow, job_id) = get_suspended_flow_info(flow_id, &mut tx).await?;

    require_owner_of_path(
        &authed,
        &flow.script_path.clone().unwrap_or_else(|| String::new()),
    )?;

    insert_resume_job(0, job_id, &flow, value, Some(authed.username), &mut tx).await?;

    resume_immediately_if_relevant(flow, job_id, &mut tx).await?;

    tx.commit().await?;
    Ok(StatusCode::CREATED)
}

pub async fn resume_suspended_job(
    authed: Option<ApiAuthed>,
    Extension(db): Extension<DB>,
    Path((w_id, job_id, resume_id, secret)): Path<(String, Uuid, u32, String)>,
    Query(approver): Query<QueryApprover>,
    QueryOrBody(value): QueryOrBody<serde_json::Value>,
) -> error::Result<StatusCode> {
    let value = value.unwrap_or(serde_json::Value::Null);
    let mut tx = db.begin().await?;
    let key = get_workspace_key(&w_id, &mut tx).await?;
    let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?;
    mac.update(job_id.as_bytes());
    mac.update(resume_id.to_be_bytes().as_ref());
    if let Some(approver) = approver.approver.clone() {
        mac.update(approver.as_bytes());
    }
    mac.verify_slice(hex::decode(secret)?.as_ref())
        .map_err(|_| anyhow::anyhow!("Invalid signature"))?;
    let parent_flow_info = get_suspended_parent_flow_info(job_id, &mut tx).await?;
    let parent_flow = get_job_internal(&db, w_id.as_str(), parent_flow_info.id).await?;
    let flow_status = parent_flow
        .flow_status()
        .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?;
    conditionally_require_authed_user(authed, flow_status)?;

    let exists = sqlx::query_scalar!(
        r#"
        SELECT EXISTS (SELECT 1 FROM resume_job WHERE id = $1)
        "#,
        Uuid::from_u128(job_id.as_u128() ^ resume_id as u128),
    )
    .fetch_one(&mut *tx)
    .await?
    .unwrap_or(false);

    if exists {
        return Err(anyhow::anyhow!("resume request already sent").into());
    }

    insert_resume_job(
        resume_id,
        job_id,
        &parent_flow_info,
        value,
        approver.approver,
        &mut tx,
    )
    .await?;

    resume_immediately_if_relevant(parent_flow_info, job_id, &mut tx).await?;

    tx.commit().await?;
    Ok(StatusCode::CREATED)
}

/* If the flow is currently waiting to be resumed (`FlowStatusModule::WaitingForEvents`)
 * the suspend column must be set to the number of resume messages waited on.
 *
 * The flow's queue row is locked in this transaction because to avoid race conditions around
 * the suspend column.
 * That is, a job needs one event but it hasn't arrived, a worker counts zero events before
 * entering WaitingForEvents.  Then this message arrives but the job isn't in WaitingForEvents
 * yet so the suspend counter isn't updated.  Then the job enters WaitingForEvents expecting
 * one event to arrive based on the count that is no longer correct. */
async fn resume_immediately_if_relevant<'c>(
    flow: FlowInfo,
    job_id: Uuid,
    tx: &mut Transaction<'c, Postgres>,
) -> error::Result<()> {
    Ok(
        if let Some(suspend) = (0 < flow.suspend).then(|| flow.suspend - 1) {
            let status =
                serde_json::from_value::<FlowStatus>(flow.flow_status.context("no flow status")?)
                    .context("deserialize flow status")?;
            if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id)
            {
                sqlx::query!(
                    "UPDATE queue SET suspend = $1 WHERE id = $2",
                    suspend,
                    flow.id,
                )
                .execute(&mut **tx)
                .await?;
            }
        },
    )
}

async fn insert_resume_job<'c>(
    resume_id: u32,
    job_id: Uuid,
    flow: &FlowInfo,
    value: serde_json::Value,
    approver: Option<String>,
    tx: &mut Transaction<'c, Postgres>,
) -> error::Result<()> {
    sqlx::query!(
        r#"
        INSERT INTO resume_job
                    (id, resume_id, job, flow, value, approver)
             VALUES ($1, $2, $3, $4, $5, $6)
        "#,
        Uuid::from_u128(job_id.as_u128() ^ resume_id as u128),
        resume_id as i32,
        job_id,
        flow.id,
        value,
        approver
    )
    .execute(&mut **tx)
    .await?;
    Ok(())
}

#[derive(sqlx::FromRow)]
struct FlowInfo {
    id: Uuid,
    flow_status: Option<serde_json::Value>,
    suspend: i32,
    script_path: Option<String>,
}

async fn get_suspended_parent_flow_info<'c>(
    job_id: Uuid,
    tx: &mut Transaction<'c, Postgres>,
) -> error::Result<FlowInfo> {
    let flow = sqlx::query_as!(
        FlowInfo,
        r#"
        SELECT id, flow_status, suspend, script_path
        FROM queue
        WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)
        FOR UPDATE
        "#,
        job_id,
    )
    .fetch_optional(&mut **tx)
    .await?
    .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?;
    Ok(flow)
}

async fn get_suspended_flow_info<'c>(
    job_id: Uuid,
    tx: &mut Transaction<'c, Postgres>,
) -> error::Result<(FlowInfo, Uuid)> {
    let flow = sqlx::query_as!(
        FlowInfo,
        r#"
        SELECT id, flow_status, suspend, script_path
        FROM queue
        WHERE id = $1
        "#,
        job_id,
    )
    .fetch_optional(&mut **tx)
    .await?
    .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?;
    let job_id = flow
        .flow_status
        .as_ref()
        .and_then(|v| serde_json::from_value::<FlowStatus>(v.clone()).ok())
        .and_then(|s| match s.modules.get(s.step as usize) {
            Some(FlowStatusModule::WaitingForEvents { job, .. }) => Some(job.to_owned()),
            _ => None,
        });

    if let Some(job_id) = job_id {
        Ok((flow, job_id))
    } else {
        Err(anyhow::anyhow!("the flow is not in a suspended state anymore").into())
    }
}

pub async fn cancel_suspended_job(
    authed: Option<ApiAuthed>,
    Extension(db): Extension<DB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>,
    Query(approver): Query<QueryApprover>,
) -> error::Result<String> {
    let mut tx = db.begin().await?;
    let key = get_workspace_key(&w_id, &mut tx).await?;
    let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?;
    mac.update(job.as_bytes());
    mac.update(resume_id.to_be_bytes().as_ref());
    if let Some(approver) = approver.approver.clone() {
        mac.update(approver.as_bytes());
    }
    mac.verify_slice(hex::decode(secret)?.as_ref())
        .map_err(|_| anyhow::anyhow!("Invalid signature"))?;

    let whom = approver.approver.unwrap_or_else(|| "unknown".to_string());
    let parent_flow_id = get_suspended_parent_flow_info(job, &mut tx).await?.id;

    let parent_flow = get_job_internal(&db, w_id.as_str(), parent_flow_id).await?;
    let flow_status = parent_flow
        .flow_status()
        .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?;
    conditionally_require_authed_user(authed, flow_status)?;

    let (mut tx, cjob) = windmill_queue::cancel_job(
        &whom,
        Some("approval request disapproved".to_string()),
        parent_flow_id,
        &w_id,
        tx,
        &db,
        rsmq,
        false,
    )
    .await?;
    if cjob.is_some() {
        audit_log(
            &mut *tx,
            &whom,
            "jobs.disapproval",
            ActionKind::Delete,
            &w_id,
            Some(&parent_flow_id.to_string()),
            None,
        )
        .await?;
        tx.commit().await?;

        Ok(format!("Flow {parent_flow_id} of job {job} cancelled"))
    } else {
        Ok(format!(
            "Flow {parent_flow_id} of job {job} was not cancellable"
        ))
    }
}

#[derive(Serialize)]
pub struct SuspendedJobFlow {
    pub job: Job,
    pub approvers: Vec<Approval>,
}

#[derive(Deserialize)]
pub struct QueryApprover {
    pub approver: Option<String>,
}

pub async fn get_suspended_job_flow(
    authed: Option<ApiAuthed>,
    Extension(db): Extension<DB>,
    Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>,
    Query(approver): Query<QueryApprover>,
) -> error::Result<Response> {
    let mut tx = db.begin().await?;
    let key = get_workspace_key(&w_id, &mut tx).await?;
    let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?;
    mac.update(job.as_bytes());
    mac.update(resume_id.to_be_bytes().as_ref());
    if let Some(approver) = approver.approver {
        mac.update(approver.as_bytes());
    }
    mac.verify_slice(hex::decode(secret)?.as_ref())
        .map_err(|_| anyhow::anyhow!("Invalid signature"))?;

    let flow_id = sqlx::query_scalar!(
        r#"
        SELECT parent_job
        FROM queue
        WHERE id = $1 AND workspace_id = $2
        UNION ALL
        SELECT parent_job
        FROM completed_job
        WHERE id = $1 AND workspace_id = $2
        "#,
        job,
        w_id
    )
    .fetch_optional(&mut *tx)
    .await?
    .flatten()
    .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?;

    let flow = get_job_internal(&db, w_id.as_str(), flow_id).await?;

    let flow_status = flow
        .flow_status()
        .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?;
    let flow_module_status = flow_status
        .modules
        .iter()
        .find(|p| p.job() == Some(job))
        .ok_or_else(|| anyhow::anyhow!("unable to find the module"))?;
    conditionally_require_authed_user(authed, flow_status.clone())?;

    let approvers_from_status = match flow_module_status {
        FlowStatusModule::Success { approvers, .. } => approvers.to_owned(),
        _ => vec![],
    };
    let approvers = if approvers_from_status.is_empty() {
        sqlx::query!(
            r#"
            SELECT resume_id, approver
            FROM resume_job
            WHERE job = $1
            "#,
            job,
        )
        .fetch_all(&mut *tx)
        .await?
        .into_iter()
        .map(|x| Approval {
            resume_id: x.resume_id as u16,
            approver: x.approver.unwrap_or_else(|| "anonymous".to_string()),
        })
        .collect()
    } else {
        approvers_from_status
    };

    Ok(Json(SuspendedJobFlow { job: flow, approvers }).into_response())
}

fn conditionally_require_authed_user(
    authed: Option<ApiAuthed>,
    flow_status: FlowStatus,
) -> error::Result<()> {
    let approval_conditions_opt = flow_status.approval_conditions;

    if approval_conditions_opt.is_none() {
        return Ok(());
    }
    let approval_conditions = approval_conditions_opt.unwrap();

    if approval_conditions.user_auth_required {
        #[cfg(not(feature = "enterprise"))]
        return Err(Error::BadRequest(
            "Approvals for logged in users is an enterprise only feature".to_string(),
        ));
        #[cfg(feature = "enterprise")]
        if authed.is_none() {
            return Err(Error::NotAuthorized(
                "Only logged in users can approve this flow step".to_string(),
            ));
        }
    }
    if authed.is_some() && !authed.as_ref().unwrap().username.eq("admin") {
        if !approval_conditions.user_groups_required.is_empty() {
            #[cfg(not(feature = "enterprise"))]
            return Err(Error::BadRequest(
                "Approvals for users in certain user groups is an enterprise only feature"
                    .to_string(),
            ));
            #[cfg(feature = "enterprise")]
            {
                for required_group in approval_conditions.user_groups_required.iter() {
                    if authed.as_ref().unwrap().groups.contains(&required_group) {
                        return Ok(());
                    }
                }
                let error_msg = format!("Only users from one of the following groups are allowed to approve this workflow: {}", 
                    approval_conditions.user_groups_required.join(", "));
                return Err(Error::PermissionDenied(error_msg));
            }
        }
    }
    Ok(())
}

pub async fn create_job_signature(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>,
    Query(approver): Query<QueryApprover>,
) -> error::Result<String> {
    let key = get_workspace_key(&w_id, &mut user_db.begin(&authed).await?).await?;
    create_signature(key, job_id, resume_id, approver.approver)
}

fn create_signature(
    key: String,
    job_id: Uuid,
    resume_id: u32,
    approver: Option<String>,
) -> Result<String, Error> {
    let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?;
    mac.update(job_id.as_bytes());
    mac.update(resume_id.to_be_bytes().as_ref());
    if let Some(approver) = approver {
        mac.update(approver.as_bytes());
    }
    Ok(hex::encode(mac.finalize().into_bytes()))
}

#[allow(non_snake_case)]
#[derive(Serialize)]
pub struct ResumeUrls {
    approvalPage: String,
    cancel: String,
    resume: String,
}

fn build_resume_url(
    op: &str,
    w_id: &str,
    job_id: &Uuid,
    resume_id: &u32,
    signature: &str,
    approver: &str,
    base_url: &str,
) -> String {
    format!("{base_url}/api/w/{w_id}/jobs_u/{op}/{job_id}/{resume_id}/{signature}{approver}")
}

pub async fn get_resume_urls(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>,
    Query(approver): Query<QueryApprover>,
) -> error::JsonResult<ResumeUrls> {
    let key = get_workspace_key(&w_id, &mut user_db.begin(&authed).await?).await?;
    let signature = create_signature(key, job_id, resume_id, approver.approver.clone())?;
    let approver = approver
        .approver
        .as_ref()
        .map(|x| format!("?approver={}", encode(x)))
        .unwrap_or_else(String::new);

    let base_url_str = BASE_URL.read().await.clone();
    let base_url = base_url_str.as_str();
    let res = ResumeUrls {
        approvalPage: format!(
            "{base_url}/approve/{w_id}/{job_id}/{resume_id}/{signature}{approver}"
        ),
        cancel: build_resume_url(
            "cancel", &w_id, &job_id, &resume_id, &signature, &approver, &base_url,
        ),
        resume: build_resume_url(
            "resume", &w_id, &job_id, &resume_id, &signature, &approver, &base_url,
        ),
    };

    Ok(Json(res))
}

#[derive(Serialize, Debug)]
#[serde(tag = "type")]
pub enum Job {
    QueuedJob(QueuedJob),
    CompletedJob(CompletedJob),
}

impl Job {
    pub fn raw_flow(&self) -> Option<FlowValue> {
        match self {
            Job::QueuedJob(job) => job
                .raw_flow
                .as_ref()
                .map(|rf| serde_json::from_str(rf.0.get()).ok())
                .flatten(),
            Job::CompletedJob(job) => job
                .raw_flow
                .as_ref()
                .map(|rf| serde_json::from_str(rf.0.get()).ok())
                .flatten(),
        }
    }
    pub fn flow_status(&self) -> Option<FlowStatus> {
        match self {
            Job::QueuedJob(job) => job
                .flow_status
                .as_ref()
                .map(|rf| serde_json::from_str(rf.0.get()).ok())
                .flatten(),
            Job::CompletedJob(job) => job
                .flow_status
                .as_ref()
                .map(|rf| serde_json::from_str(rf.0.get()).ok())
                .flatten(),
        }
    }
}

#[derive(sqlx::FromRow)]
struct UnifiedJob {
    workspace_id: String,
    typ: String,
    id: Uuid,
    parent_job: Option<Uuid>,
    created_by: String,
    created_at: chrono::DateTime<chrono::Utc>,
    started_at: Option<chrono::DateTime<chrono::Utc>>,
    scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
    running: Option<bool>,
    script_hash: Option<ScriptHash>,
    script_path: Option<String>,
    duration_ms: Option<i64>,
    success: Option<bool>,
    deleted: bool,
    canceled: bool,
    canceled_by: Option<String>,
    job_kind: JobKind,
    schedule_path: Option<String>,
    permissioned_as: String,
    is_flow_step: bool,
    language: Option<ScriptLang>,
    is_skipped: bool,
    email: String,
    visible_to_owner: bool,
    suspend: Option<i32>,
    mem_peak: Option<i32>,
    tag: String,
    concurrent_limit: Option<i32>,
    concurrency_time_window_s: Option<i32>,
    priority: Option<i16>,
}

impl<'a> From<UnifiedJob> for Job {
    fn from(uj: UnifiedJob) -> Self {
        match uj.typ.as_ref() {
            "CompletedJob" => Job::CompletedJob(CompletedJob {
                workspace_id: uj.workspace_id,
                id: uj.id,
                parent_job: uj.parent_job,
                created_by: uj.created_by,
                created_at: uj.created_at,
                started_at: uj.started_at.unwrap_or(uj.created_at),
                duration_ms: uj.duration_ms.unwrap(),
                success: uj.success.unwrap(),
                script_hash: uj.script_hash,
                script_path: uj.script_path,
                args: None,
                result: None,
                logs: None,
                flow_status: None,
                deleted: uj.deleted,
                canceled: uj.canceled,
                canceled_by: uj.canceled_by,
                raw_code: None,
                canceled_reason: None,
                job_kind: uj.job_kind,
                schedule_path: uj.schedule_path,
                permissioned_as: uj.permissioned_as,
                raw_flow: None,
                is_flow_step: uj.is_flow_step,
                language: uj.language,
                is_skipped: uj.is_skipped,
                email: uj.email,
                visible_to_owner: uj.visible_to_owner,
                mem_peak: uj.mem_peak,
                tag: uj.tag,
                priority: uj.priority,
            }),
            "QueuedJob" => Job::QueuedJob(QueuedJob {
                workspace_id: uj.workspace_id,
                id: uj.id,
                parent_job: uj.parent_job,
                created_by: uj.created_by,
                created_at: uj.created_at,
                started_at: uj.started_at,
                script_hash: uj.script_hash,
                script_path: uj.script_path,
                args: None,
                running: uj.running.unwrap(),
                scheduled_for: uj.scheduled_for.unwrap(),
                logs: None,
                flow_status: None,
                raw_code: None,
                raw_lock: None,
                canceled: uj.canceled,
                canceled_by: uj.canceled_by,
                canceled_reason: None,
                last_ping: None,
                job_kind: uj.job_kind,
                schedule_path: uj.schedule_path,
                permissioned_as: uj.permissioned_as,
                raw_flow: None,
                is_flow_step: uj.is_flow_step,
                language: uj.language,
                same_worker: false,
                pre_run_error: None,
                email: uj.email,
                visible_to_owner: uj.visible_to_owner,
                suspend: uj.suspend,
                mem_peak: uj.mem_peak,
                root_job: None,
                leaf_jobs: None,
                tag: uj.tag,
                concurrent_limit: uj.concurrent_limit,
                concurrency_time_window_s: uj.concurrency_time_window_s,
                timeout: None,
                flow_step_id: None,
                cache_ttl: None,
                priority: uj.priority,
            }),
            t => panic!("job type {} not valid", t),
        }
    }
}
#[derive(Deserialize)]
struct CancelJob {
    reason: Option<String>,
}

#[derive(Deserialize)]
#[serde(rename_all = "lowercase")]
enum PreviewKind {
    Code,
    Identity,
    Http,
    Noop,
}
#[derive(Deserialize)]
struct Preview {
    content: Option<String>,
    kind: Option<PreviewKind>,
    path: Option<String>,
    args: Option<Box<JsonRawValue>>,
    language: Option<ScriptLang>,
    tag: Option<String>,
    dedicated_worker: Option<bool>,
}

#[derive(Deserialize)]
struct PreviewFlow {
    value: FlowValue,
    path: Option<String>,
    args: Option<Box<JsonRawValue>>,
    tag: Option<String>,
    restarted_from: Option<RestartedFrom>,
}

pub struct QueryOrBody<D>(pub Option<D>);

#[axum::async_trait]
impl<S, D> FromRequest<S, axum::body::Body> for QueryOrBody<D>
where
    D: DeserializeOwned,
    S: Send + Sync,
{
    type Rejection = Response;

    async fn from_request(
        req: Request<axum::body::Body>,
        state: &S,
    ) -> std::result::Result<Self, Self::Rejection> {
        return if req.method() == axum::http::Method::GET {
            let Query(InPayload { payload }) = Query::from_request(req, state)
                .await
                .map_err(IntoResponse::into_response)?;
            payload
                .map(|p| {
                    decode_payload(p)
                        .map(QueryOrBody)
                        .map_err(|err| (StatusCode::BAD_REQUEST, format!("{err:#?}")))
                        .map_err(IntoResponse::into_response)
                })
                .unwrap_or(Ok(QueryOrBody(None)))
        } else {
            Json::from_request(req, state)
                .await
                .map(|Json(v)| QueryOrBody(Some(v)))
                .map_err(IntoResponse::into_response)
        };

        #[derive(Deserialize)]
        struct InPayload {
            payload: Option<String>,
        }
    }
}

fn decode_payload<D: DeserializeOwned>(t: String) -> anyhow::Result<D> {
    let vec = base64::engine::general_purpose::URL_SAFE
        .decode(t)
        .context("invalid base64")?;
    serde_json::from_slice(vec.as_slice()).context("invalid json")
}

#[derive(Deserialize)]
pub struct DecodeQuery {
    pub include_query: Option<String>,
}

#[derive(Deserialize)]
pub struct IncludeQuery {
    pub include_query: Option<String>,
}

pub struct DecodeQueries(pub HashMap<String, Box<RawValue>>);

#[axum::async_trait]
impl<S> FromRequest<S, axum::body::Body> for DecodeQueries
where
    S: Send + Sync,
{
    type Rejection = Response;

    async fn from_request(
        req: Request<axum::body::Body>,
        _state: &S,
    ) -> Result<Self, Self::Rejection> {
        let query = req.uri().query().unwrap_or("");
        let include_query = serde_urlencoded::from_str::<IncludeQuery>(query)
            .map(|x| x.include_query)
            .ok()
            .flatten()
            .unwrap_or_default();
        let parse_query_args = include_query
            .split(",")
            .map(|s| s.to_string())
            .collect::<Vec<_>>();
        let mut args = HashMap::new();
        if !parse_query_args.is_empty() {
            let queries =
                serde_urlencoded::from_str::<HashMap<String, String>>(query).unwrap_or_default();
            parse_query_args.iter().for_each(|h| {
                if let Some(v) = queries.get(h) {
                    args.insert(h.to_string(), to_raw_value(v));
                }
            });
        }
        Ok(DecodeQueries(args))
    }
}

pub fn add_raw_string(
    raw_string: Option<String>,
    mut args: serde_json::Map<String, serde_json::Value>,
) -> serde_json::Map<String, serde_json::Value> {
    if let Some(raw_string) = raw_string {
        args.insert(
            "raw_string".to_string(),
            serde_json::Value::String(raw_string),
        );
    }
    return args;
}

async fn check_tag_available_for_workspace(w_id: &str, tag: &Option<String>) -> error::Result<()> {
    if let Some(tag) = tag {
        if tag == "" {
            return Ok(());
        }
        let custom_tags_per_w = CUSTOM_TAGS_PER_WORKSPACE.read().await;
        if custom_tags_per_w.0.contains(&tag.to_string()) {
            Ok(())
        } else if custom_tags_per_w.1.contains_key(tag)
            && custom_tags_per_w
                .1
                .get(tag)
                .unwrap()
                .contains(&w_id.to_string())
        {
            Ok(())
        } else {
            return Err(error::Error::BadRequest(format!(
                "Tag {tag} cannot be used on workspace {w_id}: (CUSTOM_TAGS: {:?})",
                custom_tags_per_w
            )));
        }
    } else {
        Ok(())
    }
}

#[cfg(feature = "enterprise")]
pub async fn check_license_key_valid() -> error::Result<()> {
    use windmill_common::ee::LICENSE_KEY_VALID;

    let valid = *LICENSE_KEY_VALID.read().await;
    if !valid {
        return Err(error::Error::BadRequest(format!(
            "License key is not valid. Go to your superadmin settings to update your license key.",
        )));
    }
    Ok(())
}

pub async fn run_flow_by_path(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path((w_id, flow_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
) -> error::Result<(StatusCode, String)> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;
    let flow_path = flow_path.to_path();
    check_scopes(&authed, || format!("run:flow/{flow_path}"))?;

    let (tag, dedicated_worker) = sqlx::query!(
        "SELECT tag, dedicated_worker from flow WHERE path = $1 and workspace_id = $2",
        flow_path,
        w_id
    )
    .fetch_optional(&db)
    .await?
    .map(|x| (x.tag, x.dedicated_worker))
    .unwrap_or_else(|| (None, None));

    check_tag_available_for_workspace(&w_id, &tag).await?;
    let scheduled_for = run_query.get_scheduled_for(&db).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        JobPayload::Flow { path: flow_path.to_string(), dedicated_worker },
        args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        scheduled_for,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        tag,
        None,
        None,
        None,
    )
    .await?;
    tx.commit().await?;
    Ok((StatusCode::CREATED, uuid.to_string()))
}

pub async fn restart_flow(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path((w_id, job_id, step_id, branch_or_iteration_n)): Path<(
        String,
        Uuid,
        String,
        Option<usize>,
    )>,
    Query(run_query): Query<RunJobQuery>,
) -> error::Result<(StatusCode, String)> {
    #[cfg(not(feature = "enterprise"))]
    {
        return Err(Error::BadRequest(
            "Restarting a flow is a feature only available in enterprise version".to_string(),
        ));
    }

    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    let completed_job = sqlx::query_as::<_, CompletedJob>(
        "SELECT * from completed_job WHERE id = $1 and workspace_id = $2",
    )
    .bind(job_id)
    .bind(&w_id)
    .fetch_optional(&db)
    .await?
    .with_context(|| "Unable to find completed job with the given job UUID")?;

    let flow_path = completed_job
        .script_path
        .with_context(|| "No flow path set for completed flow job")?;
    check_scopes(&authed, || format!("run:flow/{flow_path}"))?;

    let push_args = completed_job
        .args
        .map(|json| PushArgs { args: json.clone(), extra: json.0 });

    let scheduled_for = run_query.get_scheduled_for(&db).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        JobPayload::RestartedFlow {
            completed_job_id: job_id,
            step_id: step_id,
            branch_or_iteration_n: branch_or_iteration_n,
        },
        push_args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        scheduled_for,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        Some(completed_job.tag),
        None,
        None,
        completed_job.priority,
    )
    .await?;
    tx.commit().await?;
    Ok((StatusCode::CREATED, uuid.to_string()))
}

pub async fn run_job_by_path(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path((w_id, script_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
) -> error::Result<(StatusCode, String)> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    let script_path = script_path.to_path();

    check_scopes(&authed, || format!("run:script/{script_path}"))?;

    let (job_payload, tag, _delete_after_use, timeout) =
        script_path_to_payload(script_path, &db, &w_id).await?;
    let scheduled_for = run_query.get_scheduled_for(&db).await?;

    check_tag_available_for_workspace(&w_id, &tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        job_payload,
        args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        scheduled_for,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        tag,
        timeout,
        None,
        None,
    )
    .await?;
    tx.commit().await?;
    Ok((StatusCode::CREATED, uuid.to_string()))
}

struct Guard {
    done: bool,
    id: Uuid,
    w_id: String,
    db: DB,
}

impl Drop for Guard {
    fn drop(&mut self) {
        if !&self.done {
            let id = self.id;
            let w_id = self.w_id.clone();
            let db = self.db.clone();

            tracing::info!("http connection broke, marking job {id} as canceled");
            tokio::spawn(async move {
                let _ = sqlx::query!(
                "UPDATE queue SET canceled = true, canceled_reason = 'http connection broke', canceled_by = queue.created_by WHERE id = $1 AND workspace_id = $2",
                id,
                w_id
            )
            .execute(&db)
            .await;
            });
        }
    }
}

#[derive(Deserialize)]
pub struct WindmillCompositeResult {
    windmill_status_code: Option<u16>,
    windmill_content_type: Option<String>,
    result: Option<Box<RawValue>>,
}
async fn run_wait_result<T>(
    db: &DB,
    uuid: Uuid,
    Path((w_id, _)): Path<(String, T)>,
    node_id_for_empty_return: Option<String>,
) -> error::Result<Response> {
    let mut result;
    let timeout = SERVER_CONFIG.read().await.timeout_wait_result.clone();
    let timeout_ms = if timeout <= 0 {
        2000
    } else {
        (timeout * 1000) as u64
    };

    let mut g = Guard { done: false, id: uuid, w_id: w_id.clone(), db: db.clone() };

    let fast_poll_duration = *WAIT_RESULT_FAST_POLL_DURATION_SECS as u64 * 1000;
    let mut accumulated_delay = 0 as u64;

    loop {
        if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() {
            result = get_result_by_id_from_running_flow(
                &db,
                &w_id,
                &uuid,
                node_id_for_empty_return,
                None,
            )
            .await
            .ok();
        } else {
            let row =
                sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2")
                    .bind(uuid)
                    .bind(&w_id)
                    .fetch_optional(db)
                    .await?;
            if let Some(row) = row {
                result = Some(RawResult::from_row(&row)?.result.to_owned());
            } else {
                result = None;
            }
        }

        if result.is_some() {
            break;
        }

        let delay = if accumulated_delay <= fast_poll_duration {
            *WAIT_RESULT_FAST_POLL_INTERVAL_MS
        } else {
            *WAIT_RESULT_SLOW_POLL_INTERVAL_MS
        };
        accumulated_delay += delay;
        if accumulated_delay > timeout_ms {
            break;
        };
        tokio::time::sleep(core::time::Duration::from_millis(delay)).await;
    }
    if let Some(result) = result {
        g.done = true;

        let composite_result = serde_json::from_str::<WindmillCompositeResult>(result.get());
        match composite_result {
            Ok(WindmillCompositeResult {
                windmill_status_code,
                windmill_content_type,
                result: result_value,
            }) => {
                if windmill_content_type.is_none() && windmill_status_code.is_none() {
                    return Ok(Json(result).into_response());
                }

                let status_code_or_default = windmill_status_code
                    .map(|val| match StatusCode::from_u16(val) {
                        Ok(sc) => Ok(sc),
                        Err(_) => Err(Error::ExecutionErr("Invalid status code".to_string())),
                    })
                    .unwrap_or(if result_value.is_some() {
                        Ok(StatusCode::OK)
                    } else {
                        Ok(StatusCode::NO_CONTENT)
                    })?;

                if windmill_content_type.is_some() {
                    let serialized_json_result = result_value
                        .map(|val| val.get().to_owned())
                        .unwrap_or_else(String::new);
                    // if the `result` was just a single string, the below removes the surrounding quotes by parsing it as a string.
                    // it falls back to the original serialized JSON if it doesn't work.
                    let serialized_result =
                        serde_json::from_str::<String>(serialized_json_result.as_str())
                            .ok()
                            .unwrap_or(serialized_json_result);
                    return Ok((
                        status_code_or_default,
                        [(
                            http::header::CONTENT_TYPE,
                            HeaderValue::from_str(windmill_content_type.unwrap().as_str()).unwrap(),
                        )],
                        serialized_result,
                    )
                        .into_response());
                }
                return Ok((
                    status_code_or_default,
                    Json(result_value), // default to JSON result if no content type is provided
                )
                    .into_response());
            }
            _ => Ok(Json(result).into_response()),
        }
    } else {
        Err(Error::ExecutionErr(format!("timeout after {}s", timeout)))
    }
}

async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> {
    sqlx::query!(
        "UPDATE completed_job
        SET logs = '##DELETED##', args = '{}'::jsonb, result = '{}'::jsonb
        WHERE id = $1",
        job_uuid,
    )
    .execute(db)
    .await?;
    Ok(())
}

pub async fn check_queue_too_long(db: &DB, queue_limit: Option<i64>) -> error::Result<()> {
    if let Some(limit) = queue_limit {
        let count = sqlx::query_scalar!(
            "SELECT COUNT(*) FROM queue WHERE  canceled = false AND (scheduled_for <= now()
        OR (suspend_until IS NOT NULL
            AND (   suspend <= 0
                 OR suspend_until <= now())))",
        )
        .fetch_one(db)
        .await?
        .unwrap_or(0);

        if count > queue_limit.unwrap() {
            return Err(Error::InternalErr(format!(
                "Number of queued job is too high: {count} > {limit}"
            )));
        }
    }
    Ok(())
}

lazy_static::lazy_static! {
    pub static ref QUEUE_LIMIT_WAIT_RESULT: Option<i64> = std::env::var("QUEUE_LIMIT_WAIT_RESULT")
        .ok()
        .and_then(|x| x.parse().ok());
    pub static ref WAIT_RESULT_FAST_POLL_INTERVAL_MS: u64 = std::env::var("WAIT_RESULT_FAST_POLL_INTERVAL_MS")
        .ok()
        .and_then(|x| x.parse().ok())
        .unwrap_or(50);
    pub static ref WAIT_RESULT_FAST_POLL_DURATION_SECS: u16 = std::env::var("WAIT_RESULT_FAST_POLL_DURATION_SECS")
        .ok()
        .and_then(|x| x.parse().ok())
        .unwrap_or(2);
    pub static ref WAIT_RESULT_SLOW_POLL_INTERVAL_MS: u64 = std::env::var("WAIT_RESULT_SLOW_POLL_INTERVAL_MS")
        .ok()
        .and_then(|x| x.parse().ok())
        .unwrap_or(200);
}

pub async fn run_wait_result_job_by_path_get(
    method: hyper::http::Method,
    authed: ApiAuthed,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Extension(user_db): Extension<UserDB>,
    Extension(db): Extension<DB>,
    Path((w_id, script_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
    DecodeQueries(queries): DecodeQueries,
) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    if method == http::Method::HEAD {
        return Ok(Json(serde_json::json!("")).into_response());
    }
    let payload_r = run_query
        .payload
        .map(decode_payload)
        .map(|x| x.map_err(|e| Error::InternalErr(e.to_string())));

    let mut payload_args = if let Some(payload) = payload_r {
        payload?
    } else {
        HashMap::new()
    };
    queries.iter().for_each(|(k, v)| {
        payload_args.insert(k.to_string(), v.clone());
    });

    let inner_args: HashMap<String, Box<RawValue>> = HashMap::new();
    let args = PushArgs { extra: payload_args, args: sqlx::types::Json(inner_args) };

    check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
    let script_path = script_path.to_path();
    check_scopes(&authed, || format!("run:script/{script_path}"))?;

    let (job_payload, tag, delete_after_use, timeout) =
        script_path_to_payload(script_path, &db, &w_id).await?;
    check_tag_available_for_workspace(&w_id, &tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        job_payload,
        args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        None,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        tag,
        timeout,
        None,
        None,
    )
    .await?;
    tx.commit().await?;

    let wait_result = run_wait_result(&db, uuid, Path((w_id, script_path)), None).await;
    if delete_after_use.unwrap_or(false) {
        delete_job_metadata_after_use(&db, uuid).await?;
    }
    return wait_result;
}

pub async fn run_wait_result_flow_by_path_get(
    method: hyper::http::Method,
    authed: ApiAuthed,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Extension(user_db): Extension<UserDB>,
    Extension(db): Extension<DB>,
    Path((w_id, flow_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
    DecodeQueries(queries): DecodeQueries,
) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    if method == http::Method::HEAD {
        return Ok(Json(serde_json::json!("")).into_response());
    }
    let payload_r = run_query
        .payload
        .clone()
        .map(decode_payload)
        .map(|x| x.map_err(|e| Error::InternalErr(e.to_string())));

    let mut payload_args = if let Some(payload) = payload_r {
        payload?
    } else {
        HashMap::new()
    };

    queries.iter().for_each(|(k, v)| {
        payload_args.insert(k.to_string(), v.clone());
    });

    let args = PushArgs { extra: payload_args, args: sqlx::types::Json(HashMap::new()) };

    run_wait_result_flow_by_path_internal(
        db, run_query, flow_path, authed, rsmq, user_db, args, w_id,
    )
    .await
}

pub async fn run_wait_result_script_by_path(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Extension(db): Extension<DB>,
    Path((w_id, script_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    run_wait_result_script_by_path_internal(
        db,
        run_query,
        script_path,
        authed,
        rsmq,
        user_db,
        w_id,
        args,
    )
    .await
}

async fn run_wait_result_script_by_path_internal(
    db: sqlx::Pool<Postgres>,
    run_query: RunJobQuery,
    script_path: StripPath,
    authed: ApiAuthed,
    rsmq: Option<rsmq_async::MultiplexedRsmq>,
    user_db: UserDB,
    w_id: String,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
) -> error::Result<Response> {
    check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
    let script_path = script_path.to_path();
    check_scopes(&authed, || format!("run:script/{script_path}"))?;

    let (job_payload, tag, delete_after_use, timeout) =
        script_path_to_payload(script_path, &db, &w_id).await?;

    check_tag_available_for_workspace(&w_id, &tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        job_payload,
        args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        None,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        tag,
        timeout,
        None,
        None,
    )
    .await?;
    tx.commit().await?;

    let wait_result = run_wait_result(&db, uuid, Path((w_id, script_path)), None).await;
    if delete_after_use.unwrap_or(false) {
        delete_job_metadata_after_use(&db, uuid).await?;
    }
    return wait_result;
}

pub async fn run_wait_result_script_by_hash(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Extension(db): Extension<DB>,
    Path((w_id, script_hash)): Path<(String, ScriptHash)>,
    Query(run_query): Query<RunJobQuery>,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    check_queue_too_long(&db, run_query.queue_limit).await?;

    let hash = script_hash.0;
    let (
        path,
        tag,
        concurrent_limit,
        concurrency_time_window_s,
        cache_ttl,
        language,
        dedicated_worker,
        priority,
        delete_after_use,
        timeout,
    ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
    check_scopes(&authed, || format!("run:script/{path}"))?;

    check_tag_available_for_workspace(&w_id, &tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        JobPayload::ScriptHash {
            hash: ScriptHash(hash),
            path: path,
            concurrent_limit: concurrent_limit,
            concurrency_time_window_s: concurrency_time_window_s,
            cache_ttl,
            language,
            dedicated_worker,
            priority,
        },
        args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        None,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        tag,
        timeout,
        None,
        None,
    )
    .await?;
    tx.commit().await?;

    let wait_result = run_wait_result(&db, uuid, Path((w_id, script_hash)), None).await;
    if delete_after_use.unwrap_or(false) {
        delete_job_metadata_after_use(&db, uuid).await?;
    }
    return wait_result;
}

pub async fn run_wait_result_flow_by_path(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Extension(db): Extension<DB>,
    Path((w_id, flow_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    run_wait_result_flow_by_path_internal(
        db, run_query, flow_path, authed, rsmq, user_db, args, w_id,
    )
    .await
}

async fn run_wait_result_flow_by_path_internal(
    db: sqlx::Pool<Postgres>,
    run_query: RunJobQuery,
    flow_path: StripPath,
    authed: ApiAuthed,
    rsmq: Option<rsmq_async::MultiplexedRsmq>,
    user_db: UserDB,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
    w_id: String,
) -> error::Result<Response> {
    check_queue_too_long(&db, run_query.queue_limit).await?;

    let flow_path = flow_path.to_path();
    check_scopes(&authed, || format!("run:flow/{flow_path}"))?;

    let scheduled_for = run_query.get_scheduled_for(&db).await?;

    let (tag, dedicated_worker, early_return) = sqlx::query!(
        "SELECT tag, dedicated_worker, value->>'early_return' as early_return from flow WHERE path = $1 and workspace_id = $2",
        flow_path,
        w_id
    )
    .fetch_optional(&db)
    .await?
    .map(|x| (x.tag, x.dedicated_worker, x.early_return))
    .unwrap_or_else(|| (None, None, None));

    check_tag_available_for_workspace(&w_id, &tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        JobPayload::Flow { path: flow_path.to_string(), dedicated_worker },
        args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        scheduled_for,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        tag,
        None,
        None,
        None,
    )
    .await?;
    tx.commit().await?;

    run_wait_result(&db, uuid, Path((w_id, flow_path)), early_return).await
}

async fn run_preview_job(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path(w_id): Path<String>,
    Query(run_query): Query<RunJobQuery>,
    Json(preview): Json<Preview>,
) -> error::Result<(StatusCode, String)> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    check_scopes(&authed, || format!("runscript"))?;
    if authed.is_operator {
        return Err(error::Error::NotAuthorized(
            "Operators cannot run preview jobs for security reasons".to_string(),
        ));
    }
    let scheduled_for = run_query.get_scheduled_for(&db).await?;
    check_tag_available_for_workspace(&w_id, &preview.tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        match preview.kind {
            Some(PreviewKind::Identity) => JobPayload::Identity,
            Some(PreviewKind::Noop) => JobPayload::Noop,
            _ => JobPayload::Code(RawCode {
                content: preview.content.unwrap_or_default(),
                path: preview.path,
                language: preview.language.unwrap_or(ScriptLang::Deno),
                lock: None,
                concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
                concurrency_time_window_s: None, // TODO(gbouv): same as above
                cache_ttl: None,
                dedicated_worker: preview.dedicated_worker,
            }),
        },
        preview.args.unwrap_or_default(),
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        scheduled_for,
        None,
        None,
        None,
        run_query.job_id,
        false,
        false,
        None,
        true,
        preview.tag,
        None,
        None,
        None,
    )
    .await?;
    tx.commit().await?;

    Ok((StatusCode::CREATED, uuid.to_string()))
}

#[derive(Deserialize)]
struct BatchInfo {
    kind: String,
    flow_value: Option<FlowValue>,
    path: Option<String>,
}

#[tracing::instrument(level = "trace", skip_all)]
async fn add_batch_jobs(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path((w_id, n)): Path<(String, i32)>,
    Json(batch_info): Json<BatchInfo>,
) -> error::JsonResult<Vec<Uuid>> {
    require_super_admin(&db, &authed.email).await?;

    let (
        hash,
        path,
        job_kind,
        language,
        dedicated_worker,
        concurrent_limit,
        concurrent_time_window_s,
        timeout,
    ) = match batch_info.kind.as_str() {
        "script" => {
            if let Some(path) = batch_info.path {
                let (
                    script_hash,
                    _tag,
                    concurrent_limit,
                    concurrency_time_window_s,
                    _cache_ttl,
                    language,
                    dedicated_worker,
                    _priority,
                    _delete_after_use,
                    timeout,
                ) = get_latest_deployed_hash_for_path(&db, &w_id, &path).await?;
                (
                    Some(script_hash),
                    Some(path),
                    JobKind::Script,
                    Some(language),
                    dedicated_worker,
                    concurrent_limit,
                    concurrency_time_window_s,
                    timeout,
                )
            } else {
                Err(anyhow::anyhow!(
                    "Path is required if no value is not provided"
                ))?
            }
        }
        "flow" => {
            let mut tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);

            let mut uuids: Vec<Uuid> = Vec::new();
            let payload = if let Some(ref fv) = batch_info.flow_value {
                JobPayload::RawFlow { value: fv.clone(), path: None, restarted_from: None }
            } else {
                if let Some(path) = batch_info.path.as_ref() {
                    JobPayload::Flow { path: path.to_string(), dedicated_worker: None }
                } else {
                    Err(anyhow::anyhow!(
                        "Path is required if no value is not provided"
                    ))?
                }
            };
            for _ in 0..n {
                let (uuid, ntx) = push(
                    &db,
                    tx,
                    &w_id,
                    payload.clone(),
                    PushArgs::empty(),
                    &authed.username,
                    &authed.email,
                    username_to_permissioned_as(&authed.username),
                    None,
                    None,
                    None,
                    None,
                    None,
                    false,
                    false,
                    None,
                    true,
                    None,
                    None,
                    None,
                    None,
                )
                .await?;
                tx = PushIsolationLevel::Transaction(ntx);
                uuids.push(uuid);
            }
            match tx {
                PushIsolationLevel::Transaction(tx) => {
                    tx.commit().await?;
                }
                _ => (),
            }
            return Ok(Json(uuids));
        }
        "noop" => (None, None, JobKind::Noop, None, None, None, None, None),
        _ => {
            return Err(error::Error::BadRequest(format!(
                "Invalid batch kind: {}",
                batch_info.kind
            )))
        }
    };

    let language = language.unwrap_or(ScriptLang::Deno);

    let tag = if let Some(dedicated_worker) = dedicated_worker {
        if dedicated_worker && path.is_some() {
            format!("{}:{}", w_id, path.clone().unwrap())
        } else {
            format!("{}", language.as_str())
        }
    } else {
        format!("{}", language.as_str())
    };

    let uuids = sqlx::query_scalar!(
        r#"WITH uuid_table as (
            select gen_random_uuid() as uuid from generate_series(1, $11)
        )
        INSERT INTO queue 
            (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout)
            (SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14 FROM uuid_table) 
        RETURNING id"#,
            hash.map(|h| h.0),
            path,
            job_kind.clone() as JobKind,
            language as ScriptLang,
            tag,
            authed.username,
            username_to_permissioned_as(&authed.username),
            authed.email,
            Utc::now(),
            w_id,
            n,
            concurrent_limit,
            concurrent_time_window_s,
            timeout
        )
        .fetch_all(&db)
        .await?;

    Ok(Json(uuids))
}

async fn run_preview_flow_job(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path(w_id): Path<String>,
    Query(run_query): Query<RunJobQuery>,
    Json(raw_flow): Json<PreviewFlow>,
) -> error::Result<(StatusCode, String)> {
    check_scopes(&authed, || format!("runflow"))?;
    if authed.is_operator {
        return Err(error::Error::NotAuthorized(
            "Operators cannot run preview jobs for security reasons".to_string(),
        ));
    }
    let scheduled_for = run_query.get_scheduled_for(&db).await?;
    check_tag_available_for_workspace(&w_id, &raw_flow.tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        JobPayload::RawFlow {
            value: raw_flow.value,
            path: raw_flow.path,
            restarted_from: raw_flow.restarted_from,
        },
        raw_flow.args.unwrap_or_default(),
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        scheduled_for,
        None,
        None,
        None,
        run_query.job_id,
        false,
        false,
        None,
        true,
        raw_flow.tag,
        None,
        None,
        None,
    )
    .await?;
    tx.commit().await?;

    Ok((StatusCode::CREATED, uuid.to_string()))
}

pub async fn run_job_by_hash(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,

    Extension(user_db): Extension<UserDB>,
    Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
    Path((w_id, script_hash)): Path<(String, ScriptHash)>,
    Query(run_query): Query<RunJobQuery>,
    args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
) -> error::Result<(StatusCode, String)> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

    let hash = script_hash.0;
    let (
        path,
        tag,
        concurrent_limit,
        concurrency_time_window_s,
        cache_ttl,
        language,
        dedicated_worker,
        priority,
        _delete_after_use, // not taken into account in async endpoints
        timeout,
    ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
    check_scopes(&authed, || format!("run:script/{path}"))?;

    let scheduled_for = run_query.get_scheduled_for(&db).await?;

    check_tag_available_for_workspace(&w_id, &tag).await?;
    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);

    let (uuid, tx) = push(
        &db,
        tx,
        &w_id,
        JobPayload::ScriptHash {
            hash: ScriptHash(hash),
            path: path,
            concurrent_limit: concurrent_limit,
            concurrency_time_window_s: concurrency_time_window_s,
            cache_ttl,
            language,
            dedicated_worker,
            priority,
        },
        args,
        &authed.username,
        &authed.email,
        username_to_permissioned_as(&authed.username),
        scheduled_for,
        None,
        run_query.parent_job,
        run_query.parent_job,
        run_query.job_id,
        false,
        false,
        None,
        !run_query.invisible_to_owner.unwrap_or(false),
        tag,
        timeout,
        None,
        None,
    )
    .await?;
    tx.commit().await?;

    Ok((StatusCode::CREATED, uuid.to_string()))
}

#[derive(Deserialize)]
pub struct JobUpdateQuery {
    pub running: bool,
    pub log_offset: i32,
}

#[derive(Serialize)]
pub struct JobUpdate {
    pub running: Option<bool>,
    pub completed: Option<bool>,
    pub new_logs: Option<String>,
    pub mem_peak: Option<i32>,
}

async fn get_job_update(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
    Query(JobUpdateQuery { running, log_offset }): Query<JobUpdateQuery>,
) -> error::JsonResult<JobUpdate> {
    let record = sqlx::query!(
        "SELECT running, substr(logs, $1) as logs, mem_peak FROM queue WHERE workspace_id = $2 AND id = $3",
        log_offset,
        &w_id,
        &id
    )
    .fetch_optional(&db)
    .await?;

    if let Some(record) = record {
        Ok(Json(JobUpdate {
            running: if !running && record.running {
                Some(true)
            } else {
                None
            },
            completed: None,
            new_logs: record.logs,
            mem_peak: record.mem_peak,
        }))
    } else {
        let logs = query_scalar!(
            "SELECT substr(logs, $1) as logs FROM completed_job WHERE workspace_id = $2 AND id = \
             $3",
            log_offset,
            &w_id,
            &id
        )
        .fetch_optional(&db)
        .await?;
        let logs = not_found_if_none(logs, "Job Update", id.to_string())?;
        Ok(Json(JobUpdate {
            running: Some(false),
            completed: Some(true),
            new_logs: logs,
            mem_peak: record.map(|r| r.mem_peak).flatten(),
        }))
    }
}

fn list_completed_jobs_query(
    w_id: &str,
    per_page: usize,
    offset: usize,
    lq: &ListCompletedQuery,
    fields: &[&str],
) -> SqlBuilder {
    let mut sqlb = SqlBuilder::select_from("completed_job")
        .fields(fields)
        .order_by("created_at", lq.order_desc.unwrap_or(true))
        .and_where_eq("workspace_id", "?".bind(&w_id))
        .offset(offset)
        .limit(per_page)
        .clone();

    if let Some(p) = &lq.schedule_path {
        sqlb.and_where_eq("schedule_path", "?".bind(p));
    }

    if let Some(ps) = &lq.script_path_start {
        sqlb.and_where_like_left("script_path", ps);
    }
    if let Some(p) = &lq.script_path_exact {
        sqlb.and_where_eq("script_path", "?".bind(p));
    }
    if let Some(h) = &lq.script_hash {
        sqlb.and_where_eq("script_hash", "?".bind(h));
    }
    if let Some(t) = &lq.tag {
        sqlb.and_where_eq("tag", "?".bind(t));
    }
    if let Some(cb) = &lq.created_by {
        sqlb.and_where_eq("created_by", "?".bind(cb));
    }
    if let Some(r) = &lq.success {
        sqlb.and_where_eq("success", r);
    }
    if let Some(pj) = &lq.parent_job {
        sqlb.and_where_eq("parent_job", "?".bind(pj));
    }
    if let Some(dt) = &lq.started_before {
        sqlb.and_where_le("started_at", format!("to_timestamp({})", dt.timestamp()));
    }
    if let Some(dt) = &lq.started_after {
        sqlb.and_where_ge("started_at", format!("to_timestamp({})", dt.timestamp()));
    }

    if let Some(dt) = &lq.created_or_started_before {
        sqlb.and_where_le("started_at", format!("to_timestamp({})", dt.timestamp()));
    }
    if let Some(dt) = &lq.created_or_started_after {
        sqlb.and_where_ge("started_at", format!("to_timestamp({})", dt.timestamp()));
    }

    if let Some(sk) = &lq.is_skipped {
        sqlb.and_where_eq("is_skipped", sk);
    }
    if let Some(fs) = &lq.is_flow_step {
        sqlb.and_where_eq("is_flow_step", fs);
    }
    if let Some(jk) = &lq.job_kinds {
        sqlb.and_where_in(
            "job_kind",
            &jk.split(',').into_iter().map(quote).collect::<Vec<_>>(),
        );
    }

    if let Some(args) = &lq.args {
        sqlb.and_where("args @> ?".bind(&args.replace("'", "''")));
    }

    if let Some(result) = &lq.result {
        sqlb.and_where("result @> ?".bind(&result.replace("'", "''")));
    }

    sqlb
}
#[derive(Deserialize, Clone)]
pub struct ListCompletedQuery {
    pub script_path_start: Option<String>,
    pub script_path_exact: Option<String>,
    pub script_hash: Option<String>,
    pub created_by: Option<String>,
    pub started_before: Option<chrono::DateTime<chrono::Utc>>,
    pub started_after: Option<chrono::DateTime<chrono::Utc>>,
    pub created_before: Option<chrono::DateTime<chrono::Utc>>,
    pub created_after: Option<chrono::DateTime<chrono::Utc>>,
    pub created_or_started_before: Option<chrono::DateTime<chrono::Utc>>,
    pub created_or_started_after: Option<chrono::DateTime<chrono::Utc>>,
    pub success: Option<bool>,
    pub running: Option<bool>,
    pub parent_job: Option<String>,
    pub order_desc: Option<bool>,
    pub job_kinds: Option<String>,
    pub is_skipped: Option<bool>,
    pub is_flow_step: Option<bool>,
    pub suspended: Option<bool>,
    pub schedule_path: Option<String>,
    // filter by matching a subset of the args using base64 encoded json subset
    pub args: Option<String>,
    // filter by matching a subset of the result using base64 encoded json subset
    pub result: Option<String>,
    pub tag: Option<String>,
}

async fn list_completed_jobs(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path(w_id): Path<String>,
    Query(pagination): Query<Pagination>,
    Query(lq): Query<ListCompletedQuery>,
) -> error::JsonResult<Vec<ListableCompletedJob>> {
    check_scopes(&authed, || format!("listjobs"))?;

    let (per_page, offset) = paginate(pagination);

    let sql = list_completed_jobs_query(
        &w_id,
        per_page,
        offset,
        &lq,
        &[
            "id",
            "workspace_id",
            "parent_job",
            "created_by",
            "created_at",
            "started_at",
            "duration_ms",
            "success",
            "script_hash",
            "script_path",
            "deleted",
            "canceled",
            "canceled_by",
            "canceled_reason",
            "job_kind",
            "schedule_path",
            "permissioned_as",
            "null as raw_code",
            "null as flow_status",
            "null as raw_flow",
            "is_flow_step",
            "language",
            "is_skipped",
            "email",
            "visible_to_owner",
            "mem_peak",
            "tag",
            "priority",
            "'CompletedJob' as type",
        ],
    )
    .sql()?;
    let jobs = sqlx::query_as::<_, ListableCompletedJob>(&sql)
        .fetch_all(&db)
        .await?;
    Ok(Json(jobs))
}

async fn get_completed_job<'a>(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<Response> {
    let job_o = sqlx::query("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, 
    CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, right(logs, 20000000) as logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id,
    schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped,
    raw_lock, email, visible_to_owner, mem_peak, tag, priority FROM completed_job WHERE id = $1 AND workspace_id = $2")
        .bind(id)
        .bind(w_id)
        .fetch_optional(&db)
        .await?;

    let job = not_found_if_none(job_o, "Completed Job", id.to_string())?;
    let response = Json(CompletedJob::from_row(&job)?).into_response();
    Ok(response)
}

#[derive(FromRow)]
pub struct RawResult<'a> {
    pub result: &'a JsonRawValue,
}

impl<'a> IntoResponse for RawResult<'a> {
    fn into_response(self) -> Response {
        Json(self.result).into_response()
    }
}

async fn get_completed_job_result(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
    Query(JsonPath { json_path }): Query<JsonPath>,
) -> error::Result<Response> {
    let result_o = if let Some(json_path) = json_path {
        sqlx::query(
            "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
        )
        .bind(id)
        .bind(w_id)
        .bind(
            json_path
                .split(".")
                .map(|x| x.to_string())
                .collect::<Vec<_>>(),
        )
        .fetch_optional(&db)
        .await?
    } else {
        sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2")
            .bind(id)
            .bind(w_id)
            .fetch_optional(&db)
            .await?
    };

    let result = not_found_if_none(result_o, "Completed Job", id.to_string())?;
    Ok(RawResult::from_row(&result)?.into_response())
}

#[derive(Serialize)]
struct CompletedJobResult<'c> {
    started: Option<bool>,
    completed: bool,
    result: Option<&'c JsonRawValue>,
}

#[derive(Deserialize)]
struct GetCompletedJobQuery {
    get_started: Option<bool>,
}

async fn get_completed_job_result_maybe(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
    Query(GetCompletedJobQuery { get_started }): Query<GetCompletedJobQuery>,
) -> error::Result<Response> {
    let result_o =
        sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2")
            .bind(id)
            .bind(&w_id)
            .fetch_optional(&db)
            .await?;

    if let Some(result) = result_o {
        let res = RawResult::from_row(&result)?;
        Ok(Json(CompletedJobResult {
            started: Some(true),
            completed: true,
            result: Some(res.result),
        })
        .into_response())
    } else if get_started.is_some_and(|x| x) {
        let started = sqlx::query_scalar!(
            "SELECT running FROM queue WHERE id = $1 AND workspace_id = $2",
            id,
            w_id
        )
        .fetch_optional(&db)
        .await?
        .unwrap_or(false);
        Ok(
            Json(CompletedJobResult { started: Some(started), completed: false, result: None })
                .into_response(),
        )
    } else {
        Ok(
            Json(CompletedJobResult { started: None, completed: false, result: None })
                .into_response(),
        )
    }
}

async fn delete_completed_job<'a>(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<Response> {
    check_scopes(&authed, || format!("deletejob"))?;

    let mut tx = user_db.begin(&authed).await?;

    require_admin(authed.is_admin, &authed.username)?;
    let job_o = sqlx::query(
        "UPDATE completed_job SET args = null, logs = '', result = null, deleted = true WHERE id = $1 AND workspace_id = $2 \
         RETURNING *",
    )
    .bind(id)
    .bind(&w_id)
    .fetch_optional(&mut *tx)
    .await?;

    let job = not_found_if_none(job_o, "Completed Job", id.to_string())?;

    audit_log(
        &mut *tx,
        &authed.username,
        "jobs.delete",
        ActionKind::Delete,
        &w_id,
        Some(&id.to_string()),
        None,
    )
    .await?;

    tx.commit().await?;
    let response = Json(CompletedJob::from_row(&job)?).into_response();
    Ok(response)
}
