[JET-2026] feature: use dynamic repo to execute action

config/config.exs

@@ -1,5 +1,5 @@
 import Config
 
-config :resource_kit, ecto_repos: [ResourceKit.Repo]
+config :resource_kit, ecto_repos: [ResourceKitCLI.Repo]
 
 import_config "#{config_env()}.exs"

config/dev.exs

@@ -2,4 +2,12 @@ import Config
 
 config :resource_kit, ResourceKit.Deref, adapter: ResourceKit.Deref.Local
 
-config :resource_kit, ResourceKitCLI.Endpoint, server: true
+config :resource_kit, ResourceKit.Repo, adapter: ResourceKitCLI.Repo
+
+config :resource_kit, ResourceKitCLI.Repo,
+  hostname: "localhost",
+  database: "resource_kit_dev",
+  username: "postgres",
+  password: "postgres"
+
+config :resource_kit, ResourceKitCLI.Server, server: true

config/prod.exs

@@ -0,0 +1,5 @@
+import Config
+
+config :resource_kit, ResourceKit.Repo, adapter: ResourceKitCLI.Repo
+
+config :resource_kit, ResourceKitCLI.Server, server: true

config/test.exs

@@ -3,7 +3,9 @@ import Config
 config :resource_kit, ResourceKit.Deref,
   adapter: {ResourceKit.Deref.Local, directory: "test/fixtures"}
 
-config :resource_kit, ResourceKit.Repo,
+config :resource_kit, ResourceKit.Repo, adapter: ResourceKitCLI.Repo
+
+config :resource_kit, ResourceKitCLI.Repo,
   hostname: "localhost",
   database: "resource_kit_test",
   username: "postgres",

lib/resource_kit/action/skeleton.ex

@@ -74,8 +74,7 @@ defmodule ResourceKit.Action.Skeleton do
       end
 
       defp __execute__(action, references, params, opts) do
-        root = Keyword.fetch!(opts, :root)
-        context = %ExecuteToken.Context{root: root, current: root}
+        context = ExecuteToken.Context.new(opts)
 
         %ExecuteToken{action: action, references: references, params: params, context: context}
         |> Pluggable.run([&__MODULE__.execute(&1, [])])

lib/resource_kit/pipeline/execute/run.ex

@@ -9,24 +9,23 @@ defmodule ResourceKit.Pipeline.Execute.Run do
 
   @behaviour Pluggable
 
-  use TypedStruct
-
-  typedstruct module: Options do
-    field :repo, module(), default: ResourceKit.Repo
-  end
-
   alias ResourceKit.Pipeline.Execute.Token
 
   @impl Pluggable
-  def init(args), do: struct(Options, args)
+  def init(args), do: args
 
   @impl Pluggable
   def call(%Token{halted: true} = token, _opts), do: token
 
-  def call(%Token{} = token, %Options{repo: repo}) do
-    token
-    |> Token.fetch_assign!(:multi)
-    |> repo.transaction()
+  def call(%Token{} = token, _opts) do
+    %Token{context: %Token.Context{dynamic_repo: dynamic_repo}} = token
+
+    dynamic_repo
+    |> ResourceKit.Repo.execute(fn repo ->
+      token
+      |> Token.fetch_assign!(:multi)
+      |> repo.transaction()
+    end)
     |> case do
       {:ok, changes} -> Token.put_assign(token, :changes, changes)
       {:error, _operation, reason, _changes} -> Token.put_error(token, reason)

lib/resource_kit/pipeline/execute/token.ex

@@ -1,16 +1,29 @@
 defmodule ResourceKit.Pipeline.Execute.Token do
   @moduledoc false
 
-  use TypedStruct
   use ResourceKit.Pipeline.Token
 
   alias ResourceKit.Types
 
   alias ResourceKit.Schema.Ref
 
-  typedstruct module: Context do
-    field :root, URI.t(), enforce: true
-    field :current, URI.t(), enforce: true
+  defmodule Context do
+    @moduledoc false
+
+    use TypedStruct
+
+    typedstruct do
+      field :root, URI.t(), enforce: true
+      field :current, URI.t(), enforce: true
+      field :dynamic_repo, ResourceKit.Repo.dynamic_repo(), enforce: true
+    end
+
+    def new(args) do
+      root = Keyword.fetch!(args, :root)
+      dynamic_repo = Keyword.fetch!(args, :dynamic_repo)
+
+      struct(__MODULE__, root: root, current: root, dynamic_repo: dynamic_repo)
+    end
   end
 
   token do

lib/resource_kit/repo.ex

@@ -1,5 +1,23 @@
 defmodule ResourceKit.Repo do
-  use Ecto.Repo,
-    otp_app: :resource_kit,
-    adapter: Ecto.Adapters.Postgres
+  @moduledoc false
+
+  @adapter Application.compile_env!(:resource_kit, [__MODULE__, :adapter])
+
+  @type dynamic_repo() :: atom() | pid()
+
+  @spec adapter() :: module()
+  def adapter, do: @adapter
+
+  @spec execute(dynamic_repo :: dynamic_repo(), callback :: (Ecto.Repo.t() -> result)) :: result
+        when result: var
+  def execute(dynamic_repo, callback) do
+    default = @adapter.get_dynamic_repo()
+
+    try do
+      @adapter.put_dynamic_repo(dynamic_repo)
+      callback.(@adapter)
+    after
+      @adapter.put_dynamic_repo(default)
+    end
+  end
 end

lib/resource_kit_cli/application.ex

@@ -0,0 +1,12 @@
+defmodule ResourceKitCLI.Application do
+  @moduledoc false
+
+  use Application
+
+  @impl Application
+  def start(_type, _args) do
+    children = [ResourceKitCLI.Server]
+
+    Supervisor.start_link(children, strategy: :one_for_one, name: __MODULE__)
+  end
+end

lib/resource_kit_cli/endpoint.ex

@@ -1,28 +1,15 @@
 defmodule ResourceKitCLI.Endpoint do
   @moduledoc false
 
-  use Supervisor
-
-  @spec start_link(args :: keyword()) :: Supervisor.on_start()
-  def start_link(args) do
-    Supervisor.start_link(__MODULE__, configuration(args), name: __MODULE__)
-  end
-
-  @impl Supervisor
-  def init(args) do
-    {server, options} = Keyword.pop(args, :server, false)
-
-    if server do
-      Supervisor.init([{Bandit, options}], strategy: :one_for_one)
-    else
-      :ignore
-    end
+  @spec child_spec(args :: keyword()) :: Supervisor.child_spec()
+  def child_spec(args) do
+    Supervisor.child_spec({Bandit, configuration(args)}, id: __MODULE__)
   end
 
   defp configuration(args) do
     :resource_kit
     |> Application.get_env(__MODULE__, [])
     |> Keyword.merge(args)
-    |> Keyword.put(:plug, ResourceKitPlug.Router)
+    |> Keyword.put(:plug, ResourceKitCLI.Plug)
   end
 end

lib/resource_kit_cli/plug.ex

@@ -0,0 +1,12 @@
+defmodule ResourceKitCLI.Plug do
+  @moduledoc false
+
+  use Plug.Builder
+
+  plug :put_dynamic
+  plug ResourceKitPlug.Router
+
+  defp put_dynamic(conn, _opts) do
+    ResourceKitPlug.Router.put_dynamic(conn, ResourceKit.Repo.adapter())
+  end
+end

lib/resource_kit_cli/repo.ex

@@ -0,0 +1,5 @@
+defmodule ResourceKitCLI.Repo do
+  use Ecto.Repo,
+    otp_app: :resource_kit,
+    adapter: Ecto.Adapters.Postgres
+end

lib/resource_kit_cli/server.ex

@@ -0,0 +1,25 @@
+defmodule ResourceKitCLI.Server do
+  @moduledoc false
+
+  use Supervisor
+
+  @spec start_link(args :: keyword()) :: Supervisor.on_start()
+  def start_link(args) do
+    Supervisor.start_link(__MODULE__, configuration(args), name: __MODULE__)
+  end
+
+  @impl Supervisor
+  def init(args) do
+    if Keyword.get(args, :server, false) do
+      Supervisor.init([ResourceKitCLI.Repo, ResourceKitCLI.Endpoint], strategy: :one_for_one)
+    else
+      :ignore
+    end
+  end
+
+  defp configuration(args) do
+    :resource_kit
+    |> Application.get_env(__MODULE__, [])
+    |> Keyword.merge(args)
+  end
+end

lib/resource_kit_plug/controller.ex

@@ -6,10 +6,12 @@ defmodule ResourceKitPlug.Controller do
 
   for type <- [:insert, :list] do
     @spec unquote(type)(request :: map(), ctx :: PhxJsonRpc.Router.Context.t()) :: map()
-    def unquote(type)(request, _ctx) do
+    def unquote(type)(request, %PhxJsonRpc.Router.Context{
+          meta_data: %{dynamic_repo: dynamic_repo}
+        }) do
       with {:ok, request} <- cast_request(request),
            {:ok, action} <- fetch_action(request),
-           {:ok, result} <- run(request, unquote(type), action) do
+           {:ok, result} <- run(request, unquote(type), action, dynamic_repo: dynamic_repo) do
         result
       end
     end
@@ -53,8 +55,8 @@ defmodule ResourceKitPlug.Controller do
     end
   end
 
-  defp run(%Request{uri: uri, params: params}, type, action) do
-    case apply(ResourceKit, type, [action, params, [root: uri]]) do
+  defp run(%Request{uri: uri, params: params}, type, action, opts) do
+    case apply(ResourceKit, type, [action, params, Keyword.put(opts, :root, uri)]) do
       {:ok, result} ->
         {:ok, result}
 

lib/resource_kit_plug/router.ex

@@ -3,19 +3,56 @@ defmodule ResourceKitPlug.Router do
 
   use Plug.Router
 
+  @dynamic_private :resource_kit_plug_dynamic_repo
+
   plug Plug.Logger
   plug Plug.Parsers, parsers: [{:json, json_decoder: Jason}]
   plug :match
   plug :dispatch
 
+  @spec put_dynamic(conn :: Plug.Conn.t(), dynamic_repo :: ResourceKit.Repo.dynamic_repo()) ::
+          Plug.Conn.t()
+  def put_dynamic(%Plug.Conn{} = conn, dynamic_repo) do
+    put_private(conn, @dynamic_private, dynamic_repo)
+  end
+
   post "/rpc/actions", do: handle(conn, ResourceKitPlug.Service)
 
   defp handle(%Plug.Conn{} = conn, service) do
     import PhxJsonRpcWeb.Views.Helpers
 
-    conn
-    |> put_resp_header("content-type", "application/json; charset=utf-8")
-    |> send_json(conn.params |> service.handle() |> render_json())
+    case fetch_dynamic(conn) do
+      {:ok, dynamic_repo} ->
+        conn
+        |> put_resp_header("content-type", "application/json; charset=utf-8")
+        |> send_json(
+          conn.params
+          |> service.handle(%{dynamic_repo: dynamic_repo})
+          |> render_json()
+        )
+
+      :error ->
+        raise RuntimeError, """
+        Could not get dynamic repo from conn, please set dynamic repo before ResourceKitPlug.Router:
+
+        ```
+        defmodule MyApp.Plug do
+          use Plug.Builder
+
+          plug :put_dynamic
+          plug ResourceKitPlug.Router
+
+          defp put_dynamic(conn, _opts) do
+            ResourceKitPlug.Router.put_dynamic_repo(conn, MyApp.Repo)
+          end
+        end
+        ```
+        """
+    end
+  end
+
+  defp fetch_dynamic(%Plug.Conn{private: private}) do
+    Map.fetch(private, @dynamic_private)
   end
 
   defp send_json(%Plug.Conn{} = conn, data) do

mix.exs

@@ -17,7 +17,7 @@ defmodule ResourceKit.MixProject do
   def application do
     [
       extra_applications: [:logger],
-      mod: {ResourceKit.Application, []}
+      mod: {ResourceKitCLI.Application, []}
     ]
   end
 

test/resource_kit/action/insert_movie_test.exs

@@ -17,11 +17,10 @@ defmodule ResourceKit.Action.InsertMovieTest do
 
   setup :setup_tables
   setup :load_jsons
+  setup :setup_options
 
   @tag jsons: [action: "actions/insert_movie.json"]
-  test "works", %{action: action} do
-    root = URI.new!("actions/insert_movie.json")
-
+  test "works", %{action: action, opts: opts} do
     params = %{
       "title" => "Spy x Family Code: White",
       "likes" => 2878,
@@ -39,13 +38,11 @@ defmodule ResourceKit.Action.InsertMovieTest do
               "release_date" => ~D[2024-04-30],
               "created_at" => ~U[2023-12-22 14:23:07.000000Z],
               "tags" => ["Animation", "Comedy"]
-            }} = ResourceKit.insert(action, params, root: root)
+            }} = ResourceKit.insert(action, params, opts)
   end
 
   @tag jsons: [action: "actions/insert_movie.json"]
-  test "failed", %{action: action} do
-    root = URI.new!("actions/insert_movie.json")
-
+  test "failed", %{action: action, opts: opts} do
     params = %{
       "title" => "Spy x Family Code: White",
       "likes" => 2878,
@@ -54,7 +51,13 @@ defmodule ResourceKit.Action.InsertMovieTest do
       "tags" => ["Animation", "Comedy"]
     }
 
-    assert {:error, changeset} = ResourceKit.insert(action, params, root: root)
+    assert {:error, changeset} = ResourceKit.insert(action, params, opts)
     assert match?(%{release_date: ["is invalid"]}, errors_on(changeset))
   end
+
+  defp setup_options(%{jsons: jsons}) do
+    uri = jsons |> Keyword.fetch!(:action) |> URI.new!()
+
+    [opts: [root: uri, dynamic_repo: ResourceKit.Repo.adapter()]]
+  end
 end