chore(migrations): device migrations

- Added device migrations to add `purge_properties_compression_format`
  column to `devices` table in the database

Signed-off-by: Luca Zaninotto <luca.zaninotto@secomind.com>

Author: lusergit

apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/data_updater/impl.ex

@@ -22,6 +22,7 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
   alias Astarte.Core.CQLUtils
   alias Astarte.DataUpdaterPlant.Config
   alias Astarte.Core.Device
+  alias Astarte.Core.Device.Capabilities
   alias Astarte.Core.InterfaceDescriptor
   alias Astarte.Core.Mapping
   alias Astarte.Core.Mapping.EndpointsAutomaton
@@ -69,6 +70,10 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
     # TODO change this, we want extended device IDs to fall in the same process
     {realm, device_id} = sharding_key
 
+    capabilities = %Capabilities{
+      purge_properties_compression_format: :zlib
+    }
+
     state = %State{
       realm: realm,
       device_id: device_id,
@@ -91,7 +96,7 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
       discard_messages: false,
       last_deletion_in_progress_refresh: 0,
       last_datastream_maximum_retention_refresh: 0,
-      purge_properties_compression_format: :zlib
+      capabilities: capabilities
     }
 
     encoded_device_id = Device.encode_device_id(device_id)
@@ -105,9 +110,12 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
 
     {:ok, ttl} = Queries.fetch_datastream_maximum_storage_retention(db_client)
 
+    {:ok, capabilities} = Queries.fetch_device_capabilities(db_client, device_id)
+
     new_state =
       Map.merge(state, stats_and_introspection)
       |> Map.put(:datastream_maximum_storage_retention, ttl)
+      |> Map.put(:capabilities, capabilities)
 
     {:ok, new_state}
   end
@@ -1613,12 +1621,36 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
   end
 
   def handle_capabilities(state, payload, _timestamp) do
+    %{device_id: device_id} = state
+
     with {:ok, bson_payload} <- decode_bson_payload(state, payload),
-         {:ok, new_state} <- maybe_purge_properties_compression(state, bson_payload) do
+         changeset = Capabilities.changeset(state.capabilities, bson_payload),
+         {:ok, capabilities} <- update_capabilities(changeset, state) do
+      {:ok, db_client} = Database.connect(realm: state.realm)
+      Queries.set_device_capabilities(db_client, device_id, capabilities)
+
+      new_state = %State{state | capabilities: capabilities}
       {:ack, :ok, new_state}
     end
   end
 
+  defp update_capabilities(changeset, state) do
+    with {:error, error} <- Ecto.Changeset.apply_action(changeset, :update) do
+      Logger.warning(
+        "Error while updating capabilities: #{error}",
+        tag: "capabilities_error"
+      )
+
+      :telemetry.execute(
+        [:astarte, :data_updater_plant, :data_updater, :discarded_message],
+        %{},
+        %{realm: state.realm}
+      )
+
+      {:discard, error, state}
+    end
+  end
+
   def handle_control(%State{discard_messages: true} = state, _, _, _) do
     {:ack, :discard_messages, state}
   end
@@ -1753,34 +1785,6 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
     end
   end
 
-  defp maybe_purge_properties_compression(state, bson) do
-    case Map.fetch(bson, "purge_properties_compression_format") do
-      # capability not found in bson, the user did not set it
-      :error ->
-        {:ok, state}
-
-      {:ok, "plaintext"} ->
-        {:ok, Map.put(state, :purge_properties_compression_format, :plaintext)}
-
-      {:ok, "zlib"} ->
-        {:ok, Map.put(state, :purge_properties_compression_format, :zlib)}
-
-      invalid ->
-        Logger.warning(
-          "Invalid capabilities purge_properties_compression_format value, ignoring: #{invalid}",
-          tag: "capabilities_error"
-        )
-
-        :telemetry.execute(
-          [:astarte, :data_updater_plant, :data_updater, :discarded_message],
-          %{},
-          %{realm: state.realm}
-        )
-
-        {:discard, :capabilities_error, state}
-    end
-  end
-
   defp inflate_purge_properties_payload(<<255, 255, 255, 255, payload::binary>>) do
     _ =
       Logger.debug("Received plaintext purge properties properties payload",

apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/data_updater/queries.ex

@@ -19,6 +19,7 @@
 defmodule Astarte.DataUpdaterPlant.DataUpdater.Queries do
   alias Astarte.Core.CQLUtils
   alias Astarte.Core.Device
+  alias Astarte.Core.Device.Capabilities
   alias Astarte.Core.InterfaceDescriptor
   alias Astarte.Core.Mapping
   alias Astarte.DataUpdaterPlant.Config
@@ -450,6 +451,53 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Queries do
     end
   end
 
+  def fetch_device_capabilities(db_client, device_id) do
+    device_capabilities_statement = """
+    SELECT purge_properties_compression_format
+    FROM devices
+    WHERE device_id=:device_id
+    """
+
+    device_capabilities_query =
+      DatabaseQuery.new()
+      |> DatabaseQuery.statement(device_capabilities_statement)
+      |> DatabaseQuery.put(:device_id, device_id)
+
+    with {:ok, result} <- DatabaseQuery.call(db_client, device_capabilities_query) do
+      capabilities_row =
+        result
+        |> DatabaseResult.head()
+        |> Enum.into(%{})
+
+      capabilities =
+        %Capabilities{}
+        |> Capabilities.changeset(capabilities_row)
+        |> Ecto.Changeset.apply_changes()
+
+      {:ok, capabilities}
+    end
+  end
+
+  def set_device_capabilities(
+        db_client,
+        device_id,
+        %Capabilities{purge_properties_compression_format: format} = _capabilities
+      ) do
+    device_capabilities_statement = """
+    UPDATE devices
+    SET purge_properties_compression_format=:format
+    WHERE device_id=:device_id
+    """
+
+    device_capabilities_query =
+      DatabaseQuery.new()
+      |> DatabaseQuery.statement(device_capabilities_statement)
+      |> DatabaseQuery.put(:format, format)
+      |> DatabaseQuery.put(:device_id, device_id)
+
+    DatabaseQuery.call!(db_client, device_capabilities_query)
+  end
+
   def set_device_connected!(db_client, device_id, timestamp_ms, ip_address) do
     set_connection_info!(db_client, device_id, timestamp_ms, ip_address)
 

apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/data_updater/state.ex

@@ -47,6 +47,6 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.State do
     :discard_messages,
     :last_deletion_in_progress_refresh,
     :last_datastream_maximum_retention_refresh,
-    :purge_properties_compression_format
+    :capabilities
   ]
 end

apps/astarte_data_updater_plant/test/astarte_data_updater_plant/data_updater_test.exs

@@ -22,6 +22,7 @@ defmodule Astarte.DataUpdaterPlant.DataUpdaterTest do
 
   alias Astarte.DataUpdaterPlant.Config
   alias Astarte.Core.Device
+  alias Astarte.Core.Device.Capabilities
   alias Astarte.Core.Triggers.SimpleEvents.DeviceConnectedEvent
   alias Astarte.Core.Triggers.SimpleEvents.IncomingDataEvent
   alias Astarte.Core.Triggers.SimpleEvents.PathRemovedEvent
@@ -1761,8 +1762,8 @@ defmodule Astarte.DataUpdaterPlant.DataUpdaterTest do
 
     DatabaseTestHelper.insert_device(device_id)
 
-    assert %State{purge_properties_compression_format: :zlib} =
-             dump_state(realm, encoded_device_id)
+    state = dump_state(realm, encoded_device_id)
+    assert %Capabilities{purge_properties_compression_format: :zlib} = state.capabilities
 
     timestamp_us_x_10 = make_timestamp("2025-01-20T14:00:32+00:00")
 
@@ -1776,8 +1777,8 @@ defmodule Astarte.DataUpdaterPlant.DataUpdaterTest do
       timestamp_us_x_10
     )
 
-    assert %State{purge_properties_compression_format: :plaintext} =
-             dump_state(realm, encoded_device_id)
+    new_state = dump_state(realm, encoded_device_id)
+    assert %Capabilities{purge_properties_compression_format: :plaintext} = new_state.capabilities
   end
 
   defp retrieve_endpoint_id(client, interface_name, interface_major, path) do

apps/astarte_data_updater_plant/test/support/database_test_helper.exs

@@ -63,15 +63,16 @@ defmodule Astarte.DataUpdaterPlant.DatabaseTestHelper do
         last_pairing_ip inet,
         last_seen_ip inet,
         groups map<text, timeuuid>,
+        purge_properties_compression_format varchar,
 
         PRIMARY KEY (device_id)
     );
   """
 
   @insert_device """
-        INSERT INTO #{CQLUtils.realm_name_to_keyspace_name("autotestrealm", Config.astarte_instance_id!())}.devices (device_id, connected, last_connection, last_disconnection, first_pairing, last_seen_ip, last_pairing_ip, total_received_msgs, total_received_bytes, introspection, groups)
+        INSERT INTO #{CQLUtils.realm_name_to_keyspace_name("autotestrealm", Config.astarte_instance_id!())}.devices (device_id, connected, last_connection, last_disconnection, first_pairing, last_seen_ip, last_pairing_ip, total_received_msgs, total_received_bytes, introspection, groups, purge_properties_compression_format)
           VALUES (:device_id, false, :last_connection, :last_disconnection, :first_pairing,
-          :last_seen_ip, :last_pairing_ip, :total_received_msgs, :total_received_bytes, :introspection, :groups);
+          :last_seen_ip, :last_pairing_ip, :total_received_msgs, :total_received_bytes, :introspection, :groups, :purge_properties_compression_format);
   """
 
   @create_interfaces_table """
@@ -803,6 +804,13 @@ defmodule Astarte.DataUpdaterPlant.DatabaseTestHelper do
     groups = Keyword.get(opts, :groups, [])
     groups_map = for group <- groups, do: {group, UUID.uuid1()}
 
+    compression_format =
+      case Keyword.get(opts, :purge_properties_compression_format) do
+        nil -> "zlib"
+        :zlib -> "zlib"
+        :plaintext -> "plaintext"
+      end
+
     query =
       DatabaseQuery.new()
       |> DatabaseQuery.statement(@insert_device)
@@ -816,6 +824,7 @@ defmodule Astarte.DataUpdaterPlant.DatabaseTestHelper do
       |> DatabaseQuery.put(:total_received_bytes, total_received_bytes)
       |> DatabaseQuery.put(:introspection, introspection)
       |> DatabaseQuery.put(:groups, groups_map)
+      |> DatabaseQuery.put(:purge_properties_compression_format, compression_format)
 
     DatabaseQuery.call(client, query)
   end

apps/astarte_housekeeping/priv/migrations/realm/0006_devices_add_purge_properties_format.sql

@@ -0,0 +1,4 @@
+ALTER TABLE devices
+ADD (
+    purge_properties_compression_format tinyint
+);