Merge pull request #1129 from noaccOS/refactor/datetimes

refactor(appengine): use datetimes for data insertion

Author: Annopaolo

apps/astarte_appengine_api/lib/astarte_appengine_api/datetime.ex

@@ -0,0 +1,53 @@
+#
+# 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.DateTime do
+  @moduledoc """
+  Ecto type for DateTimes with millisecond precision.
+  """
+  use Ecto.Type
+
+  @type t :: DateTime.t()
+
+  @doc false
+  def type, do: :utc_datetime_msec
+
+  @spec load(t() | any()) :: {:ok, t()} | :error
+  def load(%DateTime{} = datetime), do: {:ok, datetime}
+  def load(timestamp) when is_integer(timestamp), do: {:ok, DateTime.from_unix!(timestamp)}
+  def load(_other), do: :error
+
+  # xandra accepts both integers and datetimes, it can do the job for us
+  @spec dump(t() | any()) :: {:ok, t()} | :error
+  def dump(%DateTime{} = datetime), do: {:ok, datetime}
+  def dump(timestamp) when is_integer(timestamp), do: {:ok, timestamp}
+  def dump(_other), do: :error
+
+  @spec cast(t() | any()) :: {:ok, t()} | :error
+  def cast(datetime_or_timestamp_or_any), do: load(datetime_or_timestamp_or_any)
+
+  def split_submillis(timestamp) do
+    timestamp_ms = DateTime.truncate(timestamp, :millisecond)
+    submillis = timestamp |> DateTime.to_unix(:microsecond) |> rem(1000)
+    # `DateTime`s are microsecond precision, individual_properties's submillis
+    # have one extra digit. Keep the unit consistent with DUP
+    decimicrosecond_submillis = submillis * 10
+
+    {timestamp_ms, decimicrosecond_submillis}
+  end
+end

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

@@ -221,10 +221,6 @@ defmodule Astarte.AppEngine.API.Device do
 
       now = DateTime.utc_now()
 
-      timestamp_micro =
-        now
-        |> DateTime.to_unix(:microsecond)
-
       db_max_ttl =
         if mapping.database_retention_policy == :use_ttl do
           min(realm_max_ttl, mapping.database_retention_ttl)
@@ -249,7 +245,7 @@ defmodule Astarte.AppEngine.API.Device do
         mapping,
         path,
         value,
-        timestamp_micro,
+        now,
         opts
       )
 
@@ -376,10 +372,6 @@ defmodule Astarte.AppEngine.API.Device do
        ) do
     now = DateTime.utc_now()
 
-    timestamp_micro =
-      now
-      |> DateTime.to_unix(:microsecond)
-
     with {:ok, mappings} <-
            Mappings.fetch_interface_mappings(
              realm_name,
@@ -425,7 +417,7 @@ defmodule Astarte.AppEngine.API.Device do
         nil,
         path,
         value,
-        timestamp_micro,
+        now,
         opts
       )
 

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

@@ -20,6 +20,7 @@ defmodule Astarte.AppEngine.API.Device.Queries do
   import Ecto.Query
 
   alias Astarte.AppEngine.API.Config
+  alias Astarte.AppEngine.API.DateTime, as: DateTimeMs
   alias Astarte.AppEngine.API.Device.DeletionInProgress
   alias Astarte.AppEngine.API.Device.DeviceStatus
   alias Astarte.AppEngine.API.Device.DevicesList
@@ -181,19 +182,17 @@ defmodule Astarte.AppEngine.API.Device.Queries do
 
     keyspace = DataAccessRealm.keyspace_name(realm_name)
 
-    # Ecto expects microsecond precision
-    {reception, reception_submillis} = split_datetime_to_ms_and_submillis(reception_timestamp)
-    value_timestamp = value_timestamp |> DateTime.truncate(:millisecond) |> pad_usec()
+    value =
+      %IndividualProperty{
+        device_id: device_id,
+        interface_id: interface_descriptor.interface_id,
+        endpoint_id: endpoint_id,
+        path: path,
+        reception: reception_timestamp,
+        datetime_value: value_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
-    }
+    value = value |> IndividualProperty.prepare_for_db()
 
     ttl = opts[:ttl]
 
@@ -262,7 +261,7 @@ defmodule Astarte.AppEngine.API.Device.Queries do
     value_column = CQLUtils.type_to_db_column_name(endpoint.value_type)
     keyspace = DataAccessRealm.keyspace_name(realm_name)
 
-    {timestamp_ms, timestamp_submillis} = split_ms_and_submillis(timestamp)
+    {timestamp_ms, timestamp_submillis} = DateTimeMs.split_submillis(timestamp)
 
     # TODO: :reception_timestamp_submillis is just a place holder right now
     interface_storage_attributes = %{
@@ -299,7 +298,7 @@ defmodule Astarte.AppEngine.API.Device.Queries do
       ) do
     value_column = CQLUtils.type_to_db_column_name(endpoint.value_type)
     keyspace = DataAccessRealm.keyspace_name(realm_name)
-    {timestamp_ms, timestamp_submillis} = split_ms_and_submillis(timestamp)
+    {timestamp_ms, timestamp_submillis} = DateTimeMs.split_submillis(timestamp)
 
     attributes = %{
       value_column => to_db_friendly_type(value),
@@ -312,7 +311,12 @@ defmodule Astarte.AppEngine.API.Device.Queries do
       reception_timestamp_submillis: timestamp_submillis
     }
 
-    Repo.insert_all(interface_descriptor.storage, [attributes], prefix: keyspace, ttl: opts[:ttl])
+    {1, _} =
+      Repo.insert_all(interface_descriptor.storage, [attributes],
+        prefix: keyspace,
+        ttl: opts[:ttl]
+      )
+
     :ok
   end
 
@@ -348,14 +352,12 @@ defmodule Astarte.AppEngine.API.Device.Queries do
         {endpoint_name, %{name: column_name, type: endpoint.value_type}}
       end)
 
-    {timestamp_ms, submillis} = split_ms_and_submillis(timestamp)
-
     base_attributes = %{
       device_id: device_id,
       path: path
     }
 
-    timestamp_attributes = timestamp_attributes(explicit_timestamp?, timestamp_ms, submillis)
+    timestamp_attributes = timestamp_attributes(explicit_timestamp?, timestamp)
     value_attributes = value_attributes(column_meta, value)
 
     object_datastream =
@@ -372,15 +374,21 @@ defmodule Astarte.AppEngine.API.Device.Queries do
     :ok
   end
 
-  defp timestamp_attributes(true = _explicit_timestamp?, timestamp, submillis) do
+  defp timestamp_attributes(true = _explicit_timestamp?, timestamp) do
+    {timestamp, submillis} =
+      Astarte.AppEngine.API.DateTime.split_submillis(timestamp)
+
     %{
       value_timestamp: timestamp,
       reception_timestamp: timestamp,
       reception_timestamp_submillis: submillis
     }
   end
 
-  defp timestamp_attributes(_nil_or_false_explicit_timestamp?, timestamp, submillis) do
+  defp timestamp_attributes(_nil_or_false_explicit_timestamp?, timestamp) do
+    {timestamp, submillis} =
+      Astarte.AppEngine.API.DateTime.split_submillis(timestamp)
+
     %{reception_timestamp: timestamp, reception_timestamp_submillis: submillis}
   end
 
@@ -1099,29 +1107,4 @@ 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
-
-  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/devices/device.ex

@@ -18,6 +18,7 @@
 
 defmodule Astarte.AppEngine.API.Devices.Device do
   use TypedEctoSchema
+  alias Astarte.AppEngine.API.DateTime, as: DateTimeMs
   alias Astarte.AppEngine.API.UUID
 
   @primary_key {:device_id, UUID, autogenerate: false}
@@ -39,16 +40,16 @@ defmodule Astarte.AppEngine.API.Devices.Device do
       types: [:string, :integer],
       value: :integer
 
-    field :first_credentials_request, :utc_datetime_usec
-    field :first_registration, :utc_datetime_usec
+    field :first_credentials_request, DateTimeMs
+    field :first_registration, DateTimeMs
     field :groups, Exandra.Map, key: :string, value: UUID
     field :inhibit_credentials_request, :boolean
     field :introspection, Exandra.Map, key: :string, value: :integer
     field :introspection_minor, Exandra.Map, key: :string, value: :integer
-    field :last_connection, :utc_datetime_usec
+    field :last_connection, DateTimeMs
     field :last_credentials_request_ip, Exandra.Inet
 
-    field :last_disconnection, :utc_datetime_usec
+    field :last_disconnection, DateTimeMs
     field :last_seen_ip, Exandra.Inet
 
     field :old_introspection,

apps/astarte_appengine_api/lib/astarte_appengine_api/realms/individual_datastream.ex

@@ -0,0 +1,61 @@
+#
+# 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.IndividualDatastream do
+  use TypedEctoSchema
+  alias Astarte.AppEngine.API.DateTime, as: DateTimeMs
+  alias Astarte.AppEngine.API.UUID
+
+  @primary_key false
+  typed_schema "individual_datastreams" do
+    field :reception, DateTimeMs, 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 :value_timestamp, DateTimeMs, primary_key: true
+    field :reception_timestamp, DateTimeMs, primary_key: true
+    field :reception_timestamp_submillis, :integer, primary_key: true
+    field :binaryblob_value, :binary
+    field :binaryblobarray_value, {:array, :binary}
+    field :boolean_value, :boolean
+    field :booleanarray_value, {:array, :boolean}
+    field :datetime_value, DateTimeMs
+    field :datetimearray_value, {:array, DateTimeMs}
+    field :double_value, :float
+    field :doublearray_value, {:array, :float}
+    field :integer_value, :integer
+    field :integerarray_value, {:array, :integer}
+    field :longinteger_value, :integer
+    field :longintegerarray_value, {:array, :integer}
+    field :string_value, :string
+    field :stringarray_value, {:array, :string}
+  end
+
+  def prepare_for_db(%{reception: nil} = individual_datastream), do: individual_datastream
+
+  def prepare_for_db(individual_datastream) do
+    {reception_ms, submillis} = DateTimeMs.split_submillis(individual_datastream.reception)
+
+    %{
+      individual_datastream
+      | reception_timestamp: reception_ms,
+        reception_timestamp_submillis: submillis
+    }
+  end
+end

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

@@ -18,40 +18,43 @@
 
 defmodule Astarte.AppEngine.API.Realms.IndividualProperty do
   use TypedEctoSchema
+  alias Astarte.AppEngine.API.DateTime, as: DateTimeMs
   alias Astarte.AppEngine.API.UUID
 
   @primary_key false
   typed_schema "individual_properties" do
-    field :reception, :utc_datetime_usec, virtual: true
+    field :reception, DateTimeMs, 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, DateTimeMs
     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 :datetime_value, DateTimeMs
     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}
+    field :datetimearray_value, {:array, DateTimeMs}
   end
 
-  def reception(individual_property) do
-    nanos =
-      individual_property.reception_timestamp_submillis
-      |> Kernel.||(0)
-      |> Kernel.*(100)
+  def prepare_for_db(%{reception: nil} = individual_property), do: individual_property
 
-    individual_property.reception_timestamp
-    |> DateTime.add(nanos, :nanosecond)
+  def prepare_for_db(individual_property) do
+    {reception_ms, submillis} = DateTimeMs.split_submillis(individual_property.reception)
+
+    %{
+      individual_property
+      | reception_timestamp: reception_ms,
+        reception_timestamp_submillis: submillis
+    }
   end
 end