Remove support for string-encoded dataclips in step:complete

CHANGELOG.md

@@ -22,6 +22,9 @@ and this project adheres to
   button), closes any other open panel, deselects the current node, and drops
   any run-viewing context, landing on the bare canvas.
   [#4984](https://github.com/OpenFn/lightning/pull/4984)
+- `step:complete` now accepts `output_dataclip` as a decoded value, not just a
+  JSON string, while staying compatible with workers that still send a string.
+  [#5098](https://github.com/OpenFn/lightning/pull/5098)
 
 ### Changed
 

lib/lightning/runs/handlers.ex

@@ -405,6 +405,10 @@ defmodule Lightning.Runs.Handlers do
   defmodule CompleteStep do
     @moduledoc """
     Schema to validate the input attributes of a completed step.
+
+    `output_dataclip` arrives already decoded (a map, list, or scalar) rather
+    than as a JSON-encoded string. Map values are stored as-is; non-map
+    values are wrapped as `%{"value" => x}` before persistence.
     """
     use Lightning.Schema
     import Ecto.Query
@@ -415,7 +419,7 @@ defmodule Lightning.Runs.Handlers do
     embedded_schema do
       field :project_id, Ecto.UUID
       field :run_id, Ecto.UUID
-      field :output_dataclip, :string
+      field :output_dataclip, :any, virtual: true
       field :output_dataclip_id, Ecto.UUID
       field :reason, :string
       field :error_type, :string
@@ -550,7 +554,7 @@ defmodule Lightning.Runs.Handlers do
       Dataclip.new(%{
         id: dataclip_id,
         project_id: project_id,
-        body: output_dataclip |> Jason.decode!() |> ensure_map(),
+        body: output_dataclip |> ensure_map(),
         type: :step_result
       })
       |> Repo.insert()

lib/lightning/setup_utils.ex

@@ -508,7 +508,7 @@ defmodule Lightning.SetupUtils do
           [CLI] ✔ Done in 223ms! ✨
           """),
         input_dataclip_id: dataclip.id,
-        output_dataclip: %{data: http_body, references: []} |> Jason.encode!()
+        output_dataclip: %{"data" => http_body, "references" => []}
       },
       %{
         job_id: send_to_openhim.id,
@@ -532,7 +532,7 @@ defmodule Lightning.SetupUtils do
           [CLI] ✔ Writing output to /tmp/output-1686840746-126941-i2yb2g.json
           [CLI] ✔ Done in 223ms! ✨
           """),
-        output_dataclip: %{data: http_body, references: []} |> Jason.encode!()
+        output_dataclip: %{"data" => http_body, "references" => []}
       },
       %{
         job_id: notify_upload_successful.id,
@@ -556,7 +556,7 @@ defmodule Lightning.SetupUtils do
           [CLI] ✔ Writing output to /tmp/output-1686840747-126941-16ewhef.json
           [CLI] ✔ Done in 209ms! ✨
           """),
-        output_dataclip: %{data: http_body, references: []} |> Jason.encode!()
+        output_dataclip: %{"data" => http_body, "references" => []}
       }
     ]
 
@@ -733,20 +733,18 @@ defmodule Lightning.SetupUtils do
             [CMP] ℹ Added export * statement for @openfn/language-dhis2@latest
           """),
         input_dataclip_id: input_dataclip.id,
-        output_dataclip:
-          %{
-            data: %{
-              spreadsheetId: "wv5ftwhte",
-              tableRange: "A3:D3",
-              updates: %{
-                updatedCells: 4
-              }
-            },
-            references: [
-              %{}
-            ]
-          }
-          |> Jason.encode!()
+        output_dataclip: %{
+          "data" => %{
+            "spreadsheetId" => "wv5ftwhte",
+            "tableRange" => "A3:D3",
+            "updates" => %{
+              "updatedCells" => 4
+            }
+          },
+          "references" => [
+            %{}
+          ]
+        }
       },
       %{
         job_id: upload_to_google_sheet.id,

test/lightning/runs_test.exs

@@ -412,7 +412,7 @@ defmodule Lightning.RunsTest do
         Runs.complete_step(%{
           step_id: step.id,
           reason: "success",
-          output_dataclip: ~s({"foo": "bar"}),
+          output_dataclip: %{"foo" => "bar"},
           output_dataclip_id: Ecto.UUID.generate(),
           run_id: run.id,
           project_id: workflow.project_id
@@ -426,6 +426,35 @@ defmodule Lightning.RunsTest do
       assert Jason.decode!(step.output_dataclip.body) == %{"foo" => "bar"}
     end
 
+    test "wraps a scalar output_dataclip in %{\"value\" => x}" do
+      dataclip = insert(:dataclip)
+      %{triggers: [trigger], jobs: [job]} = workflow = insert(:simple_workflow)
+
+      %{runs: [run]} =
+        work_order_for(trigger, workflow: workflow, dataclip: dataclip)
+        |> insert()
+
+      step =
+        insert(:step, runs: [run], job: job, input_dataclip: dataclip)
+
+      {:ok, step} =
+        Runs.complete_step(%{
+          step_id: step.id,
+          reason: "success",
+          output_dataclip: 42,
+          output_dataclip_id: Ecto.UUID.generate(),
+          run_id: run.id,
+          project_id: workflow.project_id
+        })
+
+      step =
+        step
+        |> Repo.preload(output_dataclip: Invocation.Query.dataclip_with_body())
+
+      assert step.exit_reason == "success"
+      assert Jason.decode!(step.output_dataclip.body) == %{"value" => 42}
+    end
+
     # Regression for #4800: dataclip inserts no longer build the search_vector
     # synchronously (the AFTER INSERT trigger was dropped). Saving an output
     # dataclip via the handler must succeed and the row must be retrievable with
@@ -446,7 +475,7 @@ defmodule Lightning.RunsTest do
                Runs.complete_step(%{
                  step_id: step.id,
                  reason: "success",
-                 output_dataclip: ~s({"deferred": "indexword"}),
+                 output_dataclip: %{"deferred" => "indexword"},
                  output_dataclip_id: output_dataclip_id,
                  run_id: run.id,
                  project_id: workflow.project_id
@@ -491,7 +520,7 @@ defmodule Lightning.RunsTest do
           %{
             step_id: step.id,
             reason: "success",
-            output_dataclip: ~s({"foo": "bar"}),
+            output_dataclip: %{"foo" => "bar"},
             output_dataclip_id: Ecto.UUID.generate(),
             run_id: run.id,
             project_id: workflow.project_id
@@ -538,7 +567,7 @@ defmodule Lightning.RunsTest do
                Runs.complete_step(%{
                  step_id: Ecto.UUID.generate(),
                  reason: "success",
-                 output_dataclip: ~s({"foo": "bar"}),
+                 output_dataclip: %{"foo" => "bar"},
                  output_dataclip_id: Ecto.UUID.generate(),
                  run_id: run.id,
                  project_id: workflow.project_id

test/lightning_web/channels/run_channel_test.exs

@@ -976,7 +976,7 @@ defmodule LightningWeb.RunChannelTest do
         push(socket, "step:complete", %{
           "step_id" => step.id,
           "output_dataclip_id" => Ecto.UUID.generate(),
-          "output_dataclip" => ~s({"foo": "bar"}),
+          "output_dataclip" => %{"foo" => "bar"},
           "reason" => "normal",
           "timestamp" => to_string(timestamp)
         })
@@ -1001,7 +1001,7 @@ defmodule LightningWeb.RunChannelTest do
         push(socket, "step:complete", %{
           "step_id" => step.id,
           "output_dataclip_id" => Ecto.UUID.generate(),
-          "output_dataclip" => ~s({"foo": "bar"}),
+          "output_dataclip" => %{"foo" => "bar"},
           "reason" => "normal"
         })
 
@@ -1021,7 +1021,7 @@ defmodule LightningWeb.RunChannelTest do
         push(socket, "step:complete", %{
           "step_id" => step.id,
           "output_dataclip_id" => Ecto.UUID.generate(),
-          "output_dataclip" => ~s({"foo": "bar"}),
+          "output_dataclip" => %{"foo" => "bar"},
           "reason" => "fail"
         })
 
@@ -1043,7 +1043,7 @@ defmodule LightningWeb.RunChannelTest do
         push(socket, "step:complete", %{
           "step_id" => step_id,
           "output_dataclip_id" => dataclip_id,
-          "output_dataclip" => ~s({"foo": "bar"}),
+          "output_dataclip" => %{"foo" => "bar"},
           "reason" => "normal"
         })
 
@@ -1078,7 +1078,7 @@ defmodule LightningWeb.RunChannelTest do
         push(socket, "step:complete", %{
           "step_id" => step_id,
           "output_dataclip_id" => dataclip_id,
-          "output_dataclip" => ~s({"foo": "bar"}),
+          "output_dataclip" => %{"foo" => "bar"},
           "reason" => "normal"
         })
 
@@ -1105,7 +1105,7 @@ defmodule LightningWeb.RunChannelTest do
       ref =
         push(socket, "step:complete", %{
           "step_id" => step_id,
-          "output_dataclip" => ~s({"foo": "bar"}),
+          "output_dataclip" => %{"foo" => "bar"},
           "reason" => "normal"
         })
 
@@ -1124,7 +1124,7 @@ defmodule LightningWeb.RunChannelTest do
         push(socket, "step:complete", %{
           "step_id" => foreign_step.id,
           "output_dataclip_id" => output_dataclip_id,
-          "output_dataclip" => ~s({"leaked": "data"}),
+          "output_dataclip" => %{"leaked" => "data"},
           "reason" => "fail"
         })
 
@@ -2446,7 +2446,7 @@ defmodule LightningWeb.RunChannelTest do
       %{
         "step_id" => step.id,
         "output_dataclip_id" => Ecto.UUID.generate(),
-        "output_dataclip" => ~s({"foo": "bar"}),
+        "output_dataclip" => %{"foo" => "bar"},
         "reason" => "normal"
       }
       |> maybe_put("webhook_response", Keyword.get(opts, :webhook_response))
@@ -2956,7 +2956,7 @@ defmodule LightningWeb.RunChannelTest do
         Lightning.Runs.complete_step(
           %{
             "step_id" => step.id,
-            "output_dataclip" => Jason.encode!(%{"foo" => "bar"}),
+            "output_dataclip" => %{"foo" => "bar"},
             "output_dataclip_id" => Ecto.UUID.generate(),
             "reason" => "success",
             "finished_at" => DateTime.utc_now(),

test/lightning_web/live/run_live/show_test.exs

@@ -163,7 +163,7 @@ defmodule LightningWeb.RunLive.ShowTest do
           run_id: run_id,
           project_id: project.id,
           step_id: step.id,
-          output_dataclip: ~s({"y": 2}),
+          output_dataclip: %{"y" => 2},
           output_dataclip_id: output_dataclip_id = Ecto.UUID.generate(),
           reason: "success"
         })
@@ -207,7 +207,7 @@ defmodule LightningWeb.RunLive.ShowTest do
           run_id: run_id,
           project_id: project.id,
           step_id: step_2.id,
-          output_dataclip: ~s({"z": 2}),
+          output_dataclip: %{"z" => 2},
           output_dataclip_id: step_2_output_dataclip_id = Ecto.UUID.generate(),
           reason: "success"
         })
@@ -281,7 +281,7 @@ defmodule LightningWeb.RunLive.ShowTest do
           run_id: run_id,
           project_id: project.id,
           step_id: step.id,
-          output_dataclip: ~s({"result": 42}),
+          output_dataclip: %{"result" => 42},
           output_dataclip_id: output_dataclip_id = Ecto.UUID.generate(),
           reason: "success"
         })
@@ -465,7 +465,7 @@ defmodule LightningWeb.RunLive.ShowTest do
         run_id: run_id,
         project_id: project.id,
         step_id: step.id,
-        output_dataclip: ~s({"y": 2}),
+        output_dataclip: %{"y" => 2},
         output_dataclip_id: Ecto.UUID.generate(),
         reason: "success"
       })