publish aggregate when task is complete

lib/workload_service/consumers/quote_requested_consumer.ex

@@ -52,7 +52,7 @@ defmodule WorkloadService.Consumers.QuoteRequestedConsumer do
 
   defp handle_event(
          %{
-            "id" => %{"org_id" => org_id, "application_id" => app_id, "policy_type" => policy_type} = application_id,
+           "id" => %{"org_id" => org_id, "application_id" => app_id, "policy_type" => policy_type},
            "provider_id" => provider_id,
            "policy_details" => policy_details,
            "applicant_info" => applicant_info

lib/workload_service/handlers/task_completed_handler.ex

@@ -0,0 +1,51 @@
+defmodule WorkloadService.Handlers.TaskCompletedHandler do
+  use Commanded.Event.Handler,
+    application: WorkloadService.CommandedApp,
+    name: "TaskCompletedHandler"
+
+  require Logger
+
+  alias WorkloadService.Events.TaskCompleted
+  alias Commanded.Aggregates.Aggregate
+  alias WorkloadService.MessageBus
+
+  alias WorkloadService.Aggregates.{
+    QuoteTask,
+    SolicitationTask
+  }
+
+  def handle(%TaskCompleted{} = event, _metadata) do
+    aggregate_module =
+      case event.id.type do
+        "quote" -> QuoteTask
+        # "solicitation" -> SolicitationTask
+        _ -> nil
+      end
+
+    if aggregate_module do
+      case Aggregate.aggregate_state(
+             WorkloadService.CommandedApp,
+             aggregate_module,
+             event.id
+           ) do
+        nil ->
+          Logger.warning("TaskCompletedHandler: aggregate not found for #{event.id}")
+
+        state ->
+          MessageBus.publish(
+            "workload_service.events.quote_task_completed",
+            "quote_task_completed",
+            state
+          )
+
+          Logger.info("TaskCompletedHandler: published for #{event.id}")
+      end
+    end
+
+    :ok
+  rescue
+    e ->
+      Logger.error("TaskCompletedHandler: failed to process - #{inspect(e)}")
+      :ok
+  end
+end

lib/workload_service/message_bus.ex

@@ -1,86 +1,21 @@
 defmodule WorkloadService.MessageBus do
-  @moduledoc false
+  use AMQP
 
-  use GenServer
+  def publish(exchange, routing_key, event) do
-  require Logger
+    payload = Jason.encode!(event)
 
-  alias AMQP.{Connection, Channel, Exchange}
+    :ok =
-
+      AMQP.Basic.publish(channel(), exchange, routing_key, payload,
-  @reconnect_interval 5_000
-
-  def start_link(opts \\ []) do
-    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
-  end
-
-  def publish(routing_key, payload) when is_binary(payload) do
-    GenServer.cast(__MODULE__, {:publish, routing_key, payload})
-  end
-
-  def publish(routing_key, payload) do
-    GenServer.cast(__MODULE__, {:publish, routing_key, Jason.encode!(payload)})
-  end
-
-  @impl true
-  def init(_opts) do
-    send(self(), :connect)
-    {:ok, %{channel: nil, connection: nil}}
-  end
-
-  @impl true
-  def handle_info(:connect, _state) do
-    case connect() do
-      {:ok, connection, channel} ->
-        {:noreply, %{connection: connection, channel: channel}}
-
-      {:error, reason} ->
-        Logger.error("Failed to connect to RabbitMQ: #{inspect(reason)}")
-        schedule_reconnect()
-        {:noreply, %{channel: nil, connection: nil}}
-    end
-  end
-
-  @impl true
-  def handle_info({:DOWN, _, :process, _pid, reason}, _state) do
-    Logger.error("RabbitMQ connection lost: #{inspect(reason)}")
-    schedule_reconnect()
-    {:noreply, %{channel: nil, connection: nil}}
-  end
-
-  @impl true
-  def handle_cast({:publish, _routing_key, _payload}, %{channel: nil} = state) do
-    Logger.warning("RabbitMQ not connected, message dropped")
-    {:noreply, state}
-  end
-
-  @impl true
-  def handle_cast({:publish, routing_key, payload}, %{channel: channel} = state) do
-    Exchange.direct(channel, exchange_name(), durable: true)
-
-    AMQP.Basic.publish(channel, exchange_name(), routing_key, payload,
         content_type: "application/json",
         persistent: true
       )
-
-    {:noreply, state}
   end
 
-  defp connect do
+  defp channel do
-    config = Application.get_env(:workload_service, WorkloadService.MessageBus, [])
+    {:ok, conn} = AMQP.Connection.open(amqp_url())
-
+    {:ok, chan} = AMQP.Channel.open(conn)
-    with {:ok, connection} <- Connection.open(config),
+    chan
-         {:ok, channel} <- Channel.open(connection) do
-      Exchange.declare(channel, exchange_name(), :direct, durable: true)
-      Process.monitor(connection.pid)
-      {:ok, connection, channel}
-    end
   end
 
-  defp exchange_name do
+  defp amqp_url, do: Application.fetch_env!(:workload_service, :amqp_url)
-    Application.get_env(:workload_service, WorkloadService.MessageBus, [])
-    |> Keyword.get(:exchange, "workload_service.events")
-  end
-
-  defp schedule_reconnect do
-    Process.send_after(self(), :connect, @reconnect_interval)
-  end
 end

lib/workload_service/projectors/task_projector.ex

@@ -10,8 +10,6 @@ defmodule WorkloadService.Projectors.TaskProjector do
   import Ecto.Query
 
   project(%Events.TaskCreated{} = e, _meta, fn multi ->
-    application_id_str = to_string(e.application_id)
-
     Ecto.Multi.insert(multi, :task, %Task{
       id: to_string(e.id),
       org_id: e.id.org_id,

ops/chart/values.yaml

@@ -140,14 +140,14 @@ rawResources:
           configure: ".*"
           read: ".*"
 
-  exchange-task-completed:
+  exchange-quote-task-completed:
     enabled: true
     apiVersion: rabbitmq.com/v1beta1
     kind: Exchange
-    suffix: exchange-task-completed
+    suffix: exchange-quote-task-completed
     spec:
       spec:
-        name: workload_service.events.task_completed
+        name: workload_service.events.quote_task_completed
         type: topic
         durable: true
         vhost: "application"