Have the calculator use Oban (#1639)

Author: joshk

config/config.exs

@@ -58,6 +58,7 @@ config :nerves_hub, Oban,
     delete_archive: 1,
     delete_firmware: 1,
     device: 1,
+    device_deployment_calculations: 5,
     firmware_delta_builder: 2,
     truncate: 1,
     # temporary, will remove in November

lib/nerves_hub/deployments.ex

@@ -197,13 +197,12 @@ defmodule NervesHub.Deployments do
   end
 
   defp recalculate_devices(%{recalculation_type: :calculator_queue} = deployment, changeset) do
-    _ =
-      if Enum.any?(
-           [:conditions, :is_active, :recalculation_type],
-           &Map.has_key?(changeset.changes, &1)
-         ) do
-        create_inflight_checks(deployment)
-      end
+    if Enum.any?(
+         [:conditions, :is_active, :recalculation_type],
+         &Map.has_key?(changeset.changes, &1)
+       ) do
+      schedule_deployment_calculations(deployment)
+    end
 
     :ok
   end
@@ -334,15 +333,18 @@ defmodule NervesHub.Deployments do
 
   Also clears any previous inflight checks for this deployment.
   """
-  def create_inflight_checks(deployment) do
-    delete_inflight_checks(deployment)
-
+  def schedule_deployment_calculations(deployment) do
     query =
       Device
       |> select([d], %{
-        deployment_id: ^deployment.id,
-        device_id: d.id,
-        inserted_at: ^DateTime.utc_now()
+        worker: "NervesHub.Workers.DeviceCalculateDeployment",
+        queue: "device_deployment_calculations",
+        args:
+          fragment(
+            "json_build_object('device_id', ?, 'deployment_id', ?::integer)",
+            d.id,
+            ^deployment.id
+          )
       })
       |> where([d], not is_nil(d.connection_last_seen_at))
       |> where(
@@ -352,8 +354,9 @@ defmodule NervesHub.Deployments do
       )
       |> where([d], d.firmware_metadata["platform"] == ^deployment.firmware.platform)
       |> where([d], d.firmware_metadata["architecture"] == ^deployment.firmware.architecture)
+      |> where([d], fragment("? <@ ?", ^deployment.conditions["tags"], d.tags))
 
-    Repo.insert_all(InflightDeploymentCheck, query)
+    Repo.insert_all(Oban.Job, query)
   end
 
   @doc """

lib/nerves_hub/deployments/calculator.ex

@@ -10,7 +10,6 @@ defmodule NervesHub.Deployments.Monitor do
   alias NervesHub.DeploymentDynamicSupervisor
   alias NervesHub.Deployments
   alias NervesHub.Deployments.Orchestrator
-  alias NervesHub.InflightDeploymentCheckDynamicSupervisor
   alias Phoenix.PubSub
   alias Phoenix.Socket.Broadcast
 
@@ -37,13 +36,7 @@ defmodule NervesHub.Deployments.Monitor do
             {Deployments.Orchestrator, deployment}
           )
 
-        {:ok, calculator_pid} =
-          DynamicSupervisor.start_child(
-            InflightDeploymentCheckDynamicSupervisor,
-            {Deployments.Calculator, deployment}
-          )
-
-        {deployment.id, %{orchestrator_pid: orchestrator_pid, calculator_pid: calculator_pid}}
+        {deployment.id, %{orchestrator_pid: orchestrator_pid}}
       end)
 
     {:noreply, %{state | deployments: deployments}}
@@ -58,17 +51,8 @@ defmodule NervesHub.Deployments.Monitor do
         {Deployments.Orchestrator, deployment}
       )
 
-    {:ok, calculator_pid} =
-      DynamicSupervisor.start_child(
-        InflightDeploymentCheckDynamicSupervisor,
-        {Deployments.Calculator, deployment}
-      )
-
     deployments =
-      Map.put(state.deployments, deployment.id, %{
-        orchestrator_pid: orchestrator_pid,
-        calculator_pid: calculator_pid
-      })
+      Map.put(state.deployments, deployment.id, %{orchestrator_pid: orchestrator_pid})
 
     {:noreply, %{state | deployments: deployments}}
   end

lib/nerves_hub/deployments/monitor.ex

@@ -11,9 +11,7 @@ defmodule NervesHub.Deployments.Supervisor do
     children = [
       {Registry, keys: :unique, name: NervesHub.Deployments},
       NervesHub.Deployments.Monitor,
-      {DynamicSupervisor, strategy: :one_for_one, name: NervesHub.DeploymentDynamicSupervisor},
-      {DynamicSupervisor,
-       strategy: :one_for_one, name: NervesHub.InflightDeploymentCheckDynamicSupervisor}
+      {DynamicSupervisor, strategy: :one_for_one, name: NervesHub.DeploymentDynamicSupervisor}
     ]
 
     Supervisor.init(children, strategy: :one_for_one)

lib/nerves_hub/deployments/supervisor.ex

@@ -0,0 +1,44 @@
+defmodule NervesHub.Workers.DeviceCalculateDeployment do
+  use Oban.Worker,
+    queue: :device_deployment_calculations,
+    max_attempts: 5
+
+  alias NervesHub.Devices.Device
+  alias NervesHub.Deployments
+  alias NervesHub.Deployments.Deployment
+  alias NervesHub.Repo
+
+  @impl true
+  def perform(%Oban.Job{args: %{"device_id" => device_id, "deployment_id" => deployment_id}}) do
+    device = Repo.get!(Device, device_id)
+    deployment = Deployments.get_deployment!(deployment_id) |> Deployment.with_firmware()
+
+    if !is_nil(device.deployment_id) && device.deployment_id != deployment.id do
+      :ok
+    else
+      if deployment.is_active &&
+           !is_nil(device.connection_last_seen_at) &&
+           device.product_id == deployment.product_id &&
+           device.firmware_metadata.platform == deployment.firmware.platform &&
+           device.firmware_metadata.architecture == deployment.firmware.architecture &&
+           Enum.all?(deployment.conditions["tags"], &Enum.member?(device.tags, &1)) &&
+           Deployments.version_match?(device, deployment) do
+        device
+        |> Ecto.Changeset.change(%{deployment_id: deployment.id})
+        |> Repo.update!()
+      else
+        device
+        |> Ecto.Changeset.change(%{deployment_id: nil})
+        |> Repo.update!()
+      end
+
+      Phoenix.PubSub.broadcast(
+        NervesHub.PubSub,
+        "device:#{device.id}",
+        %Phoenix.Socket.Broadcast{event: "devices/updated"}
+      )
+    end
+
+    :ok
+  end
+end