fix: workflow as code status (#5246)

* fix: get update endpoint

* [WIP] fix: workflow as code

* fix fix

* Revert "fix: get update endpoint"

This reverts commit 7af9abf868.

* fix test

backend/.sqlx/query-2e6935811a6d818bc523f076674f794f8be6c6bad3d06e74586e8ab668d91861.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO v2_job_status (id, workflow_as_code_status)\n            VALUES ($1, JSONB_SET('{}'::JSONB, array[$2], $3))\n            ON CONFLICT (id) DO UPDATE SET\n                workflow_as_code_status = JSONB_SET(\n                    COALESCE(v2_job_status.workflow_as_code_status, '{}'::JSONB), \n                    array[$2],\n                    $3\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "2e6935811a6d818bc523f076674f794f8be6c6bad3d06e74586e8ab668d91861"
+}

backend/.sqlx/query-99f74bf675120daf965e063e5eaff808ba646f4f99d0c8837097e747b481f03a.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id FROM v2_job WHERE parent_job = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "99f74bf675120daf965e063e5eaff808ba646f4f99d0c8837097e747b481f03a"
+}

backend/tests/worker.rs

@@ -3875,6 +3875,7 @@ async fn test_workflow_as_code(db: Pool<Postgres>) {
             .await;
 
             assert_eq!(job.json_result().unwrap(), json!(["OK", 3]));
+
             let workflow_as_code_status = sqlx::query_scalar!(
                 "SELECT workflow_as_code_status FROM v2_job_completed WHERE id = $1",
                 job.id
@@ -3883,10 +3884,33 @@ async fn test_workflow_as_code(db: Pool<Postgres>) {
             .await
             .unwrap()
             .unwrap();
-            assert_eq!(
-                workflow_as_code_status.get("name"),
-                Some(&json!("send_result"))
-            );
+
+            #[derive(Deserialize)]
+            #[allow(dead_code)]
+            struct WorkflowJobStatus {
+                name: String,
+                started_at: String,
+                scheduled_for: String,
+                duration_ms: i64,
+            }
+
+            let workflow_as_code_status: std::collections::HashMap<String, WorkflowJobStatus> =
+                serde_json::from_value(workflow_as_code_status).unwrap();
+
+            let uuids = sqlx::query_scalar!("SELECT id FROM v2_job WHERE parent_job = $1", job.id)
+                .fetch_all(db)
+                .await
+                .unwrap();
+
+            assert_eq!(uuids.len(), 4);
+            for uuid in uuids {
+                let status = workflow_as_code_status.get(&uuid.to_string());
+                assert!(status.is_some());
+                assert!(
+                    status.unwrap().name == "send_result"
+                        || status.unwrap().name == "heavy_compute"
+                );
+            }
         },
         port,
     )

backend/windmill-api/src/jobs.rs

@@ -3427,8 +3427,12 @@ pub async fn run_workflow_as_code(
         sqlx::query!(
             "INSERT INTO v2_job_status (id, workflow_as_code_status)
             VALUES ($1, JSONB_SET('{}'::JSONB, array[$2], $3))
-            ON CONFLICT (id) DO UPDATE SET workflow_as_code_status =
-                COALESCE(EXCLUDED.workflow_as_code_status, '{}'::JSONB) || $3",
+            ON CONFLICT (id) DO UPDATE SET
+                workflow_as_code_status = JSONB_SET(
+                    COALESCE(v2_job_status.workflow_as_code_status, '{}'::JSONB), 
+                    array[$2],
+                    $3
+                )",
             job_id,
             uuid.to_string(),
             serde_json::json!({ "scheduled_for": Utc::now(), "name": entrypoint }),