diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index bd9e725..83bdf66 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -355,17 +355,17 @@ defmodule Aprsme.PacketConsumer do true -> packet_attrs[:sender] end - packet = %{ - sender: identifier, - latitude: packet_attrs[:lat], - longitude: packet_attrs[:lon], - received_at: packet_attrs[:received_at], - data_type: packet_attrs[:data_type], - altitude: packet_attrs[:altitude], - speed: packet_attrs[:speed], - course: packet_attrs[:course], - comment: packet_attrs[:comment] - } + # Build a rich packet payload that replaces the JSON produced by the old + # packets_notify_insert trigger. Includes latitude/longitude aliases so + # downstream consumers that expect either :lat/:lon or :latitude/:longitude work. + packet = + packet_attrs + |> Map.drop([:location, :inserted_at, :updated_at]) + |> Map.merge(%{ + sender: identifier, + latitude: packet_attrs[:lat], + 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) diff --git a/lib/aprsme/postgres_notifier.ex b/lib/aprsme/postgres_notifier.ex index 17219f2..8d1dd16 100644 --- a/lib/aprsme/postgres_notifier.ex +++ b/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 - them via Phoenix.PubSub for reactive, event-driven updates. + Listens to PostgreSQL NOTIFY events on the "aprs_events" channel (used by + 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, _ref2} = Postgrex.Notifications.listen(conn, @packet_channel) + {:ok, _ref} = Postgrex.Notifications.listen(conn, @event_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 diff --git a/priv/repo/migrations/20260417134617_drop_packets_notify_trigger.exs b/priv/repo/migrations/20260417134617_drop_packets_notify_trigger.exs new file mode 100644 index 0000000..0a2873e --- /dev/null +++ b/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