Migrate select query of register_device to Ecto

Use Ecto to compose and run the query.

Signed-off-by: Davide Briani <davide.briani@secomind.com>

Author: davidebriani

apps/astarte_pairing/lib/astarte_pairing/engine.ex

@@ -164,22 +164,12 @@ defmodule Astarte.Pairing.Engine do
     )
 
     :telemetry.execute([:astarte, :pairing, :register_new_device], %{}, %{realm: realm})
-    keyspace_name = CQLUtils.realm_name_to_keyspace_name(realm, Config.astarte_instance_id!())
-
-    cqex_options =
-      Config.cqex_options!()
-      |> Keyword.put(:keyspace, keyspace_name)
 
     with {:ok, device_id} <- Device.decode_device_id(hardware_id, allow_extended_id: true),
          :ok <- verify_can_register_device(realm, device_id),
-         {:ok, client} <-
-           Client.new(
-             Config.cassandra_node!(),
-             cqex_options
-           ),
          credentials_secret <- CredentialsSecret.generate(),
          secret_hash <- CredentialsSecret.hash(credentials_secret),
-         :ok <- Queries.register_device(client, device_id, hardware_id, secret_hash, opts) do
+         :ok <- Queries.register_device(realm, device_id, hardware_id, secret_hash, opts) do
       {:ok, credentials_secret}
     else
       {:error, :shutdown} ->

apps/astarte_pairing/lib/astarte_pairing/queries.ex

@@ -21,8 +21,8 @@ defmodule Astarte.Pairing.Queries do
   This module is responsible for the interaction with the database.
   """
 
+  alias CQEx.Client
   alias CQEx.Query
-  alias CQEx.Result
   alias Astarte.Core.CQLUtils
   alias Astarte.Pairing.Config
   alias Astarte.Pairing.Astarte.Realm
@@ -83,51 +83,42 @@ defmodule Astarte.Pairing.Queries do
     end
   end
 
-  def register_device(client, device_id, extended_id, credentials_secret, opts \\ []) do
-    statement = """
-    SELECT first_credentials_request, first_registration
-    FROM devices
-    WHERE device_id=:device_id
-    """
-
-    device_exists_query =
-      Query.new()
-      |> Query.statement(statement)
-      |> Query.put(:device_id, device_id)
-      |> Query.consistency(:quorum)
+  def register_device(realm_name, device_id, extended_id, credentials_secret, opts \\ []) do
+    case fetch_device(realm_name, device_id) do
+      {:error, :device_not_found} ->
+        Logger.info("register request for new device: #{inspect(extended_id)}")
 
-    with {:ok, res} <- Query.call(client, device_exists_query) do
-      case Result.head(res) do
-        :empty_dataset ->
-          registration_timestamp =
-            DateTime.utc_now()
-            |> DateTime.to_unix(:millisecond)
+        registration_timestamp = DateTime.utc_now()
 
-          Logger.info("register request for new device: #{inspect(extended_id)}")
-          do_register_device(client, device_id, credentials_secret, registration_timestamp, opts)
+        do_register_device(
+          realm_name,
+          device_id,
+          credentials_secret,
+          registration_timestamp,
+          opts
+        )
 
-        [first_credentials_request: nil, first_registration: registration_timestamp] ->
+      {:ok, device} ->
+        if is_nil(device.first_credentials_request) do
           Logger.info("register request for existing unconfirmed device: #{inspect(extended_id)}")
 
           do_register_unconfirmed_device(
-            client,
+            realm_name,
             device_id,
             credentials_secret,
-            registration_timestamp,
+            device.first_registration,
             opts
           )
-
-        [first_credentials_request: _timestamp, first_registration: _registration_timestamp] ->
+        else
           Logger.warning(
             "register request for existing confirmed device: #{inspect(extended_id)}"
           )
 
           {:error, :already_registered}
-      end
-    else
-      error ->
-        Logger.warning("DB error: #{inspect(error)}")
-        {:error, :database_error}
+        end
+
+      {:error, reason} ->
+        {:error, reason}
     end
   end
 
@@ -302,7 +293,13 @@ defmodule Astarte.Pairing.Queries do
     end
   end
 
-  defp do_register_device(client, device_id, credentials_secret, registration_timestamp, opts) do
+  defp do_register_device(
+         realm_name,
+         device_id,
+         credentials_secret,
+         %DateTime{} = registration_timestamp,
+         opts
+       ) do
     statement = """
     INSERT INTO devices
     (device_id, first_registration, credentials_secret, inhibit_credentials_request,
@@ -323,7 +320,7 @@ defmodule Astarte.Pairing.Queries do
       Query.new()
       |> Query.statement(statement)
       |> Query.put(:device_id, device_id)
-      |> Query.put(:first_registration, registration_timestamp)
+      |> Query.put(:first_registration, registration_timestamp |> DateTime.to_unix(:millisecond))
       |> Query.put(:credentials_secret, credentials_secret)
       |> Query.put(:inhibit_credentials_request, false)
       |> Query.put(:protocol_revision, 0)
@@ -333,21 +330,32 @@ defmodule Astarte.Pairing.Queries do
       |> Query.put(:introspection_minor, introspection_minor)
       |> Query.consistency(:quorum)
 
-    case Query.call(client, query) do
-      {:ok, _res} ->
-        :ok
+    keyspace_name =
+      CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
 
+    cqex_options =
+      Config.cqex_options!()
+      |> Keyword.put(:keyspace, keyspace_name)
+
+    with {:ok, client} <-
+           Client.new(
+             Config.cassandra_node!(),
+             cqex_options
+           ),
+         {:ok, _res} <- Query.call(client, query) do
+      :ok
+    else
       error ->
         Logger.warning("DB error: #{inspect(error)}")
         {:error, :database_error}
     end
   end
 
   defp do_register_unconfirmed_device(
-         client,
+         realm_name,
          device_id,
          credentials_secret,
-         registration_timestamp,
+         %DateTime{} = registration_timestamp,
          opts
        ) do
     statement = """
@@ -372,18 +380,29 @@ defmodule Astarte.Pairing.Queries do
       Query.new()
       |> Query.statement(statement)
       |> Query.put(:device_id, device_id)
-      |> Query.put(:first_registration, registration_timestamp)
+      |> Query.put(:first_registration, registration_timestamp |> DateTime.to_unix(:millisecond))
       |> Query.put(:credentials_secret, credentials_secret)
       |> Query.put(:inhibit_credentials_request, false)
       |> Query.put(:protocol_revision, 0)
       |> Query.put(:introspection, introspection)
       |> Query.put(:introspection_minor, introspection_minor)
       |> Query.consistency(:quorum)
 
-    case Query.call(client, query) do
-      {:ok, _res} ->
-        :ok
+    keyspace_name =
+      CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
 
+    cqex_options =
+      Config.cqex_options!()
+      |> Keyword.put(:keyspace, keyspace_name)
+
+    with {:ok, client} <-
+           Client.new(
+             Config.cassandra_node!(),
+             cqex_options
+           ),
+         {:ok, _res} <- Query.call(client, query) do
+      :ok
+    else
       error ->
         Logger.warning("DB error: #{inspect(error)}")
         {:error, :database_error}