perf: replace per-row pg_notify trigger with elixir broadcast after batch insert

lib/aprsme/packet_consumer.ex

@@ -355,17 +355,17 @@ defmodule Aprsme.PacketConsumer do
         true -> packet_attrs[:sender]
       end
 
-    packet = %{
+    # Build a rich packet payload that replaces the JSON produced by the old
-      sender: identifier,
+    # packets_notify_insert trigger. Includes latitude/longitude aliases so
-      latitude: packet_attrs[:lat],
+    # downstream consumers that expect either :lat/:lon or :latitude/:longitude work.
-      longitude: packet_attrs[:lon],
+    packet =
-      received_at: packet_attrs[:received_at],
+      packet_attrs
-      data_type: packet_attrs[:data_type],
+      |> Map.drop([:location, :inserted_at, :updated_at])
-      altitude: packet_attrs[:altitude],
+      |> Map.merge(%{
-      speed: packet_attrs[:speed],
+        sender: identifier,
-      course: packet_attrs[:course],
+        latitude: packet_attrs[:lat],
-      comment: packet_attrs[:comment]
+        longitude: packet_attrs[:lon]
-    }
+      })
 
     if cluster_enabled do
       PacketDistributor.distribute_packet(packet)
@@ -373,6 +373,24 @@ defmodule Aprsme.PacketConsumer do
       Aprsme.StreamingPacketsPubSub.broadcast_packet(packet)
       Aprsme.SpatialPubSub.broadcast_packet(packet)
     end
+
+    broadcast_legacy_topics(packet, packet_attrs)
+  end
+
+  # Replaces the `aprs_packets` pg_notify path (removed as a DB trigger).
+  defp broadcast_legacy_topics(packet, packet_attrs) do
+    Phoenix.PubSub.broadcast(Aprsme.PubSub, "postgres:aprsme_packets", {:postgres_packet, packet})
+
+    callsign = packet_attrs[:sender] || packet_attrs[:base_callsign]
+
+    if is_binary(callsign) and callsign != "" do
+      normalized = callsign |> String.trim() |> String.upcase()
+      Phoenix.PubSub.broadcast(Aprsme.PubSub, "packets:#{normalized}", {:postgres_packet, packet})
+
+      if packet_attrs[:has_weather] do
+        Phoenix.PubSub.broadcast(Aprsme.PubSub, "weather:#{normalized}", {:weather_packet, packet})
+      end
+    end
   end
 
   defp test_env? do
@@ -418,7 +436,8 @@ defmodule Aprsme.PacketConsumer do
     |> sanitize_packet_strings()
     |> Map.put(:inserted_at, current_time)
     |> Map.put(:updated_at, current_time)
-    |> Map.delete(:id)
+    # Generate id in Elixir so post-insert broadcasts can reference it
+    |> Map.put(:id, Ecto.UUID.generate())
     |> Map.delete("id")
     # Remove embedded field for batch insert
     |> Map.delete(:data_extended)

lib/aprsme/postgres_notifier.ex

@@ -1,14 +1,17 @@
 defmodule Aprsme.PostgresNotifier do
   @moduledoc """
-  Listens to PostgreSQL NOTIFY events on the "aprs_events" and "aprs_packets" channels and broadcasts
+  Listens to PostgreSQL NOTIFY events on the "aprs_events" channel (used by
-  them via Phoenix.PubSub for reactive, event-driven updates.
+  bad_packets) and broadcasts them via Phoenix.PubSub.
+
+  Packet notifications previously came through here via the "aprs_packets"
+  channel + `packets_notify_insert` trigger. That path was removed to avoid
+  per-row trigger overhead and double-broadcasting — packets are now
+  broadcast directly from `Aprsme.PacketConsumer` to the same topics.
   """
   use GenServer
 
   @event_channel "aprs_events"
   @event_topic "postgres:aprsme_events"
-  @packet_channel "aprs_packets"
-  @packet_topic "postgres:aprsme_packets"
 
   def start_link(_opts) do
     GenServer.start_link(__MODULE__, nil, name: __MODULE__)
@@ -17,8 +20,7 @@ defmodule Aprsme.PostgresNotifier do
   @impl true
   def init(_) do
     {:ok, conn} = Postgrex.Notifications.start_link(Aprsme.Repo.config())
-    {:ok, _ref1} = Postgrex.Notifications.listen(conn, @event_channel)
+    {:ok, _ref} = Postgrex.Notifications.listen(conn, @event_channel)
-    {:ok, _ref2} = Postgrex.Notifications.listen(conn, @packet_channel)
     {:ok, %{conn: conn}}
   end
 
@@ -28,50 +30,5 @@ defmodule Aprsme.PostgresNotifier do
     {:noreply, state}
   end
 
-  def handle_info({:notification, _conn, _pid, @packet_channel, payload}, state) do
-    case Jason.decode(payload) do
-      {:ok, packet} ->
-        # Broadcast to the general packet topic
-        Phoenix.PubSub.broadcast(Aprsme.PubSub, @packet_topic, {:postgres_packet, packet})
-
-        # Broadcast to callsign-specific topic
-        callsign = Map.get(packet, "sender") || Map.get(packet, :sender)
-
-        if is_binary(callsign) and callsign != "" do
-          normalized_callsign = String.upcase(String.trim(callsign))
-          Phoenix.PubSub.broadcast(Aprsme.PubSub, "packets:#{normalized_callsign}", {:postgres_packet, packet})
-        end
-
-        # Also broadcast to callsign-specific weather topic if it's a weather packet
-        broadcast_weather_packet_if_relevant(packet)
-
-      _ ->
-        :noop
-    end
-
-    {:noreply, state}
-  end
-
   def handle_info(_msg, state), do: {:noreply, state}
-
-  defp broadcast_weather_packet_if_relevant(packet) do
-    # Check if this is a weather packet using centralized weather fields
-    has_weather_data =
-      Enum.any?(Aprsme.EncodingUtils.weather_fields(), fn field ->
-        value = Map.get(packet, to_string(field))
-        not is_nil(value)
-      end)
-
-    if has_weather_data do
-      # Extract callsign from packet
-      callsign = Map.get(packet, "sender") || Map.get(packet, :sender)
-
-      if is_binary(callsign) and callsign != "" do
-        # Normalize callsign and broadcast to weather-specific topic
-        normalized_callsign = String.upcase(String.trim(callsign))
-        weather_topic = "weather:#{normalized_callsign}"
-        Phoenix.PubSub.broadcast(Aprsme.PubSub, weather_topic, {:weather_packet, packet})
-      end
-    end
-  end
 end

priv/repo/migrations/20260417134617_drop_packets_notify_trigger.exs

@@ -0,0 +1,26 @@
+defmodule Aprsme.Repo.Migrations.DropPacketsNotifyTrigger do
+  use Ecto.Migration
+
+  @moduledoc """
+  Removes the per-row `packets_notify_insert` trigger. Packet broadcasts are
+  now produced directly from `Aprsme.PacketConsumer` after batch insert,
+  eliminating DB-side JSON serialization and LISTEN/NOTIFY overhead.
+
+  The `aprs_events` channel (for bad packets) is untouched.
+  """
+
+  def up do
+    execute("SET LOCAL statement_timeout = '0'")
+
+    execute("DROP TRIGGER IF EXISTS packets_notify_insert ON packets")
+    execute("DROP FUNCTION IF EXISTS notify_packets_insert() CASCADE")
+  end
+
+  def down do
+    # Recreating the function/trigger here would require duplicating the full
+    # JSON payload definition from prior migrations. Leaving as a no-op; if a
+    # rollback is truly needed, restore from the prior migrations that defined
+    # it (e.g. 20250728230000_add_missing_fields_to_packet_notify.exs).
+    :ok
+  end
+end