From 84e65232e5143098e5756e2a44d0b5cceb206925 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 20 Feb 2026 09:04:12 -0600 Subject: [PATCH] Batch RF path queries, batch marker removals, and JS callsign index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace N+1 get_latest_packet_for_callsign calls in RF path loading with single get_latest_packets_for_callsigns batch query (DISTINCT ON) - Change batch packet processing to collect marker removal IDs and push one remove_markers_batch event instead of individual remove_marker calls - Add callsignIndex reverse map (callsign_group → Set) for O(1) lookup in new_packet/new_packets handlers instead of scanning all markers --- assets/js/map.ts | 144 ++++++++++++------ assets/js/types/map.d.ts | 1 + lib/aprsme/packets.ex | 9 ++ lib/aprsme/packets/prepared_queries.ex | 21 +++ lib/aprsme_web/live/map_live/index.ex | 30 ++-- .../live/map_live/packet_processor.ex | 31 ++-- test/aprsme/packets/prepared_queries_test.exs | 76 +++++++++ .../live/map_live/packet_processor_test.exs | 41 ++++- 8 files changed, 273 insertions(+), 80 deletions(-) diff --git a/assets/js/map.ts b/assets/js/map.ts index 5d967b5..a6c0da7 100644 --- a/assets/js/map.ts +++ b/assets/js/map.ts @@ -454,6 +454,9 @@ let MapAPRSMap = { // Track marker states to prevent unnecessary operations self.markerStates = new Map(); + // Reverse index: callsign_group → Set of marker IDs for O(1) lookup + self.callsignIndex = new Map>(); + // Store RF path lines for hover visualization self.rfPathLines = []; @@ -1043,7 +1046,7 @@ let MapAPRSMap = { data.callsign_group || data.callsign || data.id; if (incomingCallsign) { - // Find existing live markers for this callsign and convert them to historical dots + // Find existing live markers for this callsign using reverse index for O(1) lookup const markersToConvert: string[] = []; // Ensure markerStates exists @@ -1052,20 +1055,19 @@ let MapAPRSMap = { return; } - self.markerStates.forEach((state, id) => { - const stateCallsign = state.callsign_group || state.callsign; - // Convert markers for the same callsign that are either live or were the - // most-recent marker (historical loading marks all packets as historical, - // so is_most_recent_for_callsign catches the "current" marker that was - // loaded from history) - if ( - stateCallsign === incomingCallsign && - (!state.historical || state.is_most_recent_for_callsign) && - String(id) !== String(data.id) - ) { - markersToConvert.push(String(id)); + const existingIds = self.callsignIndex?.get(incomingCallsign); + if (existingIds) { + for (const id of existingIds) { + const state = self.markerStates.get(id); + if ( + state && + (!state.historical || state.is_most_recent_for_callsign) && + String(id) !== String(data.id) + ) { + markersToConvert.push(String(id)); + } } - }); + } // Convert existing live markers to historical dots by updating their icon markersToConvert.forEach((id) => { @@ -1146,46 +1148,39 @@ let MapAPRSMap = { const serverConvertKeys = payload.convert_to_historical; let markersToConvert: string[] = []; + // Build set of new packet IDs to avoid converting a marker we're about to add + const allNewIds = new Set(); + for (const data of packets) { + if (data.id) allNewIds.add(String(data.id)); + } + + // Determine which callsigns need conversion + const callsignsToConvert: Set = new Set(); if (serverConvertKeys && serverConvertKeys.length > 0) { - // Server told us which callsigns have replacements — find their - // current marker IDs by scanning only the provided callsigns - const convertSet = new Set(serverConvertKeys); - const allNewIds = new Set(); - for (const data of packets) { - if (data.id) allNewIds.add(String(data.id)); - } - self.markerStates.forEach((state, id) => { - const stateCallsign = state.callsign_group || state.callsign; - if ( - stateCallsign && - convertSet.has(stateCallsign) && - (!state.historical || state.is_most_recent_for_callsign) && - !allNewIds.has(String(id)) - ) { - markersToConvert.push(String(id)); - } - }); + for (const cs of serverConvertKeys) callsignsToConvert.add(cs); } else { - // Fallback: scan all markerStates (e.g. for batches with no replacements) - 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)); + if (cs) callsignsToConvert.add(cs); } + } - 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)); + // Use callsignIndex for O(1) lookup per callsign instead of scanning all markerStates + if (self.callsignIndex) { + for (const cs of callsignsToConvert) { + const ids = self.callsignIndex.get(cs); + if (!ids) continue; + for (const id of ids) { + if (allNewIds.has(String(id))) continue; + const state = self.markerStates.get(id); + if ( + state && + (!state.historical || state.is_most_recent_for_callsign) + ) { + markersToConvert.push(String(id)); + } } - }); + } } // Phase 2: Convert all existing live markers to historical dots @@ -1871,6 +1866,17 @@ let MapAPRSMap = { callsign: data.callsign, }); + // Update callsign reverse index + const csGroup = data.callsign_group || data.callsign; + if (csGroup && self.callsignIndex) { + let ids = self.callsignIndex.get(csGroup); + if (!ids) { + ids = new Set(); + self.callsignIndex.set(csGroup, ids); + } + ids.add(data.id); + } + // Initialize trail for new marker - always add to trail for line drawing if (self.trailManager) { const isHistoricalDot = @@ -1952,6 +1958,18 @@ let MapAPRSMap = { } } + // Update callsign reverse index + if (markerState && self.callsignIndex) { + const csGroup = markerState.callsign_group || markerState.callsign; + if (csGroup) { + const ids = self.callsignIndex.get(csGroup); + if (ids) { + ids.delete(markerId); + if (ids.size === 0) self.callsignIndex.delete(csGroup); + } + } + } + // Remove trail - use callsign_group for proper trail identification if (self.trailManager) { const trailId = @@ -2052,6 +2070,7 @@ let MapAPRSMap = { // Replace the markers and states with just the preserved ones self.markers!.clear(); self.markerStates!.clear(); + if (self.callsignIndex) self.callsignIndex.clear(); markersToPreserve.forEach((marker, id) => { self.markers!.set(id, marker); @@ -2059,6 +2078,19 @@ let MapAPRSMap = { statesToPreserve.forEach((state, id) => { self.markerStates!.set(id, state); + + // Rebuild callsign index from preserved markers + if (self.callsignIndex) { + const csGroup = state.callsign_group || state.callsign; + if (csGroup) { + let ids = self.callsignIndex.get(csGroup); + if (!ids) { + ids = new Set(); + self.callsignIndex.set(csGroup, ids); + } + ids.add(id); + } + } }); // Don't clear trails - keep them visible @@ -2143,6 +2175,9 @@ let MapAPRSMap = { const marker = self.markers!.get(markerId); if (marker) { + // Get state before deletion for index cleanup + const state = self.markerStates!.get(markerId); + try { // Remove marker from appropriate layer with safety checks if ( @@ -2175,6 +2210,18 @@ let MapAPRSMap = { self.markers!.delete(markerId); self.markerStates!.delete(markerId); } + + // Update callsign reverse index + if (state && self.callsignIndex) { + const csGroup = state.callsign_group || state.callsign; + if (csGroup) { + const ids = self.callsignIndex.get(csGroup); + if (ids) { + ids.delete(markerId); + if (ids.size === 0) self.callsignIndex.delete(csGroup); + } + } + } } }, @@ -2381,6 +2428,9 @@ let MapAPRSMap = { if (self.markerStates !== undefined) { self.markerStates!.clear(); } + if (self.callsignIndex !== undefined) { + self.callsignIndex!.clear(); + } if (self.map !== undefined) { self.map!.remove(); self.map = undefined; diff --git a/assets/js/types/map.d.ts b/assets/js/types/map.d.ts index 5ab9cbf..7f2c48f 100644 --- a/assets/js/types/map.d.ts +++ b/assets/js/types/map.d.ts @@ -12,6 +12,7 @@ export interface LiveViewHookContext { map?: LeafletMap; markers?: Map; markerStates?: Map; + callsignIndex?: Map>; markerLayer?: LayerGroup; heatLayer?: HeatLayer; trailManager?: TrailManager; diff --git a/lib/aprsme/packets.ex b/lib/aprsme/packets.ex index fc75846..644c4db 100644 --- a/lib/aprsme/packets.ex +++ b/lib/aprsme/packets.ex @@ -875,6 +875,15 @@ defmodule Aprsme.Packets do PreparedQueries.get_latest_positions_for_callsigns(callsigns) end + @doc """ + Gets the latest full packet for each callsign in a single batch query. + Returns a list of `Packet` structs with lat/lon populated. + """ + @spec get_latest_packets_for_callsigns(list(String.t())) :: [Packet.t()] + def get_latest_packets_for_callsigns(callsigns) when is_list(callsigns) do + PreparedQueries.get_latest_packets_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 a97076d..4ef6654 100644 --- a/lib/aprsme/packets/prepared_queries.ex +++ b/lib/aprsme/packets/prepared_queries.ex @@ -27,6 +27,27 @@ defmodule Aprsme.Packets.PreparedQueries do ) end + @doc """ + Get the latest full packet for each callsign in a single query. + Returns a list of `Packet` structs with lat/lon populated. + Uses DISTINCT ON to get only the most recent packet per callsign. + """ + @spec get_latest_packets_for_callsigns(list(String.t())) :: [Packet.t()] + def get_latest_packets_for_callsigns([]), do: [] + + def get_latest_packets_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, + distinct: fragment("upper(?)", p.sender), + order_by: [asc: fragment("upper(?)", p.sender), desc: p.received_at], + select: %{p | lat: fragment("ST_Y(?)", p.location), lon: fragment("ST_X(?)", p.location)} + ) + ) + end + @doc """ Get latest positions for multiple callsigns in a single query. Returns a list of `%{callsign: String.t(), lat: float(), lng: float()}` maps. diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index 98d06a2..77b7a1f 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -850,14 +850,23 @@ defmodule AprsmeWeb.MapLive.Index do end def handle_info({:packet_batch, packets}, socket) do - # 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) + # Process batch: collect marker data and removal IDs without pushing events + {socket, marker_data_list, removal_ids} = + Enum.reduce(packets, {socket, [], []}, fn packet, {acc_socket, markers, removals} -> + {new_socket, marker_data, removed_id} = process_packet_for_batch(packet, acc_socket) markers = if marker_data, do: [marker_data | markers], else: markers - {new_socket, markers} + removals = if removed_id, do: [removed_id | removals], else: removals + {new_socket, markers, removals} end) + # Push a single batched remove event for all removals + socket = + if removal_ids == [] do + socket + else + DisplayManager.remove_markers_batch(socket, removal_ids) + end + socket = assign(socket, :last_update_at, DateTime.utc_now()) # Send heat map update once if in heat map mode, otherwise batch markers @@ -930,13 +939,8 @@ defmodule AprsmeWeb.MapLive.Index do end def handle_info({:load_rf_path_station_packets, stations}, socket) do - # Load the most recent packet for each RF path station - station_packets = - stations - |> Enum.map(fn callsign -> - Packets.get_latest_packet_for_callsign(callsign) - end) - |> Enum.filter(& &1) + # Load the most recent packet for each RF path station in a single batch query + station_packets = Packets.get_latest_packets_for_callsigns(stations) socket = if Enum.any?(station_packets) do @@ -1094,7 +1098,7 @@ defmodule AprsmeWeb.MapLive.Index do socket = assign(socket, :tracked_callsign_latest_packet, packet) PacketProcessor.process_packet_for_marker_data(packet, socket) else - {socket, nil} + {socket, nil, nil} end end end diff --git a/lib/aprsme_web/live/map_live/packet_processor.ex b/lib/aprsme_web/live/map_live/packet_processor.ex index 7902ea8..455fc8c 100644 --- a/lib/aprsme_web/live/map_live/packet_processor.ex +++ b/lib/aprsme_web/live/map_live/packet_processor.ex @@ -36,10 +36,12 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do @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. + events. Returns `{socket, marker_data | nil, removed_id | nil}` where + marker_data is the data that should be included in a batched push_event, + and removed_id is a callsign key that was removed from visible_packets + (for batched remove_markers_batch events). """ - @spec process_packet_for_marker_data(map(), Socket.t()) :: {Socket.t(), map() | nil} + @spec process_packet_for_marker_data(map(), Socket.t()) :: {Socket.t(), map() | nil, String.t() | 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) @@ -116,7 +118,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do end # Batch-mode visibility: same logic as visibility_action but returns - # {socket, marker_data | nil} without pushing events + # {socket, marker_data | nil, removed_id | 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) @@ -124,19 +126,21 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do end defp batch_visibility_action(_packet, _lat, _lon, _callsign_key, socket) do - {socket, nil} + {socket, nil, nil} end - # Out of bounds but has marker — remove it (still pushes remove_marker since - # removals are rare and don't benefit from batching) + # Out of bounds but has marker — remove state only, return the removed key + # so the caller can batch all removals into one push_event defp batch_dispatch({false, true}, _packet, _lat, _lon, callsign_key, socket) do - socket = remove_marker_from_map(callsign_key, socket) - {socket, nil} + new_visible_packets = Map.delete(socket.assigns.visible_packets, callsign_key) + socket = assign(socket, :visible_packets, new_visible_packets) + {socket, nil, callsign_key} 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) + {socket, marker_data} = prepare_packet_state(packet, socket) + {socket, marker_data, nil} end # In bounds, existing marker — update if significant movement @@ -146,15 +150,16 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do 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) + {socket, marker_data} = prepare_packet_state(packet, socket) + {socket, marker_data, nil} else new_visible_packets = Map.put(socket.assigns.visible_packets, callsign_key, packet) - {assign(socket, :visible_packets, new_visible_packets), nil} + {assign(socket, :visible_packets, new_visible_packets), nil, nil} end end defp batch_dispatch(_, _packet, _lat, _lon, _callsign_key, socket) do - {socket, nil} + {socket, nil, nil} end # Shared: updates visible_packets state and builds marker data. diff --git a/test/aprsme/packets/prepared_queries_test.exs b/test/aprsme/packets/prepared_queries_test.exs index 63babb1..dafd6ef 100644 --- a/test/aprsme/packets/prepared_queries_test.exs +++ b/test/aprsme/packets/prepared_queries_test.exs @@ -24,6 +24,82 @@ defmodule Aprsme.Packets.PreparedQueriesTest do |> Repo.insert() end + describe "get_latest_packets_for_callsigns/1" do + test "returns full packet structs 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"), + comment: "Digi TX" + }) + + {:ok, _} = + create_positioned_packet(%{ + sender: "N5TXZ-10", + base_callsign: "N5TXZ", + ssid: "10", + lat: Decimal.new("33.2000"), + lon: Decimal.new("-96.5000"), + comment: "iGate" + }) + + result = PreparedQueries.get_latest_packets_for_callsigns(["K5GVL-10", "N5TXZ-10"]) + + assert length(result) == 2 + assert Enum.all?(result, &is_struct(&1, Packet)) + + senders = Enum.map(result, & &1.sender) + assert "K5GVL-10" in senders + assert "N5TXZ-10" in senders + + k5gvl = Enum.find(result, &(&1.sender == "K5GVL-10")) + assert_in_delta k5gvl.lat, 33.1, 0.01 + assert_in_delta k5gvl.lon, -96.6, 0.01 + end + + test "returns empty list for empty input" do + assert PreparedQueries.get_latest_packets_for_callsigns([]) == [] + end + + test "returns empty list for nonexistent callsigns" do + result = PreparedQueries.get_latest_packets_for_callsigns(["NONEXIST-1", "FAKE-2"]) + assert result == [] + end + + test "returns latest packet when multiple 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_packets_for_callsigns(["K5GVL-10"]) + + assert length(result) == 1 + packet = hd(result) + assert is_struct(packet, Packet) + assert_in_delta packet.lat, 34.0, 0.01 + assert_in_delta packet.lon, -97.0, 0.01 + end + end + describe "get_latest_positions_for_callsigns/1" do test "returns positions for multiple callsigns" do {:ok, _} = diff --git a/test/aprsme_web/live/map_live/packet_processor_test.exs b/test/aprsme_web/live/map_live/packet_processor_test.exs index d275065..89ec831 100644 --- a/test/aprsme_web/live/map_live/packet_processor_test.exs +++ b/test/aprsme_web/live/map_live/packet_processor_test.exs @@ -44,7 +44,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessorTest do socket = build_socket() packet = build_packet() - {updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + {updated_socket, marker_data, _removed} = PacketProcessor.process_packet_for_marker_data(packet, socket) assert marker_data assert is_map(marker_data) @@ -57,7 +57,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessorTest do # 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) + {updated_socket, marker_data, _removed} = PacketProcessor.process_packet_for_marker_data(packet, socket) assert marker_data == nil assert map_size(updated_socket.assigns.visible_packets) == 0 @@ -67,7 +67,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessorTest do socket = build_socket() packet = build_packet(%{lat: nil, lon: nil}) - {updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + {updated_socket, marker_data, _removed} = PacketProcessor.process_packet_for_marker_data(packet, socket) assert marker_data == nil assert map_size(updated_socket.assigns.visible_packets) == 0 @@ -77,7 +77,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessorTest do socket = build_socket(%{map_zoom: 7}) packet = build_packet() - {_updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + {_updated_socket, marker_data, _removed} = PacketProcessor.process_packet_for_marker_data(packet, socket) assert marker_data == nil end @@ -86,7 +86,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessorTest do socket = build_socket() packet = build_packet() - {updated_socket, _marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + {updated_socket, _marker_data, _removed} = 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] @@ -104,7 +104,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessorTest do # Second packet for same callsign (also no :id so key = sender) second_packet = %{lat: 33.2, lon: -96.4} |> build_packet() |> Map.delete(:id) - {_updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(second_packet, socket) + {_updated_socket, marker_data, _removed} = PacketProcessor.process_packet_for_marker_data(second_packet, socket) assert marker_data assert marker_data["convert_from"] == "K5GVL-10" @@ -114,10 +114,37 @@ defmodule AprsmeWeb.MapLive.PacketProcessorTest do socket = build_socket() packet = build_packet() - {_updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket) + {_updated_socket, marker_data, _removed} = PacketProcessor.process_packet_for_marker_data(packet, socket) assert marker_data refute Map.has_key?(marker_data, "convert_from") end + + test "returns removed callsign key when out-of-bounds packet had existing marker" do + # Set up socket with an existing visible packet + existing_packet = %{lat: 33.1, lon: -96.5} |> build_packet() |> Map.delete(:id) + socket = build_socket(%{visible_packets: %{"K5GVL-10" => existing_packet}}) + + # New packet from same callsign is out of bounds + out_of_bounds_packet = %{lat: 40.0, lon: -80.0} |> build_packet() |> Map.delete(:id) + + {updated_socket, marker_data, removed_id} = + PacketProcessor.process_packet_for_marker_data(out_of_bounds_packet, socket) + + assert marker_data == nil + assert removed_id == "K5GVL-10" + # The marker should be removed from visible_packets + refute Map.has_key?(updated_socket.assigns.visible_packets, "K5GVL-10") + end + + test "returns nil removed_id for normal in-bounds packets" do + socket = build_socket() + packet = build_packet() + + {_updated_socket, _marker_data, removed_id} = + PacketProcessor.process_packet_for_marker_data(packet, socket) + + assert removed_id == nil + end end end