From dcabb744df6a59942cb93554d0c6af395a30fcb4 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 20 Feb 2026 08:29:10 -0600 Subject: [PATCH] Batch push_events, RF path queries, and ETS tuning for hot paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Batch real-time packet push_events: collect marker data across the packet batch and send a single "new_packets" WebSocket message instead of one per packet. JS handler processes all live→historical conversions before adding new markers to avoid intra-batch conflicts. - Batch RF path station queries: replace N+1 get_latest_packet_for_callsign calls with a single DISTINCT ON query selecting only callsign + lat/lng. - Add read_concurrency: true to streaming_packets_subscribers and :aprsme ETS tables (read-heavy, rarely written). - Remove dead GC code in PacketConsumer (threshold > 1000 unreachable with max batch_size of 100). --- assets/js/map.ts | 83 +++++++++++++ lib/aprsme/application.ex | 2 +- lib/aprsme/packet_consumer.ex | 6 - lib/aprsme/packets.ex | 9 ++ lib/aprsme/packets/prepared_queries.ex | 22 ++++ lib/aprsme/streaming_packets_pubsub.ex | 2 +- lib/aprsme_web/live/map_live/index.ex | 52 ++++++-- .../live/map_live/packet_processor.ex | 117 +++++++++++++----- lib/aprsme_web/live/map_live/rf_path.ex | 23 +--- test/aprsme/packets/prepared_queries_test.exs | 113 +++++++++++++++++ .../live/map_live/packet_processor_test.exs | 97 +++++++++++++++ 11 files changed, 462 insertions(+), 64 deletions(-) create mode 100644 test/aprsme/packets/prepared_queries_test.exs create mode 100644 test/aprsme_web/live/map_live/packet_processor_test.exs diff --git a/assets/js/map.ts b/assets/js/map.ts index 1c69cae..4677e28 100644 --- a/assets/js/map.ts +++ b/assets/js/map.ts @@ -1125,6 +1125,89 @@ let MapAPRSMap = { } }); + // Handle batched new packets from LiveView (performance optimization) + self.handleEvent( + "new_packets", + (payload: { packets: MarkerData[] }) => { + try { + if (!self || !self.map || self.isDestroyed) return; + if (!self.map.hasLayer) return; + if (self.heatLayer && self.map.hasLayer(self.heatLayer)) return; + if (!self.markerStates || !self.markers) return; + + const packets = payload.packets; + if (!packets || packets.length === 0) return; + + // Phase 1: Collect ALL markers to convert across the entire batch + // This prevents converting a marker that was just added by an earlier packet + const allNewCallsigns = new Set(); + const allNewIds = new Set(); + for (const data of packets) { + const cs = data.callsign_group || data.callsign || data.id; + if (cs) allNewCallsigns.add(cs); + if (data.id) allNewIds.add(String(data.id)); + } + + const markersToConvert: string[] = []; + self.markerStates.forEach((state, id) => { + const stateCallsign = state.callsign_group || state.callsign; + if ( + stateCallsign && + allNewCallsigns.has(stateCallsign) && + (!state.historical || state.is_most_recent_for_callsign) && + !allNewIds.has(String(id)) + ) { + markersToConvert.push(String(id)); + } + }); + + // Phase 2: Convert all existing live markers to historical dots + for (const id of markersToConvert) { + const existingMarker = self.markers.get(id); + const existingState = self.markerStates.get(id); + if (existingMarker && existingState) { + existingState.historical = true; + existingState.is_most_recent_for_callsign = false; + const incomingCallsign = + existingState.callsign_group || existingState.callsign; + const historicalIconData = { + id: String(id), + lat: existingState.lat, + lng: existingState.lng, + callsign: existingState.callsign || incomingCallsign, + callsign_group: + existingState.callsign_group || incomingCallsign, + symbol_table_id: existingState.symbol_table, + symbol_code: existingState.symbol_code, + historical: true, + is_most_recent_for_callsign: false, + popup: existingState.popup, + }; + existingMarker.setIcon(self.createMarkerIcon(historicalIconData)); + (existingMarker as APRSMarker)._isHistorical = true; + } + } + + // Phase 3: Add all new markers + for (const data of packets) { + const shouldOpenPopup = data.openPopup !== false; + const incomingCallsign = + data.callsign_group || data.callsign || data.id; + self.addMarker({ + ...data, + historical: false, + is_most_recent_for_callsign: true, + callsign_group: incomingCallsign, + popup: data.popup || self.buildPopupContent(data), + openPopup: shouldOpenPopup, + }); + } + } catch (error) { + console.error("Error in new_packets handler:", error); + } + }, + ); + // Handle highlighting the latest packet (open its popup) self.currentPopupMarkerId = null; self.handleEvent("highlight_packet", (data: { id: string }) => { diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index fe73553..f73d032 100644 --- a/lib/aprsme/application.ex +++ b/lib/aprsme/application.ex @@ -185,7 +185,7 @@ defmodule Aprsme.Application do :ets.new(:query_cache, [:set, :public, :named_table, read_concurrency: true]) :ets.new(:device_cache, [:set, :public, :named_table, read_concurrency: true]) :ets.new(:symbol_cache, [:set, :public, :named_table, read_concurrency: true]) - :ets.new(:aprsme, [:set, :public, :named_table, write_concurrency: true]) + :ets.new(:aprsme, [:set, :public, :named_table, read_concurrency: true]) :ets.insert(:aprsme, {:message_number, 0}) [ diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 50a3a4d..b24a789 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -232,12 +232,6 @@ defmodule Aprsme.PacketConsumer do ) end - # Only do minor GC after very large batches to prevent memory accumulation - # This should rarely trigger with batch_size of 100 - if length(packets) > 1000 do - :erlang.garbage_collect(self(), type: :minor) - end - :telemetry.execute( [ :aprsme, diff --git a/lib/aprsme/packets.ex b/lib/aprsme/packets.ex index be23db9..fc75846 100644 --- a/lib/aprsme/packets.ex +++ b/lib/aprsme/packets.ex @@ -866,6 +866,15 @@ defmodule Aprsme.Packets do PreparedQueries.get_latest_packet_for_callsign(callsign) end + @doc """ + Gets latest positions for multiple callsigns in a single batch query. + Returns a list of `%{callsign: String.t(), lat: float(), lng: float()}` maps. + """ + @spec get_latest_positions_for_callsigns(list(String.t())) :: [map()] + def get_latest_positions_for_callsigns(callsigns) when is_list(callsigns) do + PreparedQueries.get_latest_positions_for_callsigns(callsigns) + end + @doc """ Gets the most recent weather packet for a callsign. Looks for any packet containing weather data fields. diff --git a/lib/aprsme/packets/prepared_queries.ex b/lib/aprsme/packets/prepared_queries.ex index 051f3b2..a97076d 100644 --- a/lib/aprsme/packets/prepared_queries.ex +++ b/lib/aprsme/packets/prepared_queries.ex @@ -27,6 +27,28 @@ defmodule Aprsme.Packets.PreparedQueries do ) end + @doc """ + Get latest positions for multiple callsigns in a single query. + Returns a list of `%{callsign: String.t(), lat: float(), lng: float()}` maps. + Uses DISTINCT ON to get only the most recent packet per callsign. + """ + @spec get_latest_positions_for_callsigns(list(String.t())) :: [map()] + def get_latest_positions_for_callsigns([]), do: [] + + def get_latest_positions_for_callsigns(callsigns) when is_list(callsigns) do + normalized = Enum.map(callsigns, &String.upcase(String.trim(&1))) + + Repo.all( + from(p in Packet, + where: fragment("upper(?)", p.sender) in ^normalized, + where: not is_nil(p.location), + distinct: fragment("upper(?)", p.sender), + order_by: [asc: fragment("upper(?)", p.sender), desc: p.received_at], + select: %{callsign: p.sender, lat: fragment("ST_Y(?)", p.location), lng: fragment("ST_X(?)", p.location)} + ) + ) + end + @doc """ Get recent packets within bounds using prepared statement. """ diff --git a/lib/aprsme/streaming_packets_pubsub.ex b/lib/aprsme/streaming_packets_pubsub.ex index 27fbccb..b72c912 100644 --- a/lib/aprsme/streaming_packets_pubsub.ex +++ b/lib/aprsme/streaming_packets_pubsub.ex @@ -61,7 +61,7 @@ defmodule Aprsme.StreamingPacketsPubSub do @impl true def init(_opts) do # Create ETS table for fast lookups - :ets.new(@table_name, [:set, :protected, :named_table]) + :ets.new(@table_name, [:set, :protected, :named_table, read_concurrency: true]) # Monitor subscribers for cleanup Process.flag(:trap_exit, true) diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index 0316029..e371c67 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -850,15 +850,38 @@ defmodule AprsmeWeb.MapLive.Index do end def handle_info({:packet_batch, packets}, socket) do - # Process batch of packets efficiently - socket = - Enum.reduce(packets, socket, fn packet, acc_socket -> - case handle_info_postgres_packet(packet, acc_socket) do - {:noreply, new_socket} -> new_socket - _ -> acc_socket - end + # Process batch: collect marker data without pushing events, then send one batch push + {socket, marker_data_list} = + Enum.reduce(packets, {socket, []}, fn packet, {acc_socket, markers} -> + {new_socket, marker_data} = process_packet_for_batch(packet, acc_socket) + markers = if marker_data, do: [marker_data | markers], else: markers + {new_socket, markers} end) + socket = assign(socket, :last_update_at, DateTime.utc_now()) + + # Send heat map update once if in heat map mode, otherwise batch markers + socket = + cond do + socket.assigns.map_zoom <= 8 and marker_data_list != [] -> + DisplayManager.send_heat_map_for_current_bounds(socket) + + marker_data_list != [] -> + markers = + Enum.map(Enum.reverse(marker_data_list), fn data -> + if socket.assigns.station_popup_open do + Map.put(data, :openPopup, false) + else + data + end + end) + + push_event(socket, "new_packets", %{packets: markers}) + + true -> + socket + end + {:noreply, socket} end @@ -1047,6 +1070,21 @@ defmodule AprsmeWeb.MapLive.Index do PacketProcessor.process_packet_for_display(packet, socket) end + defp process_packet_for_batch(packet, socket) do + if socket.assigns.tracked_callsign == "" do + PacketProcessor.process_packet_for_marker_data(packet, socket) + else + packet_sender = Map.get(packet, :sender, Map.get(packet, "sender", "")) + + if String.upcase(packet_sender) == String.upcase(socket.assigns.tracked_callsign) do + socket = assign(socket, :tracked_callsign_latest_packet, packet) + PacketProcessor.process_packet_for_marker_data(packet, socket) + else + {socket, nil} + end + end + end + # Handle replaying the next historical packet @impl true diff --git a/lib/aprsme_web/live/map_live/packet_processor.ex b/lib/aprsme_web/live/map_live/packet_processor.ex index bc0760c..fbee0ec 100644 --- a/lib/aprsme_web/live/map_live/packet_processor.ex +++ b/lib/aprsme_web/live/map_live/packet_processor.ex @@ -9,8 +9,8 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do alias AprsmeWeb.Live.Shared.BoundsUtils alias AprsmeWeb.Live.Shared.CoordinateUtils alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils + alias AprsmeWeb.MapLive.DataBuilder alias AprsmeWeb.MapLive.DisplayManager - alias AprsmeWeb.MapLive.PacketUtils alias Phoenix.LiveView alias Phoenix.LiveView.Socket @@ -18,6 +18,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do @doc """ Process a packet for display on the map. + Pushes events directly to the socket (used for single-packet path). """ @spec process_packet_for_display(map(), Socket.t()) :: {:noreply, Socket.t()} def process_packet_for_display(packet, socket) do @@ -28,12 +29,24 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do socket = handle_packet_visibility(packet, lat, lon, callsign_key, socket) # Update last update timestamp for real-time display in map sidebar - # This timestamp is shown to users to indicate when data was last refreshed socket = assign(socket, :last_update_at, DateTime.utc_now()) {:noreply, socket} end + @doc """ + Process a packet for batch display. Updates socket state but does NOT push + events. Returns `{socket, marker_data | nil}` where marker_data is the data + that should be included in a batched push_event, or nil if no marker needed. + """ + @spec process_packet_for_marker_data(map(), Socket.t()) :: {Socket.t(), map() | nil} + def process_packet_for_marker_data(packet, socket) do + {lat, lon, _data_extended} = CoordinateUtils.get_coordinates(packet) + callsign_key = SharedPacketUtils.get_callsign_key(packet) + + batch_visibility_action(packet, lat, lon, callsign_key, socket) + end + @doc """ Handle packet visibility logic based on bounds and existing markers. """ @@ -78,34 +91,17 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do """ @spec handle_valid_postgres_packet(map(), number() | nil, number() | nil, Socket.t()) :: Socket.t() def handle_valid_postgres_packet(packet, _lat, _lon, socket) do - # Add the packet to visible_packets and push a marker immediately - callsign_key = SharedPacketUtils.get_callsign_key(packet) + {socket, marker_data} = prepare_packet_state(packet, socket) - new_visible_packets = Map.put(socket.assigns.visible_packets, callsign_key, packet) + cond do + socket.assigns.map_zoom <= 8 -> + send_heat_map_for_current_bounds(socket) - # Enforce memory limits - new_visible_packets = - if map_size(new_visible_packets) > @max_visible_packets do - SharedPacketUtils.prune_oldest_packets(new_visible_packets, @max_visible_packets) - else - new_visible_packets - end - - socket = assign(socket, :visible_packets, new_visible_packets) - - # Check zoom level to decide how to display the packet - if socket.assigns.map_zoom <= 8 do - # We're in heat map mode - update the heat map with all current data - send_heat_map_for_current_bounds(socket) - else - # We're in marker mode - send individual marker - marker_data = PacketUtils.build_packet_data(packet, true, get_locale(socket)) - - if marker_data do + marker_data -> send_marker_with_popup_check(socket, marker_data) - else + + true -> socket - end end end @@ -119,14 +115,78 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do assign(socket, :visible_packets, new_visible_packets) end + # Batch-mode visibility: same logic as visibility_action but returns + # {socket, marker_data | nil} without pushing events + defp batch_visibility_action(packet, lat, lon, callsign_key, socket) when not is_nil(lat) and not is_nil(lon) do + in_bounds = within_bounds?(%{lat: lat, lon: lon}, socket.assigns.map_bounds) + has_marker = Map.has_key?(socket.assigns.visible_packets, callsign_key) + batch_dispatch({in_bounds, has_marker}, packet, lat, lon, callsign_key, socket) + end + + defp batch_visibility_action(_packet, _lat, _lon, _callsign_key, socket) do + {socket, nil} + end + + # Out of bounds but has marker — remove it (still pushes remove_marker since + # removals are rare and don't benefit from batching) + defp batch_dispatch({false, true}, _packet, _lat, _lon, callsign_key, socket) do + socket = remove_marker_from_map(callsign_key, socket) + {socket, nil} + end + + # In bounds, no existing marker — add it + defp batch_dispatch({true, false}, packet, _lat, _lon, _callsign_key, socket) do + prepare_packet_state(packet, socket) + end + + # In bounds, existing marker — update if significant movement + defp batch_dispatch({true, true}, packet, lat, lon, callsign_key, socket) do + existing_packet = socket.assigns.visible_packets[callsign_key] + {existing_lat, existing_lon, _} = CoordinateUtils.get_coordinates(existing_packet) + + if is_number(existing_lat) and is_number(existing_lon) and + GeoUtils.significant_movement?(existing_lat, existing_lon, lat, lon, 50) do + prepare_packet_state(packet, socket) + else + new_visible_packets = Map.put(socket.assigns.visible_packets, callsign_key, packet) + {assign(socket, :visible_packets, new_visible_packets), nil} + end + end + + defp batch_dispatch(_, _packet, _lat, _lon, _callsign_key, socket) do + {socket, nil} + end + + # Shared: updates visible_packets state and builds marker data. + # Returns {socket, marker_data | nil}. Does NOT push events. + defp prepare_packet_state(packet, socket) do + callsign_key = SharedPacketUtils.get_callsign_key(packet) + + new_visible_packets = Map.put(socket.assigns.visible_packets, callsign_key, packet) + + new_visible_packets = + if map_size(new_visible_packets) > @max_visible_packets do + SharedPacketUtils.prune_oldest_packets(new_visible_packets, @max_visible_packets) + else + new_visible_packets + end + + socket = assign(socket, :visible_packets, new_visible_packets) + + marker_data = + if socket.assigns.map_zoom > 8 do + DataBuilder.build_packet_data(packet, true, get_locale(socket)) + end + + {socket, marker_data} + end + # Private helper functions - # Use shared bounds utility defp within_bounds?(coords, bounds) do BoundsUtils.within_bounds?(coords, bounds) end - # Helper to get locale from socket defp get_locale(socket) do Map.get(socket.assigns, :locale, "en") end @@ -137,7 +197,6 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do defp send_marker_with_popup_check(socket, marker_data) do if socket.assigns.station_popup_open do - # Send without opening popup to avoid interrupting user LiveView.push_event(socket, "new_packet", Map.put(marker_data, :openPopup, false)) else LiveView.push_event(socket, "new_packet", marker_data) diff --git a/lib/aprsme_web/live/map_live/rf_path.ex b/lib/aprsme_web/live/map_live/rf_path.ex index da4207f..a0deee7 100644 --- a/lib/aprsme_web/live/map_live/rf_path.ex +++ b/lib/aprsme_web/live/map_live/rf_path.ex @@ -46,12 +46,9 @@ defmodule AprsmeWeb.MapLive.RfPath do """ @spec get_path_station_positions(list(binary()), Phoenix.LiveView.Socket.t()) :: list(map()) def get_path_station_positions(path_stations, _socket) do - # Limit to prevent excessive database queries - limited_stations = Enum.take(path_stations, 10) - - limited_stations - |> Enum.map(&get_station_position/1) - |> Enum.filter(& &1) + path_stations + |> Enum.take(10) + |> Packets.get_latest_positions_for_callsigns() end defp split_at_q_construct(elements) do @@ -95,18 +92,4 @@ defmodule AprsmeWeb.MapLive.RfPath do "" end end - - defp get_station_position(callsign) do - case Packets.get_latest_packet_for_callsign(callsign) do - %{lat: lat, lon: lon} when is_number(lat) and is_number(lon) -> - %{ - callsign: callsign, - lat: lat, - lng: lon - } - - _ -> - nil - end - end end diff --git a/test/aprsme/packets/prepared_queries_test.exs b/test/aprsme/packets/prepared_queries_test.exs new file mode 100644 index 0000000..63babb1 --- /dev/null +++ b/test/aprsme/packets/prepared_queries_test.exs @@ -0,0 +1,113 @@ +defmodule Aprsme.Packets.PreparedQueriesTest do + use Aprsme.DataCase, async: false + + alias Aprsme.Packet + alias Aprsme.Packets.PreparedQueries + alias Aprsme.Repo + + defp create_positioned_packet(attrs) do + defaults = %{ + sender: "TEST-1", + base_callsign: "TEST", + ssid: "1", + destination: "APRS", + data_type: "position", + lat: Decimal.new("33.0000"), + lon: Decimal.new("-96.0000"), + received_at: DateTime.truncate(DateTime.utc_now(), :second) + } + + merged = Map.merge(defaults, attrs) + + %Packet{} + |> Packet.changeset(Map.from_struct(Map.merge(%Packet{}, merged))) + |> Repo.insert() + end + + describe "get_latest_positions_for_callsigns/1" do + test "returns positions for multiple callsigns" do + {:ok, _} = + create_positioned_packet(%{ + sender: "K5GVL-10", + base_callsign: "K5GVL", + ssid: "10", + lat: Decimal.new("33.1000"), + lon: Decimal.new("-96.6000") + }) + + {:ok, _} = + create_positioned_packet(%{ + sender: "N5TXZ-10", + base_callsign: "N5TXZ", + ssid: "10", + lat: Decimal.new("33.2000"), + lon: Decimal.new("-96.5000") + }) + + result = PreparedQueries.get_latest_positions_for_callsigns(["K5GVL-10", "N5TXZ-10"]) + + assert length(result) == 2 + callsigns = Enum.map(result, & &1.callsign) + assert "K5GVL-10" in callsigns + assert "N5TXZ-10" in callsigns + + k5gvl = Enum.find(result, &(&1.callsign == "K5GVL-10")) + assert_in_delta k5gvl.lat, 33.1, 0.01 + assert_in_delta k5gvl.lng, -96.6, 0.01 + end + + test "returns empty list for empty input" do + assert PreparedQueries.get_latest_positions_for_callsigns([]) == [] + end + + test "returns empty list for nonexistent callsigns" do + result = PreparedQueries.get_latest_positions_for_callsigns(["NONEXIST-1", "FAKE-2"]) + assert result == [] + end + + test "is case-insensitive" do + {:ok, _} = + create_positioned_packet(%{ + sender: "K5GVL-10", + base_callsign: "K5GVL", + ssid: "10", + lat: Decimal.new("33.1000"), + lon: Decimal.new("-96.6000") + }) + + result = PreparedQueries.get_latest_positions_for_callsigns(["k5gvl-10"]) + + assert length(result) == 1 + assert hd(result).callsign == "K5GVL-10" + end + + test "returns latest position when multiple packets exist for a callsign" do + {:ok, _} = + create_positioned_packet(%{ + sender: "K5GVL-10", + base_callsign: "K5GVL", + ssid: "10", + lat: Decimal.new("33.1000"), + lon: Decimal.new("-96.6000"), + received_at: DateTime.add(DateTime.utc_now(), -3600, :second) + }) + + {:ok, _} = + create_positioned_packet(%{ + sender: "K5GVL-10", + base_callsign: "K5GVL", + ssid: "10", + lat: Decimal.new("34.0000"), + lon: Decimal.new("-97.0000"), + received_at: DateTime.truncate(DateTime.utc_now(), :second) + }) + + result = PreparedQueries.get_latest_positions_for_callsigns(["K5GVL-10"]) + + assert length(result) == 1 + position = hd(result) + assert_in_delta position.lat, 34.0, 0.01 + assert_in_delta position.lng, -97.0, 0.01 + end + end +end diff --git a/test/aprsme_web/live/map_live/packet_processor_test.exs b/test/aprsme_web/live/map_live/packet_processor_test.exs new file mode 100644 index 0000000..042aa1b --- /dev/null +++ b/test/aprsme_web/live/map_live/packet_processor_test.exs @@ -0,0 +1,97 @@ +defmodule AprsmeWeb.MapLive.PacketProcessorTest do + use ExUnit.Case, async: true + + alias AprsmeWeb.MapLive.PacketProcessor + + defp build_socket(assigns \\ %{}) do + defaults = %{ + map_bounds: %{north: 34.0, south: 32.0, east: -95.0, west: -97.0}, + visible_packets: %{}, + map_zoom: 10, + station_popup_open: false, + locale: "en" + } + + merged = Map.merge(defaults, assigns) + + socket = %Phoenix.LiveView.Socket{ + assigns: Map.put(merged, :__changed__, %{}) + } + + socket + end + + defp build_packet(overrides \\ %{}) do + Map.merge( + %{ + id: "test-packet-1", + sender: "K5GVL-10", + base_callsign: "K5GVL", + ssid: "10", + lat: 33.1, + lon: -96.5, + symbol_table_id: "/", + symbol_code: "-", + comment: "Test station", + received_at: DateTime.utc_now() + }, + overrides + ) + end + + describe "process_packet_for_marker_data/2" do + test "returns marker data for in-bounds packet at marker zoom level" do + socket = build_socket() + packet = build_packet() + + {updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + + assert marker_data + assert is_map(marker_data) + # Socket state should be updated + assert map_size(updated_socket.assigns.visible_packets) == 1 + end + + test "returns nil marker data for out-of-bounds packet" do + socket = build_socket() + # Packet outside the bounds (north: 34, south: 32, east: -95, west: -97) + packet = build_packet(%{lat: 40.0, lon: -80.0}) + + {updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + + assert marker_data == nil + assert map_size(updated_socket.assigns.visible_packets) == 0 + end + + test "returns nil marker data for packet with nil coordinates" do + socket = build_socket() + packet = build_packet(%{lat: nil, lon: nil}) + + {updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + + assert marker_data == nil + assert map_size(updated_socket.assigns.visible_packets) == 0 + end + + test "returns nil marker data in heat map mode (zoom <= 8)" do + socket = build_socket(%{map_zoom: 7}) + packet = build_packet() + + {_updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + + assert marker_data == nil + end + + test "does not push any events to socket" do + socket = build_socket() + packet = build_packet() + + {updated_socket, _marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + + # Socket should not have any push_events queued + # In LiveView, push_event adds to socket.private[:push_events] + push_events = get_in(updated_socket.private, [:push_events]) || [] + assert push_events == [] + end + end +end