From 561e13e51ee7ffcf20bc524c22d756ea582d546e Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Wed, 17 Aug 2022 15:56:49 +0200
Subject: [PATCH] fix: handle 0 length for-loops in the backend (#440)

* wip

* TODO

* my fix works I think
---
 backend/src/worker.rs      | 54 ++++++++++++++++++++++++++++++++++++++
 backend/src/worker_flow.rs | 37 +++++++++++++++++++++++++-
 2 files changed, 90 insertions(+), 1 deletion(-)

diff --git a/backend/src/worker.rs b/backend/src/worker.rs
index 4ed160a0e9..82ebc24700 100644
--- a/backend/src/worker.rs
+++ b/backend/src/worker.rs
@@ -1390,6 +1390,60 @@ def main():
         assert_eq!(result, serde_json::json!("hello world"));
     }
 
+
+    #[sqlx::test(fixtures("base"))]
+    async fn test_empty_loop(db: DB) {
+        initialize_tracing().await;
+
+        let flow: FlowValue = serde_json::from_value(serde_json::json!({
+            "modules": [
+                {
+                    "value": {
+                        "type": "forloopflow",
+                        "iterator": { "type": "static", "value": [] },
+                        "value": {
+                            "modules": [
+                                {
+                                    "input_transform": {
+                                        "n": {
+                                            "type": "javascript",
+                                            "expr": "previous_result.iter.value",
+                                        },
+                                    },
+                                    "value": {
+                                        "type": "rawscript",
+                                        "language": "python3",
+                                        "content": "def main(n): return n",
+                                    },
+                                }
+                            ],
+                        }
+                    },
+                },
+                {
+                    "input_transform": {
+                        "items": {
+                            "type": "javascript",
+                            "expr": "previous_result",
+                        },
+                    },
+                    "value": {
+                        "type": "rawscript",
+                        "language": "python3",
+                        "content": "def main(items): return sum(items)",
+                    },
+                },
+            ],
+        }))
+        .unwrap();
+
+        let flow = JobPayload::RawFlow { value: flow, path: None };
+        let result = run_job_in_new_worker_until_complete(&db, flow).await;
+
+        assert_eq!(result, serde_json::json!(0));
+    }
+
+
     #[sqlx::test(fixtures("base"))]
     async fn test_step_after_loop(db: DB) {
         initialize_tracing().await;
diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs
index e7eb3bb31f..fd5e157c7f 100644
--- a/backend/src/worker_flow.rs
+++ b/backend/src/worker_flow.rs
@@ -90,7 +90,7 @@ pub async fn update_flow_status_after_job_completion(
         module_status @ FlowStatusModule::InProgress {
             iterator: Some(Iterator { index, itered, .. }),
             ..
-        } if (index.to_owned() as usize) < itered.len() - 1 && (success || skip_loop_failures) => {
+        } if (index.to_owned() as usize) + 1 < itered.len() && (success || skip_loop_failures) => {
             (old_status.step, module_status.clone())
         }
         module_status @ _ => {
@@ -384,6 +384,7 @@ pub async fn handle_flow(
     Ok(())
 }
 
+#[async_recursion]
 #[instrument(level = "trace", skip_all)]
 async fn push_next_flow_job(
     flow_job: &QueuedJob,
@@ -520,6 +521,40 @@ async fn push_next_flow_job(
             _ => (true, None, None),
         };
 
+        if let Some((_, vec, _)) = &forloop_iterator {
+            if vec.len() == 0 {
+                let new_job = sqlx::query_as::<_, QueuedJob>(&format!(
+                    "UPDATE queue
+            SET 
+                flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {}}}', $1), \
+             '{{\"step\"}}', $2)
+            WHERE id = $3
+            RETURNING *",
+                    i
+                ))
+                .bind(serde_json::json!(FlowStatusModule::Success {
+                    job: flow_job.id,
+                    forloop_jobs: Some(vec![])
+                }))
+                .bind(serde_json::json!(if flow.modules.len() > i + 1 {
+                    i + 1
+                } else {
+                    i
+                }))
+                .bind(flow_job.id)
+                .fetch_one(&mut tx)
+                .await?;
+                tx.commit().await?;
+                if flow.modules.len() > i + 1 {
+                    return Ok(
+                        push_next_flow_job(&new_job, flow, schedule_path, db, json!([])).await?,
+                    );
+                } else {
+                    return Ok(());
+                }
+            }
+        }
+
         let mut args = if compute_input_transform {
             let steps = status
                 .modules