Migrate update_device_after_credentials_request query 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

@@ -28,7 +28,6 @@ defmodule Astarte.Pairing.Engine do
   alias Astarte.Pairing.CredentialsSecret
   alias Astarte.Pairing.Queries
   alias CQEx.Client
-  alias Astarte.Core.CQLUtils
 
   require Logger
 
@@ -71,19 +70,9 @@ defmodule Astarte.Pairing.Engine do
     )
 
     :telemetry.execute([:astarte, :pairing, :get_credentials], %{}, %{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, ip_tuple} <- parse_ip(device_ip),
-         {:ok, client} <-
-           Client.new(
-             Config.cassandra_node!(),
-             cqex_options
-           ),
          {:ok, device} <- Queries.fetch_device(realm, device_id),
          {:authorized?, true} <-
            {:authorized?, CredentialsSecret.verify(credentials_secret, device.credentials_secret)},
@@ -93,10 +82,10 @@ defmodule Astarte.Pairing.Engine do
          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),
-         :ok <-
+         {:ok, _device} <-
            Queries.update_device_after_credentials_request(
-             client,
-             device_id,
+             realm,
+             device,
              cert_data,
              ip_tuple,
              device.first_credentials_request
@@ -109,9 +98,6 @@ defmodule Astarte.Pairing.Engine do
       {:credentials_inhibited?, true} ->
         {:error, :credentials_request_inhibited}
 
-      {:error, :shutdown} ->
-        {:error, :realm_not_found}
-
       {:error, reason} ->
         {:error, reason}
     end

apps/astarte_pairing/lib/astarte_pairing/queries.ex

@@ -32,7 +32,6 @@ defmodule Astarte.Pairing.Queries do
   require Logger
   import Ecto.Query
 
-  @protocol_revision 1
   @keyspace_does_not_exist_regex ~r/Keyspace (.*) does not exist/
 
   def get_agent_public_key_pems(realm_name) do
@@ -170,54 +169,36 @@ defmodule Astarte.Pairing.Queries do
     end
   end
 
-  def update_device_after_credentials_request(client, device_id, cert_data, device_ip, nil) do
+  def update_device_after_credentials_request(realm_name, device, cert_data, device_ip, nil) do
     first_credentials_request_timestamp = DateTime.utc_now()
 
     update_device_after_credentials_request(
-      client,
-      device_id,
+      realm_name,
+      device,
       cert_data,
       device_ip,
       first_credentials_request_timestamp
     )
   end
 
   def update_device_after_credentials_request(
-        client,
-        device_id,
+        realm_name,
+        %Device{} = device,
         %{serial: serial, aki: aki} = _cert_data,
         device_ip,
         %DateTime{} = first_credentials_request_timestamp
       ) do
-    statement = """
-    UPDATE devices
-    SET cert_aki=:cert_aki, cert_serial=:cert_serial, last_credentials_request_ip=:last_credentials_request_ip,
-    first_credentials_request=:first_credentials_request
-    WHERE device_id=:device_id
-    """
-
-    query =
-      Query.new()
-      |> Query.statement(statement)
-      |> Query.put(:device_id, device_id)
-      |> Query.put(:cert_aki, aki)
-      |> Query.put(:cert_serial, serial)
-      |> Query.put(:last_credentials_request_ip, device_ip)
-      |> Query.put(
-        :first_credentials_request,
-        first_credentials_request_timestamp |> DateTime.to_unix(:millisecond)
-      )
-      |> Query.put(:protocol_revision, @protocol_revision)
-      |> 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!())
 
-      error ->
-        Logger.warning("DB error: #{inspect(error)}")
-        {:error, :database_error}
-    end
+    device
+    |> Ecto.Changeset.change(%{
+      cert_aki: aki,
+      cert_serial: serial,
+      last_credentials_request_ip: device_ip,
+      first_credentials_request: first_credentials_request_timestamp
+    })
+    |> Repo.update(prefix: keyspace_name, consistency: :quorum)
   end
 
   def fetch_device_registration_limit(realm_name) do