backend: re-work inputs API impl after v2

Author: uael

backend/windmill-api/src/inputs.rs

@@ -15,18 +15,16 @@ use axum::{
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use sqlx::{types::Uuid, FromRow};
-use std::{
-    fmt::{Display, Formatter},
-    vec,
-};
+use sqlx::types::Uuid;
+use std::fmt::{Display, Formatter};
 use windmill_common::{
     db::UserDB,
     error::JsonResult,
     jobs::JobKind,
     scripts::to_i64,
     utils::{not_found_if_none, paginate, Pagination},
 };
+
 pub fn workspaced_service() -> Router {
     Router::new()
         .route("/history", get(get_input_history))
@@ -40,20 +38,7 @@ pub fn workspaced_service() -> Router {
         )
 }
 
-#[derive(Debug, sqlx::FromRow, Serialize, Deserialize)]
-pub struct InputRow {
-    pub id: Uuid,
-    pub workspace_id: String,
-    pub runnable_id: String,
-    pub runnable_type: RunnableType,
-    pub name: String,
-    pub args: sqlx::types::Json<Box<serde_json::value::RawValue>>,
-    pub created_at: DateTime<Utc>,
-    pub created_by: String,
-    pub is_public: bool,
-}
-
-#[derive(Debug, Serialize, Deserialize, sqlx::Type)]
+#[derive(Debug, Serialize, Deserialize, sqlx::Type, Copy, Clone)]
 #[sqlx(type_name = "runnable_type")]
 pub enum RunnableType {
     ScriptHash,
@@ -71,24 +56,6 @@ impl Display for RunnableType {
     }
 }
 
-impl RunnableType {
-    fn job_kind(&self) -> JobKind {
-        match self {
-            RunnableType::ScriptHash => JobKind::Script,
-            RunnableType::ScriptPath => JobKind::Script,
-            RunnableType::FlowPath => JobKind::Flow,
-        }
-    }
-
-    fn column_name(&self) -> &'static str {
-        match self {
-            RunnableType::ScriptHash => "script_hash",
-            RunnableType::ScriptPath => "script_path",
-            RunnableType::FlowPath => "script_path",
-        }
-    }
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct RunnableParams {
     pub runnable_id: String,
@@ -99,22 +66,13 @@ pub struct RunnableParams {
 pub struct Input {
     id: Uuid,
     name: String,
-    created_at: chrono::DateTime<chrono::Utc>,
-    args: sqlx::types::Json<Box<serde_json::value::RawValue>>,
+    created_at: DateTime<Utc>,
+    args: Value,
     created_by: String,
     is_public: bool,
     success: bool,
 }
 
-#[derive(Debug, Serialize, Deserialize, FromRow)]
-pub struct CompletedJobMini {
-    id: Uuid,
-    created_at: chrono::DateTime<chrono::Utc>,
-    args: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
-    created_by: String,
-    success: bool,
-}
-
 #[derive(Deserialize)]
 struct GetInputHistory {
     include_preview: Option<bool>,
@@ -132,60 +90,43 @@ async fn get_input_history(
 
     let mut tx = user_db.begin(&authed).await?;
 
-    let sql = &format!(
-        "select id, created_at, created_by, 'null'::jsonb as args, success from v2_completed_job \
-        where {} = $1 and job_kind = any($2) and workspace_id = $3 \
-        order by created_at desc limit $4 offset $5",
-        r.runnable_type.column_name()
-    );
-
-    let query = sqlx::query_as::<_, CompletedJobMini>(sql);
-
-    let query = match r.runnable_type {
-        RunnableType::ScriptHash => query.bind(to_i64(&r.runnable_id)?),
-        _ => query.bind(&r.runnable_id),
+    let job_kinds = match (r.runnable_type, g.include_preview.unwrap_or(false)) {
+        (RunnableType::FlowPath, true) => [JobKind::Flow, JobKind::FlowPreview].as_slice(),
+        (RunnableType::FlowPath, _) => [JobKind::Flow].as_slice(),
+        (_, true) => [JobKind::Script, JobKind::Preview].as_slice(),
+        _ => [JobKind::Script].as_slice(),
     };
 
-    let job_kinds = match r.runnable_type.job_kind() {
-        kind @ JobKind::Script if g.include_preview.unwrap_or(false) => {
-            vec![kind, JobKind::Preview]
-        }
-        kind @ JobKind::Flow if g.include_preview.unwrap_or(false) => {
-            vec![kind, JobKind::FlowPreview]
-        }
-        kind => vec![kind],
+    let (runnable_id, runnable_path) = match r.runnable_type {
+        RunnableType::ScriptHash => (Some(to_i64(&r.runnable_id)?), None),
+        _ => (None, Some(&r.runnable_id)),
     };
 
-    let rows = query
-        .bind(job_kinds)
-        .bind(&w_id)
-        .bind(per_page as i32)
-        .bind(offset as i32)
-        .fetch_all(&mut *tx)
-        .await?;
+    let inputs = sqlx::query!(
+        "SELECT c.id, created_at, created_by, status = 'success'::job_status AS \"success!\"
+         FROM v2_job_completed c JOIN v2_job USING (id)
+         WHERE (runnable_id = $1 OR runnable_path = $2) AND kind = ANY($3) AND c.workspace_id = $4
+         ORDER BY created_at DESC LIMIT $5 OFFSET $6",
+        runnable_id,
+        runnable_path,
+        job_kinds as &[JobKind],
+        &w_id,
+        per_page as i32,
+        offset as i32
+    )
+    .map(|r| Input {
+        id: r.id,
+        name: format!("{} {}", r.created_at.format("%H:%M %-d/%-m"), r.created_by),
+        created_at: r.created_at,
+        args: Value::Null,
+        created_by: r.created_by,
+        is_public: true,
+        success: r.success,
+    })
+    .fetch_all(&mut *tx)
+    .await?;
 
     tx.commit().await?;
-
-    let mut inputs = vec![];
-
-    for row in rows {
-        inputs.push(Input {
-            id: row.id,
-            name: format!(
-                "{} {}",
-                row.created_at.format("%H:%M %-d/%-m"),
-                row.created_by
-            ),
-            created_at: row.created_at,
-            args: row.args.unwrap_or(sqlx::types::Json(
-                serde_json::value::RawValue::from_string("null".to_string()).unwrap(),
-            )),
-            created_by: row.created_by,
-            is_public: true,
-            success: row.success,
-        });
-    }
-
     Ok(Json(inputs))
 }
 
@@ -194,6 +135,7 @@ struct GetArgs {
     input: Option<bool>,
     allow_large: Option<bool>,
 }
+
 async fn get_args_from_history_or_saved_input(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
@@ -252,37 +194,26 @@ async fn list_saved_inputs(
 
     let mut tx = user_db.begin(&authed).await?;
 
-    let rows = sqlx::query_as::<_, InputRow>(
-        "select id, workspace_id, runnable_id, runnable_type, name, 'null'::jsonb as args, created_at, created_by, is_public from input \
-         where runnable_id = $1 and runnable_type = $2 and workspace_id = $3 \
-         and (is_public IS true OR created_by = $4) \
-         order by created_at desc limit $5 offset $6",
+    let inputs = sqlx::query_as!(
+        Input,
+        "SELECT id, name, 'null'::JSONB AS args, created_by, created_at, is_public,
+             TRUE AS \"success!\"
+         FROM input
+         WHERE runnable_id = $1 AND runnable_type = $2 AND workspace_id = $3
+             AND (is_public IS TRUE OR created_by = $4)
+         ORDER BY created_at DESC LIMIT $5 OFFSET $6",
+        &r.runnable_id,
+        &r.runnable_type as &RunnableType,
+        &w_id,
+        &authed.username,
+        per_page as i32,
+        offset as i32
     )
-    .bind(&r.runnable_id)
-    .bind(&r.runnable_type)
-    .bind(&w_id)
-    .bind(&authed.username)
-    .bind(per_page as i32)
-    .bind(offset as i32)
     .fetch_all(&mut *tx)
     .await?;
 
     tx.commit().await?;
 
-    let mut inputs: Vec<Input> = Vec::new();
-
-    for row in rows {
-        inputs.push(Input {
-            id: row.id,
-            name: row.name,
-            args: row.args,
-            created_by: row.created_by,
-            created_at: row.created_at,
-            is_public: row.is_public,
-            success: true,
-        })
-    }
-
     Ok(Json(inputs))
 }
 
@@ -303,16 +234,17 @@ async fn create_input(
 
     let id = Uuid::new_v4();
 
-    sqlx::query(
-        "INSERT INTO input (id, workspace_id, runnable_id, runnable_type, name, args, created_by) VALUES ($1, $2, $3, $4, $5, $6, $7)",
+    sqlx::query!(
+        "INSERT INTO input (id, workspace_id, runnable_id, runnable_type, name, args, created_by)
+         VALUES ($1, $2, $3, $4, $5, $6, $7)",
+        &id,
+        &w_id,
+        &r.runnable_id,
+        &r.runnable_type as &RunnableType,
+        &input.name,
+        sqlx::types::Json(&input.args) as sqlx::types::Json<&Box<serde_json::value::RawValue>>,
+        &authed.username
     )
-    .bind(&id)
-    .bind(&w_id)
-    .bind(&r.runnable_id)
-    .bind(&r.runnable_type)
-    .bind(&input.name)
-    .bind(sqlx::types::Json(&input.args))
-    .bind(&authed.username)
     .execute(&mut *tx)
     .await?;
 
@@ -336,13 +268,15 @@ async fn update_input(
 ) -> JsonResult<String> {
     let mut tx = user_db.begin(&authed).await?;
 
-    sqlx::query("UPDATE input SET name = $1, is_public = $2 WHERE id = $3 and workspace_id = $4")
-        .bind(&input.name)
-        .bind(&input.is_public)
-        .bind(&input.id)
-        .bind(&w_id)
-        .execute(&mut *tx)
-        .await?;
+    sqlx::query!(
+        "UPDATE input SET name = $1, is_public = $2 WHERE id = $3 and workspace_id = $4",
+        &input.name,
+        &input.is_public,
+        &input.id,
+        &w_id
+    )
+    .execute(&mut *tx)
+    .await?;
 
     tx.commit().await?;
 
@@ -356,11 +290,13 @@ async fn delete_input(
 ) -> JsonResult<String> {
     let mut tx = user_db.begin(&authed).await?;
 
-    sqlx::query("DELETE FROM input WHERE id = $1 and workspace_id = $2")
-        .bind(&i_id)
-        .bind(&w_id)
-        .execute(&mut *tx)
-        .await?;
+    sqlx::query!(
+        "DELETE FROM input WHERE id = $1 and workspace_id = $2",
+        &i_id,
+        &w_id
+    )
+    .execute(&mut *tx)
+    .await?;
 
     tx.commit().await?;