refactoring(exandra): CQEx to Exandra

Signed-off-by: Gabriele Ghio <gabriele.ghio@secomind.com>

Author: shinnokdisengir

apps/astarte_trigger_engine/.formatter.exs

@@ -1,4 +1,22 @@
+#
+# This file is part of Astarte.
+#
+# 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.
+# 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.
+#
+
 [
   inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"],
-  import_deps: [:plug, :skogsra]
+  import_deps: [:plug, :skogsra, :ecto]
 ]

apps/astarte_trigger_engine/config/config.exs

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 Ispirata 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.
@@ -37,4 +37,8 @@ config :astarte_trigger_engine, :events_consumer, Astarte.TriggerEngine.EventsCo
 
 config :astarte_trigger_engine, :amqp_adapter, ExRabbitPool.RabbitMQ
 
+config :astarte_trigger_engine, ecto_repos: [Astarte.TriggerEngine.Repo]
+
+config :astarte_trigger_engine, Astarte.TriggerEngine.Repo, []
+
 import_config "#{config_env()}.exs"

apps/astarte_trigger_engine/lib/astarte_trigger_engine/application.ex

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 Ispirata 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.
@@ -40,13 +40,13 @@ defmodule Astarte.TriggerEngine.Application do
     Config.validate!()
     DataAccessConfig.validate!()
 
-    xandra_options =
-      Config.xandra_options!()
-      |> Keyword.put(:name, :xandra)
+    xandra_options = Config.xandra_options!()
+    xandra_cluster_options = Keyword.put(xandra_options, :name, :xandra)
 
     children = [
       Astarte.TriggerEngineWeb.Telemetry,
-      {Xandra.Cluster, xandra_options},
+      {Xandra.Cluster, xandra_cluster_options},
+      {Astarte.TriggerEngine.Repo, xandra_options},
       DeliverySupervisor
     ]
 

apps/astarte_trigger_engine/lib/astarte_trigger_engine/events_consumer.ex

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017-2024 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.
@@ -17,14 +17,18 @@
 #
 
 defmodule Astarte.TriggerEngine.EventsConsumer do
+  require Logger
+
+  import Ecto.Query
+
+  alias Astarte.Core.CQLUtils
   alias Astarte.Core.Triggers.SimpleEvents.IncomingIntrospectionEvent
   alias Astarte.Core.Triggers.SimpleEvents.InterfaceVersion
   alias Astarte.Core.Triggers.SimpleEvents.SimpleEvent
   alias Astarte.Core.Triggers.Trigger
-  alias Astarte.DataAccess.Database
-  alias CQEx.Query, as: DatabaseQuery
-  alias CQEx.Result, as: DatabaseResult
-  require Logger
+  alias Astarte.TriggerEngine.Config
+  alias Astarte.TriggerEngine.Repo
+  alias Astarte.TriggerEngine.KvStore
 
   defmodule Behaviour do
     @callback consume(payload :: binary, headers :: map) :: :ok | {:error, reason :: atom}
@@ -312,27 +316,27 @@ defmodule Astarte.TriggerEngine.EventsConsumer do
   end
 
   defp retrieve_trigger_configuration(realm_name, trigger_id) do
+    keyspace_name =
+      CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
+
     query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement(
-        "SELECT value FROM kv_store WHERE group='triggers' AND key=:trigger_id;"
-      )
-      |> DatabaseQuery.put(:trigger_id, trigger_id)
-
-    with {:ok, client} <- Database.connect(realm: realm_name),
-         {:ok, result} <- DatabaseQuery.call(client, query),
-         [value: trigger_data] <- DatabaseResult.head(result),
-         trigger <- Trigger.decode(trigger_data),
+      from kvstore in KvStore,
+        prefix: ^keyspace_name,
+        where: kvstore.group == "triggers" and kvstore.key == ^trigger_id,
+        select: kvstore.value
+
+    with encoded_trigger when encoded_trigger != nil <- Repo.one(query),
+         trigger <- Trigger.decode(encoded_trigger),
          {:ok, action} <- Jason.decode(trigger.action) do
       {:ok, %{action: action, trigger_name: trigger.name}}
     else
-      {:error, :database_connection_error} ->
-        Logger.warning("Database connection error.")
-        {:error, :database_connection_error}
-
-      error ->
-        Logger.warning("Error while processing event: #{inspect(error)}")
+      nil ->
+        Logger.warning("Trigger not found: #{inspect(trigger_id)}")
         {:error, :trigger_not_found}
+
+      _ ->
+        Logger.warning("Error while decoding trigger: #{inspect(trigger_id)}")
+        {:error, :trigger_decoding_error}
     end
   end
 end

apps/astarte_trigger_engine/lib/astarte_trigger_engine/kv_store.ex

@@ -0,0 +1,27 @@
+# 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.TriggerEngine.KvStore do
+  use TypedEctoSchema
+
+  @primary_key false
+  typed_schema "kv_store" do
+    field :group, :string, primary_key: true
+    field :key, :string, primary_key: true
+    field :value, :binary
+  end
+end

apps/astarte_trigger_engine/lib/repo.ex

@@ -0,0 +1,31 @@
+#
+# 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.TriggerEngine.Repo do
+  use Ecto.Repo, otp_app: :astarte_trigger_engine, adapter: Exandra
+  alias Astarte.TriggerEngine.Config
+
+  @impl Ecto.Repo
+  def init(_context, config) do
+    config =
+      Config.xandra_options!()
+      |> Keyword.merge(config)
+
+    {:ok, config}
+  end
+end

apps/astarte_trigger_engine/mix.exs

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017-2021 Ispirata 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.
@@ -80,21 +80,24 @@ defmodule Astarte.TriggerEngine.Mixfile do
       {:cyanide, "~> 2.0"},
       {:httpoison, "~> 1.6"},
       {:jason, "~> 1.2"},
-      {:excoveralls, "~> 0.15", only: :test},
-      # hex.pm package and esl/ex_rabbit_pool do not support amqp version 2.1.
-      # This fork is supporting amqp ~> 2.0 and also ~> 3.0.
-      {:ex_rabbit_pool, github: "leductam/ex_rabbit_pool"},
       {:plug_cowboy, "~> 2.1"},
       {:telemetry_metrics_prometheus_core, "~> 0.4"},
       {:telemetry_metrics, "~> 0.4"},
       {:telemetry_poller, "~> 0.4"},
-      {:mox, "~> 0.5", only: :test},
+      {:typed_ecto_schema, "~> 0.4"},
+      {:ecto, "~> 3.12"},
+      {:exandra, "~> 0.12"},
       {:pretty_log, "~> 0.1"},
       {:telemetry, "~> 0.4"},
-      {:xandra, "~> 0.13"},
+      {:xandra, "~> 0.19"},
       {:skogsra, "~> 2.2"},
       {:observer_cli, "~> 1.5"},
+      # hex.pm package and esl/ex_rabbit_pool do not support amqp version 2.1.
+      # This fork is supporting amqp ~> 2.0 and also ~> 3.0.
+      {:ex_rabbit_pool, github: "leductam/ex_rabbit_pool"},
       {:dialyxir, "~> 1.0", only: [:dev, :ci], runtime: false},
+      {:excoveralls, "~> 0.15", only: :test},
+      {:mox, "~> 0.5", only: :test},
       # Workaround for Elixir 1.15 / ssl_verify_fun issue
       # See also: https://github.com/deadtrickster/ssl_verify_fun.erl/pull/27
       {:ssl_verify_fun, "~> 1.1.0", manager: :rebar3, override: true}