From 33171ab32e1f7dbeb521eea298a1c83ec5fc964b Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 2 Apr 2026 16:26:22 -0500 Subject: [PATCH] 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 | 1 + lib/microwaveprop/repo_listener.ex | 73 +++++++++++++++++++ lib/microwaveprop_web/live/backfill_live.ex | 21 +++--- ...2212323_add_enrichment_notify_triggers.exs | 62 ++++++++++++++++ 4 files changed, 145 insertions(+), 12 deletions(-) create mode 100644 lib/microwaveprop/repo_listener.ex create mode 100644 priv/repo/migrations/20260402212323_add_enrichment_notify_triggers.exs diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 4ce6a22e..71708f97 100644 --- a/lib/microwaveprop/application.ex +++ b/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 diff --git a/lib/microwaveprop/repo_listener.ex b/lib/microwaveprop/repo_listener.ex new file mode 100644 index 00000000..807481b5 --- /dev/null +++ b/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 diff --git a/lib/microwaveprop_web/live/backfill_live.ex b/lib/microwaveprop_web/live/backfill_live.ex index 7b43434c..022123d9 100644 --- a/lib/microwaveprop_web/live/backfill_live.ex +++ b/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 - Process.send_after(self(), :refresh_stats, @refresh_interval) - + def handle_info({:oban_job_changed, _data}, socket) do {:noreply, assign(socket, stats: fetch_stats(), - unprocessed: count_unprocessed(), - db_stats: fetch_db_stats() + unprocessed: count_unprocessed() )} end @@ -446,7 +443,7 @@ defmodule MicrowavepropWeb.BackfillLive do -

Auto-refreshes every 2 seconds.

+

Live updates via database notifications.

""" end diff --git a/priv/repo/migrations/20260402212323_add_enrichment_notify_triggers.exs b/priv/repo/migrations/20260402212323_add_enrichment_notify_triggers.exs new file mode 100644 index 00000000..b2ece920 --- /dev/null +++ b/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