Batch RF path queries, batch marker removals, and JS callsign index

- 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<markerID>) for O(1)
  lookup in new_packet/new_packets handlers instead of scanning all markers

assets/js/map.ts

@@ -454,6 +454,9 @@ let MapAPRSMap = {
     // Track marker states to prevent unnecessary operations
     self.markerStates = new Map<string, MarkerState>();
 
+    // Reverse index: callsign_group → Set of marker IDs for O(1) lookup
+    self.callsignIndex = new Map<string, Set<string>>();
+
     // 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<string>();
+          for (const data of packets) {
+            if (data.id) allNewIds.add(String(data.id));
+          }
+
+          // Determine which callsigns need conversion
+          const callsignsToConvert: Set<string> = 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<string>();
-            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<string>();
-            const allNewIds = new Set<string>();
             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<string>();
+        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<string>();
+            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;

assets/js/types/map.d.ts

@@ -12,6 +12,7 @@ export interface LiveViewHookContext {
   map?: LeafletMap;
   markers?: Map<string, Marker>;
   markerStates?: Map<string, MarkerState>;
+  callsignIndex?: Map<string, Set<string>>;
   markerLayer?: LayerGroup;
   heatLayer?: HeatLayer;
   trailManager?: TrailManager;

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.

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.

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

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.

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, _} =

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