pubsub

lib/aprs/application.ex

@@ -31,7 +31,9 @@ defmodule Aprs.Application do
       {Cluster.Supervisor, [topologies, [name: Aprs.ClusterSupervisor]]},
       # Start Oban for background jobs
       {Oban, :aprs |> Application.get_env(Oban, []) |> Keyword.put(:queues, default: 10, maintenance: 2)},
-      Aprs.Presence
+      Aprs.Presence,
+      Aprs.AprsIsConnection,
+      Aprs.PostgresNotifier
     ]
 
     children =

lib/aprs/aprs_is_connection.ex

@@ -0,0 +1,123 @@
+defmodule Aprs.AprsIsConnection do
+  @moduledoc """
+  Maintains a supervised TCP connection to APRS-IS, with reconnection logic and
+  exponential backoff. Broadcasts each received line via Phoenix.PubSub and emits
+  telemetry events for connection, disconnection, errors, and packet receipt.
+  """
+  use GenServer
+
+  require Logger
+
+  @type state :: %{
+          socket: port() | nil,
+          backoff: non_neg_integer()
+        }
+
+  @reconnect_initial 2_000
+  @reconnect_max 60_000
+  @pubsub_topic "aprs_is:raw"
+
+  # Public API
+  @spec start_link(Keyword.t()) :: GenServer.on_start()
+  def start_link(_opts) do
+    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  end
+
+  @doc """
+  Send a raw string to APRS-IS.
+  """
+  @spec send_packet(String.t()) :: :ok | {:error, term()}
+  def send_packet(packet) do
+    GenServer.call(__MODULE__, {:send, packet})
+  end
+
+  # GenServer callbacks
+  @impl true
+  def init(state) do
+    schedule_connect(0)
+    {:ok, Map.merge(%{socket: nil, backoff: @reconnect_initial}, state)}
+  end
+
+  @impl true
+  def handle_info(:connect, state) do
+    case connect_aprs_is() do
+      {:ok, socket} ->
+        Logger.info("Connected to APRS-IS")
+        :telemetry.execute([:aprs, :is, :connected], %{}, %{})
+        {:noreply, %{state | socket: socket, backoff: @reconnect_initial}}
+
+      {:error, reason} ->
+        Logger.error("APRS-IS connection failed: #{inspect(reason)}. Retrying in #{state.backoff} ms.")
+
+        :telemetry.execute([:aprs, :is, :connect_error], %{}, %{reason: reason})
+        schedule_connect(state.backoff)
+        {:noreply, %{state | socket: nil, backoff: min(state.backoff * 2, @reconnect_max)}}
+    end
+  end
+
+  @impl true
+  def handle_info({:tcp, _socket, data}, state) do
+    # Each line received from APRS-IS
+    Phoenix.PubSub.broadcast(Aprs.PubSub, @pubsub_topic, {:aprs_is_line, data})
+    :telemetry.execute([:aprs, :is, :packet], %{size: byte_size(data)}, %{data: data})
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info({:tcp_closed, _socket}, state) do
+    Logger.warning("APRS-IS connection closed. Reconnecting...")
+    :telemetry.execute([:aprs, :is, :disconnected], %{}, %{})
+    schedule_connect(@reconnect_initial)
+    {:noreply, %{state | socket: nil, backoff: @reconnect_initial}}
+  end
+
+  @impl true
+  def handle_info({:tcp_error, _socket, reason}, state) do
+    Logger.error("APRS-IS TCP error: #{inspect(reason)}. Reconnecting...")
+    :telemetry.execute([:aprs, :is, :tcp_error], %{}, %{reason: reason})
+    schedule_connect(@reconnect_initial)
+    {:noreply, %{state | socket: nil, backoff: @reconnect_initial}}
+  end
+
+  @impl true
+  def handle_call({:send, packet}, _from, %{socket: socket} = state) when is_port(socket) do
+    :ok = :gen_tcp.send(socket, packet <> "\r\n")
+    {:reply, :ok, state}
+  end
+
+  def handle_call({:send, _packet}, _from, state) do
+    {:reply, {:error, :not_connected}, state}
+  end
+
+  @impl true
+  def terminate(_reason, %{socket: socket}) when is_port(socket) do
+    :gen_tcp.close(socket)
+    :ok
+  end
+
+  def terminate(_reason, _state), do: :ok
+
+  defp schedule_connect(delay) do
+    Process.send_after(self(), :connect, delay)
+  end
+
+  defp connect_aprs_is do
+    host = Application.get_env(:aprs, :aprs_is_host, ~c"rotate.aprs2.net")
+    port = Application.get_env(:aprs, :aprs_is_port, 14_580)
+    callsign = Application.get_env(:aprs, :aprs_is_callsign, "N0CALL")
+    passcode = Application.get_env(:aprs, :aprs_is_passcode, "00000")
+    filter = Application.get_env(:aprs, :aprs_is_filter, "")
+
+    opts = [:binary, active: true, packet: :line, keepalive: true]
+
+    case :gen_tcp.connect(host, port, opts) do
+      {:ok, socket} ->
+        login = "user #{callsign} pass #{passcode} vers aprs.me 0.1 #{filter}\r\n"
+        :ok = :gen_tcp.send(socket, login)
+        {:ok, socket}
+
+      error ->
+        error
+    end
+  end
+end

lib/aprs/postgres_notifier.ex

@@ -0,0 +1,44 @@
+defmodule Aprs.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.
+  """
+  use GenServer
+
+  @event_channel "aprs_events"
+  @event_topic "postgres:aprs_events"
+  @packet_channel "aprs_packets"
+  @packet_topic "postgres:aprs_packets"
+
+  def start_link(_opts) do
+    GenServer.start_link(__MODULE__, nil, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_) do
+    {:ok, conn} = Postgrex.Notifications.start_link(Aprs.Repo.config())
+    {:ok, _ref1} = Postgrex.Notifications.listen(conn, @event_channel)
+    {:ok, _ref2} = Postgrex.Notifications.listen(conn, @packet_channel)
+    {:ok, %{conn: conn}}
+  end
+
+  @impl true
+  def handle_info({:notification, _conn, _pid, @event_channel, payload}, state) do
+    Phoenix.PubSub.broadcast(Aprs.PubSub, @event_topic, {:postgres_notify, payload})
+    {:noreply, state}
+  end
+
+  def handle_info({:notification, _conn, _pid, @packet_channel, payload}, state) do
+    case Jason.decode(payload) do
+      {:ok, packet} ->
+        Phoenix.PubSub.broadcast(Aprs.PubSub, @packet_topic, {:postgres_packet, packet})
+
+      _ ->
+        :noop
+    end
+
+    {:noreply, state}
+  end
+
+  def handle_info(_msg, state), do: {:noreply, state}
+end

lib/aprs_web/live/bad_packets_live/index.ex

@@ -10,13 +10,19 @@ defmodule AprsWeb.BadPacketsLive.Index do
   @impl true
   def mount(_params, _session, socket) do
     if connected?(socket) do
+      # Subscribe to Postgres notifications for bad packets
+      Phoenix.PubSub.subscribe(Aprs.PubSub, "postgres:aprs_events")
       # Load initial bad packets
       bad_packets = fetch_bad_packets()
       # Extra safeguard to ensure we never show more than 100
       limited_packets = Enum.take(bad_packets, 100)
-      # Schedule automatic refresh every 5 seconds
-      :timer.send_interval(5000, self(), :refresh_bad_packets)
-      {:ok, assign(socket, bad_packets: limited_packets, loading: false, last_updated: DateTime.utc_now())}
+
+      {:ok,
+       assign(socket,
+         bad_packets: limited_packets,
+         loading: false,
+         last_updated: DateTime.utc_now()
+       )}
     else
       {:ok, assign(socket, bad_packets: [], loading: false, last_updated: nil)}
     end
@@ -38,7 +44,8 @@ defmodule AprsWeb.BadPacketsLive.Index do
   end
 
   @impl true
-  def handle_info(:refresh_bad_packets, socket) do
+  def handle_info({:postgres_notify, _payload}, socket) do
+    # Optionally filter payload for bad packet events
     send(self(), :do_refresh)
     {:noreply, socket}
   end
@@ -48,7 +55,13 @@ defmodule AprsWeb.BadPacketsLive.Index do
     bad_packets = fetch_bad_packets()
     # Extra safeguard to ensure we never show more than 100
     limited_packets = Enum.take(bad_packets, 100)
-    {:noreply, assign(socket, bad_packets: limited_packets, loading: false, last_updated: DateTime.utc_now())}
+
+    {:noreply,
+     assign(socket,
+       bad_packets: limited_packets,
+       loading: false,
+       last_updated: DateTime.utc_now()
+     )}
   end
 
   defp fetch_bad_packets(limit \\ 100) do

lib/aprs_web/live/map_live/index.ex

@@ -4,7 +4,6 @@ defmodule AprsWeb.MapLive.Index do
   """
   use AprsWeb, :live_view
 
-  alias Aprs.EncodingUtils
   alias AprsWeb.Endpoint
   alias AprsWeb.Helpers.AprsSymbols
   alias Parser.Types.MicE
@@ -15,15 +14,18 @@ defmodule AprsWeb.MapLive.Index do
   @ip_api_url "https://ip-api.com/json/"
   @finch_name Aprs.Finch
   @default_replay_speed 1000
+  @debounce_interval 200
 
   @impl true
   def mount(_params, _session, socket) do
     one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second)
 
     socket = assign_defaults(socket, one_hour_ago)
+    socket = assign(socket, packet_buffer: [], buffer_timer: nil)
 
     if connected?(socket) do
       Endpoint.subscribe("aprs_messages")
+      Phoenix.PubSub.subscribe(Aprs.PubSub, "postgres:aprs_packets")
       maybe_start_geolocation(socket)
       schedule_timers()
     end
@@ -278,61 +280,23 @@ defmodule AprsWeb.MapLive.Index do
 
   @spec process_bounds_update(map(), Socket.t()) :: Socket.t()
   defp process_bounds_update(map_bounds, socket) do
-    # Filter visible packets to only include those within the new bounds and time threshold
+    # Remove out-of-bounds packets and markers immediately
     new_visible_packets =
       socket.assigns.visible_packets
-      |> Enum.filter(fn {_callsign, packet} ->
-        within_bounds?(packet, map_bounds) &&
-          packet_within_time_threshold?(packet, socket.assigns.packet_age_threshold)
-      end)
+      |> Enum.filter(fn {_k, packet} -> within_bounds?(packet, map_bounds) end)
       |> Map.new()
 
-    # Get packets that are no longer visible (to remove from map)
     packets_to_remove =
       socket.assigns.visible_packets
-      |> Enum.reject(fn {_callsign, packet} ->
-        within_bounds?(packet, map_bounds)
+      |> Enum.reject(fn {_k, packet} -> within_bounds?(packet, map_bounds) end)
+      |> Enum.map(fn {k, _} -> k end)
+
+    socket =
+      Enum.reduce(packets_to_remove, socket, fn k, acc ->
+        push_event(acc, "remove_marker", %{id: k})
       end)
-      |> Enum.map(fn {callsign, _packet} -> callsign end)
 
-    # Clear markers that are outside the new bounds
-    socket =
-      if Enum.any?(packets_to_remove) do
-        # Remove markers that are outside bounds
-        socket =
-          Enum.reduce(packets_to_remove, socket, fn callsign, acc_socket ->
-            push_event(acc_socket, "remove_marker", %{id: callsign})
-          end)
-
-        # Also send a general filter event to clean up any remaining out-of-bounds markers
-        push_event(socket, "filter_markers_by_bounds", %{bounds: map_bounds})
-      else
-        socket
-      end
-
-    # If replay is not active, update the replay packets based on the new bounds
-    socket =
-      if socket.assigns.replay_active do
-        socket
-      else
-        # Clear any existing replay data when bounds change significantly
-        socket =
-          assign(socket,
-            replay_packets: [],
-            replay_index: 0,
-            historical_packets: %{},
-            map_ready: true
-          )
-
-        socket
-      end
-
-    assign(socket,
-      map_bounds: map_bounds,
-      visible_packets: new_visible_packets,
-      bounds_update_timer: nil,
-      pending_bounds: nil
-    )
+    assign(socket, map_bounds: map_bounds, visible_packets: new_visible_packets)
   end
 
   @impl true
@@ -392,46 +356,60 @@ defmodule AprsWeb.MapLive.Index do
         # Clean up packets older than 1 hour from the map display
         handle_cleanup_old_packets(socket)
 
-      %{event: "packet", payload: payload} ->
-        # Sanitize the packet to prevent encoding errors
-        sanitized_packet = EncodingUtils.sanitize_packet(payload)
-
-        # Add received timestamp if not present
-        sanitized_packet = Map.put_new(sanitized_packet, :received_at, DateTime.utc_now())
-
+      {:postgres_packet, packet} ->
         # Only process packets with position data that are within current map bounds
-        # AND are not older than 1 hour
-        if has_position_data?(sanitized_packet) &&
-             within_bounds?(sanitized_packet, socket.assigns.map_bounds) &&
-             packet_within_time_threshold?(sanitized_packet, socket.assigns.packet_age_threshold) do
-          # Convert to a simple map structure for JSON encoding
-          packet_data = build_packet_data(sanitized_packet)
-
-          # Only push if we have valid packet data
-          if packet_data do
-            # Generate a unique key for this packet
-            callsign_key =
-              "#{sanitized_packet.base_callsign}#{if sanitized_packet.ssid, do: "-#{sanitized_packet.ssid}", else: ""}"
-
-            # Update visible packets tracking
-            visible_packets =
-              Map.put(socket.assigns.visible_packets, callsign_key, sanitized_packet)
-
-            # Push the packet to the client-side JavaScript
-            socket =
+        if within_bounds?(packet, socket.assigns.map_bounds) do
+          # Add to buffer for debounced batch update
+          buffer = [packet | socket.assigns.packet_buffer]
+          socket = assign(socket, packet_buffer: buffer)
+          # If no timer, start one
+          socket =
+            if socket.assigns.buffer_timer == nil do
+              timer = Process.send_after(self(), :flush_packet_buffer, @debounce_interval)
+              assign(socket, buffer_timer: timer)
+            else
               socket
-              |> push_event("new_packet", packet_data)
-              |> assign(visible_packets: visible_packets)
+            end
 
-            {:noreply, socket}
-          else
-            # Invalid packet data, skip it
-            {:noreply, socket}
-          end
+          {:noreply, socket}
         else
-          # Ignore packets without position data, outside bounds, or too old
           {:noreply, socket}
         end
+
+      :flush_packet_buffer ->
+        packets = Enum.reverse(socket.assigns.packet_buffer)
+        visible_packets = socket.assigns.visible_packets
+
+        {new_visible_packets, events} =
+          Enum.reduce(packets, {visible_packets, []}, fn packet, {vis, evs} ->
+            packet_data = build_packet_data(packet)
+
+            if packet_data do
+              callsign_key =
+                if Map.has_key?(packet, "id"),
+                  do: to_string(packet["id"]),
+                  else: System.unique_integer([:positive])
+
+              {Map.put(vis, callsign_key, packet), [{:new_packet, packet_data} | evs]}
+            else
+              {vis, evs}
+            end
+          end)
+
+        # Push all new packets in one event (or as a batch)
+        socket =
+          Enum.reduce(events, socket, fn {:new_packet, data}, acc ->
+            push_event(acc, "new_packet", data)
+          end)
+
+        socket =
+          assign(socket,
+            visible_packets: new_visible_packets,
+            packet_buffer: [],
+            buffer_timer: nil
+          )
+
+        {:noreply, socket}
     end
   end
 
@@ -800,23 +778,6 @@ defmodule AprsWeb.MapLive.Index do
     Enum.sort_by(packets, fn packet -> packet.received_at end)
   end
 
-  @spec has_position_data?(map() | struct()) :: boolean()
-  defp has_position_data?(packet) do
-    case packet.data_extended do
-      %MicE{} = mic_e ->
-        # MicE packets have lat/lon in separate components
-        is_number(mic_e.lat_degrees) && is_number(mic_e.lat_minutes) &&
-          is_number(mic_e.lon_degrees) && is_number(mic_e.lon_minutes)
-
-      %{latitude: lat, longitude: lon} ->
-        # Regular position packets have decimal lat/lon
-        is_number(lat) && is_number(lon)
-
-      _ ->
-        false
-    end
-  end
-
   @spec within_bounds?(map() | struct(), map()) :: boolean()
   defp within_bounds?(packet, bounds) do
     {lat, lng} = get_coordinates(packet)
@@ -1040,6 +1001,7 @@ defmodule AprsWeb.MapLive.Index do
 
   @impl true
   def terminate(_reason, socket) do
+    if socket.assigns.buffer_timer, do: Process.cancel_timer(socket.assigns.buffer_timer)
     # Clean up any pending bounds update timer
     if socket.assigns[:bounds_update_timer] do
       Process.cancel_timer(socket.assigns.bounds_update_timer)

priv/repo/migrations/20250617190000_notify_on_badpackets_insert.exs

@@ -0,0 +1,27 @@
+defmodule Aprs.Repo.Migrations.NotifyOnBadpacketsInsert do
+  use Ecto.Migration
+
+  def up do
+    execute """
+    CREATE OR REPLACE FUNCTION notify_badpackets_insert() RETURNS trigger AS $$
+    BEGIN
+      PERFORM pg_notify('aprs_events', 'bad_packet');
+      RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+    """
+
+    execute "DROP TRIGGER IF EXISTS badpackets_notify_insert ON badpackets;"
+
+    execute """
+    CREATE TRIGGER badpackets_notify_insert
+    AFTER INSERT ON badpackets
+    FOR EACH ROW EXECUTE FUNCTION notify_badpackets_insert();
+    """
+  end
+
+  def down do
+    execute "DROP TRIGGER IF EXISTS badpackets_notify_insert ON badpackets;"
+    execute "DROP FUNCTION IF EXISTS notify_badpackets_insert();"
+  end
+end

priv/repo/migrations/20250617191000_notify_on_packets_insert.exs

@@ -0,0 +1,35 @@
+defmodule Aprs.Repo.Migrations.NotifyOnPacketsInsert do
+  use Ecto.Migration
+
+  def up do
+    execute """
+    CREATE OR REPLACE FUNCTION notify_packets_insert() RETURNS trigger AS $$
+    DECLARE
+      payload TEXT;
+    BEGIN
+      payload := json_build_object(
+        'id', NEW.id,
+        'lat', NEW.lat,
+        'lon', NEW.lon,
+        'inserted_at', NEW.inserted_at
+      )::text;
+      PERFORM pg_notify('aprs_packets', payload);
+      RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+    """
+
+    execute "DROP TRIGGER IF EXISTS packets_notify_insert ON packets;"
+
+    execute """
+    CREATE TRIGGER packets_notify_insert
+    AFTER INSERT ON packets
+    FOR EACH ROW EXECUTE FUNCTION notify_packets_insert();
+    """
+  end
+
+  def down do
+    execute "DROP TRIGGER IF EXISTS packets_notify_insert ON packets;"
+    execute "DROP FUNCTION IF EXISTS notify_packets_insert();"
+  end
+end

priv/repo/migrations/20250617200000_recreate_notify_packets_insert_function.exs

@@ -0,0 +1,36 @@
+defmodule Aprs.Repo.Migrations.RecreateNotifyPacketsInsertFunction do
+  use Ecto.Migration
+
+  def up do
+    execute "DROP TRIGGER IF EXISTS packets_notify_insert ON packets;"
+    execute "DROP FUNCTION IF EXISTS notify_packets_insert();"
+
+    execute """
+    CREATE OR REPLACE FUNCTION notify_packets_insert() RETURNS trigger AS $$
+    DECLARE
+      payload TEXT;
+    BEGIN
+      payload := json_build_object(
+        'id', NEW.id,
+        'lat', NEW.lat,
+        'lon', NEW.lon,
+        'inserted_at', NEW.inserted_at
+      )::text;
+      PERFORM pg_notify('aprs_packets', payload);
+      RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+    """
+
+    execute """
+    CREATE TRIGGER packets_notify_insert
+    AFTER INSERT ON packets
+    FOR EACH ROW EXECUTE FUNCTION notify_packets_insert();
+    """
+  end
+
+  def down do
+    execute "DROP TRIGGER IF EXISTS packets_notify_insert ON packets;"
+    execute "DROP FUNCTION IF EXISTS notify_packets_insert();"
+  end
+end