From 21d28afe8e92513acd431ab48172cb51844a451f Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 22 Mar 2026 17:20:27 -0500 Subject: [PATCH] refactor: reduce map live complexity --- lib/aprsme/streaming_packets_pubsub.ex | 51 +++++--- .../live/map_live/display_manager.ex | 83 +++++++------ lib/aprsme_web/live/map_live/index.ex | 117 +++++++++++------- 3 files changed, 148 insertions(+), 103 deletions(-) diff --git a/lib/aprsme/streaming_packets_pubsub.ex b/lib/aprsme/streaming_packets_pubsub.ex index 42848ce..0a6efee 100644 --- a/lib/aprsme/streaming_packets_pubsub.ex +++ b/lib/aprsme/streaming_packets_pubsub.ex @@ -112,26 +112,7 @@ defmodule Aprsme.StreamingPacketsPubSub do @impl true def handle_cast({:broadcast, packet}, state) do - # Get packet coordinates - lat = packet[:latitude] || packet[:lat] - lon = packet[:longitude] || packet[:lon] || packet[:lng] - - if lat && lon do - # Get all subscribers — table is small (one entry per LiveView client) - subscribers = :ets.tab2list(@table_name) - - if test_env?() do - send_to_matching_subscribers(subscribers, lat, lon, packet, self()) - else - # Send to matching subscribers using BroadcastTaskSupervisor - # Collect dead pids to clean up - server_pid = self() - - Aprsme.BroadcastTaskSupervisor.async_execute(fn -> - send_to_matching_subscribers(subscribers, lat, lon, packet, server_pid) - end) - end - end + maybe_broadcast_packet(packet) {:noreply, state} end @@ -216,4 +197,34 @@ defmodule Aprsme.StreamingPacketsPubSub do default end end + + defp maybe_broadcast_packet(packet) do + case packet_coordinates(packet) do + {lat, lon} -> + subscribers = :ets.tab2list(@table_name) + dispatch_packet_to_subscribers(subscribers, lat, lon, packet) + + nil -> + :ok + end + end + + defp packet_coordinates(packet) do + lat = packet[:latitude] || packet[:lat] + lon = packet[:longitude] || packet[:lon] || packet[:lng] + + if lat && lon, do: {lat, lon} + end + + defp dispatch_packet_to_subscribers(subscribers, lat, lon, packet) do + if test_env?() do + send_to_matching_subscribers(subscribers, lat, lon, packet, self()) + else + server_pid = self() + + Aprsme.BroadcastTaskSupervisor.async_execute(fn -> + send_to_matching_subscribers(subscribers, lat, lon, packet, server_pid) + end) + end + end end diff --git a/lib/aprsme_web/live/map_live/display_manager.ex b/lib/aprsme_web/live/map_live/display_manager.ex index d150333..a89a8b7 100644 --- a/lib/aprsme_web/live/map_live/display_manager.ex +++ b/lib/aprsme_web/live/map_live/display_manager.ex @@ -90,45 +90,12 @@ defmodule AprsmeWeb.MapLive.DisplayManager do def send_trail_line_for_tracked_callsign(socket) do callsign = socket.assigns.tracked_callsign threshold = socket.assigns.packet_age_threshold + packets = trail_packets_for_callsign(socket.assigns.visible_packets, callsign, threshold) - packets = - socket.assigns.visible_packets - |> Map.values() - |> Enum.filter(fn packet -> - sender = Map.get(packet, :sender) || Map.get(packet, "sender") - received_at = Map.get(packet, :received_at) || Map.get(packet, "received_at") - - String.upcase(sender) == String.upcase(callsign) && - DateTime.compare(received_at, threshold) != :lt - end) - |> Enum.sort_by( - fn packet -> - received_at = Map.get(packet, :received_at) || Map.get(packet, "received_at") - DateTime.to_unix(received_at, :microsecond) - end, - :asc - ) - - if Enum.empty?(packets) do + if packets == [] do socket else - points = - Enum.map(packets, fn packet -> - lat = Map.get(packet, :lat) || Map.get(packet, "lat") - lon = Map.get(packet, :lon) || Map.get(packet, "lon") - received_at = Map.get(packet, :received_at) || Map.get(packet, "received_at") - - %{ - lat: Aprsme.EncodingUtils.to_float(lat) || 0.0, - lng: Aprsme.EncodingUtils.to_float(lon) || 0.0, - timestamp: DateTime.to_iso8601(received_at) - } - end) - - LiveView.push_event(socket, "show_trail_line", %{ - callsign: callsign, - points: points - }) + push_trail_line(socket, callsign, packets) end end @@ -184,4 +151,48 @@ defmodule AprsmeWeb.MapLive.DisplayManager do # This function should be moved to HistoricalLoader module socket end + + defp trail_packets_for_callsign(visible_packets, callsign, threshold) do + visible_packets + |> Map.values() + |> Enum.filter(&tracked_packet?(&1, callsign, threshold)) + |> Enum.sort_by(&packet_timestamp/1, :asc) + end + + defp tracked_packet?(packet, callsign, threshold) do + sender = Map.get(packet, :sender) || Map.get(packet, "sender") + received_at = packet_received_at(packet) + + String.upcase(sender) == String.upcase(callsign) && + DateTime.compare(received_at, threshold) != :lt + end + + defp push_trail_line(socket, callsign, packets) do + LiveView.push_event(socket, "show_trail_line", %{ + callsign: callsign, + points: Enum.map(packets, &trail_point/1) + }) + end + + defp trail_point(packet) do + lat = Map.get(packet, :lat) || Map.get(packet, "lat") + lon = Map.get(packet, :lon) || Map.get(packet, "lon") + received_at = packet_received_at(packet) + + %{ + lat: Aprsme.EncodingUtils.to_float(lat) || 0.0, + lng: Aprsme.EncodingUtils.to_float(lon) || 0.0, + timestamp: DateTime.to_iso8601(received_at) + } + end + + defp packet_received_at(packet) do + Map.get(packet, :received_at) || Map.get(packet, "received_at") + end + + defp packet_timestamp(packet) do + packet + |> packet_received_at() + |> DateTime.to_unix(:microsecond) + end end diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index a40e8b9..fdb275c 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -1945,60 +1945,22 @@ defmodule AprsmeWeb.MapLive.Index do Logger.debug("process_bounds_update called with bounds: #{inspect(map_bounds)}") - # Update spatial viewport if we have a client ID - if socket.assigns[:spatial_client_id] do - Aprsme.SpatialPubSub.update_viewport(socket.assigns.spatial_client_id, map_bounds) - end + sync_pubsub_bounds(socket, map_bounds) - # Update StreamingPacketsPubSub subscription with new bounds - Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), map_bounds) + bounds_state = bounds_update_state(socket, map_bounds) + log_bounds_update_state(bounds_state) - # Check if this is the initial load or if bounds have actually changed - is_initial_load = socket.assigns[:needs_initial_historical_load] || !socket.assigns[:initial_bounds_loaded] - bounds_changed = socket.assigns.map_bounds && not BoundsUtils.compare_bounds(map_bounds, socket.assigns.map_bounds) + {new_visible_packets, packets_to_remove} = filtered_packets_for_bounds(socket.assigns.visible_packets, map_bounds) - # Check if we've completed the initial historical load - initial_historical_completed = socket.assigns[:initial_historical_completed] || false - - Logger.debug( - "is_initial_load: #{is_initial_load}, bounds_changed: #{bounds_changed}, initial_historical_completed: #{initial_historical_completed}" - ) - - # Remove out-of-bounds packets and markers immediately - new_visible_packets = BoundsUtils.filter_packets_by_bounds(socket.assigns.visible_packets, map_bounds) - packets_to_remove = BoundsUtils.reject_packets_by_bounds(socket.assigns.visible_packets, map_bounds) - - # Remove markers for out-of-bounds packets - socket = remove_markers_batch(socket, packets_to_remove) - - # Only clear historical packets if: - # 1. Bounds actually changed AND - # 2. This is not the initial load AND - # 3. We've already completed the initial historical load - socket = - if bounds_changed and not is_initial_load and initial_historical_completed do - Logger.debug("Bounds changed after initial load - clearing historical packets") - push_event(socket, "clear_historical_packets", %{}) - else - Logger.debug("Initial load or no significant change - keeping existing markers") - socket - end - - # Always filter markers by bounds - socket = push_event(socket, "filter_markers_by_bounds", %{bounds: map_bounds}) - - # Update map bounds and visible_packets with bounds-filtered result socket = socket + |> remove_markers_batch(packets_to_remove) + |> maybe_clear_historical_packets(bounds_state) + |> push_event("filter_markers_by_bounds", %{bounds: map_bounds}) |> assign(map_bounds: map_bounds, visible_packets: new_visible_packets) |> assign(needs_initial_historical_load: false) - # Load historical packets for the new bounds (now socket.assigns.map_bounds is correct) - Logger.debug("Starting progressive historical loading for new bounds") - - Logger.debug( - "Current assigns: historical_loading=#{socket.assigns.historical_loading}, map_bounds=#{inspect(socket.assigns.map_bounds)}" - ) + log_historical_reload(socket) socket = HistoricalLoader.start_progressive_historical_loading(socket) @@ -2012,10 +1974,71 @@ defmodule AprsmeWeb.MapLive.Index do end # Mark initial historical as completed if this was the initial load - if is_initial_load do + if bounds_state.is_initial_load do assign(socket, initial_historical_completed: true) else socket end end + + defp sync_pubsub_bounds(socket, map_bounds) do + if socket.assigns[:spatial_client_id] do + Aprsme.SpatialPubSub.update_viewport(socket.assigns.spatial_client_id, map_bounds) + end + + Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), map_bounds) + end + + defp bounds_update_state(socket, map_bounds) do + %{ + is_initial_load: socket.assigns[:needs_initial_historical_load] || !socket.assigns[:initial_bounds_loaded], + bounds_changed: + socket.assigns.map_bounds && not BoundsUtils.compare_bounds(map_bounds, socket.assigns.map_bounds), + initial_historical_completed: socket.assigns[:initial_historical_completed] || false + } + end + + defp log_bounds_update_state(bounds_state) do + require Logger + + Logger.debug( + "is_initial_load: #{bounds_state.is_initial_load}, bounds_changed: #{bounds_state.bounds_changed}, " <> + "initial_historical_completed: #{bounds_state.initial_historical_completed}" + ) + end + + defp filtered_packets_for_bounds(visible_packets, map_bounds) do + { + BoundsUtils.filter_packets_by_bounds(visible_packets, map_bounds), + BoundsUtils.reject_packets_by_bounds(visible_packets, map_bounds) + } + end + + defp maybe_clear_historical_packets(socket, %{ + bounds_changed: true, + is_initial_load: false, + initial_historical_completed: true + }) do + require Logger + + Logger.debug("Bounds changed after initial load - clearing historical packets") + push_event(socket, "clear_historical_packets", %{}) + end + + defp maybe_clear_historical_packets(socket, _bounds_state) do + require Logger + + Logger.debug("Initial load or no significant change - keeping existing markers") + socket + end + + defp log_historical_reload(socket) do + require Logger + + Logger.debug("Starting progressive historical loading for new bounds") + + Logger.debug( + "Current assigns: historical_loading=#{socket.assigns.historical_loading}, map_bounds=#{inspect(socket.assigns.map_bounds)}" + ) + end end