Replace polling with Postgres LISTEN/NOTIFY for live updates

- Database triggers on contacts and oban_jobs fire pg_notify on
  status/state changes
- RepoListener GenServer subscribes via Postgrex.Notifications and
  broadcasts to PubSub
- Backfill dashboard subscribes to PubSub instead of 2-second polling
- Eliminates periodic SELECT queries, updates arrive instantly

lib/microwaveprop/application.ex

@@ -12,6 +12,7 @@ defmodule Microwaveprop.Application do
       Microwaveprop.Repo,
       {Phoenix.PubSub, name: Microwaveprop.PubSub},
       {Oban, Application.fetch_env!(:microwaveprop, Oban)},
+      Microwaveprop.RepoListener,
       # Start to serve requests, typically the last entry
       MicrowavepropWeb.Endpoint,
       Microwaveprop.Propagation.FreshnessMonitor

lib/microwaveprop/repo_listener.ex

@@ -0,0 +1,73 @@
+defmodule Microwaveprop.RepoListener do
+  @moduledoc """
+  Listens for PostgreSQL NOTIFY events on enrichment status changes
+  and Oban job state changes, broadcasting via PubSub.
+  Replaces polling with event-driven updates.
+  """
+
+  use GenServer
+
+  require Logger
+
+  @channels ["contact_status_changed", "oban_job_changed"]
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_opts) do
+    config = Microwaveprop.Repo.config()
+
+    {:ok, pid} =
+      Postgrex.Notifications.start_link(
+        hostname: config[:hostname] || "localhost",
+        port: config[:port] || 5432,
+        username: config[:username],
+        password: config[:password],
+        database: config[:database]
+      )
+
+    for channel <- @channels do
+      Postgrex.Notifications.listen!(pid, channel)
+    end
+
+    Logger.info("RepoListener: listening on #{Enum.join(@channels, ", ")}")
+    {:ok, %{pid: pid}}
+  end
+
+  @impl true
+  def handle_info({:notification, _pid, _ref, "contact_status_changed", payload}, state) do
+    case Jason.decode(payload) do
+      {:ok, data} ->
+        Phoenix.PubSub.broadcast(
+          Microwaveprop.PubSub,
+          "db:contact_status",
+          {:contact_status_changed, data}
+        )
+
+      _ ->
+        :ok
+    end
+
+    {:noreply, state}
+  end
+
+  def handle_info({:notification, _pid, _ref, "oban_job_changed", payload}, state) do
+    case Jason.decode(payload) do
+      {:ok, data} ->
+        Phoenix.PubSub.broadcast(
+          Microwaveprop.PubSub,
+          "db:oban_jobs",
+          {:oban_job_changed, data}
+        )
+
+      _ ->
+        :ok
+    end
+
+    {:noreply, state}
+  end
+
+  def handle_info(_msg, state), do: {:noreply, state}
+end

lib/microwaveprop_web/live/backfill_live.ex

@@ -9,12 +9,11 @@ defmodule MicrowavepropWeb.BackfillLive do
   alias Microwaveprop.Repo
   alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
 
-  @refresh_interval 2_000
-
   @impl true
   def mount(_params, _session, socket) do
     if connected?(socket) do
-      Process.send_after(self(), :refresh_stats, @refresh_interval)
+      Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:contact_status")
+      Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:oban_jobs")
     end
 
     stats = fetch_stats()
@@ -56,24 +55,22 @@ defmodule MicrowavepropWeb.BackfillLive do
 
   @impl true
   def handle_info({:enqueued, count}, socket) do
+    {:noreply, assign(socket, last_enqueued: count, enqueuing: false)}
+  end
+
+  def handle_info({:contact_status_changed, _data}, socket) do
     {:noreply,
      assign(socket,
-       last_enqueued: count,
-       enqueuing: false,
-       stats: fetch_stats(),
        unprocessed: count_unprocessed(),
        db_stats: fetch_db_stats()
      )}
   end
 
-  def handle_info(:refresh_stats, socket) do
+  def handle_info({:oban_job_changed, _data}, socket) do
-    Process.send_after(self(), :refresh_stats, @refresh_interval)
-
     {:noreply,
      assign(socket,
        stats: fetch_stats(),
-       unprocessed: count_unprocessed(),
+       unprocessed: count_unprocessed()
-       db_stats: fetch_db_stats()
      )}
   end
 
@@ -446,7 +443,7 @@ defmodule MicrowavepropWeb.BackfillLive do
         </table>
       </div>
 
-      <p class="text-xs text-base-content/50 mt-4">Auto-refreshes every 2 seconds.</p>
+      <p class="text-xs text-base-content/50 mt-4">Live updates via database notifications.</p>
     </Layouts.app>
     """
   end

priv/repo/migrations/20260402212323_add_enrichment_notify_triggers.exs

@@ -0,0 +1,62 @@
+defmodule Microwaveprop.Repo.Migrations.AddEnrichmentNotifyTriggers do
+  use Ecto.Migration
+
+  def up do
+    # Notify on contact enrichment status changes
+    execute """
+    CREATE OR REPLACE FUNCTION notify_contact_status_change() RETURNS trigger AS $$
+    BEGIN
+      IF OLD.hrrr_status IS DISTINCT FROM NEW.hrrr_status
+        OR OLD.weather_status IS DISTINCT FROM NEW.weather_status
+        OR OLD.terrain_status IS DISTINCT FROM NEW.terrain_status
+        OR OLD.iemre_status IS DISTINCT FROM NEW.iemre_status THEN
+        PERFORM pg_notify('contact_status_changed', json_build_object(
+          'id', NEW.id,
+          'hrrr_status', NEW.hrrr_status,
+          'weather_status', NEW.weather_status,
+          'terrain_status', NEW.terrain_status,
+          'iemre_status', NEW.iemre_status
+        )::text);
+      END IF;
+      RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+    """
+
+    execute """
+    CREATE TRIGGER contact_status_notify
+    AFTER UPDATE ON contacts
+    FOR EACH ROW EXECUTE FUNCTION notify_contact_status_change();
+    """
+
+    # Notify on Oban job state changes
+    execute """
+    CREATE OR REPLACE FUNCTION notify_oban_job_change() RETURNS trigger AS $$
+    BEGIN
+      IF OLD.state IS DISTINCT FROM NEW.state THEN
+        PERFORM pg_notify('oban_job_changed', json_build_object(
+          'id', NEW.id,
+          'worker', NEW.worker,
+          'queue', NEW.queue,
+          'state', NEW.state
+        )::text);
+      END IF;
+      RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+    """
+
+    execute """
+    CREATE TRIGGER oban_job_notify
+    AFTER UPDATE ON oban_jobs
+    FOR EACH ROW EXECUTE FUNCTION notify_oban_job_change();
+    """
+  end
+
+  def down do
+    execute "DROP TRIGGER IF EXISTS contact_status_notify ON contacts"
+    execute "DROP FUNCTION IF EXISTS notify_contact_status_change()"
+    execute "DROP TRIGGER IF EXISTS oban_job_notify ON oban_jobs"
+    execute "DROP FUNCTION IF EXISTS notify_oban_job_change()"
+  end
+end