diff --git a/CHANGELOG.md b/CHANGELOG.md index 672f70f5a30..b5bb42c9a0a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/lib/lightning/runs/handlers.ex b/lib/lightning/runs/handlers.ex index ebb4c526cd8..712a64a0a03 100644 --- a/lib/lightning/runs/handlers.ex +++ b/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() diff --git a/lib/lightning/setup_utils.ex b/lib/lightning/setup_utils.ex index 6c817c1b70b..28f3f3f42eb 100644 --- a/lib/lightning/setup_utils.ex +++ b/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, diff --git a/test/lightning/runs_test.exs b/test/lightning/runs_test.exs index 3fac793b4a2..1ae745e675d 100644 --- a/test/lightning/runs_test.exs +++ b/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 diff --git a/test/lightning_web/channels/run_channel_test.exs b/test/lightning_web/channels/run_channel_test.exs index 316fd759fad..36e68028151 100644 --- a/test/lightning_web/channels/run_channel_test.exs +++ b/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(), diff --git a/test/lightning_web/live/run_live/show_test.exs b/test/lightning_web/live/run_live/show_test.exs index 1d3527874c0..8d667043103 100644 --- a/test/lightning_web/live/run_live/show_test.exs +++ b/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" })