Merge pull request #1114 from noaccOS/refactor/insert-value-to-db

refactor(appengine): port insert_values_to_db inserts to exandra

Author: Annopaolo

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

@@ -278,7 +278,6 @@ defmodule Astarte.AppEngine.API.Device do
 
       Queries.insert_value_into_db(
         realm_name,
-        client,
         device_id,
         interface_descriptor,
         endpoint_id,
@@ -454,7 +453,6 @@ defmodule Astarte.AppEngine.API.Device do
 
       Queries.insert_value_into_db(
         realm_name,
-        client,
         device_id,
         interface_descriptor,
         nil,
@@ -718,8 +716,7 @@ defmodule Astarte.AppEngine.API.Device do
   # TODO: we should probably allow delete for every path regardless of the interface type
   # just for maintenance reasons
   def delete_interface_values(realm_name, encoded_device_id, interface, no_prefix_path) do
-    with {:ok, client} <- Database.connect(realm: realm_name),
-         {:ok, device_id} <- Device.decode_device_id(encoded_device_id),
+    with {:ok, device_id} <- Device.decode_device_id(encoded_device_id),
          {:ok, major_version} <-
            DeviceQueries.interface_version(realm_name, device_id, interface),
          {:ok, interface_row} <-
@@ -733,7 +730,6 @@ defmodule Astarte.AppEngine.API.Device do
 
       Queries.insert_value_into_db(
         realm_name,
-        client,
         device_id,
         interface_descriptor,
         endpoint_id,
@@ -1562,7 +1558,6 @@ defmodule Astarte.AppEngine.API.Device do
 
   defp timestamp_column(explicit_timestamp?) do
     case explicit_timestamp? do
-      nil -> :reception_timestamp
       false -> :reception_timestamp
       true -> :value_timestamp
     end

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

@@ -116,12 +116,18 @@ defmodule Astarte.AppEngine.API.Device.Queries do
   end
 
   def do_interface_has_explicit_timestamp?(keyspace, interface_id) do
-    from(d in DatabaseEndpoint,
-      where: [interface_id: ^interface_id],
-      select: d.explicit_timestamp,
-      limit: 1
-    )
-    |> Repo.one!(prefix: keyspace)
+    interface_explicit_timestamp =
+      from(d in DatabaseEndpoint,
+        where: [interface_id: ^interface_id],
+        select: d.explicit_timestamp,
+        limit: 1
+      )
+      |> Repo.one!(prefix: keyspace)
+
+    # ensure boolean value
+    with nil <- interface_explicit_timestamp do
+      false
+    end
   end
 
   def fetch_datastream_maximum_storage_retention(realm_name) do
@@ -195,15 +201,17 @@ defmodule Astarte.AppEngine.API.Device.Queries do
         :reception_timestamp_submillis, :datetime_value) #{ttl_string};
     """
 
+    {reception_ms, reception_submillis} = split_ms_and_submillis(reception_timestamp)
+
     insert_query =
       DatabaseQuery.new()
       |> DatabaseQuery.statement(insert_statement)
       |> DatabaseQuery.put(:device_id, device_id)
       |> DatabaseQuery.put(:interface_id, interface_descriptor.interface_id)
       |> DatabaseQuery.put(:endpoint_id, endpoint_id)
       |> DatabaseQuery.put(:path, path)
-      |> DatabaseQuery.put(:reception_timestamp, div(reception_timestamp, 1000))
-      |> DatabaseQuery.put(:reception_timestamp_submillis, rem(reception_timestamp, 1000))
+      |> DatabaseQuery.put(:reception_timestamp, reception_ms)
+      |> DatabaseQuery.put(:reception_timestamp_submillis, reception_submillis)
       |> DatabaseQuery.put(:datetime_value, value_timestamp)
 
     DatabaseQuery.call!(db_client, insert_query)
@@ -214,7 +222,6 @@ 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} =
           interface_descriptor,
@@ -258,8 +265,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,
+        realm_name,
         device_id,
         %InterfaceDescriptor{storage_type: :multi_interface_individual_properties_dbtable} =
           interface_descriptor,
@@ -270,35 +276,34 @@ defmodule Astarte.AppEngine.API.Device.Queries do
         timestamp,
         opts
       ) do
-    ttl_string = get_ttl_string(opts)
+    value_column = CQLUtils.type_to_db_column_name(endpoint.value_type)
+    keyspace = keyspace_name(realm_name)
+
+    {timestamp_ms, timestamp_submillis} = split_ms_and_submillis(timestamp)
 
     # TODO: :reception_timestamp_submillis is just a place holder right now
-    insert_query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement("""
-      INSERT INTO #{interface_descriptor.storage}
-        (device_id, interface_id, endpoint_id, path, reception_timestamp,
-          #{CQLUtils.type_to_db_column_name(endpoint.value_type)})
-        VALUES (:device_id, :interface_id, :endpoint_id, :path, :reception_timestamp,
-          :value) #{ttl_string};
-      """)
-      |> DatabaseQuery.put(:device_id, device_id)
-      |> DatabaseQuery.put(:interface_id, interface_descriptor.interface_id)
-      |> DatabaseQuery.put(:endpoint_id, endpoint_id)
-      |> DatabaseQuery.put(:path, path)
-      |> DatabaseQuery.put(:reception_timestamp, div(timestamp, 1000))
-      |> DatabaseQuery.put(:reception_timestamp_submillis, div(timestamp, 100))
-      |> DatabaseQuery.put(:value, to_db_friendly_type(value))
+    interface_storage_attributes = %{
+      value_column => to_db_friendly_type(value),
+      device_id: device_id,
+      interface_id: interface_descriptor.interface_id,
+      endpoint_id: endpoint_id,
+      path: path,
+      reception_timestamp: timestamp_ms,
+      reception_timestamp_submillis: timestamp_submillis
+    }
 
-    DatabaseQuery.call!(db_client, insert_query)
+    {1, _} =
+      Repo.insert_all(interface_descriptor.storage, [interface_storage_attributes],
+        prefix: keyspace,
+        ttl: opts[:ttl]
+      )
 
     :ok
   end
 
   # TODO Copy&pasted from data updater plant, make it a library
   def insert_value_into_db(
-        _realm_name,
-        db_client,
+        realm_name,
         device_id,
         %InterfaceDescriptor{storage_type: :multi_interface_individual_datastream_dbtable} =
           interface_descriptor,
@@ -309,37 +314,28 @@ defmodule Astarte.AppEngine.API.Device.Queries do
         timestamp,
         opts
       ) do
-    ttl_string = get_ttl_string(opts)
-
-    insert_query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement("""
-      INSERT INTO #{interface_descriptor.storage}
-        (device_id, interface_id, endpoint_id, path, value_timestamp, reception_timestamp, reception_timestamp_submillis,
-          #{CQLUtils.type_to_db_column_name(endpoint.value_type)})
-        VALUES (:device_id, :interface_id, :endpoint_id, :path, :value_timestamp, :reception_timestamp,
-          :reception_timestamp_submillis, :value) #{ttl_string};
-      """)
-      |> DatabaseQuery.put(:device_id, device_id)
-      |> DatabaseQuery.put(:interface_id, interface_descriptor.interface_id)
-      |> DatabaseQuery.put(:endpoint_id, endpoint.endpoint_id)
-      |> DatabaseQuery.put(:path, path)
-      |> DatabaseQuery.put(:value_timestamp, div(timestamp, 1000))
-      |> DatabaseQuery.put(:reception_timestamp, div(timestamp, 1000))
-      |> DatabaseQuery.put(:reception_timestamp_submillis, rem(timestamp, 1000))
-      |> DatabaseQuery.put(:value, to_db_friendly_type(value))
-
-    # TODO: |> DatabaseQuery.consistency(insert_consistency(interface_descriptor, endpoint))
+    value_column = CQLUtils.type_to_db_column_name(endpoint.value_type)
+    keyspace = keyspace_name(realm_name)
+    {timestamp_ms, timestamp_submillis} = split_ms_and_submillis(timestamp)
 
-    DatabaseQuery.call!(db_client, insert_query)
+    attributes = %{
+      value_column => to_db_friendly_type(value),
+      device_id: device_id,
+      interface_id: interface_descriptor.interface_id,
+      endpoint_id: endpoint.endpoint_id,
+      path: path,
+      value_timestamp: timestamp_ms,
+      reception_timestamp: timestamp_ms,
+      reception_timestamp_submillis: timestamp_submillis
+    }
 
+    Repo.insert_all(interface_descriptor.storage, [attributes], prefix: keyspace, ttl: opts[:ttl])
     :ok
   end
 
   # 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,
         _endpoint_id,
@@ -349,7 +345,6 @@ defmodule Astarte.AppEngine.API.Device.Queries do
         timestamp,
         opts
       ) do
-    ttl_string = get_ttl_string(opts)
     keyspace = keyspace_name(realm_name)
     interface_id = interface_descriptor.interface_id
 
@@ -360,70 +355,74 @@ defmodule Astarte.AppEngine.API.Device.Queries do
       )
       |> Repo.all(prefix: keyspace)
 
-    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
-          |> String.split("/")
-          |> List.last()
+    explicit_timestamp? = do_interface_has_explicit_timestamp?(keyspace, interface_id)
 
+    column_meta =
+      endpoint_rows
+      |> Map.new(fn endpoint ->
+        endpoint_name = endpoint.endpoint |> String.split("/") |> List.last()
         column_name = CQLUtils.endpoint_to_db_column_name(endpoint_name)
-
-        Map.put(column_atoms_acc, endpoint_name, String.to_atom(column_name))
+        {endpoint_name, %{name: column_name, type: endpoint.value_type}}
       end)
 
-    {query_values, placeholders, query_columns} =
-      Enum.reduce(value, {%{}, "", ""}, fn {obj_key, obj_value},
-                                           {query_values_acc, placeholders_acc, query_acc} ->
-        if column_atoms[obj_key] != nil do
-          column_name = CQLUtils.endpoint_to_db_column_name(obj_key)
-
-          db_value = to_db_friendly_type(obj_value)
-          next_query_values_acc = Map.put(query_values_acc, column_atoms[obj_key], db_value)
-          next_placeholders_acc = "#{placeholders_acc} :#{to_string(column_atoms[obj_key])},"
-          next_query_acc = "#{query_acc} #{column_name}, "
-
-          {next_query_values_acc, next_placeholders_acc, next_query_acc}
-        else
-          Logger.warning(
-            "Unexpected object key #{inspect(obj_key)} with value #{inspect(obj_value)}."
-          )
+    {timestamp_ms, submillis} = split_ms_and_submillis(timestamp)
 
-          query_values_acc
-        end
-      end)
+    base_attributes = %{
+      device_id: device_id,
+      path: path
+    }
 
-    {query_columns, placeholders} =
-      if explicit_timestamp do
-        {"value_timestamp, #{query_columns}", ":value_timestamp, #{placeholders}"}
-      else
-        {query_columns, placeholders}
-      end
+    timestamp_attributes = timestamp_attributes(explicit_timestamp?, timestamp_ms, submillis)
+    value_attributes = value_attributes(column_meta, value)
 
-    # TODO: :reception_timestamp_submillis is just a place holder right now
-    insert_query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement("""
-      INSERT INTO #{interface_descriptor.storage} (device_id, path, #{query_columns} reception_timestamp, reception_timestamp_submillis)
-        VALUES (:device_id, :path, #{placeholders} :reception_timestamp, :reception_timestamp_submillis) #{ttl_string};
-      """)
-      |> DatabaseQuery.put(:device_id, device_id)
-      |> DatabaseQuery.put(:path, path)
-      |> DatabaseQuery.put(:value_timestamp, div(timestamp, 1000))
-      |> DatabaseQuery.put(:reception_timestamp, div(timestamp, 1000))
-      |> DatabaseQuery.put(:reception_timestamp_submillis, rem(timestamp, 1000))
-      |> DatabaseQuery.merge(query_values)
+    object_datastream =
+      base_attributes
+      |> Map.merge(timestamp_attributes)
+      |> Map.merge(value_attributes)
 
-    # TODO: |> DatabaseQuery.consistency(insert_consistency(interface_descriptor, endpoint))
+    ttl = Keyword.get(opts, :ttl)
+    # TODO: consistency = insert_consistency(interface_descriptor, endpoint)
+    opts = [prefix: keyspace, ttl: ttl, returning: false]
 
-    DatabaseQuery.call!(db_client, insert_query)
+    Repo.insert_all(interface_descriptor.storage, [object_datastream], opts)
 
     :ok
   end
 
+  defp timestamp_attributes(true = _explicit_timestamp?, timestamp, submillis) do
+    %{
+      value_timestamp: timestamp,
+      reception_timestamp: timestamp,
+      reception_timestamp_submillis: submillis
+    }
+  end
+
+  defp timestamp_attributes(_nil_or_false_explicit_timestamp?, timestamp, submillis) do
+    %{reception_timestamp: timestamp, reception_timestamp_submillis: submillis}
+  end
+
+  defp value_attributes(column_meta, value) do
+    value =
+      value
+      |> Enum.flat_map(fn {key, value} ->
+        # filter map
+        case Map.fetch(column_meta, key) do
+          {:ok, meta} ->
+            %{name: name, type: type} = meta
+            data = %{type: type, value: value}
+            [{name, data}]
+
+          :error ->
+            Logger.warning("Unexpected object key #{inspect(key)} with value #{inspect(value)}.")
+
+            []
+        end
+      end)
+
+    value
+    |> Map.new(fn {column, data} -> {column, data.value} end)
+  end
+
   # TODO Copy&pasted from data updater plant, make it a library
   defp to_db_friendly_type(array) when is_list(array) do
     # If we have an array, we convert its elements to a db friendly type
@@ -1030,4 +1029,11 @@ defmodule Astarte.AppEngine.API.Device.Queries do
         {:error, :database_error}
     end
   end
+
+  defp split_ms_and_submillis(timestamp_micro) do
+    timestamp_ms = div(timestamp_micro, 1000)
+    timestamp_submillis = rem(timestamp_micro, 1000)
+
+    {timestamp_ms, timestamp_submillis}
+  end
 end