Start support for AWS IoT setup

config/dev.exs

@@ -108,3 +108,38 @@ config :nerves_hub, NervesHub.SwooshMailer, adapter: Swoosh.Adapters.Local
 config :nerves_hub, NervesHub.RateLimit, limit: 10
 
 config :sentry, environment_name: :development
+
+broker_opts = [
+  name: NervesHub.AWSIoT.PintBroker,
+  rules: [{"nh/device_messages", &Broadway.test_message(:nerves_hub_iot_messages, &1.payload)}],
+  on_connect: fn client_id ->
+    payload = %{clientId: client_id, eventType: :connected}
+    Broadway.test_message(:nerves_hub_iot_messages, Jason.encode!(payload))
+  end,
+  on_disconnect: fn client_id ->
+    payload = %{
+      clientId: client_id,
+      eventType: :disconnected,
+      disconnectReason: "CONNECTION_LOST"
+    }
+
+    Broadway.test_message(:nerves_hub_iot_messages, Jason.encode!(payload))
+  end
+]
+
+config :nerves_hub, NervesHub.AWSIoT,
+  # Run a PintBroker for local process and/or device connections
+  local_broker: {PintBroker, broker_opts},
+  queues: [
+    [
+      name: :nerves_hub_iot_messages,
+      producer: [
+        module: {Broadway.DummyProducer, []}
+        # To test fetching from development queues registered with AWS, use the producer
+        # below. You may need to set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY ¬
+        # module: {BroadwaySQS.Producer, queue_url: "nerves-hub-iot-messages"}
+      ],
+      processors: [default: []],
+      batchers: [default: [batch_size: 10, batch_timeout: 2000]]
+    ]
+  ]

config/runtime.exs

@@ -140,6 +140,23 @@ if config_env() == :prod do
           ]
         ]
       ]
+
+    if System.get_env("AWS_IOT_ENABLED") in ["1", "true", "t"] do
+      config :nerves_hub, NervesHub.AWSIoT,
+        queues: [
+          [
+            # AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY must be set
+            name: :nerves_hub_iot_messages,
+            producer: [
+              module:
+                {BroadwaySQS.Producer,
+                 queue_url: System.get_env("AWS_IOT_SQS_QUEUE", "nerves-hub-iot-messages")}
+            ],
+            processors: [default: []],
+            batchers: [default: [batch_size: 10, batch_timeout: 2000]]
+          ]
+        ]
+    end
   end
 end
 

config/test.exs

@@ -82,3 +82,36 @@ config :nerves_hub, NervesHub.SwooshMailer, adapter: Swoosh.Adapters.Test
 config :nerves_hub, NervesHub.RateLimit, limit: 100
 
 config :sentry, environment_name: :test
+
+## AWS IoT
+broker_opts = [
+  name: NervesHub.AWSIoT.PintBroker,
+  rules: [{"nh/device_messages", &Broadway.test_message(:nerves_hub_iot_messages, &1.payload)}],
+  on_connect: fn client_id ->
+    payload = %{clientId: client_id, eventType: :connected}
+    Broadway.test_message(:nerves_hub_iot_messages, Jason.encode!(payload))
+  end,
+  on_disconnect: fn client_id ->
+    payload = %{
+      clientId: client_id,
+      eventType: :disconnected,
+      disconnectReason: "CONNECTION_LOST"
+    }
+
+    Broadway.test_message(:nerves_hub_iot_messages, Jason.encode!(payload))
+  end
+]
+
+config :nerves_hub, NervesHub.AWSIoT,
+  # Use PintBroker for local device connections in tests
+  local_broker: {PintBroker, broker_opts},
+  queues: [
+    [
+      name: :nerves_hub_iot_messages,
+      producer: [
+        module: {Broadway.DummyProducer, []}
+      ],
+      processors: [default: []],
+      batchers: [default: [batch_size: 10, batch_timeout: 2000]]
+    ]
+  ]

lib/nerves_hub/application.ex

@@ -30,7 +30,8 @@ defmodule NervesHub.Application do
           {Phoenix.PubSub, name: NervesHub.PubSub},
           {Cluster.Supervisor, [topologies]},
           {Task.Supervisor, name: NervesHub.TaskSupervisor},
-          {Oban, Application.fetch_env!(:nerves_hub, Oban)}
+          {Oban, Application.fetch_env!(:nerves_hub, Oban)},
+          NervesHub.AWSIoT
         ] ++
         deployments_supervisor(deploy_env()) ++
         endpoints(deploy_env())

lib/nerves_hub/aws_iot.ex

@@ -0,0 +1,164 @@
+defmodule NervesHub.AWSIoT do
+  @moduledoc """
+  Support for common AWS IOT infrastructure including MQTT and SQS
+
+  Requires `:queues` to be defined in the application config or
+  the supervisor is simply ignored
+
+  See docs.nerves-hub.org for a general overview of the architecture
+  """
+  use Supervisor
+
+  alias NervesHub.Tracker
+
+  @type opt :: {:queues, [keyword()]}
+  @spec start_link([opt]) :: Supervisor.on_start()
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl Supervisor
+  def init(opts) do
+    opts =
+      Application.get_env(:nerves_hub, __MODULE__, [])
+      |> Keyword.merge(opts)
+
+    case opts[:queues] do
+      queues when is_list(queues) and length(queues) > 0 ->
+        children =
+          Enum.map(queues, &{__MODULE__.SQS, &1})
+          |> maybe_add_local_broker(opts)
+
+        Supervisor.init(children, strategy: :one_for_one)
+
+      _ ->
+        :ignore
+    end
+  end
+
+  defp maybe_add_local_broker(children, opts) do
+    if broker_spec = opts[:local_broker] do
+      [broker_spec | children]
+    else
+      children
+    end
+  end
+
+  if Application.compile_env(:nerves_hub, [__MODULE__, :local_broker], false) do
+    def publish(serial, event, payload) do
+      data = Jason.encode!(%{event: event, payload: payload})
+      PintBroker.publish(__MODULE__.PintBroker, "nh/#{serial}", data)
+    end
+  else
+    def publish(serial, event, payload) do
+      # TODO: Topic and data may change soon
+      # Stubbing out initial idea here for now
+      data = %{event: event, payload: payload}
+      topic = "/topics/nh/#{serial}"
+
+      ExAws.Operation.JSON.new(:iot_data, %{path: topic, data: data})
+      |> ExAws.request()
+    end
+  end
+
+  defmodule SQS do
+    @moduledoc """
+    Consumer for AWS SQS messages
+
+    This is the ingestion point of devices coming from the MQTT
+    broker. A message from a device must include the `"identifier"`
+    key either in the payload or pulled from the topic via the
+    AWS IoT rule that forwards to the queue.
+
+    The system must also be setup with a rule to forward [AWS Lifecycle
+    events](https://docs.aws.amazon.com/iot/latest/developerguide/life-cycle-events.html)
+    to a queue for tracking device online/offline presence
+
+    Right now, all configured queues are handled by this module.
+    In the future, we may want to separate handling for each
+    queue in it's own module.
+    """
+    use Broadway
+
+    alias Broadway.Message
+    alias NervesHub.Devices
+
+    require Logger
+
+    def start_link(opts), do: Broadway.start_link(__MODULE__, opts)
+
+    @impl Broadway
+    def handle_message(_processor, %{data: raw} = msg, _context) do
+      case Jason.decode(raw) do
+        {:ok, data} ->
+          Message.put_data(msg, data)
+          |> process_message()
+
+        _ ->
+          Message.failed(msg, :malformed)
+      end
+    end
+
+    @impl Broadway
+    def handle_batch(_batcher, messages, batch_info, _context) do
+      Logger.debug("[SQS] Handled #{inspect(batch_info.size)}")
+      messages
+    end
+
+    defp process_message(%{data: %{"eventType" => "connected"} = data} = msg) do
+      # TODO: Maybe use more info from the connection?
+      # Example payload of AWS lifecycle connected event
+      # principalIdentifier is a SHA256 fingerprint of the certificate that
+      # is Base16 encoded
+      # {
+      #   "clientId": "186b5",
+      #   "timestamp": 1573002230757,
+      #   "eventType": "connected",
+      #   "sessionIdentifier": "a4666d2a7d844ae4ac5d7b38c9cb7967",
+      #   "principalIdentifier": "12345678901234567890123456789012",
+      #   "ipAddress": "192.0.2.0",
+      #   "versionNumber": 0
+      # }
+
+      with {:ok, device} <- Devices.get_by_identifier(data["clientId"]) do
+        Logger.debug("[AWS IoT] device #{device.identifier} connected")
+
+        Tracker.online(device)
+
+        msg
+      else
+        _ ->
+          Message.failed(msg, :unknown_device)
+      end
+    end
+
+    defp process_message(%{data: %{"eventType" => "disconnected"} = data} = msg) do
+      # TODO: Maybe use more of the disconnect data?
+      # Example payload of AWS lifecyle disconnect event
+      # {
+      #   "clientId": "186b5",
+      #   "timestamp": 1573002340451,
+      #   "eventType": "disconnected",
+      #   "sessionIdentifier": "a4666d2a7d844ae4ac5d7b38c9cb7967",
+      #   "principalIdentifier": "12345678901234567890123456789012",
+      #   "clientInitiatedDisconnect": true,
+      #   "disconnectReason": "CLIENT_INITIATED_DISCONNECT",
+      #   "versionNumber": 0
+      # }
+      with {:ok, device} <- Devices.get_by_identifier(data["clientId"]) do
+        Logger.debug(
+          "[AWS IoT] device #{device.identifier} disconnected: #{data["disconnectReason"]}"
+        )
+
+        Tracker.offline(device)
+      end
+
+      msg
+    end
+
+    defp process_message(msg) do
+      # TODO: Track unhandled msg
+      msg
+    end
+  end
+end

mix.exs

@@ -56,6 +56,7 @@ defmodule NervesHub.MixProject do
       {:bcrypt_elixir, "~> 3.0"},
       {:castore, "~> 1.0"},
       {:circular_buffer, "~> 0.4.1"},
+      {:broadway_sqs, "~> 0.7"},
       {:comeonin, "~> 5.3"},
       {:crontab, "~> 1.1"},
       {:decorator, "~> 1.2"},
@@ -84,6 +85,7 @@ defmodule NervesHub.MixProject do
       {:phoenix_pubsub, "~> 2.0"},
       {:phoenix_swoosh, "~> 1.0"},
       {:phoenix_view, "~> 2.0"},
+      {:pint_broker, "~> 1.0", only: [:dev, :test]},
       {:plug, "~> 1.7"},
       {:postgrex, "~> 0.14"},
       {:scrivener_ecto, "~> 2.7"},