Send an archive message to a device on connect

If the deployment has a configured archive and the device is of recent
enough version to understand the message.

lib/nerves_hub/devices/device_link.ex

@@ -8,6 +8,7 @@ defmodule NervesHub.Devices.DeviceLink do
 
   use GenServer
 
+  alias NervesHub.Archives
   alias NervesHub.AuditLogs
   alias NervesHub.Deployments
   alias NervesHub.Devices
@@ -30,6 +31,7 @@ defmodule NervesHub.Devices.DeviceLink do
       :reference_id,
       :transport_pid,
       :transport_ref,
+      :device_api_version,
       :update_started?
     ]
   end
@@ -184,6 +186,8 @@ defmodule NervesHub.Devices.DeviceLink do
         Tracer.set_attribute("nerves_hub.device.id", device.id)
         Tracer.set_attribute("nerves_hub.device.identifier", device.identifier)
 
+        state = %{state | device_api_version: Map.get(params, "device_api_version", "1.0.0")}
+
         description = "device #{device.identifier} connected to the server"
 
         AuditLogs.audit_with_ref!(
@@ -197,7 +201,7 @@ defmodule NervesHub.Devices.DeviceLink do
           device
           |> Devices.verify_deployment()
           |> Deployments.set_deployment()
-          |> Repo.preload(deployment: [:firmware])
+          |> Repo.preload(deployment: [:archive, :firmware])
 
         # clear out any inflight updates, there shouldn't be one at this point
         # we might make a new one right below it, so clear it beforehand
@@ -256,6 +260,23 @@ defmodule NervesHub.Devices.DeviceLink do
               {:error, ex}
           end
 
+        if Version.match?(state.device_api_version, ">= 2.0.0") do
+          if device.deployment && device.deployment.archive do
+            archive = device.deployment.archive
+
+            push_cb.("archive", %{
+              size: archive.size,
+              uuid: archive.uuid,
+              version: archive.version,
+              description: archive.description,
+              platform: archive.platform,
+              architecture: archive.architecture,
+              uploaded_at: archive.inserted_at,
+              url: Archives.url(archive)
+            })
+          end
+        end
+
         state =
           case monitor do
             {transport_pid, ref_id} ->

test/nerves_hub_web/channels/websocket_test.exs

@@ -93,7 +93,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(@socket_config)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
 
       device =
@@ -133,7 +133,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(config)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
 
       device =
@@ -211,7 +211,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(opts)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
 
       device =
@@ -289,7 +289,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(opts)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
 
       device =
@@ -331,7 +331,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(@socket_config)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
       update = SocketClient.wait_update(socket)
       assert %{"update_available" => true, "firmware_url" => _, "firmware_meta" => %{}} = update
@@ -370,7 +370,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(@socket_config)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
       reply = SocketClient.reply(socket)
 
@@ -405,7 +405,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(@socket_config)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
       reply = SocketClient.reply(socket)
       assert %{} = reply
@@ -449,7 +449,6 @@ defmodule NervesHubWeb.WebsocketTest do
 
       # Device has updated and no longer matches the attached deployment
       SocketClient.join(socket, "device", %{
-        "device_api_version" => "2.0.0",
         "nerves_fw_uuid" => Ecto.UUID.generate(),
         "nerves_fw_product" => "test",
         "nerves_fw_architecture" => "arm",
@@ -505,7 +504,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(opts)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
 
       device =
@@ -568,7 +567,7 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(opts)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
 
       device =
@@ -621,13 +620,13 @@ defmodule NervesHubWeb.WebsocketTest do
 
       {:ok, socket} = SocketClient.start_link(opts)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
       GenServer.stop(socket)
 
       {:ok, socket} = SocketClient.start_link(opts)
       SocketClient.wait_connect(socket)
-      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.join(socket, "device")
       SocketClient.wait_join(socket)
 
       [%{last_used: updated_last_used}] = Devices.get_ca_certificates(org)
@@ -637,4 +636,47 @@ defmodule NervesHubWeb.WebsocketTest do
       SocketClient.close(socket)
     end
   end
+
+  describe "archives" do
+    test "on connect receive an archive", %{user: user} do
+      org = Fixtures.org_fixture(user)
+      org_key = Fixtures.org_key_fixture(org)
+
+      {device, firmware} = device_fixture(user, %{identifier: @valid_serial}, org)
+
+      firmware = Repo.preload(firmware, [:product])
+      product = firmware.product
+
+      archive = Fixtures.archive_fixture(org_key, product)
+
+      deployment =
+        Fixtures.deployment_fixture(org, firmware, %{
+          name: "beta",
+          conditions: %{
+            "tags" => ["beta"]
+          }
+        })
+
+      {:ok, deployment} = Deployments.update_deployment(deployment, %{archive_id: archive.id})
+      {:ok, _deployment} = Deployments.update_deployment(deployment, %{is_active: true})
+
+      Fixtures.device_certificate_fixture(device)
+
+      {:ok, socket} = SocketClient.start_link(@socket_config)
+      SocketClient.wait_connect(socket)
+      SocketClient.join(socket, "device", %{"device_api_version" => "2.0.0"})
+      SocketClient.wait_join(socket)
+
+      device =
+        NervesHub.Repo.get(Device, device.id)
+        |> NervesHub.Repo.preload(:org)
+
+      assert Tracker.online?(device)
+
+      archive = SocketClient.wait_archive(socket)
+      assert %{"url" => _, "version" => _} = archive
+
+      SocketClient.close(socket)
+    end
+  end
 end

test/support/socket_client.ex

@@ -57,6 +57,27 @@ defmodule SocketClient do
     end
   end
 
+  def received_archive?(socket) do
+    GenServer.call(socket, :received_archive?)
+  end
+
+  def wait_archive(_, _ \\ nil)
+
+  def wait_archive(socket, nil) do
+    timeout = 2_000
+    {:ok, t_ref} = :timer.exit_after(timeout, "Timed out waiting for a firmware archive")
+    wait_archive(socket, t_ref)
+  end
+
+  def wait_archive(socket, timer) do
+    if __MODULE__.received_archive?(socket) do
+      :timer.cancel(timer)
+      GenServer.call(socket, :archive_message)
+    else
+      wait_archive(socket, timer)
+    end
+  end
+
   def received_update?(socket) do
     GenServer.call(socket, :received_update?)
   end
@@ -90,6 +111,8 @@ defmodule SocketClient do
       |> assign(:reply, nil)
       |> assign(:received_update?, false)
       |> assign(:update, nil)
+      |> assign(:received_archive?, false)
+      |> assign(:archive, nil)
 
     {:ok, socket}
   end
@@ -119,6 +142,15 @@ defmodule SocketClient do
     {:ok, socket}
   end
 
+  def handle_message("device", "archive", message, socket) do
+    socket =
+      socket
+      |> assign(:received_archive?, true)
+      |> assign(:archive, message)
+
+    {:ok, socket}
+  end
+
   @impl true
   def handle_call(:connected?, _from, socket) do
     {:reply, socket.assigns.connected?, socket}
@@ -128,6 +160,14 @@ defmodule SocketClient do
     {:reply, socket.assigns.joined?, socket}
   end
 
+  def handle_call(:received_archive?, _from, socket) do
+    {:reply, socket.assigns.received_archive?, socket}
+  end
+
+  def handle_call(:archive_message, _from, socket) do
+    {:reply, socket.assigns.archive, socket}
+  end
+
   def handle_call(:received_update?, _from, socket) do
     {:reply, socket.assigns.received_update?, socket}
   end