Merge pull request #27 from OpenFn/adaptor_service

AdaptorService

Author: taylordowns2000

.gitignore

@@ -27,4 +27,7 @@ engine-*.tar
 
 /tmp/
 
-excoveralls.json
+excoveralls.json
+
+node_modules
+package*.json

.tool-versions

@@ -0,0 +1,3 @@
+nodejs 12.20.2
+elixir 1.11.3-otp-23
+erlang 23.2.7

lib/engine.ex

@@ -21,21 +21,8 @@ defmodule Engine do
   @doc """
   DEPRECATED
   """
-  def execute_sync(%Message{} = message, %Job{} = job) do
+  def execute_sync(%Message{}, %Job{}) do
     raise "execute_sync/2 is no longer supported"
-    # {:ok, state_path} = Temp.path(%{prefix: "state", suffix: ".json"})
-    # {:ok, final_state_path} = Temp.path(%{prefix: "final_state", suffix: ".json"})
-    # {:ok, expression_path} = Temp.path(%{prefix: "expression", suffix: ".js"})
-
-    # File.write!(state_path, Jason.encode!(message.body))
-    # File.write!(expression_path, job.expression || "")
-
-    # Engine.ShellRuntime.run(%RunSpec{
-    #   state_path: state_path,
-    #   final_state_path: final_state_path,
-    #   expression_path: expression_path,
-    #   adaptor: job.adaptor
-    # })
   end
 
   def handle_message(run_broadcaster, %Message{} = message) do

lib/engine/adaptor.ex

@@ -0,0 +1,18 @@
+defmodule Engine.Adaptor do
+  @moduledoc false
+  @type install_status :: :present | :installing
+
+  @type t :: %__MODULE__{
+          name: binary(),
+          version: binary(),
+          path: binary(),
+          status: install_status()
+        }
+
+  @enforce_keys [:name, :version]
+  defstruct @enforce_keys ++ [:status, :path]
+
+  def set_present(adaptor) do
+    %{adaptor | status: :present}
+  end
+end

lib/engine/adaptor/repo.ex

@@ -0,0 +1,83 @@
+defmodule Engine.Adaptor.Repo do
+  @callback list_local(path :: String.t()) :: list(Engine.Adaptor.t())
+  def list_local(path, depth \\ 4) when is_binary(path) do
+    System.cmd("find", ~w[#{path} -maxdepth #{depth} -type f -name package.json])
+    |> case do
+      {stdout, 0} ->
+        stdout
+        |> String.trim()
+        |> String.split("\n")
+        |> filter_parent_paths()
+        |> Enum.map(fn package_json ->
+          res = Jason.decode!(File.read!(package_json))
+          get = &Map.get(res, &1)
+
+          %Engine.Adaptor{
+            name: get.("name"),
+            version: get.("version"),
+            path: String.replace_suffix(package_json, "/package.json", ""),
+            status: :present
+          }
+        end)
+
+      {stdout, _} ->
+        raise "Failed to list adaptors from path: #{path}\n#{stdout}"
+    end
+  end
+
+  @doc """
+  ```
+  |------------ alias ---------| |----- source &|| version -------|
+  @openfn/language-common-v1.2.6@npm:@openfn/language-common@1.2.6
+  ```
+  """
+  @callback install(adaptors :: list(String.t()) | String.t(), dir :: String.t()) ::
+              {Collectable.t(), exit_status :: non_neg_integer}
+  @spec install(adaptors :: list(String.t()) | String.t(), dir :: String.t()) ::
+          {Collectable.t(), exit_status :: non_neg_integer}
+  def install(adaptor, dir) when is_binary(adaptor),
+    do: install([adaptor], dir)
+
+  def install(adaptors, dir) when is_list(adaptors) do
+    System.cmd(
+      "/usr/bin/env",
+      [
+        "sh",
+        "-c",
+        """
+        npm install \
+          --no-save \
+          --ignore-scripts \
+          --no-fund \
+          --no-audit \
+          --no-package-lock \
+          --global-style \
+          --prefix #{dir} \
+          #{Enum.join(adaptors, " ")}
+        """
+      ],
+      stderr_to_stdout: true
+    )
+  end
+
+  @doc """
+  Given a list of _potentially_ nested package.json files (i.e. dependancies of
+  our adaptors), `filter_parent_paths/1` reduces the list down to the parent
+  directories by grouping directory names by their shortest common path.
+  """
+  def filter_parent_paths(paths) when is_list(paths) do
+    paths
+    |> Enum.sort(:desc)
+    |> Enum.reduce([], fn path, acc ->
+      base = path |> String.replace("package.json", "")
+
+      parent =
+        acc
+        |> Enum.find(base, fn parent -> String.contains?(base, parent) end)
+
+      acc ++ [parent]
+    end)
+    |> Enum.uniq()
+    |> Enum.map(fn folder -> "#{folder}package.json" end)
+  end
+end

lib/engine/adaptor/service.ex

@@ -0,0 +1,232 @@
+defmodule Engine.Adaptor.Service do
+  @moduledoc """
+  The Adaptor Service is use to query and install adaptors in order to run jobs.
+
+  On startup, it queries the filesystem for `package.json` files and builds up
+  a list of available adaptors.
+
+  ## Configuration
+
+  The service requires at least `:adaptors_path`, which is used to both query
+  which adaptors are installed and when to install new adaptors.
+
+  Another optional setting is: `:repo`, which must point at a module that will be
+  used to do the querying and installing.
+
+  ## Installing Adaptors
+
+  Using the `install/3` function an adaptor can be installed, which will also
+  add it to the list of available adaptors.
+
+  The adaptor is marked as `:installing`, to allow for conditional behaviour
+  elsewhere such as delaying or rejecting processing until the adaptor becomes
+  available.
+
+  ## Looking up adaptors
+
+  The module leans on Elixir's built-in `Version` module to provide version
+  lookups.
+
+  When looking up an adaptor, either a string or a tuple can be used.
+  In the case of requesting the latest version, any one of these will return
+  the latest version the service is aware of.
+
+  - `@openfn/language-http`
+  - `@openfn/language-http@latest`
+  - `{"@openfn/language-http", nil}`
+  - `{"@openfn/language-http", "latest"}`
+  - `{~r/language-http/, "latest"}`
+
+  You can also request a specific version, or use a range specification:
+
+  - `@openfn/language-http@1.2.3`
+  - `{"@openfn/language-http", "~> 1.2.0"}`
+  - `{"@openfn/language-http", "< 2.0.0"}`
+
+  > **NOTE**
+  > More complex npm style install strings like: `">=0.1.0 <0.2.0"`
+  > are not supported.
+  > Generally the tuple style is preferred when using range specifications as
+  > the npm style strings have a simplistic regex splitter.
+
+  See [Version](https://hexdocs.pm/elixir/Version.html) for more details on
+  matching versions.
+  """
+
+  @type package_spec :: {name :: String.t() | Regex.t(), version :: String.t() | nil}
+
+  use Agent
+  require Logger
+
+  defmodule State do
+    @moduledoc false
+
+    @type t :: %__MODULE__{
+            name: GenServer.server(),
+            adaptors: [Engine.Adaptor.t()],
+            adaptors_path: binary(),
+            repo: module()
+          }
+
+    @enforce_keys [:adaptors_path]
+    defstruct @enforce_keys ++ [:name, adaptors: [], repo: Engine.Adaptor.Repo]
+
+    def find_adaptor(%{adaptors: adaptors}, fun) when is_function(fun) do
+      Enum.find(adaptors, fun)
+    end
+
+    def find_adaptor(state, {package_name, version}) do
+      find_adaptor(state, fn %{name: n, version: v} ->
+        n == package_name && (v == version || is_nil(version))
+      end)
+    end
+
+    def refresh_list(state) do
+      %{state | adaptors: state.repo.list_local(state.adaptors_path)}
+    end
+
+    def add_adaptor(state, adaptor) do
+      %{state | adaptors: state.adaptors ++ [adaptor]}
+    end
+
+    def remove_adaptor(state, fun) do
+      %{state | adaptors: Enum.reject(state.adaptors, fun)}
+    end
+  end
+
+  def start_link(opts) do
+    state = struct!(State, opts) |> State.refresh_list()
+
+    Agent.start_link(fn -> state end, name: state.name || __MODULE__)
+  end
+
+  def get_adaptors(agent) do
+    Agent.get(agent, fn state -> state.adaptors end)
+  end
+
+  @spec find_adaptor(Agent.agent(), package :: String.t()) :: Adaptor.t() | nil
+  def find_adaptor(agent, package) when is_binary(package) do
+    find_adaptor(agent, resolve_package_name(package))
+  end
+
+  @spec find_adaptor(Agent.agent(), package_spec()) :: Adaptor.t() | nil
+  def find_adaptor(agent, {package_name, version}) do
+    requirement = version_to_requirement(version)
+
+    get_adaptors(agent)
+    |> Enum.filter(&by_name_and_requirement(&1, package_name, requirement))
+    |> Enum.max_by(
+      fn %{version: version} ->
+        Version.parse!(version)
+      end,
+      Version,
+      fn -> nil end
+    )
+  end
+
+  defp by_name_and_requirement(adaptor, matcher = %Regex{}, requirement) do
+    !!(Regex.match?(matcher, adaptor.name) &&
+         Version.match?(adaptor.version, requirement))
+  end
+
+  defp by_name_and_requirement(adaptor, name, requirement) do
+    !!(match?(%{name: ^name}, adaptor) &&
+         Version.match?(adaptor.version, requirement))
+  end
+
+  defp version_to_requirement(version) do
+    cond do
+      version in ["latest", nil] ->
+        "> 0.0.0"
+
+      String.match?(version, ~r/[<=>]/) ->
+        raise ArgumentError, message: "Version specs not implemented yet."
+
+      true ->
+        version
+    end
+    |> Version.parse_requirement!()
+  end
+
+  def installed?(agent, package_spec) do
+    !!find_adaptor(agent, package_spec)
+  end
+
+  @spec install(Agent.agent(), binary()) ::
+          {:ok, Engine.Adaptor.t()} | {:error, {Collectable.t(), exit_status :: non_neg_integer}}
+  def install(agent, package) when is_binary(package) do
+    install(agent, resolve_package_name(package))
+  end
+
+  @spec install(Agent.agent(), package_spec()) ::
+          {:ok, Engine.Adaptor.t()} | {:error, {Collectable.t(), exit_status :: non_neg_integer}}
+  def install(agent, package_spec) do
+    agent
+    |> find_adaptor(package_spec)
+    |> case do
+      nil -> install!(agent, package_spec)
+      existing -> {:ok, existing}
+    end
+  end
+
+  def install!(agent, {package_name, version} = package_spec) do
+    new_adaptor = %Engine.Adaptor{name: package_name, version: version, status: :installing}
+
+    agent |> Agent.update(&State.add_adaptor(&1, new_adaptor))
+
+    {repo, adaptors_path} =
+      agent
+      |> Agent.get(fn state ->
+        {state.repo, state.adaptors_path}
+      end)
+
+    repo.install(build_aliased_name(package_spec), adaptors_path)
+    |> case do
+      {_stdout, 0} ->
+        Logger.info("Refreshing Adaptor list")
+        agent |> Agent.update(&State.refresh_list/1, 30_000)
+        {:ok, agent |> Agent.get(&State.find_adaptor(&1, {package_name, version}))}
+
+      {stdout, code} ->
+        agent
+        |> Agent.update(fn state ->
+          State.remove_adaptor(state, &match?(^new_adaptor, &1))
+        end)
+
+        {:error, {stdout, code}}
+    end
+  end
+
+  def resolve_package_name(package_name) when is_binary(package_name) do
+    ~r/(@?[\/\d\n\w-]+)(?:@([\d\.\w]+))?$/
+    |> Regex.run(package_name)
+    |> case do
+      [_, name, version] ->
+        {name, version}
+
+      [_, _name] ->
+        {package_name, nil}
+
+      _ ->
+        raise ArgumentError, "Only npm style package names are currently supported"
+    end
+  end
+
+  @doc """
+  Turns a package name and version into a string for NPM.
+
+  Since multiple versions of the same package can be installed, it's important
+  to rely on npms built-in package aliasing.
+
+  E.g. `@openfn/language-http@1.2.8` turns into:
+       `@openfn/language-http-v1.2.8@npm:@openfn/language-http@1.2.8`
+
+  Which is pretty long winded but necessary for the reason above.
+
+  If using this module as a base, it's likely you would need to adaptor this
+  to suit your particular naming strategy.
+  """
+  def build_aliased_name({package, version}) do
+    "#{package}-v#{version}@npm:#{package}@#{version}"
+  end
+end

lib/engine/run/handler.ex

@@ -65,8 +65,8 @@ defmodule Engine.Run.Handler do
 
       @impl Handler
       def log_callback(log_agent, context, args) do
-        Engine.LogAgent.process_chunk(log_agent, args)
-        |> __MODULE__.on_log_emit(context)
+        res = Engine.LogAgent.process_chunk(log_agent, args)
+        res |> __MODULE__.on_log_emit(context)
 
         true
       end