diff --git a/lib/aprs/application.ex b/lib/aprs/application.ex index 5cdd389..f0bfbc1 100644 --- a/lib/aprs/application.ex +++ b/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 = diff --git a/lib/aprs/aprs_is_connection.ex b/lib/aprs/aprs_is_connection.ex new file mode 100644 index 0000000..cbfe4b1 --- /dev/null +++ b/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 diff --git a/lib/aprs/postgres_notifier.ex b/lib/aprs/postgres_notifier.ex new file mode 100644 index 0000000..b65d0fc --- /dev/null +++ b/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 diff --git a/lib/aprs_web/live/bad_packets_live/index.ex b/lib/aprs_web/live/bad_packets_live/index.ex index b2ca404..9da2ee9 100644 --- a/lib/aprs_web/live/bad_packets_live/index.ex +++ b/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 diff --git a/lib/aprs_web/live/map_live/index.ex b/lib/aprs_web/live/map_live/index.ex index 4a284fb..d53f31e 100644 --- a/lib/aprs_web/live/map_live/index.ex +++ b/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) diff --git a/priv/repo/migrations/20250617190000_notify_on_badpackets_insert.exs b/priv/repo/migrations/20250617190000_notify_on_badpackets_insert.exs new file mode 100644 index 0000000..7fe4700 --- /dev/null +++ b/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 diff --git a/priv/repo/migrations/20250617191000_notify_on_packets_insert.exs b/priv/repo/migrations/20250617191000_notify_on_packets_insert.exs new file mode 100644 index 0000000..b4ad1fa --- /dev/null +++ b/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 diff --git a/priv/repo/migrations/20250617200000_recreate_notify_packets_insert_function.exs b/priv/repo/migrations/20250617200000_recreate_notify_packets_insert_function.exs new file mode 100644 index 0000000..3dae22f --- /dev/null +++ b/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