feat(capabilities): Add capabilites endpoint

- Devices can publish on `capabilities` endpoint
- Adds the `purge_properties_compression_format` capability, setting
  whether the device should recive the purge properties compressed with
  zlib or in plain text

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

Author: lusergit

CHANGELOG.md

@@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   `booleanarray`, `longintegerarray`, `stringarray`, `datetimearray`, `binaryblobarray`.
 - [astarte_export] Added a new command for exporting by device_id. 
   `mix astarte.export $REALM $FILE_XML $DEVICE_ID`
+- [astarte_data_updater_plant] Added the `capabilities` endpoint for devices,
+  setting whether the device should get the purge properties compressed with
+  `zlib` or in plaintext
 
 ## [1.2.1] - Unreleased
 ### Changed

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

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 - 2023 SECO Mind Srl
+# Copyright 2017 - 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.
@@ -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,
@@ -90,7 +95,8 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
       trigger_id_to_policy_name: %{},
       discard_messages: false,
       last_deletion_in_progress_refresh: 0,
-      last_datastream_maximum_retention_refresh: 0
+      last_datastream_maximum_retention_refresh: 0,
+      capabilities: capabilities
     }
 
     encoded_device_id = Device.encode_device_id(device_id)
@@ -104,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
@@ -141,6 +150,9 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
         %{@control_path_header => control_path} = headers
         handle_control(state, control_path, payload, timestamp)
 
+      "capabilities" ->
+        handle_capabilities(state, payload, timestamp)
+
       _ ->
         # Ack all messages for now
         {:ack, :ok, state}
@@ -1608,6 +1620,37 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
     {:ack, :ok, final_state}
   end
 
+  def handle_capabilities(state, payload, _timestamp) do
+    %{device_id: device_id} = state
+
+    with {:ok, bson_payload} <- decode_bson_payload(state, payload),
+         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
@@ -1641,9 +1684,7 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
 
     # TODO: check payload size, to avoid anoying crashes
 
-    <<_size_header::size(32), zlib_payload::binary>> = payload
-
-    case PayloadsDecoder.safe_inflate(zlib_payload) do
+    case inflate_purge_properties_payload(payload) do
       {:ok, decoded_payload} ->
         :ok = prune_device_properties(new_state, decoded_payload, timestamp_ms)
 
@@ -1727,6 +1768,40 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
     {:discard, :unexpected_control_message, new_state, {:continue, continue_arg}}
   end
 
+  defp decode_bson_payload(state, payload) do
+    with {:error, error} <- Cyanide.decode(payload) do
+      Logger.warning(
+        "error parsing capabilities BSON message: #{error}",
+        tag: "capabilities_error"
+      )
+
+      :telemetry.execute(
+        [:astarte, :data_updater_plant, :data_updater, :discarded_message],
+        %{},
+        %{realm: state.realm}
+      )
+
+      {:discard, error, state}
+    end
+  end
+
+  defp inflate_purge_properties_payload(<<255, 255, 255, 255, payload::binary>>) do
+    _ =
+      Logger.debug("Received plaintext purge properties properties payload",
+        tag: "purge_properties"
+      )
+
+    {:ok, payload}
+  end
+
+  defp inflate_purge_properties_payload(<<_size_header::size(32), zlib_payload::binary>>) do
+    Logger.debug("Received zlib compressed purge properties properties payload",
+      tag: "purge_properties"
+    )
+
+    PayloadsDecoder.safe_inflate(zlib_payload)
+  end
+
   defp delete_volatile_trigger(
          state,
          {obj_id, _obj_type},
@@ -2673,7 +2748,12 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
 
     # TODO: use the returned byte count in stats
     with {:ok, _bytes} <-
-           send_consumer_properties_payload(state.realm, state.device_id, abs_paths_list) do
+           send_consumer_properties_payload(
+             state.realm,
+             state.device_id,
+             abs_paths_list,
+             state.purge_properties_compression_format
+           ) do
       :ok
     end
   end
@@ -2742,15 +2822,22 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.Impl do
     :ok
   end
 
-  defp send_consumer_properties_payload(realm, device_id, abs_paths_list) do
+  defp send_consumer_properties_payload(realm, device_id, abs_paths_list, compression_format) do
     topic = "#{realm}/#{Device.encode_device_id(device_id)}/control/consumer/properties"
 
     uncompressed_payload = Enum.join(abs_paths_list, ";")
 
-    payload_size = byte_size(uncompressed_payload)
-    compressed_payload = :zlib.compress(uncompressed_payload)
+    payload =
+      case compression_format do
+        :zlib ->
+          payload_size = byte_size(uncompressed_payload)
+          compressed_payload = :zlib.compress(uncompressed_payload)
 
-    payload = <<payload_size::unsigned-big-integer-size(32), compressed_payload::binary>>
+          <<payload_size::unsigned-big-integer-size(32), compressed_payload::binary>>
+
+        :plaintext ->
+          <<255, 255, 255, 255>> <> uncompressed_payload
+      end
 
     case VMQPlugin.publish(topic, payload, 2) do
       {:ok, %{local_matches: local, remote_matches: remote}} when local + remote == 1 ->

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

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 - 2023 SECO Mind Srl
+# Copyright 2017 - 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.
@@ -46,6 +46,7 @@ defmodule Astarte.DataUpdaterPlant.DataUpdater.State do
     :trigger_id_to_policy_name,
     :discard_messages,
     :last_deletion_in_progress_refresh,
-    :last_datastream_maximum_retention_refresh
+    :last_datastream_maximum_retention_refresh,
+    :capabilities
   ]
 end

apps/astarte_data_updater_plant/mix.exs

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 - 2023 SECO Mind Srl
+# Copyright 2017 - 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.

apps/astarte_data_updater_plant/test/astarte_data_updater_plant/data_updater_test.exs

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 - 2023 SECO Mind Srl
+# Copyright 2017 - 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.
@@ -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
@@ -40,6 +41,7 @@ defmodule Astarte.DataUpdaterPlant.DataUpdaterTest do
   alias Astarte.DataAccess.Database
   alias Astarte.DataUpdaterPlant.AMQPTestHelper
   alias Astarte.DataUpdaterPlant.DatabaseTestHelper
+  alias Astarte.DataUpdaterPlant.DataUpdater.State
   alias Astarte.Core.CQLUtils
   alias CQEx.Query, as: DatabaseQuery
   alias CQEx.Result, as: DatabaseResult
@@ -1750,6 +1752,35 @@ defmodule Astarte.DataUpdaterPlant.DataUpdaterTest do
     assert [%{"dup_end_ack" => true}] = dup_end_ack_result
   end
 
+  test "capabilities message correctly handled" do
+    AMQPTestHelper.clean_queue()
+
+    realm = "autotestrealm"
+
+    device_id = Device.random_device_id()
+    encoded_device_id = Device.encode_device_id(device_id)
+
+    DatabaseTestHelper.insert_device(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")
+
+    payload = Cyanide.encode!(%{"purge_properties_compression_format" => "plaintext"})
+
+    handle_capabilities(
+      realm,
+      encoded_device_id,
+      payload,
+      gen_tracking_id(),
+      timestamp_us_x_10
+    )
+
+    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
     query =
       DatabaseQuery.new()
@@ -1882,6 +1913,27 @@ defmodule Astarte.DataUpdaterPlant.DataUpdaterTest do
     ensure_message_has_been_handled(realm, device_id)
   end
 
+  defp handle_capabilities(realm, encoded_device_id, payload, _tracking_id, timestamp) do
+    {:ok, device_id} = Astarte.Core.Device.decode_device_id(encoded_device_id)
+
+    headers =
+      headers_fixture(realm, encoded_device_id, x_astarte_msg_type: "capabilities")
+
+    publish_opts = [
+      headers: headers,
+      message_id: generate_message_id(realm, encoded_device_id, timestamp),
+      timestamp: timestamp,
+      sharding_key: {realm, device_id}
+    ]
+
+    {:ok, pid} = Mississippi.Consumer.MessageTracker.get_message_tracker({realm, device_id})
+    :erlang.trace(pid, true, [:receive])
+
+    :ok = Mississippi.Producer.EventsProducer.publish(payload, publish_opts)
+
+    assert_receive {:trace, ^pid, :receive, {_, {_, _}, {:ack_delivery, _message}}}
+  end
+
   defp handle_control(realm, encoded_device_id, control_path, value, _tracking_id, timestamp) do
     {:ok, device_id} = Astarte.Core.Device.decode_device_id(encoded_device_id)