Mk/front/broadway integration

front/.gitignore

@@ -43,3 +43,8 @@ cover/
 lcov*
 
 docker-compose.override.yml
+
+# Nix
+.nix-mix
+.nix-hex
+.direnv

front/AGENTS.md

@@ -7,6 +7,27 @@
 - `test/` mirrors `lib/` with ExUnit suites, Wallaby browser specs in `test/browser`, and fixtures under `test/fixture`.
 - `priv/` serves runtime assets, and `workflow_templates/` supplies seeded YAML pipelines consumed by the UI.
 
+## Environment: Nix + direnv
+
+This repo uses **Nix flakes** with **direnv** for development environments. The `flake.nix` and `.envrc` configure Elixir 1.15, Erlang 26, and Node 22.
+
+### Running commands
+
+Shell tools like `mix`, `elixir`, `node`, etc. are only available inside the Nix dev shell. To run commands:
+
+```bash
+# Option 1: Use direnv (loads automatically when you cd into the directory)
+cd front && direnv allow && mix compile
+
+# Option 2: Use nix develop directly
+nix develop front/ -c mix compile
+
+# Option 3: Prefix with direnv exec
+direnv exec front mix compile
+```
+
+**Important:** Do not run `mix`, `elixir`, or other app-specific tools from the repo root — they won't be on PATH. Always `cd` into the app directory or use `direnv exec`.
+
 ## Build, Test, and Development Commands
 - First-time setup: `mix deps.get` and `npm install --prefix assets`.
 - `make dev.server` (Docker) launches Phoenix with Redis, RabbitMQ, and demo data preloaded.

front/Dockerfile

@@ -1,6 +1,6 @@
-ARG ELIXIR_VERSION=1.14.5
-ARG OTP_VERSION=25.3.2.21
-ARG ALPINE_VERSION=3.22.2
+ARG ELIXIR_VERSION=1.15.7
+ARG OTP_VERSION=26.2.5.17
+ARG ALPINE_VERSION=3.22.3
 ARG BUILDER_IMAGE="hexpm/elixir:${ELIXIR_VERSION}-erlang-${OTP_VERSION}-alpine-${ALPINE_VERSION}"
 ARG RUNNER_IMAGE="alpine:${ALPINE_VERSION}"
 

front/docker-compose.yml

@@ -1,4 +1,4 @@
-version: '3.6'
+version: "3.6"
 
 services:
   app:
@@ -69,5 +69,8 @@ services:
 
   rabbitmq:
     image: rabbitmq:3-management
+    ports:
+      - "5672:5672"
+      - "15672:15672"
     environment:
       RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: "-rabbit log_levels [{connection,error}]"

front/flake.nix

@@ -0,0 +1,44 @@
+{
+  description = "Billing Elixir development environment";
+
+  inputs = {
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+    flake-utils.url = "github:numtide/flake-utils";
+  };
+
+  outputs = { self, nixpkgs, flake-utils }:
+    flake-utils.lib.eachDefaultSystem (system:
+      let
+        pkgs = nixpkgs.legacyPackages.${system};
+
+        erlang = pkgs.beam.packages.erlang_26;
+        elixir = erlang.elixir_1_15;
+      in
+      {
+        devShells.default = pkgs.mkShell {
+          packages = [
+            elixir
+            pkgs.erlang_26
+            pkgs.nodejs_22
+            pkgs.gnumake
+            pkgs.chromedriver
+          ];
+
+          shellHook = ''
+            export MIX_HOME="$PWD/.nix-mix"
+            export HEX_HOME="$PWD/.nix-hex"
+            export PATH="$MIX_HOME/bin:$HEX_HOME/bin:$PATH"
+            export ERL_AFLAGS="-kernel shell_history enabled"
+
+            # Create dirs if they don't exist
+            mkdir -p "$MIX_HOME" "$HEX_HOME"
+
+            # Install hex and rebar if not present
+            if [ ! -f "$MIX_HOME/rebar3" ]; then
+              mix local.hex --force --if-missing
+              mix local.rebar --force --if-missing
+            fi
+          '';
+        };
+      });
+}

front/lib/front/feature_provider_invalidator_worker.ex

@@ -1,59 +1,97 @@
 defmodule Front.FeatureProviderInvalidatorWorker do
+  use Broadway
+
   require Logger
 
-  @doc """
-  This module consumes RabbitMQ feature and machine state change events
-  and invalidates features and machines caches.
-  """
-
-  use Tackle.Multiconsumer,
-    url: Application.get_env(:front, :amqp_url),
-    service: "front",
-    routes: [
-      {"feature_exchange", "machines_changed", :machines_changed},
-      {"feature_exchange", "organization_machines_changed", :organization_machines_changed},
-      {"feature_exchange", "features_changed", :features_changed},
-      {"feature_exchange", "organization_features_changed", :organization_features_changed}
-    ],
-    # This queue is used to consume events from the feature exchange.
-    # It is declared as non-durable, auto-delete and exclusive.
-    # This means that the queue will be deleted when the consumer disconnects.
-    # This is the desired behavior, because these events are used to invalidate pod-level caches.
-    queue: :dynamic,
-    queue_opts: [
-      durable: false,
-      auto_delete: true,
-      exclusive: true
-    ],
-    connection_id: Front.FeatureProviderInvalidatorWorker
-
-  def machines_changed(_message) do
+  @routing_keys ~w(
+    machines_changed
+    organization_machines_changed
+    features_changed
+    organization_features_changed
+  )
+
+  def start_link(_opts) do
+    Broadway.start_link(__MODULE__,
+      name: __MODULE__,
+      producer: [
+        module:
+          {BroadwayRabbitMQ.Producer,
+           queue: "",
+           connection: amqp_url(),
+           after_connect: fn channel ->
+             AMQP.Exchange.declare(channel, "feature_exchange", :direct, durable: true)
+           end,
+           declare: [
+             durable: false,
+             auto_delete: true,
+             exclusive: true
+           ],
+           bindings:
+             Enum.map(@routing_keys, fn rk ->
+               {"feature_exchange", routing_key: rk}
+             end),
+           on_failure: :reject,
+           metadata: [:routing_key]},
+        concurrency: 1
+      ],
+      processors: [
+        default: [
+          concurrency: 1,
+          max_demand: 1
+        ]
+      ]
+    )
+  end
+
+  @impl true
+  def handle_message(_processor, message, _context) do
+    case message.metadata.routing_key do
+      "machines_changed" ->
+        handle_machines_changed(message.data)
+
+      "organization_machines_changed" ->
+        handle_organization_machines_changed(message.data)
+
+      "features_changed" ->
+        handle_features_changed(message.data)
+
+      "organization_features_changed" ->
+        handle_organization_features_changed(message.data)
+
+      unknown ->
+        Logger.warning("[FEATURE PROVIDER INVALIDATOR WORKER] unknown routing key: #{unknown}")
+    end
+
+    message
+  end
+
+  defp handle_machines_changed(_payload) do
     log("invalidating machines")
     {:ok, _} = FeatureProvider.list_machines(reload: true)
-    :ok
   end
 
-  def organization_machines_changed(message) do
-    event = InternalApi.Feature.OrganizationMachinesChanged.decode(message)
+  defp handle_organization_machines_changed(payload) do
+    event = InternalApi.Feature.OrganizationMachinesChanged.decode(payload)
     log("invalidating machines for org #{event.org_id}")
     {:ok, _} = FeatureProvider.list_machines(reload: true, param: event.org_id)
-    :ok
   end
 
-  def features_changed(_message) do
+  defp handle_features_changed(_payload) do
     log("invalidating features")
     {:ok, _} = FeatureProvider.list_features(reload: true)
-    :ok
   end
 
-  def organization_features_changed(message) do
-    event = InternalApi.Feature.OrganizationFeaturesChanged.decode(message)
+  defp handle_organization_features_changed(payload) do
+    event = InternalApi.Feature.OrganizationFeaturesChanged.decode(payload)
     log("invalidating features for org #{event.org_id}")
     {:ok, _} = FeatureProvider.list_features(reload: true, param: event.org_id)
-    :ok
   end
 
   defp log(message) do
     Logger.info("[FEATURE PROVIDER INVALIDATOR WORKER] #{message}")
   end
+
+  defp amqp_url do
+    Application.get_env(:front, :amqp_url)
+  end
 end

front/lib/front/models/project_metrics.ex

@@ -175,7 +175,7 @@ defmodule Front.Models.ProjectMetrics do
         API.MetricAggregation.value(:RANGE)
 
       _ ->
-        Logger.warn("Unknown aggregate: '#{value}', defaulting to range")
+        Logger.warning("Unknown aggregate: '#{value}', defaulting to range")
         API.MetricAggregation.value(:RANGE)
     end
   end

front/lib/front/models/service_account.ex

@@ -146,7 +146,7 @@ defmodule Front.Models.ServiceAccount do
     end)
     |> Enum.map(fn
       {member, nil} ->
-        Logger.warn("Service account #{member.id} not found in service accounts list")
+        Logger.warning("Service account #{member.id} not found in service accounts list")
         nil
 
       {member, service_account} ->

front/lib/front/rbac/role_management.ex

@@ -208,7 +208,7 @@ defmodule Front.RBAC.RoleManagement do
             InternalApi.RBAC.SubjectType.value(:USER)
 
           _ ->
-            Logger.warn("Unrecognized subject type: #{subject_type}, defaulting to user")
+            Logger.warning("Unrecognized subject type: #{subject_type}, defaulting to user")
             InternalApi.RBAC.SubjectType.value(:USER)
         end
 

front/lib/front_web/controllers/deployments_controller.ex

@@ -49,7 +49,7 @@ defmodule FrontWeb.DeploymentsController do
         render_page(conn, "show.html", target_details, %{page_args: page_args})
       else
         {:exit, {%GRPC.RPCError{status: @grpc_not_found}, _stacktrace}} ->
-          Logger.warn("[DT] Target not found: target_id=#{target_id}")
+          Logger.warning("[DT] Target not found: target_id=#{target_id}")
           render_404(conn)
       end
     end)
@@ -75,7 +75,7 @@ defmodule FrontWeb.DeploymentsController do
         render_page(conn, "edit.html", changeset, resources)
       else
         {:exit, {%GRPC.RPCError{status: @grpc_not_found}, _stacktrace}} ->
-          Logger.warn("[DT] Target not found: target_id=#{target_id}")
+          Logger.warning("[DT] Target not found: target_id=#{target_id}")
           render_404(conn)
       end
     end)
@@ -142,7 +142,7 @@ defmodule FrontWeb.DeploymentsController do
             |> render_page("edit.html", changeset, resources)
 
           {:error, %GRPC.RPCError{status: @grpc_not_found}} ->
-            Logger.warn("[DT] Target not found: target_id=#{target_id}")
+            Logger.warning("[DT] Target not found: target_id=#{target_id}")
 
             conn
             |> put_flash(:alert, "Failure: deployment target was not found")
@@ -186,11 +186,11 @@ defmodule FrontWeb.DeploymentsController do
         |> redirect(to: deployments_path(conn, :index, project_name))
       else
         {:exit, {%GRPC.RPCError{status: @grpc_not_found}, _stacktrace}} ->
-          Logger.warn("[DT] Target not found: target_id=#{target_id}")
+          Logger.warning("[DT] Target not found: target_id=#{target_id}")
           render_404(conn)
 
         {:error, %GRPC.RPCError{status: @grpc_not_found}} ->
-          Logger.warn("[DT] Target not found: target_id=#{target_id}")
+          Logger.warning("[DT] Target not found: target_id=#{target_id}")
 
           conn
           |> put_flash(:alert, "Failure: deployment target was not found")
@@ -220,7 +220,7 @@ defmodule FrontWeb.DeploymentsController do
         |> redirect(to: deployments_path(conn, :index, project_name))
       else
         {:exit, {%GRPC.RPCError{status: @grpc_not_found}, _stacktrace}} ->
-          Logger.warn("[DT] Target not found: target_id=#{target_id}")
+          Logger.warning("[DT] Target not found: target_id=#{target_id}")
           render_404(conn)
 
         {:error, reason} ->

front/lib/front_web/controllers/secrets_controller.ex

@@ -177,7 +177,7 @@ defmodule FrontWeb.SecretsController do
           }
 
           require Logger
-          Logger.warn("validation errors: #{inspect(validation_errors)}")
+          Logger.warning("validation errors: #{inspect(validation_errors)}")
 
           conn
           |> put_flash(:alert, compose_alert_message(validation_errors))