refactor(appengine): use datetimes for data insertion

using integer timestamps, especially when doing calculations on them,
is error prone and we've just recently resolved a bug that was caused
by a double "microsecond to millisecond" conversion.

elixir has something more solid than that: `DateTime`s.

although currently unused, the submillis should now use the correct
number of digits.

an unused and nonfunctioning function in `IndividualProperty` has been
removed

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

Author: noaccOS

apps/astarte_appengine_api/lib/astarte_appengine_api/datetime.ex

@@ -40,4 +40,14 @@ defmodule Astarte.AppEngine.API.DateTime do
 
   @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

@@ -256,10 +256,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)
@@ -284,7 +280,7 @@ defmodule Astarte.AppEngine.API.Device do
         mapping,
         path,
         value,
-        timestamp_micro,
+        now,
         opts
       )
 
@@ -411,10 +407,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,
@@ -460,7 +452,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
 
@@ -988,29 +996,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/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

@@ -46,13 +46,15 @@ defmodule Astarte.AppEngine.API.Realms.IndividualProperty do
     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