backend: re-work job table usages after v2

Author: uael

backend/.sqlx/query-111ee00fa6661fd4d3bb1d2e567fdaa1cbd3fe93b590b97c549ff2ad0016da1a.json

@@ -8,7 +8,6 @@
 
 use axum::body::Body;
 use axum::http::HeaderValue;
-use futures::TryFutureExt;
 use itertools::Itertools;
 use quick_cache::sync::Cache;
 use serde_json::value::RawValue;
@@ -792,38 +791,25 @@ impl<'a> GetQuery<'a> {
     async fn resolve_raw_values<T>(
         &self,
         db: &DB,
-        id: Uuid,
         kind: JobKind,
-        hash: Option<ScriptHash>,
+        runnable_id: Option<i64>,
         job: &mut JobExtended<T>,
     ) {
-        let (raw_code, raw_lock, raw_flow) = (
-            job.raw_code.take(),
-            job.raw_lock.take(),
-            job.raw_flow.take(),
-        );
-        if self.with_flow {
-            // Try to fetch the flow from the cache, fallback to the preview flow.
-            // NOTE: This could check for the job kinds instead of the `or_else` but it's not
-            // necessary as `fetch_flow` return early if the job kind is not a preview one.
-            cache::job::fetch_flow(db, kind, hash)
-                .or_else(|_| cache::job::fetch_preview_flow(db, &id, raw_flow))
+        job.raw_flow = match job.raw_flow.take() {
+            _ if !self.with_flow => None,
+            flow => cache::job::fetch_flow(db, kind, runnable_id, flow)
                 .await
                 .ok()
-                .inspect(|data| job.raw_flow = Some(sqlx::types::Json(data.raw_flow.clone())));
-        }
-        if self.with_code {
-            // Try to fetch the code from the cache, fallback to the preview code.
-            // NOTE: This could check for the job kinds instead of the `or_else` but it's not
-            // necessary as `fetch_script` return early if the job kind is not a preview one.
-            cache::job::fetch_script(db, kind, hash)
-                .or_else(|_| cache::job::fetch_preview_script(db, &id, raw_lock, raw_code))
+                .map(|data| sqlx::types::Json(data.raw_flow.clone())),
+        };
+        (job.raw_code, job.raw_lock) = match (job.raw_code.take(), job.raw_lock.take()) {
+            _ if !self.with_code => (None, None),
+            (code, lock) => cache::job::fetch_script(db, kind, runnable_id, code, lock)
                 .await
                 .ok()
-                .inspect(|data| {
-                    (job.raw_lock, job.raw_code) = (data.lock.clone(), Some(data.code.clone()))
-                });
-        }
+                .map(|data| (Some(data.code.clone()), data.lock.clone()))
+                .unwrap_or_default(),
+        };
     }
 
     async fn fetch_queued(
@@ -846,7 +832,7 @@ impl<'a> GetQuery<'a> {
 
         self.check_auth(job.as_ref().map(|job| job.created_by.as_str()))?;
         if let Some(job) = job.as_mut() {
-            self.resolve_raw_values(db, job.id, job.job_kind, job.script_hash, job)
+            self.resolve_raw_values(db, job.job_kind, job.script_hash.map(|x| x.0), job)
                 .await;
         }
         if self.with_flow {
@@ -878,7 +864,7 @@ impl<'a> GetQuery<'a> {
 
         self.check_auth(cjob.as_ref().map(|job| job.created_by.as_str()))?;
         if let Some(job) = cjob.as_mut() {
-            self.resolve_raw_values(db, job.id, job.job_kind, job.script_hash, job)
+            self.resolve_raw_values(db, job.job_kind, job.script_hash.map(|x| x.0), job)
                 .await;
         }
         if self.with_flow {

backend/.sqlx/query-3af91bb37638473d3d97a3a6f7fc6c6927ac0a6f6c26f11db69fbc4aa839cf64.json

@@ -22,7 +22,6 @@ use windmill_common::{
     cache,
     error::{self, Error},
     jobs::JobKind,
-    scripts::ScriptHash,
     variables::{build_crypt, decrypt_value_with_mc},
 };
 
@@ -978,7 +977,7 @@ async fn get_modal_blocks(
     let (job_kind, script_hash, raw_flow, parent_job_id, created_at, created_by, script_path, args) = sqlx::query!(
         "SELECT
             v2_queue.job_kind AS \"job_kind!: JobKind\",
-            v2_queue.script_hash AS \"script_hash: ScriptHash\",
+            v2_queue.script_hash,
             v2_queue.raw_flow AS \"raw_flow: sqlx::types::Json<Box<RawValue>>\",
             v2_completed_job.parent_job AS \"parent_job: Uuid\",
             v2_completed_job.created_at AS \"created_at!: chrono::NaiveDateTime\",
@@ -998,17 +997,11 @@ async fn get_modal_blocks(
     .ok_or_else(|| error::Error::BadRequest("This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string()))
     .map(|r| (r.job_kind, r.script_hash, r.raw_flow, r.parent_job, r.created_at, r.created_by, r.script_path, r.args))?;
 
-    let flow_data = match cache::job::fetch_flow(&db, job_kind, script_hash).await {
+    let flow_data = match cache::job::fetch_flow(&db, job_kind, script_hash, raw_flow).await {
         Ok(data) => data,
-        Err(_) => {
-            if let Some(parent_job_id) = parent_job_id.as_ref() {
-                cache::job::fetch_preview_flow(&db, parent_job_id, raw_flow).await?
-            } else {
-                return Err(error::Error::BadRequest(
-                    "This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string(),
-                ));
-            }
-        }
+        Err(_) => return Err(error::Error::BadRequest(
+            "This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string(),
+        ))
     };
 
     let flow_value = &flow_data.flow;