Replace endpoint broadcasts (#1714)

See #1703.

Removes all broadcasting to DeviceEndpoint.

lib/nerves_hub/devices.ex

@@ -21,6 +21,7 @@ defmodule NervesHub.Devices do
   alias NervesHub.Devices.SharedSecretAuth
   alias NervesHub.Devices.InflightUpdate
   alias NervesHub.Devices.UpdatePayload
+  alias NervesHub.Extensions
   alias NervesHub.Firmwares
   alias NervesHub.Firmwares.Firmware
   alias NervesHub.Firmwares.FirmwareMetadata
@@ -1408,11 +1409,7 @@ defmodule NervesHub.Devices do
     |> Repo.update()
     |> tap(fn
       {:ok, _} ->
-        topic = "device:#{device.id}:extensions"
-
-        NervesHubWeb.DeviceEndpoint.broadcast(topic, "attach", %{
-          "extensions" => [extension_string]
-        })
+        Extensions.toggle_extension(device, "attach", extension_string)
 
       _ ->
         :nope
@@ -1426,11 +1423,7 @@ defmodule NervesHub.Devices do
     |> Repo.update()
     |> tap(fn
       {:ok, _} ->
-        topic = "device:#{device.id}:extensions"
-
-        NervesHubWeb.DeviceEndpoint.broadcast(topic, "detach", %{
-          "extensions" => [extension_string]
-        })
+        Extensions.toggle_extension(device, "detach", extension_string)
 
       _ ->
         :nope

lib/nerves_hub/extensions.ex

@@ -64,4 +64,23 @@ defmodule NervesHub.Extensions do
   def module(_key, _ver) do
     :unsupported
   end
+
+  def toggle_extension(target, event, extension) do
+    device_internal_broadcast!(target, event, %{
+      "extensions" => [extension]
+    })
+  end
+
+  defp device_internal_broadcast!(target, event, payload) do
+    Phoenix.Channel.Server.broadcast_from!(
+      NervesHub.PubSub,
+      self(),
+      topic(target),
+      event,
+      payload
+    )
+  end
+
+  defp topic(%NervesHub.Devices.Device{} = device), do: "device:#{device.id}:extensions"
+  defp topic(%NervesHub.Products.Product{} = product), do: "product:#{product.id}:extensions"
 end

lib/nerves_hub/extensions/geo.ex

@@ -48,7 +48,8 @@ defmodule NervesHub.Extensions.Geo do
     {:ok, device} = Devices.update_device(device, %{connection_metadata: metadata})
 
     _ =
-      NervesHubWeb.DeviceEndpoint.broadcast(
+      Phoenix.Channel.Server.broadcast(
+        NervesHub.PubSub,
         "device:#{device.identifier}:internal",
         "location:updated",
         location

lib/nerves_hub/extensions/health.ex

@@ -100,8 +100,8 @@ defmodule NervesHub.Extensions.Health do
   end
 
   defp device_internal_broadcast!(device, event, payload) do
-    topic = "device:#{device.identifier}:extensions"
+    topic = "device:#{device.id}:extensions"
 
-    Phoenix.Channel.Server.broadcast_from(NervesHub.PubSub, self(), topic, event, payload)
+    Phoenix.Channel.Server.broadcast_from!(NervesHub.PubSub, self(), topic, event, payload)
   end
 end

lib/nerves_hub/products.ex

@@ -9,6 +9,7 @@ defmodule NervesHub.Products do
 
   alias NervesHub.Accounts.Org
   alias NervesHub.Accounts.OrgUser
+  alias NervesHub.Extensions
   alias NervesHub.Products.Product
   alias NervesHub.Products.SharedSecretAuth
   alias NervesHub.Accounts.User
@@ -277,11 +278,7 @@ defmodule NervesHub.Products do
     |> Repo.update()
     |> tap(fn
       {:ok, _} ->
-        topic = "product:#{product.id}:extensions"
-
-        NervesHubWeb.DeviceEndpoint.broadcast(topic, "attach", %{
-          "extensions" => [extension_string]
-        })
+        Extensions.toggle_extension(product, "attach", extension_string)
 
       _ ->
         :nope
@@ -295,11 +292,7 @@ defmodule NervesHub.Products do
     |> Repo.update()
     |> tap(fn
       {:ok, _} ->
-        topic = "product:#{product.id}:extensions"
-
-        NervesHubWeb.DeviceEndpoint.broadcast(topic, "detach", %{
-          "extensions" => [extension_string]
-        })
+        Extensions.toggle_extension(product, "detach", extension_string)
 
       _ ->
         :nope

lib/nerves_hub_web/channels/extensions_channel.ex

@@ -3,6 +3,7 @@ defmodule NervesHubWeb.ExtensionsChannel do
 
   alias NervesHub.Extensions
   alias NervesHub.Helpers.Logging
+  alias Phoenix.PubSub
   alias Phoenix.Socket.Broadcast
 
   require Logger
@@ -19,7 +20,7 @@ defmodule NervesHubWeb.ExtensionsChannel do
     end
 
     topic = "device:#{socket.assigns.device.id}:extensions"
-    NervesHubWeb.DeviceEndpoint.subscribe(topic)
+    PubSub.subscribe(NervesHub.PubSub, topic)
 
     {:ok, attach_list, socket}
   end
@@ -94,7 +95,7 @@ defmodule NervesHubWeb.ExtensionsChannel do
   @impl Phoenix.Channel
   def handle_info(:init_extensions, socket) do
     topic = "product:#{socket.assigns.device.product.id}:extensions"
-    NervesHubWeb.DeviceEndpoint.subscribe(topic)
+    PubSub.subscribe(NervesHub.PubSub, topic)
 
     {:noreply, socket}
   end

lib/nerves_hub_web/live/devices/device_health.ex

@@ -4,6 +4,7 @@ defmodule NervesHubWeb.Live.Devices.DeviceHealth do
   alias NervesHub.Devices
   alias NervesHub.Devices.Connections
   alias NervesHub.Devices.Metrics
+  alias NervesHub.Extensions.Health
 
   alias NervesHubWeb.Components.HealthHeader
 
@@ -86,8 +87,7 @@ defmodule NervesHubWeb.Live.Devices.DeviceHealth do
   def handle_info(:check_health_interval, socket) do
     timer_ref = Process.send_after(self(), :check_health_interval, @check_health_interval)
 
-    topic = "device:#{socket.assigns.device.id}:extensions"
-    socket.endpoint.broadcast(topic, "health:check", %{})
+    Health.request_health_check(socket.assigns.device)
 
     socket
     |> assign(:health_check_timer, timer_ref)