event pubsub

lib/towerops/application.ex

@@ -20,6 +20,8 @@ defmodule Towerops.Application do
       Towerops.Repo,
       {DNSCluster, query: Application.get_env(:towerops, :dns_cluster_query) || :ignore},
       {Phoenix.PubSub, name: Towerops.PubSub},
+      # Start event logger (subscribes to PubSub)
+      Towerops.Equipment.EventLogger,
       # Start monitoring supervisor
       Towerops.Monitoring.Supervisor,
       # Start a worker by calling: Towerops.Worker.start_link(arg)

lib/towerops/equipment/event_logger.ex

@@ -0,0 +1,51 @@
+defmodule Towerops.Equipment.EventLogger do
+  @moduledoc """
+  GenServer that subscribes to equipment events via PubSub and logs them to the database.
+
+  This decouples event detection from event storage, allowing:
+  - Faster polling without database write blocking
+  - Multiple event consumers (logging, alerts, webhooks)
+  - Better failure isolation
+  """
+  use GenServer
+
+  alias Towerops.Equipment
+
+  require Logger
+
+  # Client API
+
+  def start_link(_opts) do
+    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  end
+
+  # Server Callbacks
+
+  @impl true
+  def init(state) do
+    # Subscribe to equipment events topic
+    Phoenix.PubSub.subscribe(Towerops.PubSub, "equipment:events")
+    Logger.info("EventLogger started and subscribed to equipment events")
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_info({:equipment_event, event_attrs}, state) do
+    case Equipment.create_event(event_attrs) do
+      {:ok, event} ->
+        Logger.debug("Logged event: #{event.message}")
+
+      {:error, changeset} ->
+        Logger.error("Failed to log event: #{inspect(changeset.errors)}")
+        Logger.error("Event data: #{inspect(event_attrs)}")
+    end
+
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info(_msg, state) do
+    # Ignore other messages
+    {:noreply, state}
+  end
+end

lib/towerops/snmp/poller_worker.ex

@@ -389,17 +389,17 @@ defmodule Towerops.Snmp.PollerWorker do
         changes
       end
 
-    # If there are changes, create events and update interface
+    # If there are changes, broadcast events and update interface
     if length(changes) > 0 do
-      # Create all events
+      # Broadcast all events via PubSub
       Enum.each(changes, fn {:event, event_attrs} ->
-        case Equipment.create_event(event_attrs) do
-          {:ok, _event} ->
-            Logger.info("Created event: #{event_attrs.message}")
+        Phoenix.PubSub.broadcast(
+          Towerops.PubSub,
+          "equipment:events",
+          {:equipment_event, event_attrs}
+        )
 
-          {:error, changeset} ->
-            Logger.error("Failed to create event: #{inspect(changeset.errors)}")
-        end
+        Logger.debug("Broadcast event: #{event_attrs.message}")
       end)
 
       # Update interface with new attributes