refactor(appengine): port Device to exandra part 5

final effort in porting all cqex select queries in Device to exandra

- select in `delete_attribute`
- select in `insert_value_into_db`

Signed-off-by: Francesco Noacco <francesco.noacco@secomind.com>

Author: noaccOS

apps/astarte_appengine_api/lib/astarte_appengine_api/device/device.ex

@@ -81,7 +81,7 @@ defmodule Astarte.AppEngine.API.Device do
          aliases_change = Map.get(changeset.changes, :aliases, %{}),
          attributes_change = Map.get(changeset.changes, :attributes, %{}),
          :ok <- update_aliases(realm_name, client, device_id, aliases_change),
-         :ok <- update_attributes(client, device_id, attributes_change) do
+         :ok <- update_attributes(realm_name, client, device_id, attributes_change) do
       # Manually merge aliases since changesets don't perform maps deep merge
       merged_aliases = merge_data(device_status.aliases, updated_device_status.aliases)
       merged_attributes = merge_data(device_status.attributes, updated_device_status.attributes)
@@ -95,7 +95,7 @@ defmodule Astarte.AppEngine.API.Device do
     end
   end
 
-  defp update_attributes(client, device_id, attributes) do
+  defp update_attributes(realm_name, client, device_id, attributes) do
     Enum.reduce_while(attributes, :ok, fn
       {"", _attribute_value}, _acc ->
         Logger.warning("Attribute key cannot be an empty string.",
@@ -105,7 +105,7 @@ defmodule Astarte.AppEngine.API.Device do
         {:halt, {:error, :invalid_attributes}}
 
       {attribute_key, nil}, _acc ->
-        case Queries.delete_attribute(client, device_id, attribute_key) do
+        case Queries.delete_attribute(realm_name, client, device_id, attribute_key) do
           :ok ->
             {:cont, :ok}
 
@@ -276,6 +276,7 @@ defmodule Astarte.AppEngine.API.Device do
         end
 
       Queries.insert_value_into_db(
+        realm_name,
         client,
         device_id,
         interface_descriptor,
@@ -451,6 +452,7 @@ defmodule Astarte.AppEngine.API.Device do
         end
 
       Queries.insert_value_into_db(
+        realm_name,
         client,
         device_id,
         interface_descriptor,
@@ -729,6 +731,7 @@ defmodule Astarte.AppEngine.API.Device do
         Queries.retrieve_mapping(realm_name, interface_descriptor.interface_id, endpoint_id)
 
       Queries.insert_value_into_db(
+        realm_name,
         client,
         device_id,
         interface_descriptor,

apps/astarte_appengine_api/lib/astarte_appengine_api/device/queries.ex

@@ -107,7 +107,10 @@ defmodule Astarte.AppEngine.API.Device.Queries do
 
   def interface_has_explicit_timestamp?(realm_name, interface_id) do
     keyspace = keyspace_name(realm_name)
+    do_interface_has_explicit_timestamp?(keyspace, interface_id)
+  end
 
+  def do_interface_has_explicit_timestamp?(keyspace, interface_id) do
     from(d in DatabaseEndpoint,
       where: [interface_id: ^interface_id],
       select: d.explicit_timestamp,
@@ -205,6 +208,7 @@ defmodule Astarte.AppEngine.API.Device.Queries do
 
   # TODO Copy&pasted from data updater plant, make it a library
   def insert_value_into_db(
+        _realm_name,
         db_client,
         device_id,
         %InterfaceDescriptor{storage_type: :multi_interface_individual_properties_dbtable} =
@@ -243,6 +247,7 @@ defmodule Astarte.AppEngine.API.Device.Queries do
 
   # TODO Copy&pasted from data updater plant, make it a library
   def insert_value_into_db(
+        _realm_name,
         db_client,
         device_id,
         %InterfaceDescriptor{storage_type: :multi_interface_individual_properties_dbtable} =
@@ -281,6 +286,7 @@ defmodule Astarte.AppEngine.API.Device.Queries do
 
   # TODO Copy&pasted from data updater plant, make it a library
   def insert_value_into_db(
+        _realm_name,
         db_client,
         device_id,
         %InterfaceDescriptor{storage_type: :multi_interface_individual_datastream_dbtable} =
@@ -321,6 +327,7 @@ defmodule Astarte.AppEngine.API.Device.Queries do
 
   # TODO Copy&pasted from data updater plant, make it a library
   def insert_value_into_db(
+        realm_name,
         db_client,
         device_id,
         %InterfaceDescriptor{storage_type: :one_object_datastream_dbtable} = interface_descriptor,
@@ -332,32 +339,23 @@ defmodule Astarte.AppEngine.API.Device.Queries do
         opts
       ) do
     ttl_string = get_ttl_string(opts)
+    keyspace = keyspace_name(realm_name)
+    interface_id = interface_descriptor.interface_id
 
-    endpoint_query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement(
-        "SELECT endpoint, value_type FROM endpoints WHERE interface_id=:interface_id;"
-      )
-      |> DatabaseQuery.put(:interface_id, interface_descriptor.interface_id)
-
-    endpoint_rows = DatabaseQuery.call!(db_client, endpoint_query)
-
-    explicit_timestamp_query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement(
-        "SELECT explicit_timestamp FROM endpoints WHERE interface_id=:interface_id LIMIT 1;"
+    endpoint_rows =
+      from(DatabaseEndpoint,
+        where: [interface_id: ^interface_id],
+        select: [:endpoint, :value_type]
       )
-      |> DatabaseQuery.put(:interface_id, interface_descriptor.interface_id)
+      |> Repo.all(prefix: keyspace)
 
-    [explicit_timestamp: explicit_timestamp] =
-      DatabaseQuery.call!(db_client, explicit_timestamp_query)
-      |> DatabaseResult.head()
+    explicit_timestamp = do_interface_has_explicit_timestamp?(keyspace, interface_id)
 
     # FIXME: new atoms are created here, we should avoid this. We need to replace CQEx.
     column_atoms =
       Enum.reduce(endpoint_rows, %{}, fn endpoint, column_atoms_acc ->
         endpoint_name =
-          endpoint[:endpoint]
+          endpoint.endpoint
           |> String.split("/")
           |> List.last()
 
@@ -571,21 +569,13 @@ defmodule Astarte.AppEngine.API.Device.Queries do
     end
   end
 
-  def delete_attribute(client, device_id, attribute_key) do
-    retrieve_attribute_statement = """
-    SELECT attributes FROM devices WHERE device_id = :device_id
-    """
-
-    retrieve_attribute_query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement(retrieve_attribute_statement)
-      |> DatabaseQuery.put(:device_id, device_id)
-      |> DatabaseQuery.consistency(:quorum)
+  def delete_attribute(realm_name, client, device_id, attribute_key) do
+    keyspace = keyspace_name(realm_name)
+    query = from(d in DatabaseDevice, select: d.attributes)
+    opts = [prefix: keyspace, consistency: :quorum]
 
-    with {:ok, result} <- DatabaseQuery.call(client, retrieve_attribute_query),
-         [attributes: attributes] <- DatabaseResult.head(result),
-         {^attribute_key, _attribute_value} <-
-           Enum.find(attributes || [], fn m -> match?({^attribute_key, _}, m) end) do
+    with {:ok, attributes} <- Repo.fetch(query, device_id, opts),
+         {:ok, _} <- get_value(attributes, attribute_key, :attribute_key_not_found) do
       delete_attribute_statement = """
         DELETE attributes[:attribute_key]
         FROM devices
@@ -611,17 +601,6 @@ defmodule Astarte.AppEngine.API.Device.Queries do
           _ = Logger.warning("Database error, reason: #{inspect(reason)}.", tag: "db_error")
           {:error, :database_error}
       end
-    else
-      nil ->
-        {:error, :attribute_key_not_found}
-
-      %{acc: _, msg: error_message} ->
-        _ = Logger.warning("Database error: #{error_message}.", tag: "db_error")
-        {:error, :database_error}
-
-      {:error, reason} ->
-        _ = Logger.warning("Database error, reason: #{inspect(reason)}.", tag: "db_error")
-        {:error, :database_error}
     end
   end