+ <.modal id={@id} width="xl:min-w-1/3 min-w-1/2 max-w-full">
+ <:title>
+
+
+ <%= if @mode == :create,
+ do: "Create Collection",
+ else: "Edit Collection" %>
+
+
+ Close
+
+
+
+ <.form
+ :let={f}
+ for={@changeset}
+ id={"collection-form-#{@collection.id || "new"}"}
+ phx-target={@myself}
+ phx-change="validate"
+ phx-submit="save"
+ >
+
+
+ <.input
+ type="text"
+ field={f[:raw_name]}
+ value={@name}
+ label="Name"
+ required="true"
+ />
+ <.input type="hidden" field={f[:name]} />
+
+ <%= if to_string(f[:name].value) != "" do %>
+ Your collection will be named
+ <%= @name %> .
+ <% end %>
+
+
+
+ <.input
+ type="select"
+ field={f[:project_id]}
+ label="Project"
+ options={@projects_options}
+ required="true"
+ />
+
+
+ <.modal_footer class="mt-6 mx-6">
+
+
+ Save
+
+
+ Cancel
+
+
+
+
+
+
+ <.button
+ id={"#{@id}_confirm_button"}
+ type="button"
+ phx-click="delete_collection"
+ phx-value-collection={@collection.id}
+ color_class="bg-red-600 hover:bg-red-700 text-white"
+ phx-disable-with="Deleting..."
+ >
+ Delete
+
+
+ Cancel
+
+
+
+ """
+ end
+
+ defp table_title(assigns) do
+ ~H"""
+
+ <.button
+ id={"remove_project_user_#{project_user.id}_button"}
+ type="button"
+ phx-click={show_modal("remove_#{project_user.id}_modal")}
+ color_class="bg-white text-gray-900 hover:bg-gray-50 disabled:bg-gray-100"
+ class="gap-x-2 rounded-md px-3.5 py-2.5 text-sm shadow-sm ring-1 ring-inset ring-gray-300 disabled:cursor-not-allowed"
+ tooltip={
+ remove_user_tooltip(
+ project_user,
+ @current_user,
+ @can_remove_project_user
+ )
+ }
+ disabled={
+ !user_removable?(
+ project_user,
+ @current_user,
+ @can_remove_project_user
+ )
+ }
+ >
+ <.icon name="hero-minus-circle" class="w-5 h-5" />
+ Remove Collaborator
+
+
<.confirm_user_removal_modal
:if={
user_removable?(
diff --git a/lib/lightning_web/plugs/api_auth.ex b/lib/lightning_web/plugs/api_auth.ex
new file mode 100644
index 0000000000..b37b2d7eed
--- /dev/null
+++ b/lib/lightning_web/plugs/api_auth.ex
@@ -0,0 +1,48 @@
+defmodule LightningWeb.Plugs.ApiAuth do
+ @moduledoc """
+ Authenticates api calls based on JWT bearer token.
+ """
+ use Phoenix.Controller
+ import Plug.Conn
+
+ def init(opts) do
+ opts
+ end
+
+ def call(conn, _opts) do
+ with {:ok, bearer_token} <- get_bearer_token(conn),
+ {:ok, claims} <- Lightning.Tokens.verify(bearer_token) do
+ conn
+ |> assign(:claims, claims)
+ |> put_subject()
+ else
+ {:error, _reason} ->
+ deny_access(conn)
+ end
+ end
+
+ defp get_bearer_token(conn) do
+ conn
+ |> get_req_header("authorization")
+ |> case do
+ ["Bearer " <> bearer] -> {:ok, bearer}
+ _none_or_many -> {:error, :token_not_found}
+ end
+ end
+
+ defp put_subject(conn) do
+ conn.assigns.claims
+ |> Lightning.Tokens.get_subject()
+ |> then(fn subject ->
+ conn |> assign(:subject, subject)
+ end)
+ end
+
+ defp deny_access(conn) do
+ conn
+ |> put_status(:unauthorized)
+ |> put_view(LightningWeb.ErrorView)
+ |> render(:"401")
+ |> halt()
+ end
+end
diff --git a/lib/lightning_web/route_helpers.ex b/lib/lightning_web/route_helpers.ex
index f255c3e901..33df348577 100644
--- a/lib/lightning_web/route_helpers.ex
+++ b/lib/lightning_web/route_helpers.ex
@@ -4,15 +4,6 @@ defmodule LightningWeb.RouteHelpers do
"""
alias LightningWeb.Router.Helpers, as: Routes
- def show_run_url(project_id, run_id) do
- Routes.project_run_show_url(
- LightningWeb.Endpoint,
- :show,
- project_id,
- run_id
- )
- end
-
def project_dashboard_url(project_id) do
Routes.project_workflow_index_url(
LightningWeb.Endpoint,
diff --git a/lib/lightning_web/router.ex b/lib/lightning_web/router.ex
index 84b25ab718..50587825f9 100644
--- a/lib/lightning_web/router.ex
+++ b/lib/lightning_web/router.ex
@@ -42,6 +42,11 @@ defmodule LightningWeb.Router do
plug :accepts, ["json"]
end
+ pipeline :authenticated_api do
+ plug :accepts, ["json"]
+ plug LightningWeb.Plugs.ApiAuth
+ end
+
scope "/", LightningWeb do
pipe_through [:browser]
@@ -84,6 +89,18 @@ defmodule LightningWeb.Router do
# resources "/runs", API.RunController, only: [:index, :show]
end
+ ## Collections
+ scope "/collections", LightningWeb do
+ pipe_through [:authenticated_api]
+
+ get "/:name", CollectionsController, :stream
+ get "/:name/:key", CollectionsController, :get
+ put "/:name/:key", CollectionsController, :put
+ post "/:name", CollectionsController, :put_all
+ delete "/:name/:key", CollectionsController, :delete
+ delete "/:name", CollectionsController, :delete_all
+ end
+
## Authentication routes
scope "/", LightningWeb do
@@ -154,6 +171,8 @@ defmodule LightningWeb.Router do
live "/settings/authentication", AuthProvidersLive.Index, :edit
live "/settings/authentication/new", AuthProvidersLive.Index, :new
+
+ live "/settings/collections", CollectionLive.Index, :index
end
live_session :default, on_mount: LightningWeb.InitAssigns do
diff --git a/lib/lightning_web/views/error_view.ex b/lib/lightning_web/views/error_view.ex
index 3db6b9f58c..e050068202 100644
--- a/lib/lightning_web/views/error_view.ex
+++ b/lib/lightning_web/views/error_view.ex
@@ -1,4 +1,6 @@
defmodule LightningWeb.ErrorView do
+ @moduledoc false
+
# This module needs to be changed to use Layouts
use LightningWeb, :view
@@ -31,7 +33,10 @@ defmodule LightningWeb.ErrorView do
+
<%= for {k,v} <- @error do %>
<%= k %>
<%= v %>
@@ -43,10 +48,6 @@ defmodule LightningWeb.ErrorView do
"""
end
- def render("403.json", %{error: error}) do
- %{"error" => error}
- end
-
defp logo_bar(assigns) do
~H"""
@@ -70,10 +71,14 @@ defmodule LightningWeb.ErrorView do
# By default, Phoenix returns the status message from
# the template name. For example, "404.html" becomes
# "Not Found".
- def template_not_found(template, _assigns) do
+ def template_not_found(template, assigns) do
if String.match?(template, ~r/.json$/) do
%{
- "error" => Phoenix.Controller.status_message_from_template(template)
+ "error" =>
+ case assigns do
+ %{error: error} -> error
+ _ -> Phoenix.Controller.status_message_from_template(template)
+ end
}
else
Phoenix.Controller.status_message_from_template(template)
diff --git a/lib/mix/tasks/gen_worker_keys.ex b/lib/mix/tasks/gen_worker_keys.ex
index 1c2a1edec6..59362ed180 100644
--- a/lib/mix/tasks/gen_worker_keys.ex
+++ b/lib/mix/tasks/gen_worker_keys.ex
@@ -6,6 +6,8 @@ defmodule Mix.Tasks.Lightning.GenWorkerKeys do
use Mix.Task
+ alias Lightning.Utils
+
@footer """
To use these keys, use the above output to set the environment variables.
@@ -25,70 +27,21 @@ defmodule Mix.Tasks.Lightning.GenWorkerKeys do
# looks like we may need "try" with this "with"
# https://hexdocs.pm/credo/Credo.Check.Readability.PreferImplicitTry.html
# credo:disable-for-next-line
- try do
- with {:ok, private_key} <- create_private_key(),
- {:ok, public_key} <- abstract_public_key(private_key) do
- IO.puts("""
- WORKER_RUNS_PRIVATE_KEY="#{private_key |> Base.encode64(padding: false)}"
-
- WORKER_SECRET="#{generate_hs256_key()}"
+ {private_key, public_key} = Utils.Crypto.generate_rsa_key_pair()
- WORKER_LIGHTNING_PUBLIC_KEY="#{public_key |> Base.encode64(padding: false)}"
+ IO.puts("""
+ WORKER_RUNS_PRIVATE_KEY="#{private_key |> Base.encode64(padding: false)}"
+ WORKER_SECRET="#{Utils.Crypto.generate_hs256_key()}"
- #{@footer}
- """)
- end
- rescue
- e ->
- case e do
- %{original: :enoent} ->
- IO.puts("openssl not found in PATH")
+ WORKER_LIGHTNING_PUBLIC_KEY="#{public_key |> Base.encode64(padding: false)}"
- e ->
- IO.puts("Error: #{inspect(e)}")
- end
-
- exit({:shutdown, 1})
- end
- end
-
- defp call_openssl(args) do
- System.cmd("openssl", args, stderr_to_stdout: true)
- |> case do
- {_, 0} ->
- :ok
-
- {stdout, status} ->
- {:error, status, stdout}
- end
- end
-
- defp create_private_key do
- filename = Path.join(System.tmp_dir!(), "jwtRSA256-private.pem")
-
- with :ok <- call_openssl(~w[genrsa -out #{filename} 2048]),
- {:ok, contents} <- File.read(filename),
- :ok <- File.rm(filename) do
- {:ok, contents}
- end
- end
-
- defp abstract_public_key(private_key) do
- filename = Path.join(System.tmp_dir!(), "jwtRSA256.pem")
-
- with :ok <- File.write(filename, private_key),
- :ok <-
- call_openssl(
- ~w[rsa -in #{filename} -pubout -outform PEM -out #{filename}]
- ),
- {:ok, contents} <- File.read(filename),
- :ok <- File.rm(filename) do
- {:ok, contents}
- end
- end
- defp generate_hs256_key do
- 32 |> :crypto.strong_rand_bytes() |> Base.encode64()
+ #{@footer}
+ """)
+ rescue
+ e ->
+ IO.puts("Error: #{inspect(e)}")
+ exit({:shutdown, 1})
end
end
diff --git a/mix.exs b/mix.exs
index c5a2129bac..2bfaf785f9 100644
--- a/mix.exs
+++ b/mix.exs
@@ -150,7 +150,8 @@ defmodule Lightning.MixProject do
# Github API Secret Encoding
{:enacl, github: "aeternity/enacl", branch: "master"},
{:earmark, "~> 1.4"},
- {:eventually, "~> 1.1", only: [:test]}
+ {:eventually, "~> 1.1", only: [:test]},
+ {:benchee, "~> 1.3.1", only: :dev}
]
end
diff --git a/mix.lock b/mix.lock
index e4b2dd4d72..15d31ce9cf 100644
--- a/mix.lock
+++ b/mix.lock
@@ -1,6 +1,7 @@
%{
"acceptor_pool": {:hex, :acceptor_pool, "1.0.0", "43c20d2acae35f0c2bcd64f9d2bde267e459f0f3fd23dab26485bf518c281b21", [:rebar3], [], "hexpm", "0cbcd83fdc8b9ad2eee2067ef8b91a14858a5883cb7cd800e6fcd5803e158788"},
"bcrypt_elixir": {:hex, :bcrypt_elixir, "2.3.1", "5114d780459a04f2b4aeef52307de23de961b69e13a5cd98a911e39fda13f420", [:make, :mix], [{:comeonin, "~> 5.3", [hex: :comeonin, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "42182d5f46764def15bf9af83739e3bf4ad22661b1c34fc3e88558efced07279"},
+ "benchee": {:hex, :benchee, "1.3.1", "c786e6a76321121a44229dde3988fc772bca73ea75170a73fd5f4ddf1af95ccf", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}, {:statistex, "~> 1.0", [hex: :statistex, repo: "hexpm", optional: false]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "76224c58ea1d0391c8309a8ecbfe27d71062878f59bd41a390266bf4ac1cc56d"},
"bodyguard": {:hex, :bodyguard, "2.4.2", "e91ecd43fae949f5dca7d5054bedf612b5afd6fb4a84376c9909d14e69ab15a7", [:mix], [{:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "4683ae363d98dee29aac30bda14af51287898f49880fb25ea961227d2dbf6f27"},
"briefly": {:hex, :briefly, "0.5.1", "ee10d48da7f79ed2aebdc3e536d5f9a0c3e36ff76c0ad0d4254653a152b13a8a", [:mix], [], "hexpm", "bd684aa92ad8b7b4e0d92c31200993c4bc1469fc68cd6d5f15144041bd15cb57"},
"broadway": {:hex, :broadway, "1.1.0", "8ed3aea01fd6f5640b3e1515b90eca51c4fc1fac15fb954cdcf75dc054ae719c", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "25e315ef1afe823129485d981dcc6d9b221cea30e625fd5439e9b05f44fb60e4"},
@@ -27,6 +28,7 @@
"ctx": {:hex, :ctx, "0.6.0", "8ff88b70e6400c4df90142e7f130625b82086077a45364a78d208ed3ed53c7fe", [:rebar3], [], "hexpm", "a14ed2d1b67723dbebbe423b28d7615eb0bdcba6ff28f2d1f1b0a7e1d4aa5fc2"},
"db_connection": {:hex, :db_connection, "2.7.0", "b99faa9291bb09892c7da373bb82cba59aefa9b36300f6145c5f201c7adf48ec", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "dcf08f31b2701f857dfc787fbad78223d61a32204f217f15e881dd93e4bdd3ff"},
"decimal": {:hex, :decimal, "2.1.1", "5611dca5d4b2c3dd497dec8f68751f1f1a54755e8ed2a966c2633cf885973ad6", [:mix], [], "hexpm", "53cfe5f497ed0e7771ae1a475575603d77425099ba5faef9394932b35020ffcc"},
+ "deep_merge": {:hex, :deep_merge, "1.0.0", "b4aa1a0d1acac393bdf38b2291af38cb1d4a52806cf7a4906f718e1feb5ee961", [:mix], [], "hexpm", "ce708e5f094b9cd4e8f2be4f00d2f4250c4095be93f8cd6d018c753894885430"},
"dialyxir": {:hex, :dialyxir, "1.4.4", "fb3ce8741edeaea59c9ae84d5cec75da00fa89fe401c72d6e047d11a61f65f70", [:mix], [{:erlex, ">= 0.2.7", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "cd6111e8017ccd563e65621a4d9a4a1c5cd333df30cebc7face8029cacb4eff6"},
"dotenvy": {:hex, :dotenvy, "0.8.0", "777486ad485668317c56afc53a7cbcd74f43e4e34588ba8e95a73e15a360050e", [:mix], [], "hexpm", "1f535066282388cbd109743d337ac46ff0708195780d4b5778bb83491ab1b654"},
"earmark": {:hex, :earmark, "1.4.46", "8c7287bd3137e99d26ae4643e5b7ef2129a260e3dcf41f251750cb4563c8fb81", [:mix], [], "hexpm", "798d86db3d79964e759ddc0c077d5eb254968ed426399fbf5a62de2b5ff8910a"},
@@ -143,6 +145,7 @@
"snappyer": {:hex, :snappyer, "1.2.9", "9cc58470798648ce34c662ca0aa6daae31367667714c9a543384430a3586e5d3", [:rebar3], [], "hexpm", "18d00ca218ae613416e6eecafe1078db86342a66f86277bd45c95f05bf1c8b29"},
"sobelow": {:hex, :sobelow, "0.13.0", "218afe9075904793f5c64b8837cc356e493d88fddde126a463839351870b8d1e", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "cd6e9026b85fc35d7529da14f95e85a078d9dd1907a9097b3ba6ac7ebbe34a0d"},
"ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
+ "statistex": {:hex, :statistex, "1.0.0", "f3dc93f3c0c6c92e5f291704cf62b99b553253d7969e9a5fa713e5481cd858a5", [:mix], [], "hexpm", "ff9d8bee7035028ab4742ff52fc80a2aa35cece833cf5319009b52f1b5a86c27"},
"sweet_xml": {:hex, :sweet_xml, "0.7.4", "a8b7e1ce7ecd775c7e8a65d501bc2cd933bff3a9c41ab763f5105688ef485d08", [:mix], [], "hexpm", "e7c4b0bdbf460c928234951def54fe87edf1a170f6896675443279e2dbeba167"},
"swoosh": {:hex, :swoosh, "1.12.0", "ecc85ee12947932986243299b8d28e6cdfc192c8d9e24c4c64f6738efdf344cb", [:mix], [{:cowboy, "~> 1.1 or ~> 2.4", [hex: :cowboy, repo: "hexpm", optional: true]}, {:ex_aws, "~> 2.1", [hex: :ex_aws, repo: "hexpm", optional: true]}, {:finch, "~> 0.6", [hex: :finch, repo: "hexpm", optional: true]}, {:gen_smtp, "~> 0.13 or ~> 1.0", [hex: :gen_smtp, repo: "hexpm", optional: true]}, {:hackney, "~> 1.9", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mail, "~> 0.2", [hex: :mail, repo: "hexpm", optional: true]}, {:mime, "~> 1.1 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 1.0.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "87db7ab0f35e358ba5eac3afc7422ed0c8c168a2d219d2a83ad8cb7a424f6cc9"},
"table_rex": {:hex, :table_rex, "3.1.1", "0c67164d1714b5e806d5067c1e96ff098ba7ae79413cc075973e17c38a587caa", [:mix], [], "hexpm", "678a23aba4d670419c23c17790f9dcd635a4a89022040df7d5d772cb21012490"},
diff --git a/priv/bench/collections.exs b/priv/bench/collections.exs
new file mode 100644
index 0000000000..3a8853f581
--- /dev/null
+++ b/priv/bench/collections.exs
@@ -0,0 +1,169 @@
+alias Lightning.Repo
+alias Lightning.Collections
+alias Lightning.Projects
+
+keys_count = 5_000
+
+Repo.delete_all(Collections.Collection)
+
+project =
+ with nil <- Repo.get_by(Projects.Project, name: "benchee") do
+ user = Repo.get_by(Lightning.Accounts.User, email: "demo@openfn.org")
+ {:ok, project} = Projects.create_project(%{name: "benchee", project_users: [%{user_id: user.id, role: :owner}]})
+ project
+ end
+
+{:ok, collection} =
+ with {:error, :not_found} <- Collections.get_collection("benchee") do
+ Collections.create_collection(project.id, "benchee")
+ end
+
+IO.puts("\n### Setup:")
+IO.puts("Generating items for benchee collection...")
+
+record = fn prefix, i ->
+ i_str = String.pad_leading(to_string(i), 5, "0")
+ {"#{prefix}:foo#{i_str}:bar#{i_str}", Jason.encode!(%{fieldA: "value#{1_000_000 + i}"})}
+end
+
+sampleA = Enum.map(1..keys_count, fn i -> record.("keyA", i) end)
+sampleB = Enum.map(1..keys_count, fn i -> record.("keyB", i) end)
+{sampleC1, sampleC2} = Enum.map(1..keys_count * 2, fn i -> record.("keyC", i) end) |> Enum.split(keys_count)
+
+:timer.tc(fn ->
+ [sampleA, sampleB, sampleC1]
+ |> Enum.map(fn sample ->
+ Task.async(fn ->
+ Enum.with_index(sample, fn {key, value}, idx ->
+ if rem(idx, 50) == 0, do: IO.puts("Inserting " <> key)
+ :ok = Collections.put(collection, key, value)
+ end)
+ :ok
+ end)
+ end)
+ |> Task.await_many(:infinity)
+end)
+|> tap(fn {duration, _res} ->
+ IO.puts("Inserted 3 x #{keys_count} shuffled items (w/ unsorted keys).")
+ IO.puts("elapsed time: #{div(duration, 1_000)}ms\n")
+end)
+
+IO.puts("Inserting #{length(sampleC2)} items with put_all...")
+:timer.tc(fn ->
+ :ok = Collections.put_all(collection, sampleC2)
+end)
+|> tap(fn {duration, _res} ->
+ IO.puts("elapsed time: #{div(duration, 1_000)}ms\n")
+end)
+
+sampleD = Enum.map(1..keys_count, fn i -> record.("keyD", i) end)
+
+IO.puts("Inserting sampleD (w/ sorted keys)...")
+:timer.tc(fn ->
+ sampleD
+ |> Enum.chunk_every(1000)
+ |> Enum.map(fn sample ->
+ Task.async(fn ->
+ Enum.each(sample, fn {k, v} -> Collections.put(collection, k, v) end)
+ end)
+ end)
+ |> Task.await_many(:infinity)
+end)
+|> tap(fn {duration, _res} ->
+ IO.puts("elapsed time: #{div(duration, 1000)}ms\n")
+end)
+
+IO.puts("Upserting sampleD...")
+:timer.tc(fn ->
+ sampleD
+ |> Enum.chunk_every(1000)
+ |> Enum.map(fn sample ->
+ Task.async(fn ->
+ Enum.each(sample, fn {k, v} -> Collections.put(collection, k, v) end)
+ end)
+ end)
+ |> Task.await_many(:infinity)
+end)
+|> tap(fn {duration, _res} ->
+ IO.puts("elapsed time: #{div(duration, 1000)}ms\n")
+end)
+
+stream_all =
+ fn ->
+ Stream.unfold(nil, fn cursor ->
+ case Repo.transaction(fn -> Collections.stream_all(collection, cursor) |> Enum.to_list() end) do
+ {:ok, []} -> nil
+ {:ok, list} -> {list, List.last(list).updated_at}
+ end
+ end)
+ |> Enum.to_list()
+ |> List.flatten()
+ end
+
+stream_match_all =
+ fn ->
+ Stream.unfold(nil, fn cursor ->
+ case Repo.transaction(fn -> Collections.stream_match(collection, "key*", cursor) |> Enum.to_list() end) do
+ {:ok, []} -> nil
+ {:ok, list} -> {list, List.last(list).updated_at}
+ end
+ end)
+ |> Enum.to_list()
+ |> List.flatten()
+ end
+
+stream_match_prefix =
+ fn ->
+ Stream.unfold(nil, fn cursor ->
+ case Repo.transaction(fn -> Collections.stream_match(collection, "keyA*", cursor) |> Enum.to_list() end) do
+ {:ok, []} -> nil
+ {:ok, list} -> {list, List.last(list).updated_at}
+ end
+ end)
+ |> Enum.to_list()
+ |> List.flatten()
+ end
+
+stream_match_trigram =
+ fn ->
+ Stream.unfold(nil, fn cursor ->
+ case Repo.transaction(fn -> Collections.stream_match(collection, "keyB*bar*", cursor) |> Enum.to_list() end) do
+ {:ok, []} -> nil
+ {:ok, list} -> {list, List.last(list).updated_at}
+ end
+ end)
+ |> Enum.to_list()
+ |> List.flatten()
+ end
+
+
+IO.puts("\n### Round record count ({microsecs, count}):")
+:timer.tc(fn -> stream_all.() |> Enum.count() end) |> IO.inspect(label: "stream_all")
+:timer.tc(fn -> stream_match_all.() |> Enum.count() end) |> IO.inspect(label: "stream_match_all")
+:timer.tc(fn -> stream_match_prefix.() |> Enum.count() end) |> IO.inspect(label: "stream_match_prefix")
+:timer.tc(fn -> stream_match_trigram.() |> Enum.count() end) |> IO.inspect(label: "stream_match_trigram")
+IO.puts("\n")
+
+Benchee.run(
+ %{
+ "stream_all" => stream_all,
+ "stream_match_all" => stream_match_all,
+ "stream_match_prefix" => stream_match_prefix,
+ "stream_match_trigram" => stream_match_trigram
+ },
+ warmup: 2,
+ time: 5,
+ parallel: 1
+)
+
+Benchee.run(
+ %{
+ "stream_all" => stream_all,
+ "stream_match_all" => stream_match_all,
+ "stream_match_prefix" => stream_match_prefix,
+ "stream_match_trigram" => stream_match_trigram
+ },
+ warmup: 2,
+ time: 5,
+ parallel: 4
+)
diff --git a/priv/repo/migrations/20241022021209_create_collections.exs b/priv/repo/migrations/20241022021209_create_collections.exs
new file mode 100644
index 0000000000..5e87809bd0
--- /dev/null
+++ b/priv/repo/migrations/20241022021209_create_collections.exs
@@ -0,0 +1,36 @@
+defmodule Lightning.Repo.Migrations.CreateCollections do
+ use Ecto.Migration
+
+ def change do
+ create table(:collections, primary_key: false) do
+ add :id, :binary_id, primary_key: true
+ add :name, :string
+
+ add :project_id,
+ references(:projects, on_delete: :delete_all, type: :binary_id, null: false)
+
+ timestamps()
+ end
+
+ create unique_index(:collections, [:name])
+
+ create table(:collections_items, primary_key: false) do
+ add :collection_id,
+ references(:collections, type: :binary_id, on_delete: :delete_all, null: false)
+
+ add :key, :string
+ add :value, :string
+
+ timestamps(type: :naive_datetime_usec)
+ end
+
+ execute "CREATE EXTENSION IF NOT EXISTS pg_trgm",
+ "DROP EXTENSION IF EXISTS pg_trgm"
+
+ create index(:collections_items, [:updated_at])
+ create unique_index(:collections_items, [:collection_id, :key])
+
+ execute "CREATE INDEX collections_items_key_trgm_idx ON collections_items USING GIN (key gin_trgm_ops)",
+ "DROP INDEX IF EXISTS collections_items_key_trgm_idx"
+ end
+end
diff --git a/test/integration/web_and_worker_test.exs b/test/integration/web_and_worker_test.exs
index 0cfe9d9877..64daa3dae4 100644
--- a/test/integration/web_and_worker_test.exs
+++ b/test/integration/web_and_worker_test.exs
@@ -248,7 +248,7 @@ defmodule Lightning.WebAndWorkerTest do
end)
assert version_logs =~ "▸ node.js 18.17"
- assert version_logs =~ "▸ worker 1.7"
+ assert version_logs =~ "▸ worker 1.8"
assert version_logs =~ "▸ @openfn/language-http 3.1.12"
expected_lines =
diff --git a/test/lightning/accounts_test.exs b/test/lightning/accounts_test.exs
index 4de50f2e19..636f2e1be9 100644
--- a/test/lightning/accounts_test.exs
+++ b/test/lightning/accounts_test.exs
@@ -1044,7 +1044,10 @@ defmodule Lightning.AccountsTest do
assert user_token = Repo.get_by(UserToken, token: token)
assert user_token.context == "api"
- Lightning.Accounts.UserToken.verify_and_validate!(token)
+ Lightning.Tokens.PersonalAccessToken.verify_and_validate!(
+ token,
+ Lightning.Config.token_signer()
+ )
# Creating the same token for another user should fail
assert_raise Ecto.ConstraintError, fn ->
diff --git a/test/lightning/collection_test.exs b/test/lightning/collection_test.exs
new file mode 100644
index 0000000000..cdaf76a815
--- /dev/null
+++ b/test/lightning/collection_test.exs
@@ -0,0 +1,85 @@
+defmodule Lightning.CollectionTest do
+ use Lightning.DataCase, async: true
+
+ import Lightning.Factories
+
+ alias Lightning.Collections.Collection
+
+ setup do
+ project = insert(:project, name: "Test Project")
+ {:ok, project: project}
+ end
+
+ @valid_name "valid-name"
+
+ describe "changeset/2" do
+ test "valid attributes create a valid changeset", %{project: project} do
+ valid_attrs = %{
+ "project_id" => project.id,
+ "name" => @valid_name
+ }
+
+ changeset = Collection.changeset(%Collection{}, valid_attrs)
+ assert changeset.valid?
+ end
+
+ test "missing required fields result in errors" do
+ invalid_attrs = %{"project_id" => nil, "name" => nil}
+
+ changeset = Collection.changeset(%Collection{}, invalid_attrs)
+ refute changeset.valid?
+
+ assert %{
+ project_id: ["can't be blank"],
+ name: ["can't be blank"]
+ } = errors_on(changeset)
+ end
+
+ test "name must be URL-safe (valid format)", %{project: project} do
+ valid_names = [
+ "valid-name",
+ "collection_123",
+ "my.collection",
+ "valid-collection-name"
+ ]
+
+ for name <- valid_names do
+ attrs = %{"project_id" => project.id, "name" => name}
+ changeset = Collection.changeset(%Collection{}, attrs)
+ assert changeset.valid?, "Expected #{name} to be valid"
+ end
+ end
+
+ test "name with invalid characters fails validation", %{project: project} do
+ invalid_names = [
+ "invalid name",
+ "invalid_name!",
+ "invalid/name",
+ "-invalid",
+ "invalid-",
+ "invalid--name"
+ ]
+
+ for name <- invalid_names do
+ attrs = %{"project_id" => project.id, "name" => name}
+ changeset = Collection.changeset(%Collection{}, attrs)
+ refute changeset.valid?, "Expected #{name} to be invalid"
+
+ assert %{name: ["Collection name must be URL safe"]} =
+ errors_on(changeset)
+ end
+ end
+
+ test "name uniqueness constraint adds error", %{project: project} do
+ insert(:collection, project: project, name: "existing-name")
+
+ attrs = %{"project_id" => project.id, "name" => "existing-name"}
+ changeset = Collection.changeset(%Collection{}, attrs)
+
+ assert {:error, changeset} = Lightning.Repo.insert(changeset)
+
+ assert %{name: ["A collection with this name already exists"]} =
+ errors_on(changeset)
+ end
+ end
+end
diff --git a/test/lightning/collections_test.exs b/test/lightning/collections_test.exs
new file mode 100644
index 0000000000..8aa2514f7f
--- /dev/null
+++ b/test/lightning/collections_test.exs
@@ -0,0 +1,562 @@
+defmodule Lightning.CollectionsTest do
+ use Lightning.DataCase
+
+ alias Lightning.Collections
+ alias Lightning.Collections.Collection
+ alias Lightning.Collections.Item
+
+ describe "get_collection/1" do
+ test "get a collection" do
+ %{id: collection_id, name: collection_name} = insert(:collection)
+
+ assert {:ok, %Collection{id: ^collection_id}} =
+ Collections.get_collection(collection_name)
+ end
+
+ test "returns an error when the collection does not exist" do
+ assert {:error, :not_found} =
+ Collections.get_collection("nonexistent")
+ end
+ end
+
+ describe "create_collection/2" do
+ test "creates a new collection" do
+ %{id: project_id} = insert(:project)
+ name = "col1_project1"
+
+ assert {:ok, %Collection{project_id: ^project_id, name: ^name}} =
+ Collections.create_collection(project_id, name)
+ end
+
+ test "returns an error when collection name is taken" do
+ %{id: project_id1} = insert(:project)
+ %{id: project_id2} = insert(:project)
+ name = "col1_project1"
+
+ assert {:ok, %Collection{project_id: ^project_id1, name: ^name}} =
+ Collections.create_collection(project_id1, name)
+
+ assert {:error,
+ %{
+ errors: [
+ name:
+ {"A collection with this name already exists",
+ [
+ constraint: :unique,
+ constraint_name: "collections_name_index"
+ ]}
+ ]
+ }} = Collections.create_collection(project_id2, name)
+ end
+ end
+
+ describe "delete_collection/1" do
+ test "deletes a collection" do
+ %{id: collection_id} = insert(:collection)
+
+ assert {:ok, %Collection{id: ^collection_id}} =
+ Collections.delete_collection(collection_id)
+ end
+
+ test "returns an error when collection does not exist" do
+ assert {:error, :not_found} =
+ Collections.delete_collection(Ecto.UUID.generate())
+ end
+ end
+
+ describe "get/2" do
+ test "returns an entry for the given collection" do
+ %{key: key, value: value, collection: collection} =
+ insert(:collection_item) |> Repo.preload(:collection)
+
+ assert %Item{key: ^key, value: ^value} = Collections.get(collection, key)
+ end
+
+ test "returns nil if the item key does not exist" do
+ collection = insert(:collection)
+
+ refute Collections.get(collection, "nonexistent")
+ end
+
+ test "returns nil if the collection does not exist" do
+ insert(:collection_item, key: "existing_key")
+
+ refute Collections.get(%{id: Ecto.UUID.generate()}, "existing_key")
+ end
+ end
+
+ describe "stream_all/3" do
+ test "returns all items for the given collection sorted by inserted_at" do
+ collection = insert(:collection)
+
+ items =
+ 1..11
+ |> Enum.map(fn _i ->
+ insert(:collection_item,
+ key: "rkey#{:rand.uniform()}",
+ collection: collection
+ )
+ end)
+
+ Repo.transaction(fn ->
+ assert stream = Collections.stream_all(collection, limit: 50)
+
+ assert stream_items =
+ stream
+ |> Stream.take(15)
+ |> Enum.to_list()
+ |> Repo.preload(collection: :project)
+
+ assert List.last(stream_items) ==
+ Enum.sort_by(items, & &1.inserted_at) |> List.last()
+
+ assert MapSet.new(stream_items) == MapSet.new(items)
+ end)
+ end
+
+ test "returns the items after a cursor up to a limited amount" do
+ collection = insert(:collection)
+
+ items =
+ Enum.map(1..30, fn _i ->
+ insert(:collection_item,
+ key: "rkey#{:rand.uniform()}",
+ collection: collection
+ )
+ end)
+
+ %{inserted_at: cursor} = Enum.at(items, 4)
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_all(collection, cursor: cursor, limit: 50)
+
+ assert stream |> Enum.to_list() |> Enum.count() == 30 - (4 + 1)
+ end)
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_all(collection, cursor: cursor, limit: 10)
+
+ assert Enum.count(stream) == 10
+ end)
+ end
+
+ test "returns empty list when collection is empty" do
+ collection = insert(:collection)
+
+ Repo.transaction(fn ->
+ assert stream = Collections.stream_all(collection, limit: 50)
+ assert Enum.count(stream) == 0
+ end)
+ end
+
+ test "returns empty list when the collection doesn't exist" do
+ insert(:collection_item, key: "existing_key")
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_all(%{id: Ecto.UUID.generate()}, limit: 50)
+
+ assert Enum.count(stream) == 0
+ end)
+ end
+
+ test "fails when outside of an explicit transaction" do
+ collection = insert(:collection)
+ _items = insert_list(5, :collection_item, collection: collection)
+
+ assert stream = Collections.stream_all(collection, limit: 50)
+
+ assert_raise RuntimeError,
+ ~r/cannot reduce stream outside of transaction/,
+ fn ->
+ Enum.take(stream, 5) |> Enum.each(&inspect/1)
+ end
+ end
+ end
+
+ describe "stream_match/3" do
+ test "returns item with exact match" do
+ collection = insert(:collection)
+ _itemA = insert(:collection_item, key: "keyA", collection: collection)
+ itemB = insert(:collection_item, key: "keyB", collection: collection)
+
+ Repo.transaction(fn ->
+ assert stream = Collections.stream_match(collection, "keyB*", limit: 50)
+
+ assert [itemB] ==
+ stream
+ |> Enum.to_list()
+ |> Repo.preload(collection: :project)
+ end)
+ end
+
+ test "returns matching items for the given collection sorted by inserted_at" do
+ collection = insert(:collection)
+
+ items =
+ 1..11
+ |> Enum.map(fn _i ->
+ insert(:collection_item,
+ key: "rkeyA#{:rand.uniform()}",
+ collection: collection
+ )
+ end)
+
+ insert(:collection_item, key: "rkeyB", collection: collection)
+
+ Repo.transaction(fn ->
+ assert stream = Collections.stream_match(collection, "rkeyA*", limit: 50)
+
+ assert stream_items =
+ Stream.take(stream, 12)
+ |> Enum.to_list()
+ |> Repo.preload(collection: :project)
+
+ assert List.last(stream_items) ==
+ Enum.sort_by(items, & &1.inserted_at) |> List.last()
+
+ assert MapSet.new(stream_items) == MapSet.new(items)
+ end)
+ end
+
+ test "returns matching items after a cursor up to a limited amount" do
+ collection = insert(:collection)
+
+ items =
+ Enum.map(1..30, fn _i ->
+ insert(:collection_item,
+ key: "rkeyA#{:rand.uniform()}",
+ collection: collection
+ )
+ end)
+
+ %{inserted_at: cursor} = Enum.at(items, 9)
+
+ insert(:collection_item, key: "rkeyB", collection: collection)
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_match(collection, "rkeyA*",
+ cursor: cursor,
+ limit: 50
+ )
+
+ assert Enum.count(stream) == 30 - (9 + 1)
+ end)
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_match(collection, "rkeyA*",
+ cursor: cursor,
+ limit: 16
+ )
+
+ assert Enum.count(stream) == 16
+ end)
+ end
+
+ test "returns empty list when collection is empty" do
+ collection = insert(:collection)
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_match(collection, "any-key", limit: 50)
+
+ assert Enum.count(stream) == 0
+ end)
+ end
+
+ test "returns empty list when the collection doesn't exist" do
+ insert(:collection_item, key: "existing_key")
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_match(
+ %{id: Ecto.UUID.generate()},
+ "existing_key",
+ limit: 50
+ )
+
+ assert Enum.count(stream) == 0
+ end)
+ end
+
+ test "returns item escaping the %" do
+ collection = insert(:collection)
+ item = insert(:collection_item, key: "keyA%", collection: collection)
+
+ Repo.transaction(fn ->
+ assert stream = Collections.stream_match(collection, "keyA%*", limit: 50)
+
+ assert [item] ==
+ stream
+ |> Enum.to_list()
+ |> Repo.preload(collection: :project)
+ end)
+
+ insert(:collection_item, key: "keyBC", collection: collection)
+
+ Repo.transaction(fn ->
+ assert stream = Collections.stream_match(collection, "keyB%", limit: 50)
+
+ assert Enum.count(stream) == 0
+ end)
+ end
+
+ test "returns item escaping the \\" do
+ collection = insert(:collection)
+ item = insert(:collection_item, key: "keyA\\", collection: collection)
+
+ Repo.transaction(fn ->
+ assert stream =
+ Collections.stream_match(collection, "keyA\\*", limit: 50)
+
+ assert [item] ==
+ stream
+ |> Enum.to_list()
+ |> Repo.preload(collection: :project)
+ end)
+ end
+
+ test "fails when outside of an explicit transaction" do
+ collection = insert(:collection)
+ _items = insert_list(5, :collection_item, collection: collection)
+
+ assert stream = Collections.stream_match(collection, "key*", limit: 50)
+
+ assert_raise RuntimeError,
+ ~r/cannot reduce stream outside of transaction/,
+ fn ->
+ Enum.take(stream, 5) |> Enum.each(&inspect/1)
+ end
+ end
+ end
+
+ describe "put/3" do
+ test "creates a new entry in the collection for the given collection" do
+ collection = insert(:collection)
+
+ assert :ok = Collections.put(collection, "some-key", "some-value")
+
+ assert %{key: "some-key", value: "some-value"} =
+ Repo.get_by!(Item, key: "some-key")
+ end
+
+ test "updates the value of an item when key exists" do
+ collection = insert(:collection)
+
+ assert :ok = Collections.put(collection, "some-key", "some-value1")
+
+ assert %{key: "some-key", value: "some-value1"} =
+ Repo.get_by!(Item, key: "some-key")
+
+ assert :ok = Collections.put(collection, "some-key", "some-value2")
+
+ assert %{key: "some-key", value: "some-value2"} =
+ Repo.get_by!(Item, key: "some-key")
+ end
+
+ test "returns an :error if the collection does not exist" do
+ assert {:error,
+ %{
+ errors: [
+ collection_id:
+ {"does not exist",
+ [
+ constraint: :foreign,
+ constraint_name: "collections_items_collection_id_fkey"
+ ]}
+ ]
+ }} = Collections.put(%{id: Ecto.UUID.generate()}, "key", "value")
+ end
+ end
+
+ describe "put_all/2" do
+ test "inserts multiple entries at once in a given collection" do
+ collection = insert(:collection)
+
+ items =
+ Enum.map(1..5, fn i -> %{"key" => "key#{i}", "value" => "value#{i}"} end)
+
+ assert {:ok, 5} = Collections.put_all(collection, items)
+
+ assert Item |> Repo.all() |> Enum.map(&Map.take(&1, [:key, :value])) ==
+ Enum.map(items, &%{key: &1["key"], value: &1["value"]})
+ end
+
+ test "replaces conflicting values and updates timestamp" do
+ collection = insert(:collection)
+
+ items =
+ Enum.map(1..5, fn i -> %{"key" => "key#{i}", "value" => "value#{i}"} end)
+
+ assert {:ok, 5} = Collections.put_all(collection, items)
+
+ assert %{updated_at: updated_at1} = Repo.get_by(Item, key: "key1")
+ assert %{updated_at: updated_at2} = Repo.get_by(Item, key: "key2")
+ assert %{updated_at: updated_at5} = Repo.get_by(Item, key: "key5")
+
+ update_items =
+ Enum.map(1..2, fn i ->
+ %{"key" => "key#{i}", "value" => "value#{10 + i}"}
+ end)
+
+ assert {:ok, 2} = Collections.put_all(collection, update_items)
+
+ assert %{value: "value11", updated_at: updated_at} =
+ Repo.get_by(Item, key: "key1")
+
+ assert updated_at > updated_at1
+
+ assert %{value: "value12", updated_at: updated_at} =
+ Repo.get_by(Item, key: "key2")
+
+ assert updated_at > updated_at2
+
+ assert %{value: "value5", updated_at: ^updated_at5} =
+ Repo.get_by(Item, key: "key5")
+ end
+ end
+
+ describe "delete/2" do
+ test "deletes an entry for the given collection" do
+ collection = insert(:collection)
+
+ %{key: key} =
+ insert(:collection_item, collection: collection)
+
+ assert :ok = Collections.delete(collection, key)
+
+ refute Collections.get(collection, key)
+ end
+
+ test "returns an :error if the collection does not exist" do
+ assert {:error, :not_found} =
+ Collections.delete(%{id: Ecto.UUID.generate()}, "key")
+ end
+
+ test "returns an :error if item does not exist" do
+ collection = insert(:collection)
+
+ assert {:error, :not_found} =
+ Collections.delete(collection, "nonexistent")
+ end
+ end
+
+ describe "delete_all/2" do
+ test "deletes all items of the given collection" do
+ collection = insert(:collection)
+
+ items = insert_list(3, :collection_item, collection: collection)
+
+ assert {:ok, 3} = Collections.delete_all(collection)
+
+ refute Enum.any?(items, &Collections.get(collection, &1.key))
+ end
+
+ test "deletes matching items of the given collection" do
+ collection = insert(:collection)
+
+ item1 =
+ insert(:collection_item, collection: collection, key: "foo:123:bar1")
+
+ item2 =
+ insert(:collection_item, collection: collection, key: "foo:234:bar2")
+
+ item3 =
+ insert(:collection_item, collection: collection, key: "foo:345:bar3")
+
+ item4 =
+ insert(:collection_item, collection: collection, key: "foo:456:zanzibar")
+
+ assert {:ok, 3} = Collections.delete_all(collection, "foo:*:bar*")
+
+ refute Collections.get(collection, item1.key)
+ refute Collections.get(collection, item2.key)
+ refute Collections.get(collection, item3.key)
+ assert Collections.get(collection, item4.key)
+ end
+
+ test "returns an :error if the collection does not exist" do
+ assert {:error, :not_found} =
+ Collections.delete(%{id: Ecto.UUID.generate()}, "key")
+ end
+
+ test "returns an :error if item does not exist" do
+ collection = insert(:collection)
+
+ assert {:error, :not_found} =
+ Collections.delete(collection, "nonexistent")
+ end
+ end
+
+ describe "list_collections/1" do
+ test "returns a list of collections with default ordering and preloading" do
+ collection1 = insert(:collection, name: "B Collection")
+ collection2 = insert(:collection, name: "A Collection")
+
+ result = Collections.list_collections()
+
+ assert Enum.map(result, & &1.id) == [collection2.id, collection1.id]
+ end
+
+ test "returns collections ordered by specified field" do
+ collection1 = insert(:collection, inserted_at: ~N[2024-01-01 00:00:00])
+ collection2 = insert(:collection, inserted_at: ~N[2024-02-01 00:00:00])
+
+ result = Collections.list_collections(order_by: [asc: :inserted_at])
+
+ assert Enum.map(result, & &1.id) == [collection1.id, collection2.id]
+ end
+
+ test "preloads specified associations" do
+ project = insert(:project)
+ insert(:collection, project: project)
+
+ result = Collections.list_collections(preload: [:project])
+
+ assert Enum.map(result, & &1.project.id) == [project.id]
+ end
+ end
+
+ describe "create_collection/1" do
+ test "creates a new collection with valid attributes" do
+ %{id: project_id} = insert(:project)
+ attrs = %{name: "new-collection", project_id: project_id}
+
+ assert {:ok, %Collection{name: "new-collection"}} =
+ Collections.create_collection(attrs)
+ end
+
+ test "returns an error if invalid attributes are provided" do
+ attrs = %{name: nil}
+
+ assert {:error, changeset} = Collections.create_collection(attrs)
+
+ assert %{name: ["can't be blank"], project_id: ["can't be blank"]} ==
+ errors_on(changeset)
+ end
+ end
+
+ describe "update_collection/2" do
+ test "updates an existing collection with valid attributes" do
+ collection = insert(:collection, name: "Old Name")
+ attrs = %{name: "updated-name"}
+
+ assert {:ok, %Collection{name: "updated-name"}} =
+ Collections.update_collection(collection, attrs)
+ end
+
+ test "returns an error if invalid attributes are provided" do
+ collection = insert(:collection)
+ attrs = %{name: nil}
+
+ assert {:error, changeset} =
+ Collections.update_collection(collection, attrs)
+
+ assert %{name: ["can't be blank"]} == errors_on(changeset)
+ end
+ end
+end
diff --git a/test/lightning/failure_alert_test.exs b/test/lightning/failure_alert_test.exs
index 3b7d38884a..7b0f4db91e 100644
--- a/test/lightning/failure_alert_test.exs
+++ b/test/lightning/failure_alert_test.exs
@@ -258,7 +258,7 @@ defmodule Lightning.FailureAlertTest do
Lightning.Stub.reset_time()
{:ok, bearer, claims} =
- Workers.Token.generate_and_sign(
+ Workers.WorkerToken.generate_and_sign(
%{},
Lightning.Config.worker_token_signer()
)
@@ -269,6 +269,7 @@ defmodule Lightning.FailureAlertTest do
UsageLimiter.get_run_options(%Context{
project_id: run.work_order.workflow.project_id
})
+ |> Map.new()
{:ok, %{}, socket} =
LightningWeb.WorkerSocket
diff --git a/test/lightning/helpers_test.exs b/test/lightning/helpers_test.exs
index de87090f4a..5ae1f17e48 100644
--- a/test/lightning/helpers_test.exs
+++ b/test/lightning/helpers_test.exs
@@ -3,6 +3,9 @@ defmodule Lightning.HelpersTest do
import Lightning.Helpers, only: [coerce_json_field: 2]
+ alias Ecto.Changeset
+ alias Lightning.Helpers
+
test "coerce_json_field/2 will transform a json string inside a map by it's key" do
input = %{
"body" =>
@@ -54,4 +57,110 @@ defmodule Lightning.HelpersTest do
goals: 123
}
end
+
+ describe "copy_error/4" do
+ test "copies an error from one key to another" do
+ changeset = %Changeset{errors: [name: {"has already been taken", []}]}
+ updated_changeset = Helpers.copy_error(changeset, :name, :raw_name)
+
+ assert updated_changeset.errors[:name] == {"has already been taken", []}
+
+ assert updated_changeset.errors[:raw_name] ==
+ {"has already been taken", []}
+ end
+
+ test "returns the changeset unchanged if original_key does not exist" do
+ changeset = %Changeset{errors: [email: {"is invalid", []}]}
+ updated_changeset = Helpers.copy_error(changeset, :name, :raw_name)
+
+ assert updated_changeset == changeset
+ refute Keyword.has_key?(updated_changeset.errors, :raw_name)
+ end
+
+ test "overwrites the new_key error if it exists and overwrite is true" do
+ changeset = %Changeset{
+ errors: [
+ name: {"has already been taken", []},
+ raw_name: {"is invalid", []}
+ ]
+ }
+
+ updated_changeset = Helpers.copy_error(changeset, :name, :raw_name)
+
+ assert updated_changeset.errors[:raw_name] ==
+ {"has already been taken", []}
+ end
+
+ test "does not overwrite the new_key error if overwrite is false" do
+ changeset = %Changeset{
+ errors: [
+ name: {"has already been taken", []},
+ raw_name: {"is invalid", []}
+ ]
+ }
+
+ updated_changeset =
+ Helpers.copy_error(changeset, :name, :raw_name, overwrite: false)
+
+ assert updated_changeset.errors[:raw_name] == {"is invalid", []}
+ end
+
+ test "returns the changeset unchanged if new_key already exists and overwrite is false" do
+ changeset = %Changeset{
+ errors: [
+ name: {"has already been taken", []},
+ raw_name: {"is invalid", []}
+ ]
+ }
+
+ updated_changeset =
+ Helpers.copy_error(changeset, :name, :raw_name, overwrite: false)
+
+ assert updated_changeset.errors[:raw_name] == {"is invalid", []}
+ end
+ end
+
+ describe "url_safe_name/1" do
+ test "returns an empty string when given nil" do
+ assert Helpers.url_safe_name(nil) == ""
+ end
+
+ test "converts a simple string to lowercase and replaces spaces" do
+ assert Helpers.url_safe_name("My Project") == "my-project"
+ end
+
+ test "removes special characters and replaces them with hyphens" do
+ assert Helpers.url_safe_name("My@#Project!!") == "my-project"
+ end
+
+ test "trims leading and trailing hyphens" do
+ assert Helpers.url_safe_name("--My Project--") == "my-project"
+ end
+
+ test "preserves international characters" do
+ assert Helpers.url_safe_name("Éléphant") == "éléphant"
+ end
+
+ test "handles a string with multiple special characters" do
+ assert Helpers.url_safe_name("Hello, World! 123") == "hello-world-123"
+ end
+
+ test "handles a string with underscores and periods" do
+ assert Helpers.url_safe_name("file_name.version_1.0") ==
+ "file_name.version_1.0"
+ end
+
+ test "replaces multiple spaces or special characters with a single hyphen" do
+ assert Helpers.url_safe_name("My Project") == "my-project"
+ assert Helpers.url_safe_name("My--Project!!") == "my--project"
+ end
+
+ test "handles a string with only special characters and removes them" do
+ assert Helpers.url_safe_name("###!!!") == ""
+ end
+
+ test "keeps numbers intact in the string" do
+ assert Helpers.url_safe_name("Project 2023") == "project-2023"
+ end
+ end
end
diff --git a/test/lightning/tokens_test.exs b/test/lightning/tokens_test.exs
new file mode 100644
index 0000000000..ce09342727
--- /dev/null
+++ b/test/lightning/tokens_test.exs
@@ -0,0 +1,107 @@
+defmodule Lightning.TokensTest do
+ use Lightning.DataCase, async: true
+
+ import Lightning.Factories
+
+ alias Lightning.Tokens
+
+ setup do
+ Mox.stub_with(Lightning.MockConfig, Lightning.Config.API)
+ Mox.stub_with(LightningMock, Lightning.Stub)
+
+ :ok
+ end
+
+ describe "UserToken" do
+ test "verify a valid token" do
+ user = insert(:user)
+
+ Lightning.Stub.freeze_time(~U[2024-01-01 00:00:00Z])
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ assert {:ok, claims} = Tokens.verify(token)
+
+ expected_sub = "user:#{user.id}"
+
+ assert %{
+ "iss" => "Lightning",
+ "iat" => 1_704_067_200,
+ "sub" => ^expected_sub,
+ "jti" => _
+ } = claims
+ end
+
+ test "verify a forged/invalid token" do
+ user = insert(:user)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ replace_signer(:token_signer)
+
+ assert {:error, :signature_error} = Tokens.verify(token)
+ end
+
+ test "retrieving the subject from the token" do
+ Lightning.Stub.freeze_time(DateTime.utc_now())
+ user = insert(:user)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ assert {:ok, claims} = Tokens.verify(token)
+
+ assert Tokens.get_subject(claims) == user |> Repo.reload!()
+ end
+ end
+
+ describe "RunToken" do
+ test "verify a valid token" do
+ Lightning.Stub.freeze_time(~U[2024-01-01 00:00:00Z])
+
+ token =
+ Lightning.Workers.generate_run_token(%{id: run_id = Ecto.UUID.generate()})
+
+ assert {:ok, claims} = Tokens.verify(token)
+
+ assert %{
+ "exp" => 1_704_067_270,
+ "id" => run_id,
+ "iss" => "Lightning",
+ "nbf" => 1_704_067_200,
+ "sub" => "run:#{run_id}"
+ } == claims
+ end
+
+ test "verify a forged/invalid token" do
+ token =
+ Lightning.Workers.generate_run_token(%{id: Ecto.UUID.generate()})
+
+ replace_signer(:run_token_signer)
+
+ assert {:error, :signature_error} = Tokens.verify(token)
+ end
+
+ test "verify an expired token" do
+ Lightning.Stub.freeze_time(~U[2024-01-01 00:00:00Z])
+
+ token =
+ Lightning.Workers.generate_run_token(%{id: Ecto.UUID.generate()})
+
+ Lightning.Stub.freeze_time(~U[2024-02-01 00:00:00Z])
+
+ assert {
+ :error,
+ [message: "Invalid token", claim: "exp", claim_val: 1_704_067_270]
+ } = Tokens.verify(token)
+ end
+ end
+
+ # Generate a new RSA cert that is different to the one the token was
+ # signed with.
+ defp replace_signer(key) do
+ Mox.stub(Lightning.MockConfig, key, fn ->
+ {pvt, _pub} = Lightning.Utils.Crypto.generate_rsa_key_pair()
+ Joken.Signer.create("RS256", %{"pem" => pvt})
+ end)
+ end
+end
diff --git a/test/lightning/workers_test.exs b/test/lightning/workers_test.exs
index a19b067712..9ef97d93d0 100644
--- a/test/lightning/workers_test.exs
+++ b/test/lightning/workers_test.exs
@@ -1,24 +1,26 @@
defmodule Lightning.WorkersTest do
use ExUnit.Case, async: true
- alias Lightning.Workers.Token
+ alias Lightning.Workers.RunToken
+ alias Lightning.Workers.WorkerToken
setup do
Mox.stub_with(LightningMock, Lightning.API)
+ Mox.stub_with(Lightning.MockConfig, Lightning.Config.API)
- :ok
+ %{run_token_signer: Lightning.Config.run_token_signer()}
end
- describe "Token" do
+ describe "WorkerToken" do
test "can generate a token" do
{:ok, token, claims} =
- Token.generate_and_sign(%{"id" => id = Ecto.UUID.generate()})
+ WorkerToken.generate_and_sign(%{"id" => id = Ecto.UUID.generate()})
assert %{"id" => ^id, "nbf" => nbf} = claims
- assert nbf <= DateTime.utc_now() |> DateTime.to_unix()
+ assert nbf <= Lightning.current_time() |> DateTime.to_unix()
assert token != ""
- assert {:ok, claims} = Token.verify(token)
+ assert {:ok, claims} = WorkerToken.verify(token)
assert {:error,
[
@@ -26,9 +28,47 @@ defmodule Lightning.WorkersTest do
{:claim, "nbf"},
{:claim_val, _time}
]} =
- Token.validate(claims, %{
+ WorkerToken.validate(claims, %{
current_time: DateTime.utc_now() |> DateTime.add(-5, :second)
})
end
end
+
+ describe "RunToken" do
+ test "can generate a token", %{run_token_signer: run_token_signer} do
+ {:ok, token, claims} =
+ RunToken.generate_and_sign(
+ %{"id" => id = Ecto.UUID.generate()},
+ run_token_signer
+ )
+
+ assert %{"id" => ^id, "nbf" => nbf} = claims
+ assert nbf <= Lightning.current_time() |> DateTime.to_unix()
+ assert token != ""
+
+ assert {:ok, ^claims} =
+ RunToken.verify(token, run_token_signer)
+ end
+
+ test "validating with a run_id" do
+ {:ok, claims} =
+ RunToken.generate_claims(%{"id" => id = Ecto.UUID.generate()})
+
+ assert {:ok, ^claims} =
+ RunToken.validate(claims, %{
+ id: id,
+ current_time: Lightning.current_time()
+ })
+ end
+
+ test "validating without a run_id" do
+ {:ok, claims} =
+ RunToken.generate_claims(%{"id" => _id = Ecto.UUID.generate()})
+
+ assert {:ok, ^claims} =
+ RunToken.validate(claims, %{
+ current_time: Lightning.current_time()
+ })
+ end
+ end
end
diff --git a/test/lightning_web/channels/run_channel_test.exs b/test/lightning_web/channels/run_channel_test.exs
index 74af8e9bb0..6cdfb386ce 100644
--- a/test/lightning_web/channels/run_channel_test.exs
+++ b/test/lightning_web/channels/run_channel_test.exs
@@ -59,7 +59,7 @@ defmodule LightningWeb.RunChannelTest do
# A valid token, but nbf hasn't been reached yet
{:ok, bearer, _} =
- Workers.Token.generate_and_sign(
+ Workers.WorkerToken.generate_and_sign(
%{
"nbf" =>
DateTime.utc_now()
@@ -84,7 +84,9 @@ defmodule LightningWeb.RunChannelTest do
other_id = Ecto.UUID.generate()
bearer =
- Workers.generate_run_token(%{id: id}, run_timeout_ms: 1000)
+ Workers.generate_run_token(%{id: id}, %{
+ run_timeout_ms: 1000
+ })
assert {:error, %{reason: "unauthorized"}} =
socket
@@ -99,7 +101,7 @@ defmodule LightningWeb.RunChannelTest do
id = Ecto.UUID.generate()
bearer =
- Workers.generate_run_token(%{id: id}, run_timeout_ms: 1000)
+ Workers.generate_run_token(%{id: id}, %{run_timeout_ms: 1000})
assert {:error, %{reason: "not_found"}} =
socket
@@ -567,6 +569,7 @@ defmodule LightningWeb.RunChannelTest do
Lightning.Extensions.MockUsageLimiter.get_run_options(%Context{
project_id: project.id
})
+ |> Enum.into(%{})
{:ok, _, socket} =
context.socket
@@ -1379,7 +1382,7 @@ defmodule LightningWeb.RunChannelTest do
defp create_socket(context) do
{:ok, bearer, claims} =
- Workers.Token.generate_and_sign(
+ Workers.WorkerToken.generate_and_sign(
%{},
Lightning.Config.worker_token_signer()
)
@@ -1403,7 +1406,12 @@ defmodule LightningWeb.RunChannelTest do
|> subscribe_and_join(
LightningWeb.RunChannel,
"run:#{run.id}",
- %{"token" => Workers.generate_run_token(run, run_timeout_ms: 2)}
+ %{
+ "token" =>
+ Workers.generate_run_token(run, %Lightning.Runs.RunOptions{
+ run_timeout_ms: 2
+ })
+ }
)
%{socket: socket}
diff --git a/test/lightning_web/channels/worker_channel_test.exs b/test/lightning_web/channels/worker_channel_test.exs
index 3e790ad424..abebafbb53 100644
--- a/test/lightning_web/channels/worker_channel_test.exs
+++ b/test/lightning_web/channels/worker_channel_test.exs
@@ -22,7 +22,7 @@ defmodule LightningWeb.WorkerChannelTest do
describe "worker:queue channel" do
setup do
{:ok, bearer, claims} =
- Workers.Token.generate_and_sign(
+ Workers.WorkerToken.generate_and_sign(
%{},
Lightning.Config.worker_token_signer()
)
@@ -82,7 +82,9 @@ defmodule LightningWeb.WorkerChannelTest do
ref = push(socket, "claim", %{"demand" => 4})
assert_reply ref, :ok, %{runs: runs}
- assert runs |> Enum.map(& &1["id"]) == rest |> Enum.map(& &1.id)
+ assert runs |> Enum.map(& &1["id"]) |> MapSet.new() ==
+ rest |> Enum.map(& &1.id) |> MapSet.new()
+
assert length(runs) == 2, "only 2 runs should be returned"
end
end
diff --git a/test/lightning_web/channels/worker_socket_test.exs b/test/lightning_web/channels/worker_socket_test.exs
index ad42144d5a..66acd5cc7c 100644
--- a/test/lightning_web/channels/worker_socket_test.exs
+++ b/test/lightning_web/channels/worker_socket_test.exs
@@ -12,7 +12,7 @@ defmodule LightningWeb.WorkerSocketTest do
test "with a valid token" do
{:ok, bearer, _} =
- Workers.Token.generate_and_sign(
+ Workers.WorkerToken.generate_and_sign(
%{},
Lightning.Config.worker_token_signer()
)
diff --git a/test/lightning_web/collections_controller_test.exs b/test/lightning_web/collections_controller_test.exs
new file mode 100644
index 0000000000..9d9a9aad43
--- /dev/null
+++ b/test/lightning_web/collections_controller_test.exs
@@ -0,0 +1,725 @@
+defmodule LightningWeb.API.CollectionsControllerTest do
+ use LightningWeb.ConnCase, async: true
+
+ import Lightning.Factories
+
+ alias Lightning.Collections
+
+ @default_limit Application.compile_env!(
+ :lightning,
+ LightningWeb.CollectionsController
+ )[
+ :stream_limit
+ ]
+
+ setup %{conn: conn} do
+ {:ok, conn: put_req_header(conn, "accept", "application/json")}
+ end
+
+ test "without a token", %{conn: conn} do
+ conn = get(conn, ~p"/collections/foo")
+
+ assert %{"error" => "Unauthorized"} == json_response(conn, 401)
+ end
+
+ describe "authenticating with a run token" do
+ test "with a token that is invalid", %{conn: conn} do
+ workflow = insert(:simple_workflow)
+ workorder = insert(:workorder, dataclip: insert(:dataclip))
+
+ collection = insert(:collection, project: workflow.project)
+
+ run =
+ insert(:run,
+ work_order: workorder,
+ dataclip: workorder.dataclip,
+ starting_trigger: workflow.triggers |> hd()
+ )
+
+ token = Lightning.Workers.generate_run_token(run)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}")
+
+ assert json_response(conn, 401) == %{"error" => "Unauthorized"}
+ end
+ end
+
+ describe "authenticating with a personal access token" do
+ test "with a project they don't have access to", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [])
+
+ collection = insert(:collection, project: project)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}")
+
+ assert json_response(conn, 401) == %{"error" => "Unauthorized"}
+ end
+ end
+
+ describe "GET /collections/:name/:key" do
+ test "returns the item", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: [%{key: "foo", value: "bar"}]
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}/foo")
+
+ item = hd(collection.items)
+
+ assert json_response(conn, 200) == %{
+ "key" => item.key,
+ "value" => item.value,
+ "created" => DateTime.to_iso8601(item.inserted_at),
+ "updated" => DateTime.to_iso8601(item.updated_at)
+ }
+ end
+
+ test "returns 404 when the collection doesn't exist", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ _another_collection = insert(:collection, project: project)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/misspelled-collection/foo")
+
+ assert json_response(conn, 404) == %{"error" => "Not Found"}
+ end
+
+ test "returns 404 when the item doesn't exist", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection = insert(:collection, project: project)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}/some-unexisting-key")
+
+ assert json_response(conn, 204) == nil
+ end
+ end
+
+ describe "PUT /collections/:name/:key" do
+ test "inserts an item", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: [%{key: "foo", value: "bar"}]
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> put(~p"/collections/#{collection.name}/baz", value: "qux")
+
+ assert json_response(conn, 200) == %{
+ "upserted" => 1,
+ "error" => nil
+ }
+ end
+
+ test "updates an item", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: [%{key: "foo", value: "bar"}]
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> put(~p"/collections/#{collection.name}/foo", %{value: "qux2"})
+
+ assert json_response(conn, 200) == %{
+ "upserted" => 1,
+ "error" => nil
+ }
+ end
+
+ test "returns 404 when the collection doesn't exist", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ _another_collection = insert(:collection, project: project)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> post(~p"/collections/misspelled-collection/baz", value: "qux")
+
+ assert json_response(conn, 404) == %{"error" => "Not Found"}
+ end
+ end
+
+ describe "POST /collections/:name" do
+ test "upserted multiple items", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: Enum.map(1..3, &%{key: "foo#{&1}", value: "bar#{&1}"})
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> post(~p"/collections/#{collection.name}", %{
+ items: Enum.map(1..10, &%{key: "foo#{&1}", value: "bar#{&1}"})
+ })
+
+ assert json_response(conn, 200) == %{
+ "upserted" => 10,
+ "error" => nil
+ }
+ end
+
+ test "returns 404 when the collection doesn't exist", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ _another_collection = insert(:collection, project: project)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> post(~p"/collections/misspelled-collection", %{
+ items: [%{key: "baz", value: "qux"}]
+ })
+
+ assert json_response(conn, 404) == %{"error" => "Not Found"}
+ end
+ end
+
+ describe "DELETE" do
+ test "deletes an item", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: [%{key: "foo", value: "bar"}]
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> delete(~p"/collections/#{collection.name}/foo")
+
+ assert json_response(conn, 200) == %{
+ "key" => "foo",
+ "deleted" => 1,
+ "error" => nil
+ }
+ end
+
+ test "deletes matching items", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: [
+ %{key: "foo:123:bar1", value: "value1"},
+ %{key: "foo:234:boo", value: "value2"},
+ %{key: "foo:345:bar2", value: "value3"}
+ ]
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> delete(~p"/collections/#{collection.name}", key: "foo:*:bar*")
+
+ assert json_response(conn, 200) == %{
+ "key" => "foo:*:bar*",
+ "deleted" => 2,
+ "error" => nil
+ }
+ end
+
+ test "returns 404 when the collection doesn't exist", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ _another_collection = insert(:collection, project: project)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> delete(~p"/collections/misspelled-collection/foo")
+
+ assert json_response(conn, 404) == %{"error" => "Not Found"}
+ end
+ end
+
+ describe "GET /collections/:name" do
+ test "with no results", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection = insert(:collection, project: project)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}")
+
+ assert json_response(conn, 200) == %{"items" => [], "cursor" => nil}
+ end
+
+ test "using a key pattern", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection = insert(:collection, project: project)
+
+ insert(:collection_item, collection: collection, key: "foo:bar:baz")
+ insert(:collection_item, collection: collection, key: "foo:bar:baz:quux")
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn = assign_bearer(conn, token)
+
+ assert %{state: :chunked} =
+ conn =
+ get(
+ conn,
+ ~p"/collections/#{collection.name}?#{%{key: "foo:bar:*"}}"
+ )
+
+ assert %{
+ "items" => [
+ %{"key" => "foo:bar:baz", "value" => _},
+ %{"key" => "foo:bar:baz:quux", "value" => _}
+ ],
+ "cursor" => nil
+ } = json_response(conn, 200)
+
+ assert %{state: :chunked} =
+ conn =
+ get(
+ conn,
+ ~p"/collections/#{collection.name}?#{%{key: "foo:*:baz"}}"
+ )
+
+ assert %{
+ "items" => [%{"key" => "foo:bar:baz", "value" => _}],
+ "cursor" => nil
+ } = json_response(conn, 200)
+ end
+
+ test "using a key pattern and creation filters", %{conn: conn} do
+ user = insert(:user)
+
+ project =
+ insert(:project, project_users: [%{user: user}])
+
+ collection = insert(:collection, project: project)
+
+ before_insert = DateTime.utc_now() |> DateTime.add(-1, :microsecond)
+
+ insert(:collection_item, collection: collection, key: "foo:bar:baz")
+ insert(:collection_item, collection: collection, key: "foo:moon:baz")
+ insert(:collection_item, collection: collection, key: "foo:bar:baz:out")
+
+ after_insert = DateTime.utc_now() |> DateTime.add(1, :microsecond)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn = assign_bearer(conn, token)
+
+ assert %{
+ "items" => [],
+ "cursor" => nil
+ } =
+ conn
+ |> get(~p"/collections/#{collection.name}",
+ key: "foo:*:baz",
+ created_after: after_insert
+ )
+ |> json_response(200)
+
+ assert %{
+ "items" => [],
+ "cursor" => nil
+ } =
+ conn
+ |> get(~p"/collections/#{collection.name}",
+ key: "foo:*:baz",
+ created_before: before_insert
+ )
+ |> json_response(200)
+
+ assert %{
+ "items" => [
+ %{"key" => "foo:bar:baz", "value" => _},
+ %{"key" => "foo:moon:baz", "value" => _}
+ ],
+ "cursor" => nil
+ } =
+ conn
+ |> get(~p"/collections/#{collection.name}",
+ key: "foo:*:baz",
+ created_after: before_insert,
+ created_before: after_insert
+ )
+ |> json_response(200)
+ end
+
+ test "up exactly to the limit", %{conn: conn} do
+ user = insert(:user)
+ project = insert(:project, project_users: [%{user: user}])
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: insert_list(3, :collection_item)
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}")
+
+ assert conn.state == :chunked
+
+ expected_items =
+ collection.items
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ assert json_response(conn, 200) == %{
+ "items" => expected_items,
+ "cursor" => nil
+ }
+ end
+
+ test "using creation filters", %{conn: conn} do
+ user = insert(:user)
+ project = insert(:project, project_users: [%{user: user}])
+
+ before_insert = DateTime.utc_now() |> DateTime.add(-1, :microsecond)
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: insert_list(3, :collection_item)
+ )
+
+ after_insert = DateTime.utc_now() |> DateTime.add(1, :microsecond)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn = assign_bearer(conn, token)
+
+ assert %{
+ "items" => [],
+ "cursor" => nil
+ } =
+ conn
+ |> get(~p"/collections/#{collection.name}",
+ created_after: after_insert
+ )
+ |> json_response(200)
+
+ assert %{
+ "items" => [],
+ "cursor" => nil
+ } =
+ conn
+ |> get(~p"/collections/#{collection.name}",
+ created_before: before_insert
+ )
+ |> json_response(200)
+
+ items =
+ collection.items
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ assert %{
+ "items" => ^items,
+ "cursor" => nil
+ } =
+ conn
+ |> get(~p"/collections/#{collection.name}")
+ |> json_response(200)
+ end
+
+ test "returns items up to a custom limit", %{conn: conn} do
+ user = insert(:user)
+ project = insert(:project, project_users: [%{user: user}])
+ limit = 10
+
+ collection =
+ insert(:collection,
+ project: project,
+ items: insert_list(limit + 1, :collection_item)
+ )
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}", limit: limit)
+
+ assert conn.state == :chunked
+
+ expected_items =
+ collection.items
+ |> Enum.take(limit)
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ last_item =
+ collection.items
+ |> Enum.take(limit)
+ |> List.last()
+
+ assert json_response(conn, 200) == %{
+ "items" => expected_items,
+ "cursor" =>
+ Base.encode64(DateTime.to_iso8601(last_item.inserted_at))
+ }
+ end
+ end
+
+ describe "GET /collections/:name with cursors" do
+ test "up to the limit and returning a cursor", %{conn: conn} do
+ user = insert(:user)
+ project = insert(:project, project_users: [%{user: user}])
+ collection = insert(:collection, project: project)
+
+ items =
+ insert_list(@default_limit + 1, :collection_item, collection: collection)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn =
+ conn
+ |> assign_bearer(token)
+ |> get(~p"/collections/#{collection.name}")
+
+ assert conn.state == :chunked
+
+ expected_items = Enum.take(items, @default_limit)
+ last_item = List.last(expected_items)
+
+ assert %{
+ "items" => items,
+ "cursor" => cursor
+ } = json_response(conn, 200)
+
+ assert items ==
+ expected_items
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ assert cursor == Base.encode64(DateTime.to_iso8601(last_item.inserted_at))
+ end
+
+ test "up to the limit from a cursor returning a cursor", %{conn: conn} do
+ user = insert(:user)
+ project = insert(:project, project_users: [%{user: user}])
+ collection = insert(:collection, project: project)
+ limit = 100
+
+ all_items =
+ Enum.map(1..(2 * limit + 1), fn i ->
+ insert(:collection_item,
+ collection: collection,
+ key: "key#{i}",
+ value: "value#{i}"
+ )
+ end)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn = assign_bearer(conn, token)
+
+ assert %{state: :chunked} =
+ conn =
+ get(conn, ~p"/collections/#{collection.name}", limit: limit)
+
+ expected_items =
+ all_items
+ |> Enum.take(limit)
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ assert %{
+ "items" => ^expected_items,
+ "cursor" => cursor
+ } = json_response(conn, 200)
+
+ assert %{state: :chunked} =
+ conn =
+ get(conn, ~p"/collections/#{collection.name}",
+ cursor: cursor,
+ limit: limit
+ )
+
+ expected_items =
+ all_items
+ |> Enum.drop(limit)
+ |> Enum.take(limit)
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ assert %{
+ "items" => ^expected_items,
+ "cursor" => cursor
+ } = json_response(conn, 200)
+
+ %{inserted_at: last_inserted_at} =
+ Repo.get_by(Collections.Item,
+ collection_id: collection.id,
+ key: List.last(expected_items)["key"]
+ )
+
+ assert {:ok, ^last_inserted_at, 0} =
+ cursor |> Base.decode64!() |> DateTime.from_iso8601()
+ end
+
+ test "up exactly to the limit from a cursor", %{conn: conn} do
+ user = insert(:user)
+ project = insert(:project, project_users: [%{user: user}])
+ collection = insert(:collection, project: project)
+ limit = 100
+
+ all_items =
+ Enum.map(1..(2 * limit), fn i ->
+ insert(:collection_item,
+ collection: collection,
+ key: "key#{i}",
+ value: "value#{i}"
+ )
+ end)
+
+ token = Lightning.Accounts.generate_api_token(user)
+
+ conn = assign_bearer(conn, token)
+
+ assert %{state: :chunked} =
+ conn =
+ get(conn, ~p"/collections/#{collection.name}", limit: limit)
+
+ expected_items =
+ all_items
+ |> Enum.take(limit)
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ assert %{
+ "items" => response_items,
+ "cursor" => cursor
+ } = json_response(conn, 200)
+
+ assert response_items == expected_items
+
+ assert %{state: :chunked} =
+ conn =
+ get(conn, ~p"/collections/#{collection.name}",
+ cursor: cursor,
+ limit: limit
+ )
+
+ expected_items =
+ all_items
+ |> Enum.drop(limit)
+ |> Enum.map(&Jason.encode!/1)
+ |> Enum.map(&Jason.decode!/1)
+
+ assert %{
+ "items" => ^expected_items,
+ "cursor" => nil
+ } = json_response(conn, 200)
+ end
+ end
+end
diff --git a/test/lightning_web/live/collection_live_test.exs b/test/lightning_web/live/collection_live_test.exs
new file mode 100644
index 0000000000..a745e08a43
--- /dev/null
+++ b/test/lightning_web/live/collection_live_test.exs
@@ -0,0 +1,195 @@
+defmodule LightningWeb.CollectionLiveTest do
+ use LightningWeb.ConnCase, async: true
+
+ import Phoenix.LiveViewTest
+ import Lightning.Factories
+
+ describe "Index as a regular user" do
+ setup :register_and_log_in_user
+
+ test "Regular user cannot access the collections page", %{conn: conn} do
+ {:ok, _view, html} =
+ live(conn, ~p"/settings/collections")
+ |> follow_redirect(conn, ~p"/projects")
+
+ assert html =~ "No Access"
+ end
+ end
+
+ describe "Index as a superuser" do
+ setup :register_and_log_in_superuser
+
+ test "Superuser can access the collections page", %{conn: conn} do
+ {:ok, _view, html} = live(conn, ~p"/settings/collections")
+
+ assert html =~ "Collections"
+ assert html =~ "No collection found. Create a new one."
+ end
+
+ test "Collections are listed for superuser", %{conn: conn} do
+ collection_1 =
+ insert(:collection,
+ name: "Collection A",
+ project: build(:project, name: "Project A")
+ )
+
+ collection_2 =
+ insert(:collection,
+ name: "Collection B",
+ project: build(:project, name: "Project B")
+ )
+
+ {:ok, view, _html} = live(conn, ~p"/settings/collections")
+
+ assert has_element?(view, "tr#collections-table-row-#{collection_1.id}")
+ assert has_element?(view, "tr#collections-table-row-#{collection_2.id}")
+ end
+
+ test "Collections can be sorted by name for superuser", %{conn: conn} do
+ insert(:collection, name: "B Collection")
+ insert(:collection, name: "A Collection")
+
+ {:ok, view, _html} = live(conn, ~p"/settings/collections")
+
+ sorted_names = get_sorted_collection_names(view)
+ assert sorted_names == ["A Collection", "B Collection"]
+
+ view |> element("span[phx-click='sort']") |> render_click()
+ sorted_names = get_sorted_collection_names(view)
+ assert sorted_names == ["B Collection", "A Collection"]
+ end
+
+ test "Superuser can delete a collection", %{conn: conn} do
+ collection = insert(:collection, name: "Delete Me")
+
+ {:ok, view, _html} = live(conn, ~p"/settings/collections")
+
+ assert has_element?(view, "tr#collections-table-row-#{collection.id}")
+
+ {:ok, view, html} =
+ view
+ |> element("#delete-collection-#{collection.id}-modal_confirm_button")
+ |> render_click()
+ |> follow_redirect(conn, ~p"/settings/collections")
+
+ assert html =~ "Collection deleted successfully"
+
+ refute has_element?(view, "tr#collections-table-row-#{collection.id}")
+ end
+
+ test "Superuser can create a collection via the modal", %{
+ conn: conn,
+ user: user
+ } do
+ project = insert(:project, project_users: [%{user: user}])
+ {:ok, view, _html} = live(conn, ~p"/settings/collections")
+
+ assert has_element?(view, "#collection-form-new")
+
+ view
+ |> form("#collection-form-new", collection: %{raw_name: "New Collection"})
+ |> render_change()
+
+ assert has_element?(view, "input[type='text'][value='new-collection']")
+
+ {:ok, _view, html} =
+ view
+ |> form("#collection-form-new",
+ collection: %{raw_name: "New Collection", project_id: project.id}
+ )
+ |> render_submit()
+ |> follow_redirect(conn, ~p"/settings/collections")
+
+ assert html =~ "Collection created successfully"
+ assert html =~ "new-collection"
+ assert html =~ project.name
+ end
+
+ test "Canceling collection creation modal closes the modal", %{conn: conn} do
+ {:ok, view, _html} = live(conn, ~p"/settings/collections")
+
+ assert has_element?(view, "#collection-form-new")
+
+ view
+ |> form("#collection-form-new", collection: %{raw_name: "New Collection"})
+ |> render_change()
+
+ {:ok, _view, html} =
+ view
+ |> element("#cancel-collection-creation-new")
+ |> render_click()
+ |> follow_redirect(conn, ~p"/settings/collections")
+
+ refute html =~ "new-collection"
+ end
+
+ test "Superuser can update a collection via the modal", %{
+ conn: conn,
+ user: user
+ } do
+ project = insert(:project, project_users: [%{user: user}])
+ collection = insert(:collection, name: "Old Collection", project: project)
+
+ {:ok, view, _html} = live(conn, ~p"/settings/collections")
+
+ assert has_element?(view, "#collection-form-#{collection.id}")
+
+ view
+ |> form("#collection-form-#{collection.id}",
+ collection: %{raw_name: "Updated Collection"}
+ )
+ |> render_change()
+
+ assert has_element?(view, "input[type='text'][value='updated-collection']")
+
+ {:ok, _view, html} =
+ view
+ |> form("#collection-form-#{collection.id}",
+ collection: %{raw_name: "Updated Collection", project_id: project.id}
+ )
+ |> render_submit()
+ |> follow_redirect(conn, ~p"/settings/collections")
+
+ assert html =~ "Collection updated successfully"
+ assert html =~ "updated-collection"
+ assert html =~ project.name
+ end
+
+ test "Creating a collection with a name that already exists fails", %{
+ conn: conn,
+ user: user
+ } do
+ project = insert(:project, project_users: [%{user: user}])
+
+ collection = insert(:collection, name: "duplicate-name", project: project)
+
+ {:ok, view, _html} = live(conn, ~p"/settings/collections")
+
+ assert has_element?(view, "#collection-form-new")
+
+ view
+ |> form("#collection-form-new",
+ collection: %{raw_name: collection.name, project_id: project.id}
+ )
+ |> render_change()
+
+ view
+ |> form("#collection-form-new")
+ |> render_submit() =~ "A collection with this name already exists"
+ end
+ end
+
+ defp get_sorted_collection_names(view) do
+ html = render(view)
+
+ html
+ |> Floki.parse_document!()
+ |> Floki.find("#collections-table tr")
+ |> Enum.map(fn tr ->
+ Floki.find(tr, "td:nth-child(1)")
+ |> Floki.text()
+ |> String.trim()
+ end)
+ |> Enum.filter(fn name -> String.length(name) > 0 end)
+ end
+end
diff --git a/test/lightning_web/live/tokens_live_test.exs b/test/lightning_web/live/tokens_live_test.exs
index c11fad0011..516bb3302c 100644
--- a/test/lightning_web/live/tokens_live_test.exs
+++ b/test/lightning_web/live/tokens_live_test.exs
@@ -49,7 +49,7 @@ defmodule LightningWeb.TokensLiveTest do
|> Floki.parse_fragment!()
|> Floki.attribute("value")
|> Floki.text()
- |> String.length() == 275
+ |> String.length() == 539
end
test "See a list of tokens", %{conn: conn} do
diff --git a/test/support/conn_case.ex b/test/support/conn_case.ex
index 945ea06f2d..3306b2867f 100644
--- a/test/support/conn_case.ex
+++ b/test/support/conn_case.ex
@@ -27,6 +27,7 @@ defmodule LightningWeb.ConnCase do
import Plug.Conn
import Phoenix.ConnTest
import LightningWeb.ConnCase
+ import LightningWeb.ConnHelpers
alias LightningWeb.Router.Helpers, as: Routes
alias Lightning.Repo
diff --git a/test/support/conn_helpers.ex b/test/support/conn_helpers.ex
new file mode 100644
index 0000000000..e4cb38e605
--- /dev/null
+++ b/test/support/conn_helpers.ex
@@ -0,0 +1,5 @@
+defmodule LightningWeb.ConnHelpers do
+ def assign_bearer(conn, token) do
+ conn |> Plug.Conn.put_req_header("authorization", "Bearer #{token}")
+ end
+end
diff --git a/test/support/factories.ex b/test/support/factories.ex
index 2fa524a33f..a43ba3226e 100644
--- a/test/support/factories.ex
+++ b/test/support/factories.ex
@@ -318,6 +318,26 @@ defmodule Lightning.Factories do
}
end
+ def collection_factory do
+ %Lightning.Collections.Collection{
+ project: build(:project),
+ name: sequence(:name, &"collection-#{&1}")
+ }
+ end
+
+ def collection_item_factory do
+ %Lightning.Collections.Item{
+ collection: build(:collection),
+ key: sequence(:key, &"key-#{&1}"),
+ value: sequence(:value, &"value-#{&1}"),
+ inserted_at:
+ sequence(
+ :inserted_at,
+ &DateTime.add(DateTime.utc_now(), &1, :microsecond)
+ )
+ }
+ end
+
# ----------------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------------
@@ -345,7 +365,7 @@ defmodule Lightning.Factories do
sequence(:timestamp, fn i ->
DateTime.utc_now()
|> DateTime.add(ago, scale)
- |> DateTime.add(i * gap, :second)
+ |> DateTime.add(i * gap, scale)
end)
end