Merge pull request #1115 from noaccOS/refactor/insert_path_into_db

refactor(appengine): port insert_path_into_db to exandra

Author: Annopaolo

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

@@ -36,7 +36,6 @@ defmodule Astarte.AppEngine.API.Device do
   alias Astarte.Core.Mapping
   alias Astarte.Core.Mapping.EndpointsAutomaton
   alias Astarte.Core.Mapping.ValueType
-  alias Astarte.DataAccess.Database
   alias Astarte.DataAccess.Mappings
   alias Astarte.DataAccess.Device, as: DeviceQueries
   alias Astarte.DataAccess.Interface, as: InterfaceQueries
@@ -228,7 +227,6 @@ defmodule Astarte.AppEngine.API.Device do
   end
 
   defp update_individual_interface_values(
-         client,
          realm_name,
          device_id,
          interface_descriptor,
@@ -256,8 +254,10 @@ defmodule Astarte.AppEngine.API.Device do
            ) do
       realm_max_ttl = Queries.fetch_datastream_maximum_storage_retention(realm_name)
 
+      now = DateTime.utc_now()
+
       timestamp_micro =
-        DateTime.utc_now()
+        now
         |> DateTime.to_unix(:microsecond)
 
       db_max_ttl =
@@ -290,13 +290,13 @@ defmodule Astarte.AppEngine.API.Device do
 
       if interface_descriptor.type == :datastream do
         Queries.insert_path_into_db(
-          client,
+          realm_name,
           device_id,
           interface_descriptor,
           endpoint_id,
           path,
-          div(timestamp_micro, 1000),
-          timestamp_micro,
+          now,
+          now,
           opts
         )
       end
@@ -403,15 +403,16 @@ defmodule Astarte.AppEngine.API.Device do
   end
 
   defp update_object_interface_values(
-         client,
          realm_name,
          device_id,
          interface_descriptor,
          path,
          raw_value
        ) do
+    now = DateTime.utc_now()
+
     timestamp_micro =
-      DateTime.utc_now()
+      now
       |> DateTime.to_unix(:microsecond)
 
     with {:ok, mappings} <-
@@ -464,13 +465,13 @@ defmodule Astarte.AppEngine.API.Device do
       )
 
       Queries.insert_path_into_db(
-        client,
+        realm_name,
         device_id,
         interface_descriptor,
         endpoint_id,
         path,
-        div(timestamp_micro, 1000),
-        timestamp_micro,
+        now,
+        now,
         opts
       )
 
@@ -514,8 +515,7 @@ defmodule Astarte.AppEngine.API.Device do
         raw_value,
         _params
       ) 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} <-
@@ -525,7 +525,6 @@ defmodule Astarte.AppEngine.API.Device do
          path <- "/" <> no_prefix_path do
       if interface_descriptor.aggregation == :individual do
         update_individual_interface_values(
-          client,
           realm_name,
           device_id,
           interface_descriptor,
@@ -534,7 +533,6 @@ defmodule Astarte.AppEngine.API.Device do
         )
       else
         update_object_interface_values(
-          client,
           realm_name,
           device_id,
           interface_descriptor,

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

@@ -23,27 +23,21 @@ defmodule Astarte.AppEngine.API.Device.Queries do
   alias Astarte.AppEngine.API.Device.DeletionInProgress
   alias Astarte.AppEngine.API.Device.DeviceStatus
   alias Astarte.AppEngine.API.Device.DevicesList
-  alias Astarte.AppEngine.API.Device.InterfaceValuesOptions
   alias Astarte.AppEngine.API.Device.InterfaceInfo
+  alias Astarte.AppEngine.API.Realms.IndividualProperty
   alias Astarte.AppEngine.API.KvStore
   alias Astarte.AppEngine.API.Name
   alias Astarte.AppEngine.API.Repo
   alias Astarte.Core.CQLUtils
   alias Astarte.Core.Device
   alias Astarte.Core.InterfaceDescriptor
-  alias CQEx.Query, as: DatabaseQuery
-  alias CQEx.Result, as: DatabaseResult
   alias Astarte.AppEngine.API.Realm
   alias Astarte.AppEngine.API.Devices.Device, as: DatabaseDevice
   alias Astarte.AppEngine.API.Endpoint, as: DatabaseEndpoint
 
   require CQEx
   require Logger
 
-  def first_result_row(values) do
-    DatabaseResult.head(values)
-  end
-
   def retrieve_interfaces_list(realm_name, device_id) do
     keyspace = keyspace_name(realm_name)
 
@@ -165,17 +159,8 @@ defmodule Astarte.AppEngine.API.Device.Queries do
     |> Repo.all()
   end
 
-  defp get_ttl_string(opts) do
-    with {:ok, value} when is_integer(value) <- Keyword.fetch(opts, :ttl) do
-      "USING TTL #{to_string(value)}"
-    else
-      _any_error ->
-        ""
-    end
-  end
-
   def insert_path_into_db(
-        db_client,
+        realm_name,
         device_id,
         %InterfaceDescriptor{storage_type: storage_type} = interface_descriptor,
         endpoint_id,
@@ -191,30 +176,25 @@ defmodule Astarte.AppEngine.API.Device.Queries do
     # TODO: use received value_timestamp when needed
     # TODO: :reception_timestamp_submillis is just a place holder right now
 
-    ttl_string = get_ttl_string(opts)
+    keyspace = keyspace_name(realm_name)
 
-    insert_statement = """
-    INSERT INTO individual_properties
-        (device_id, interface_id, endpoint_id, path,
-        reception_timestamp, reception_timestamp_submillis, datetime_value)
-    VALUES (:device_id, :interface_id, :endpoint_id, :path, :reception_timestamp,
-        :reception_timestamp_submillis, :datetime_value) #{ttl_string};
-    """
+    # Ecto expects microsecond precision
+    {reception, reception_submillis} = split_datetime_to_ms_and_submillis(reception_timestamp)
+    value_timestamp = value_timestamp |> DateTime.truncate(:millisecond) |> pad_usec()
 
-    {reception_ms, reception_submillis} = split_ms_and_submillis(reception_timestamp)
+    value = %IndividualProperty{
+      device_id: device_id,
+      interface_id: interface_descriptor.interface_id,
+      endpoint_id: endpoint_id,
+      path: path,
+      reception_timestamp: reception,
+      reception_timestamp_submillis: reception_submillis,
+      datetime_value: value_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, reception_ms)
-      |> DatabaseQuery.put(:reception_timestamp_submillis, reception_submillis)
-      |> DatabaseQuery.put(:datetime_value, value_timestamp)
+    ttl = opts[:ttl]
 
-    DatabaseQuery.call!(db_client, insert_query)
+    Repo.insert!(value, prefix: keyspace, ttl: ttl)
 
     :ok
   end
@@ -723,28 +703,6 @@ defmodule Astarte.AppEngine.API.Device.Queries do
     {count, values}
   end
 
-  def get_results_count(_client, _count_query, %InterfaceValuesOptions{downsample_to: nil}) do
-    # Count will be ignored since there's no downsample_to
-    nil
-  end
-
-  def get_results_count(client, count_query, opts) do
-    with {:ok, result} <- DatabaseQuery.call(client, count_query),
-         [{_count_key, count}] <- DatabaseResult.head(result) do
-      limit = opts.limit || Config.max_results_limit!()
-
-      min(count, limit)
-    else
-      error ->
-        _ =
-          Logger.warning("Can't retrieve count for #{inspect(count_query)}: #{inspect(error)}.",
-            tag: "db_error"
-          )
-
-        nil
-    end
-  end
-
   def all_properties_for_endpoint!(
         realm_name,
         device_id,
@@ -1036,4 +994,22 @@ defmodule Astarte.AppEngine.API.Device.Queries do
 
     {timestamp_ms, timestamp_submillis}
   end
+
+  defp split_datetime_to_ms_and_submillis(datetime) do
+    datetime_ms = datetime |> DateTime.truncate(:millisecond) |> pad_usec()
+
+    {usec, _} = datetime.microsecond
+    submillis = usec |> rem(1000)
+
+    {datetime_ms, submillis}
+  end
+
+  defp pad_usec(nil), do: nil
+
+  defp pad_usec(timestamp) do
+    case timestamp.microsecond do
+      {_, 6} -> timestamp
+      {usec, _} -> %{timestamp | microsecond: {usec, 6}}
+    end
+  end
 end

apps/astarte_appengine_api/lib/astarte_appengine_api/realms/individual_property.ex

@@ -0,0 +1,57 @@
+#
+# This file is part of Astarte.
+#
+# Copyright 2025 SECO Mind Srl
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+defmodule Astarte.AppEngine.API.Realms.IndividualProperty do
+  use TypedEctoSchema
+  alias Astarte.AppEngine.API.UUID
+
+  @primary_key false
+  typed_schema "individual_properties" do
+    field :reception, :utc_datetime_usec, virtual: true
+    field :device_id, UUID, primary_key: true
+    field :interface_id, UUID, primary_key: true
+    field :endpoint_id, UUID, primary_key: true
+    field :path, :string, primary_key: true
+    field :reception_timestamp, :utc_datetime_usec
+    field :reception_timestamp_submillis, :integer
+    field :double_value, :float
+    field :integer_value, :integer
+    field :boolean_value, :boolean
+    field :longinteger_value, :integer
+    field :string_value, :string
+    field :binaryblob_value, :binary
+    field :datetime_value, :utc_datetime_usec
+    field :doublearray_value, {:array, :float}
+    field :integerarray_value, {:array, :integer}
+    field :booleanarray_value, {:array, :boolean}
+    field :longintegerarray_value, {:array, :integer}
+    field :stringarray_value, {:array, :string}
+    field :binaryblobarray_value, {:array, :binary}
+    field :datetimearray_value, {:array, :utc_datetime_usec}
+  end
+
+  def reception(individual_property) do
+    nanos =
+      individual_property.reception_timestamp_submillis
+      |> Kernel.||(0)
+      |> Kernel.*(100)
+
+    individual_property.reception_timestamp
+    |> DateTime.add(nanos, :nanosecond)
+  end
+end