diff --git a/lib/aprsme_web/live/packets_live/callsign_view.ex b/lib/aprsme_web/live/packets_live/callsign_view.ex index 22c3355..a3312c4 100644 --- a/lib/aprsme_web/live/packets_live/callsign_view.ex +++ b/lib/aprsme_web/live/packets_live/callsign_view.ex @@ -11,6 +11,8 @@ defmodule AprsmeWeb.PacketsLive.CallsignView do alias Aprsme.Repo alias AprsmeWeb.Live.SharedPacketHandler + @flush_interval_ms 150 + @impl true def mount(%{"callsign" => callsign}, _session, socket) do normalized = Callsign.normalize(callsign) @@ -40,6 +42,8 @@ defmodule AprsmeWeb.PacketsLive.CallsignView do |> assign(:live_packets, []) |> assign(:all_packets, stored) |> assign(:error, nil) + |> put_private(:packet_buffer, []) + |> put_private(:flush_timer, nil) end defp assign_invalid(socket, callsign) do @@ -53,28 +57,52 @@ defmodule AprsmeWeb.PacketsLive.CallsignView do @impl true def handle_info({:postgres_packet, payload}, socket) do - SharedPacketHandler.handle_packet_update(payload, socket, - filter_fn: fn _packet, _socket -> true end, - process_fn: &process_matching_packet/2 - ) + # Buffer in socket.private to avoid triggering a rerender on every packet. + # A periodic flush timer drains the buffer and updates assigns in one batch, + # which keeps client-side morphdom from being overwhelmed during high-traffic bursts. + socket = update_in(socket.private.packet_buffer, &[payload | &1]) + socket = ensure_flush_timer(socket) + {:noreply, socket} + end + + @impl true + def handle_info(:flush_packets, socket) do + packets = Enum.reverse(socket.private.packet_buffer) + socket = put_private(socket, :packet_buffer, []) + socket = put_private(socket, :flush_timer, nil) + + {updated_stored, updated_live} = + packets + # Sanitize encoding first, then batch enrich with device info + # (enrich_packets_with_device_info batches device lookups internally) + |> Enum.map(&EncodingUtils.sanitize_packet/1) + |> SharedPacketHandler.enrich_packets_with_device_info() + # Add canonical device identifier per-packet + |> Enum.map(&enrich_packet_with_device_identifier/1) + # Accumulate through the packet list update logic + |> Enum.reduce({socket.assigns.packets, socket.assigns.live_packets}, fn enriched, {stored, live} -> + update_packet_lists(stored, live, enriched) + end) + + all_packets = get_all_packets_list(updated_stored, updated_live) + + {:noreply, + assign(socket, + packets: updated_stored, + live_packets: updated_live, + all_packets: all_packets + )} end def handle_info(_message, socket), do: {:noreply, socket} - defp process_matching_packet(enriched_payload, socket) do - enriched_payload = enrich_packet_with_device_identifier(enriched_payload) - - current_live = socket.assigns.live_packets - current_stored = socket.assigns.packets - - {updated_stored, updated_live} = update_packet_lists(current_stored, current_live, enriched_payload) - all_packets = get_all_packets_list(updated_stored, updated_live) - - {:noreply, - socket - |> assign(:packets, updated_stored) - |> assign(:live_packets, updated_live) - |> assign(:all_packets, all_packets)} + defp ensure_flush_timer(socket) do + if socket.private.flush_timer do + socket + else + timer = Process.send_after(self(), :flush_packets, @flush_interval_ms) + put_private(socket, :flush_timer, timer) + end end defp enrich_packet_with_device_identifier(sanitized_payload) do diff --git a/lib/aprsme_web/live/packets_live/index.ex b/lib/aprsme_web/live/packets_live/index.ex index e6e479c..67138a5 100644 --- a/lib/aprsme_web/live/packets_live/index.ex +++ b/lib/aprsme_web/live/packets_live/index.ex @@ -5,37 +5,78 @@ defmodule AprsmeWeb.PacketsLive.Index do alias Aprsme.EncodingUtils + @flush_interval_ms 150 + @max_stream_size 100 + @impl true def mount(_params, _session, socket) do - _ = - if connected?(socket) do - Phoenix.PubSub.subscribe(Aprsme.PubSub, "postgres:aprsme_packets") - end + if connected?(socket) do + Phoenix.PubSub.subscribe(Aprsme.PubSub, "postgres:aprsme_packets") + end - {:ok, stream(socket, :packets, [], at: 0)} + {:ok, + socket + |> put_private(:packet_buffer, []) + |> put_private(:flush_timer, nil) + |> stream(:packets, [], at: 0)} end @impl true def handle_info({:postgres_packet, payload}, socket) do - sanitized_payload = EncodingUtils.sanitize_packet(payload) + # Buffer in socket.private to avoid triggering a rerender on every packet. + # A periodic flush timer drains the buffer and batch-inserts into the stream, + # which keeps client-side morphdom from being overwhelmed during high-traffic bursts. + socket = update_in(socket.private.packet_buffer, &[payload | &1]) + socket = ensure_flush_timer(socket) + {:noreply, socket} + end + @impl true + def handle_info(:flush_packets, socket) do + packets = Enum.reverse(socket.private.packet_buffer) + socket = put_private(socket, :packet_buffer, []) + socket = put_private(socket, :flush_timer, nil) + + # Batch insert all accumulated packets into the stream. + # Processing newest-first with at: 0 preserves descending-chronological order. socket = - socket - |> stream_insert(:packets, sanitized_payload, at: 0) - |> then(fn s -> - stream = s.assigns.streams.packets - - if Enum.count(stream) > 100 do - {id, _} = stream |> Enum.reverse() |> hd() - stream_delete(s, :packets, id) - else - s - end + Enum.reduce(packets, socket, fn payload, s -> + sanitized = EncodingUtils.sanitize_packet(payload) + stream_insert(s, :packets, sanitized, at: 0) end) + # Trim stream to keep at most @max_stream_size items. + socket = trim_stream(socket) + {:noreply, socket} end + defp ensure_flush_timer(socket) do + if socket.private.flush_timer do + socket + else + timer = Process.send_after(self(), :flush_packets, @flush_interval_ms) + put_private(socket, :flush_timer, timer) + end + end + + defp trim_stream(socket) do + stream = socket.assigns.streams.packets + count = Enum.count(stream) + + if count > @max_stream_size do + excess = count - @max_stream_size + # Remove excess from the end (oldest items) + to_remove = stream |> Enum.reverse() |> Enum.take(excess) + + Enum.reduce(to_remove, socket, fn {id, _}, s -> + stream_delete(s, :packets, id) + end) + else + socket + end + end + @doc """ Extract a coordinate (:lat or :lon) from a packet, checking multiple sources: 1. Direct :lat/:lon keys