Merge pull request #1082 from davidebriani/pairing-ecto-queries

refactor(pairing): move CQEx select queries to exandra

Author: Annopaolo

apps/astarte_pairing/lib/astarte_pairing/engine.ex

@@ -56,11 +56,7 @@ defmodule Astarte.Pairing.Engine do
     end
   end
 
-  def get_agent_public_key_pems(realm_name) do
-    keyspace = CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
-
-    Queries.get_agent_public_key_pems(keyspace)
-  end
+  defdelegate get_agent_public_key_pems(realm_name), to: Queries
 
   def get_credentials(
         :astarte_mqtt_v1,
@@ -88,13 +84,12 @@ defmodule Astarte.Pairing.Engine do
              Config.cassandra_node!(),
              cqex_options
            ),
-         {:ok, device_row} <- Queries.select_device_for_credentials_request(client, device_id),
+         {:ok, device} <- Queries.fetch_device(realm, device_id),
          {:authorized?, true} <-
-           {:authorized?,
-            CredentialsSecret.verify(credentials_secret, device_row[:credentials_secret])},
+           {:authorized?, CredentialsSecret.verify(credentials_secret, device.credentials_secret)},
          {:credentials_inhibited?, false} <-
-           {:credentials_inhibited?, device_row[:inhibit_credentials_request]},
-         _ <- CFSSLCredentials.revoke(device_row[:cert_serial], device_row[:cert_aki]),
+           {:credentials_inhibited?, device.inhibit_credentials_request},
+         _ <- CFSSLCredentials.revoke(device.cert_serial, device.cert_aki),
          encoded_device_id <- Device.encode_device_id(device_id),
          {:ok, %{cert: cert, aki: _aki, serial: _serial} = cert_data} <-
            CFSSLCredentials.get_certificate(csr, realm, encoded_device_id),
@@ -104,7 +99,7 @@ defmodule Astarte.Pairing.Engine do
              device_id,
              cert_data,
              ip_tuple,
-             device_row[:first_credentials_request]
+             device.first_credentials_request
            ) do
       {:ok, %{client_crt: cert}}
     else
@@ -139,23 +134,12 @@ defmodule Astarte.Pairing.Engine do
 
   def get_info(realm, hardware_id, credentials_secret) do
     Logger.debug("get_info request for device #{inspect(hardware_id)} in realm #{inspect(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, client} <-
-           Client.new(
-             Config.cassandra_node!(),
-             cqex_options
-           ),
-         {:ok, device_row} <- Queries.select_device_for_info(client, device_id),
+         {:ok, device} <- Queries.fetch_device(realm, device_id),
          {:authorized?, true} <-
-           {:authorized?,
-            CredentialsSecret.verify(credentials_secret, device_row[:credentials_secret])} do
-      device_status = device_status_string(device_row)
+           {:authorized?, CredentialsSecret.verify(credentials_secret, device.credentials_secret)} do
+      device_status = device_status_string(device)
       protocols = get_protocol_info()
 
       {:ok, %{version: @version, device_status: device_status, protocols: protocols}}
@@ -180,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} ->
@@ -207,19 +181,19 @@ defmodule Astarte.Pairing.Engine do
   end
 
   defp verify_can_register_device(realm_name, device_id) do
-    # An already existing device should always be able to retrieve a new credentials secret
-    case Queries.check_already_registered_device(realm_name, device_id) do
-      {:ok, true} ->
+    try do
+      if Queries.check_already_registered_device(realm_name, device_id) do
+        # An already existing device should always be able to retrieve a new credentials secret
         :ok
-
-      {:ok, false} ->
+      else
         verify_can_register_new_device(realm_name)
-
-      {:error, reason} ->
+      end
+    rescue
+      err ->
         # Consider a failing database as a negative answer
         _ =
           Logger.warning(
-            "Failed to verify if unconfirmed device #{Device.encode_device_id(device_id)} exists, reason: #{inspect(reason)}",
+            "Failed to verify if unconfirmed device #{Device.encode_device_id(device_id)} exists, reason: #{Exception.message(err)}",
             realm_name: realm_name
           )
 
@@ -250,26 +224,9 @@ defmodule Astarte.Pairing.Engine do
         "in realm #{inspect(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(encoded_device_id),
-         {:ok, client} <-
-           Client.new(
-             Config.cassandra_node!(),
-             cqex_options
-           ),
-         :ok <- Queries.unregister_device(client, device_id) do
+         :ok <- Queries.unregister_device(realm, device_id) do
       :ok
-    else
-      {:error, :shutdown} ->
-        {:error, :realm_not_found}
-
-      {:error, reason} ->
-        {:error, reason}
     end
   end
 
@@ -278,21 +235,10 @@ defmodule Astarte.Pairing.Engine do
       "verify_credentials request for device #{inspect(hardware_id)} in realm #{inspect(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, client} <-
-           Client.new(
-             Config.cassandra_node!(),
-             cqex_options
-           ),
-         {:ok, device_row} <- Queries.select_device_for_verify_credentials(client, device_id),
+         {:ok, device} <- Queries.fetch_device(realm, device_id),
          {:authorized?, true} <-
-           {:authorized?, CredentialsSecret.verify(secret, device_row[:credentials_secret])} do
+           {:authorized?, CredentialsSecret.verify(secret, device.credentials_secret)} do
       CertVerifier.verify(client_crt, Config.ca_cert!())
     else
       {:authorized?, false} ->
@@ -317,17 +263,12 @@ defmodule Astarte.Pairing.Engine do
     {:error, :unknown_protocol}
   end
 
-  defp device_status_string(device_row) do
+  defp device_status_string(device) do
     # The device is pending until the first credendtial request
     cond do
-      Keyword.get(device_row, :inhibit_credentials_request) ->
-        "inhibited"
-
-      Keyword.get(device_row, :first_credentials_request) ->
-        "confirmed"
-
-      true ->
-        "pending"
+      device.inhibit_credentials_request -> "inhibited"
+      device.first_credentials_request -> "confirmed"
+      true -> "pending"
     end
   end