performance improvements

lib/aprsme/application.ex

@@ -33,6 +33,12 @@ defmodule Aprsme.Application do
       Aprsme.CircuitBreaker,
       # Start device cache manager
       Aprsme.DeviceCache,
+      # Start system monitor for adaptive performance tuning
+      Aprsme.SystemMonitor,
+      # Start spatial PubSub for viewport-based filtering
+      Aprsme.SpatialPubSub,
+      # Start packet store for efficient LiveView memory usage
+      AprsmeWeb.MapLive.PacketStore,
 
       # Start the Endpoint (http/https)
       AprsmeWeb.Endpoint,

lib/aprsme/cached_queries.ex

@@ -48,7 +48,8 @@ defmodule Aprsme.CachedQueries do
   end
 
   @doc """
-  Get packet count with caching
+  Get packet count with caching.
+  Now uses the efficient packet_counters table for O(1) performance.
   """
   def get_total_packet_count_cached do
     cache_key = "total_packet_count"
@@ -58,8 +59,10 @@ defmodule Aprsme.CachedQueries do
         result
 
       _ ->
+        # This is now extremely fast due to the counter table
         result = Packets.get_total_packet_count()
-        Cachex.put(:query_cache, cache_key, result, ttl: @cache_ttl_medium)
+        # Cache for only 5 seconds since the query is now instant
+        Cachex.put(:query_cache, cache_key, result, ttl: to_timeout(second: 5))
         result
     end
   end

lib/aprsme/packet_consumer.ex

@@ -16,7 +16,8 @@ defmodule Aprsme.PacketConsumer do
 
   @impl true
   def init(opts) do
-    batch_size = opts[:batch_size] || 100
+    # Use dynamic batch sizing from system monitor
+    initial_batch_size = Aprsme.SystemMonitor.get_recommended_batch_size()
     batch_timeout = opts[:batch_timeout] || 1000
     # Maximum batch size to prevent unbounded memory growth
     max_batch_size = opts[:max_batch_size] || 1000
@@ -24,18 +25,26 @@ defmodule Aprsme.PacketConsumer do
     # Start a timer for batch processing
     timer = Process.send_after(self(), :process_batch, batch_timeout)
 
+    # Schedule periodic batch size adjustment
+    Process.send_after(self(), :adjust_batch_size, 10_000)
+
     {:consumer,
      %{
        batch: [],
-       batch_size: batch_size,
+       batch_size: initial_batch_size,
        batch_timeout: batch_timeout,
        max_batch_size: max_batch_size,
-       timer: timer
+       timer: timer,
+       last_adjustment: System.monotonic_time(:millisecond)
      }}
   end
 
   @impl true
-  def handle_events(events, _from, %{batch: batch, batch_size: batch_size, max_batch_size: max_batch_size} = state) do
+  def handle_events(events, _from, %{batch: batch, batch_size: _batch_size, max_batch_size: max_batch_size} = state) do
+    # Get current recommended batch size
+    current_batch_size = Aprsme.SystemMonitor.get_recommended_batch_size()
+    state = %{state | batch_size: current_batch_size}
+
     new_batch = batch ++ events
     new_batch_length = length(new_batch)
 
@@ -59,7 +68,7 @@ defmodule Aprsme.PacketConsumer do
 
         {:noreply, [], %{state | batch: []}}
 
-      new_batch_length >= batch_size ->
+      new_batch_length >= current_batch_size ->
         # Process the batch immediately
         process_batch(new_batch)
         {:noreply, [], %{state | batch: []}}
@@ -93,6 +102,28 @@ defmodule Aprsme.PacketConsumer do
     {:noreply, [], %{state | batch: [], timer: timer}}
   end
 
+  @impl true
+  def handle_info(:adjust_batch_size, state) do
+    # Get current system metrics and recommended batch size
+    new_batch_size = Aprsme.SystemMonitor.get_recommended_batch_size()
+
+    if new_batch_size != state.batch_size do
+      Logger.info("Adjusting batch size based on system load",
+        batch_adjustment:
+          LogSanitizer.log_data(
+            old_size: state.batch_size,
+            new_size: new_batch_size,
+            reason: "system_load_adaptation"
+          )
+      )
+    end
+
+    # Schedule next adjustment
+    Process.send_after(self(), :adjust_batch_size, 10_000)
+
+    {:noreply, [], %{state | batch_size: new_batch_size}}
+  end
+
   defp process_batch(packets) do
     require Logger
 

lib/aprsme/packet_counter.ex

@@ -0,0 +1,43 @@
+defmodule Aprsme.PacketCounter do
+  @moduledoc """
+  Schema for the packet_counters table that provides O(1) packet counting.
+  """
+  use Ecto.Schema
+
+  import Ecto.Query
+
+  alias Aprsme.Repo
+
+  schema "packet_counters" do
+    field :counter_type, :string
+    field :count, :integer, default: 0
+
+    timestamps()
+  end
+
+  @doc """
+  Gets the current packet count instantly.
+  This is an O(1) operation thanks to the counter table.
+  """
+  def get_count(counter_type \\ "total_packets") do
+    from(pc in __MODULE__,
+      where: pc.counter_type == ^counter_type,
+      select: pc.count
+    )
+    |> Repo.one()
+    |> case do
+      nil -> 0
+      count -> count
+    end
+  end
+
+  @doc """
+  Subscribe to packet count changes via PostgreSQL LISTEN/NOTIFY.
+  This allows real-time updates without polling.
+  """
+  def subscribe_to_changes do
+    # This would require setting up PostgreSQL LISTEN/NOTIFY
+    # For now, we'll rely on the existing PubSub mechanism
+    :ok
+  end
+end

lib/aprsme/packets.ex

@@ -486,21 +486,29 @@ defmodule Aprsme.Packets do
     hours_back = Map.get(opts, :hours_back, 1)
     cutoff_time = TimeUtils.hours_ago(hours_back)
 
-    # Use a CTE with window function to get the most recent packet per callsign
-    # This approach maintains proper distance ordering while deduplicating by callsign
+    # Use named prepared statement for better performance (future optimization)
+    # _statement_name = if exclude_callsign, do: "nearby_stations_exclude", else: "nearby_stations"
+
+    # Use DISTINCT ON for better performance than window functions
+    # First get most recent packet per callsign, then order by distance
+    # Add spatial bounds to use indexes effectively
     query = """
-    WITH ranked_packets AS (
-      SELECT p.*,
-             ST_Distance(p.location::geography, ST_SetSRID(ST_MakePoint($2, $1), 4326)::geography) as distance,
-             ROW_NUMBER() OVER (PARTITION BY p.base_callsign ORDER BY p.received_at DESC) as row_num
+    SELECT * FROM (
+      SELECT DISTINCT ON (p.base_callsign)
+        p.*,
+        ST_Distance(p.location::geography, ST_SetSRID(ST_MakePoint($2, $1), 4326)::geography) as distance
       FROM packets p
       WHERE p.has_position = true
         AND p.received_at >= $3
+        AND p.location IS NOT NULL
         #{if exclude_callsign, do: "AND p.sender != $4", else: ""}
-    )
-    SELECT *
-    FROM ranked_packets
-    WHERE row_num = 1
+        AND ST_DWithin(
+          p.location::geography,
+          ST_SetSRID(ST_MakePoint($2, $1), 4326)::geography,
+          100000  -- 100km radius for initial filtering
+        )
+      ORDER BY p.base_callsign, p.received_at DESC
+    ) AS recent_packets
     ORDER BY distance ASC
     LIMIT #{if exclude_callsign, do: "$5", else: "$4"}
     """
@@ -580,11 +588,32 @@ defmodule Aprsme.Packets do
 
   @doc """
   Gets the total count of stored packets in the database.
+  Uses the efficient packet_counters table with triggers for O(1) performance.
   """
   @spec get_total_packet_count() :: non_neg_integer()
   def get_total_packet_count do
-    # Use the new index for faster counting
-    Repo.one(from p in Packet, select: count(p.id)) || 0
+    # Use the PostgreSQL function for instant count retrieval
+    case Repo.query("SELECT get_packet_count()", []) do
+      {:ok, %{rows: [[count]]}} when is_integer(count) ->
+        count
+
+      {:ok, %{rows: [[nil]]}} ->
+        # Fallback to actual count if counter is not initialized
+        Repo.one(from p in Packet, select: count(p.id)) || 0
+
+      {:error, _} ->
+        # Try the faster counter table directly as a fallback
+        query = """
+        SELECT count FROM packet_counters 
+        WHERE counter_type = 'total_packets'
+        LIMIT 1
+        """
+
+        case Repo.query(query, []) do
+          {:ok, %{rows: [[count]]}} -> count
+          _ -> 0
+        end
+    end
   rescue
     DBConnection.ConnectionError ->
       require Logger

lib/aprsme/spatial_pubsub.ex

@@ -0,0 +1,396 @@
+defmodule Aprsme.SpatialPubSub do
+  @moduledoc """
+  Spatial-aware PubSub system that broadcasts packets only to clients
+  whose viewports contain the packet's location.
+  """
+  use GenServer
+
+  alias Phoenix.PubSub
+
+  require Logger
+
+  # Grid size in degrees for spatial indexing
+  @grid_size 1.0
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @doc """
+  Register a client with their current viewport bounds.
+  """
+  def register_viewport(client_id, bounds) do
+    GenServer.call(__MODULE__, {:register_viewport, client_id, bounds})
+  end
+
+  @doc """
+  Update a client's viewport bounds.
+  """
+  def update_viewport(client_id, bounds) do
+    GenServer.call(__MODULE__, {:update_viewport, client_id, bounds})
+  end
+
+  @doc """
+  Unregister a client.
+  """
+  def unregister_client(client_id) do
+    GenServer.cast(__MODULE__, {:unregister_client, client_id})
+  end
+
+  @doc """
+  Broadcast a packet to all clients whose viewports contain the packet's location.
+  """
+  def broadcast_packet(packet) do
+    GenServer.cast(__MODULE__, {:broadcast_packet, packet})
+  end
+
+  @doc """
+  Get statistics about spatial filtering.
+  """
+  def get_stats do
+    GenServer.call(__MODULE__, :get_stats)
+  end
+
+  @doc """
+  Start telemetry reporting for LiveDashboard integration.
+  """
+  def start_telemetry_reporting do
+    GenServer.cast(__MODULE__, :start_telemetry_reporting)
+  end
+
+  # Server callbacks
+
+  @impl true
+  def init(_opts) do
+    # Subscribe to packet notifications
+    PubSub.subscribe(Aprsme.PubSub, "postgres:aprsme_packets")
+
+    # Start telemetry reporting
+    Process.send_after(self(), :report_telemetry, 5_000)
+
+    state = %{
+      # client_id => %{bounds: bounds, topic: topic, pid: pid}
+      clients: %{},
+      # grid_key => MapSet of client_ids
+      spatial_index: %{},
+      # Statistics
+      stats: %{
+        total_broadcasts: 0,
+        filtered_broadcasts: 0,
+        total_packets: 0,
+        clients_count: 0
+      }
+    }
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_call({:register_viewport, client_id, bounds}, {pid, _}, state) do
+    # Create a unique topic for this client
+    topic = "spatial:#{client_id}"
+
+    # Monitor the client process
+    Process.monitor(pid)
+
+    # Update client info
+    client_info = %{
+      bounds: normalize_bounds(bounds),
+      topic: topic,
+      pid: pid
+    }
+
+    # Update spatial index
+    new_state =
+      state
+      |> put_in([:clients, client_id], client_info)
+      |> update_spatial_index(client_id, client_info.bounds)
+      |> update_in([:stats, :clients_count], &(&1 + 1))
+
+    {:reply, {:ok, topic}, new_state}
+  end
+
+  @impl true
+  def handle_call({:update_viewport, client_id, bounds}, _from, state) do
+    case Map.get(state.clients, client_id) do
+      nil ->
+        {:reply, {:error, :not_registered}, state}
+
+      client_info ->
+        # Remove old spatial index entries
+        old_bounds = client_info.bounds
+        state = remove_from_spatial_index(state, client_id, old_bounds)
+
+        # Update with new bounds
+        normalized_bounds = normalize_bounds(bounds)
+        updated_client = %{client_info | bounds: normalized_bounds}
+
+        new_state =
+          state
+          |> put_in([:clients, client_id], updated_client)
+          |> update_spatial_index(client_id, normalized_bounds)
+
+        {:reply, :ok, new_state}
+    end
+  end
+
+  @impl true
+  def handle_call(:get_stats, _from, state) do
+    stats =
+      Map.merge(state.stats, %{
+        grid_cells: map_size(state.spatial_index),
+        avg_clients_per_cell: calculate_avg_clients_per_cell(state.spatial_index)
+      })
+
+    {:reply, stats, state}
+  end
+
+  @impl true
+  def handle_cast({:unregister_client, client_id}, state) do
+    new_state = remove_client(state, client_id)
+    {:noreply, new_state}
+  end
+
+  @impl true
+  def handle_cast({:broadcast_packet, packet}, state) do
+    state = update_in(state, [:stats, :total_packets], &(&1 + 1))
+
+    case extract_location(packet) do
+      {lat, lon} when is_number(lat) and is_number(lon) ->
+        # Find all clients whose viewports contain this location
+        client_ids = find_clients_for_location(state, lat, lon)
+
+        # Broadcast to each matching client's topic
+        Enum.each(client_ids, fn client_id ->
+          case Map.get(state.clients, client_id) do
+            %{topic: topic} ->
+              PubSub.broadcast(Aprsme.PubSub, topic, {:spatial_packet, packet})
+
+            _ ->
+              :ok
+          end
+        end)
+
+        # Update statistics
+        state =
+          state
+          |> update_in([:stats, :total_broadcasts], &(&1 + length(client_ids)))
+          |> update_in([:stats, :filtered_broadcasts], &(&1 + max(0, map_size(state.clients) - length(client_ids))))
+
+        {:noreply, state}
+
+      _ ->
+        # No valid location, skip broadcasting
+        {:noreply, state}
+    end
+  end
+
+  @impl true
+  def handle_info({:postgres_packet, packet}, state) do
+    # Handle packets from Postgres notifications
+    handle_cast({:broadcast_packet, packet}, state)
+  end
+
+  @impl true
+  def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
+    # Find and remove the client associated with this pid
+    client_id =
+      Enum.find_value(state.clients, fn {id, %{pid: client_pid}} ->
+        if client_pid == pid, do: id
+      end)
+
+    new_state =
+      if client_id do
+        remove_client(state, client_id)
+      else
+        state
+      end
+
+    {:noreply, new_state}
+  end
+
+  @impl true
+  def handle_info(:report_telemetry, state) do
+    # Emit telemetry metrics
+    emit_telemetry_metrics(state)
+
+    # Schedule next report
+    Process.send_after(self(), :report_telemetry, 5_000)
+
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info(_, state), do: {:noreply, state}
+
+  # Private functions
+
+  defp normalize_bounds(%{north: n, south: s, east: e, west: w}) do
+    %{
+      north: ensure_float(n),
+      south: ensure_float(s),
+      east: ensure_float(e),
+      west: ensure_float(w)
+    }
+  end
+
+  defp ensure_float(val) when is_binary(val), do: String.to_float(val)
+  defp ensure_float(val) when is_integer(val), do: val * 1.0
+  defp ensure_float(val) when is_float(val), do: val
+  defp ensure_float(_), do: 0.0
+
+  defp update_spatial_index(state, client_id, bounds) do
+    # Get all grid cells that intersect with the bounds
+    grid_cells = get_intersecting_grid_cells(bounds)
+
+    # Add client to each grid cell
+    Enum.reduce(grid_cells, state, fn grid_key, acc_state ->
+      update_in(acc_state, [:spatial_index, grid_key], fn
+        nil -> MapSet.new([client_id])
+        set -> MapSet.put(set, client_id)
+      end)
+    end)
+  end
+
+  defp remove_from_spatial_index(state, client_id, bounds) do
+    grid_cells = get_intersecting_grid_cells(bounds)
+
+    Enum.reduce(grid_cells, state, fn grid_key, acc_state ->
+      case Map.get(acc_state.spatial_index, grid_key) do
+        nil ->
+          acc_state
+
+        set ->
+          new_set = MapSet.delete(set, client_id)
+
+          if MapSet.size(new_set) == 0 do
+            # Remove the key entirely instead of storing nil
+            update_in(acc_state, [:spatial_index], &Map.delete(&1, grid_key))
+          else
+            put_in(acc_state, [:spatial_index, grid_key], new_set)
+          end
+      end
+    end)
+  end
+
+  defp get_intersecting_grid_cells(%{north: n, south: s, east: e, west: w}) do
+    # Calculate which grid cells intersect with the bounds
+    min_lat_cell = floor(s / @grid_size)
+    max_lat_cell = floor(n / @grid_size)
+    min_lon_cell = floor(w / @grid_size)
+    max_lon_cell = floor(e / @grid_size)
+
+    for lat_cell <- min_lat_cell..max_lat_cell,
+        lon_cell <- min_lon_cell..max_lon_cell do
+      {lat_cell, lon_cell}
+    end
+  end
+
+  defp find_clients_for_location(state, lat, lon) do
+    # Get the grid cell for this location
+    grid_key = {floor(lat / @grid_size), floor(lon / @grid_size)}
+
+    # Get all clients in this grid cell
+    case Map.get(state.spatial_index, grid_key) do
+      nil ->
+        []
+
+      client_set ->
+        # Filter to only clients whose bounds actually contain the point
+        Enum.filter(client_set, fn client_id ->
+          case Map.get(state.clients, client_id) do
+            %{bounds: bounds} -> point_in_bounds?(lat, lon, bounds)
+            _ -> false
+          end
+        end)
+    end
+  end
+
+  defp point_in_bounds?(lat, lon, %{north: n, south: s, east: e, west: w}) do
+    lat >= s and lat <= n and lon >= w and lon <= e
+  end
+
+  defp extract_location(packet) do
+    case packet do
+      %{lat: lat, lon: lon} when not is_nil(lat) and not is_nil(lon) ->
+        {ensure_float(lat), ensure_float(lon)}
+
+      _ ->
+        nil
+    end
+  end
+
+  defp remove_client(state, client_id) do
+    case Map.get(state.clients, client_id) do
+      nil ->
+        state
+
+      %{bounds: bounds} ->
+        state
+        |> remove_from_spatial_index(client_id, bounds)
+        |> update_in([:clients], &Map.delete(&1, client_id))
+        |> update_in([:stats, :clients_count], &max(0, &1 - 1))
+    end
+  end
+
+  defp calculate_avg_clients_per_cell(spatial_index) do
+    non_nil_cells =
+      spatial_index
+      |> Map.values()
+      |> Enum.reject(&is_nil/1)
+
+    if Enum.empty?(non_nil_cells) do
+      0.0
+    else
+      total_clients =
+        non_nil_cells
+        |> Enum.map(&MapSet.size/1)
+        |> Enum.sum()
+
+      total_clients / length(non_nil_cells)
+    end
+  end
+
+  defp emit_telemetry_metrics(state) do
+    # Client metrics
+    :telemetry.execute(
+      [:aprsme, :spatial_pubsub, :clients],
+      %{
+        count: state.stats.clients_count,
+        grid_cells: map_size(state.spatial_index),
+        avg_clients_per_cell: calculate_avg_clients_per_cell(state.spatial_index)
+      },
+      %{}
+    )
+
+    # Broadcast metrics
+    :telemetry.execute(
+      [:aprsme, :spatial_pubsub, :broadcasts],
+      %{
+        total: state.stats.total_broadcasts,
+        filtered: state.stats.filtered_broadcasts,
+        packets: state.stats.total_packets
+      },
+      %{}
+    )
+
+    # Calculate and emit efficiency metrics
+    total_potential_broadcasts = state.stats.total_packets * state.stats.clients_count
+
+    efficiency =
+      if total_potential_broadcasts > 0 do
+        1.0 - state.stats.total_broadcasts / total_potential_broadcasts
+      else
+        0.0
+      end
+
+    :telemetry.execute(
+      [:aprsme, :spatial_pubsub, :efficiency],
+      %{
+        ratio: efficiency,
+        saved_broadcasts: state.stats.filtered_broadcasts
+      },
+      %{}
+    )
+  end
+end

lib/aprsme/system_monitor.ex

@@ -0,0 +1,298 @@
+defmodule Aprsme.SystemMonitor do
+  @moduledoc """
+  Monitors system metrics to help with adaptive performance tuning.
+  """
+  use GenServer
+
+  require Logger
+
+  @check_interval 5_000
+  @min_batch_size 50
+  @max_batch_size 500
+  @default_batch_size 100
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def get_recommended_batch_size do
+    GenServer.call(__MODULE__, :get_batch_size)
+  catch
+    :exit, {:noproc, _} -> @default_batch_size
+  end
+
+  def get_metrics do
+    GenServer.call(__MODULE__, :get_metrics)
+  catch
+    :exit, {:noproc, _} -> default_metrics()
+  end
+
+  @impl true
+  def init(_opts) do
+    schedule_check()
+
+    state = %{
+      metrics: default_metrics(),
+      batch_size: @default_batch_size,
+      history: []
+    }
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_call(:get_batch_size, _from, state) do
+    {:reply, state.batch_size, state}
+  end
+
+  @impl true
+  def handle_call(:get_metrics, _from, state) do
+    {:reply, state.metrics, state}
+  end
+
+  @impl true
+  def handle_info(:check_system, state) do
+    metrics = collect_metrics()
+    new_batch_size = calculate_optimal_batch_size(metrics, state)
+
+    # Keep history for trend analysis (last 12 data points = 1 minute)
+    history = Enum.take([metrics | state.history], 12)
+
+    # Emit telemetry events for LiveDashboard
+    emit_telemetry_events(metrics, new_batch_size)
+
+    new_state = %{state | metrics: metrics, batch_size: new_batch_size, history: history}
+
+    schedule_check()
+    {:noreply, new_state}
+  end
+
+  defp schedule_check do
+    Process.send_after(self(), :check_system, @check_interval)
+  end
+
+  defp collect_metrics do
+    # Memory metrics
+    memory_info = :erlang.memory()
+    total_memory = memory_info[:total]
+    process_memory = memory_info[:processes]
+    binary_memory = memory_info[:binary]
+
+    # CPU metrics
+    scheduler_count = :erlang.system_info(:schedulers_online)
+
+    # Parse load averages more robustly
+    load_values =
+      ~c"uptime | awk -F'load average:' '{print $2}'"
+      |> :os.cmd()
+      |> to_string()
+      |> String.trim()
+      |> String.split(",")
+      |> Enum.map(&String.trim/1)
+      |> Enum.map(&parse_float/1)
+
+    # Ensure we have 3 values, defaulting to 0.0 if missing
+    {load1, load5, load15} =
+      case load_values do
+        [l1, l5, l15 | _] -> {l1, l5, l15}
+        [l1, l5] -> {l1, l5, 0.0}
+        [l1] -> {l1, 0.0, 0.0}
+        [] -> {0.0, 0.0, 0.0}
+      end
+
+    # Process metrics
+    process_count = :erlang.system_info(:process_count)
+
+    # Database pool metrics
+    db_pool_status = get_db_pool_status()
+
+    # Calculate memory pressure (0.0 to 1.0)
+    memory_pressure = calculate_memory_pressure(memory_info)
+
+    # Calculate CPU pressure (0.0 to 1.0)
+    cpu_pressure = min(1.0, load1 / scheduler_count)
+
+    %{
+      memory: %{
+        total: total_memory,
+        process: process_memory,
+        binary: binary_memory,
+        pressure: memory_pressure
+      },
+      cpu: %{
+        load1: load1,
+        load5: load5,
+        load15: load15,
+        schedulers: scheduler_count,
+        pressure: cpu_pressure
+      },
+      processes: %{
+        count: process_count,
+        pressure: min(1.0, process_count / 50_000)
+      },
+      db_pool: db_pool_status,
+      timestamp: DateTime.utc_now()
+    }
+  end
+
+  defp parse_float(str) do
+    case Float.parse(str) do
+      {float, _} -> float
+      :error -> 0.0
+    end
+  end
+
+  defp calculate_memory_pressure(memory_info) do
+    # Get app memory for pressure calculation
+    app_memory = memory_info[:total]
+
+    # Assume 4GB available memory as baseline
+    available_memory = 4 * 1024 * 1024 * 1024
+
+    # Calculate pressure based on usage
+    min(1.0, app_memory / available_memory)
+  end
+
+  defp get_db_pool_status do
+    pool_config = Aprsme.Repo.config()[:pool_size] || 10
+
+    # Get pool telemetry if available
+    :telemetry.execute([:aprsme, :repo, :pool], %{}, %{})
+
+    %{
+      size: pool_config,
+      # Would need actual telemetry
+      available: pool_config,
+      # Placeholder
+      pressure: 0.3
+    }
+  rescue
+    _ -> %{size: 10, available: 7, pressure: 0.3}
+  end
+
+  defp calculate_optimal_batch_size(metrics, state) do
+    # Base factors
+    memory_factor = 1.0 - metrics.memory.pressure
+    cpu_factor = 1.0 - metrics.cpu.pressure
+    db_factor = 1.0 - metrics.db_pool.pressure
+
+    # Historical trend analysis
+    trend_factor = calculate_trend_factor(state.history)
+
+    # Weighted combination
+    combined_factor =
+      memory_factor * 0.4 +
+        cpu_factor * 0.3 +
+        db_factor * 0.2 +
+        trend_factor * 0.1
+
+    # Calculate new batch size
+    target_size =
+      @min_batch_size +
+        round(combined_factor * (@max_batch_size - @min_batch_size))
+
+    # Apply smoothing to avoid rapid changes
+    current_size = state.batch_size
+    step = round((target_size - current_size) * 0.3)
+    new_size = current_size + step
+
+    # Ensure within bounds
+    new_size
+    |> max(@min_batch_size)
+    |> min(@max_batch_size)
+  end
+
+  defp calculate_trend_factor(history) when length(history) < 3, do: 0.5
+
+  defp calculate_trend_factor(history) do
+    # Analyze recent pressure trends
+    recent_pressures =
+      history
+      |> Enum.take(3)
+      |> Enum.map(fn m ->
+        (m.memory.pressure + m.cpu.pressure + m.db_pool.pressure) / 3
+      end)
+
+    case recent_pressures do
+      [p1, p2, p3] when p1 > p2 and p2 > p3 ->
+        # Pressure increasing, reduce batch size
+        0.2
+
+      [p1, p2, p3] when p1 < p2 and p2 < p3 ->
+        # Pressure decreasing, increase batch size
+        0.8
+
+      _ ->
+        # Stable
+        0.5
+    end
+  end
+
+  defp default_metrics do
+    %{
+      memory: %{total: 0, process: 0, binary: 0, pressure: 0.5},
+      cpu: %{load1: 1.0, load5: 1.0, load15: 1.0, schedulers: 1, pressure: 0.5},
+      processes: %{count: 1000, pressure: 0.5},
+      db_pool: %{size: 10, available: 7, pressure: 0.3},
+      timestamp: DateTime.utc_now()
+    }
+  end
+
+  defp emit_telemetry_events(metrics, batch_size) do
+    # Memory metrics
+    :telemetry.execute(
+      [:aprsme, :system, :memory],
+      %{
+        total: metrics.memory.total,
+        process: metrics.memory.process,
+        binary: metrics.memory.binary,
+        pressure: metrics.memory.pressure
+      },
+      %{}
+    )
+
+    # CPU metrics
+    :telemetry.execute(
+      [:aprsme, :system, :cpu],
+      %{
+        load1: metrics.cpu.load1,
+        load5: metrics.cpu.load5,
+        load15: metrics.cpu.load15,
+        pressure: metrics.cpu.pressure
+      },
+      %{schedulers: metrics.cpu.schedulers}
+    )
+
+    # Process metrics
+    :telemetry.execute(
+      [:aprsme, :system, :processes],
+      %{
+        count: metrics.processes.count,
+        pressure: metrics.processes.pressure
+      },
+      %{}
+    )
+
+    # Database pool metrics
+    :telemetry.execute(
+      [:aprsme, :system, :db_pool],
+      %{
+        size: metrics.db_pool.size,
+        available: metrics.db_pool.available,
+        pressure: metrics.db_pool.pressure
+      },
+      %{}
+    )
+
+    # Batch size metrics
+    :telemetry.execute(
+      [:aprsme, :system, :batch_size],
+      %{
+        current: batch_size,
+        min: @min_batch_size,
+        max: @max_batch_size
+      },
+      %{}
+    )
+  end
+end

lib/aprsme/workers/packet_cleanup_worker.ex

@@ -200,6 +200,4 @@ defmodule Aprsme.Workers.PacketCleanupWorker do
       )
     )
   end
-
-
 end

lib/aprsme_web/live/info_live/show.ex

@@ -243,7 +243,7 @@ defmodule AprsmeWeb.InfoLive.Show do
     # This is much more efficient than fetching 100 packets and filtering in Elixir
     query = """
     WITH recent_ssids AS (
-      SELECT DISTINCT ON (sender) 
+      SELECT DISTINCT ON (sender)
         sender, ssid, received_at, id
       FROM packets
       WHERE base_callsign = $1
@@ -294,7 +294,7 @@ defmodule AprsmeWeb.InfoLive.Show do
     # In APRS, the first station with an asterisk (*) in the path is the one that heard the packet directly
     query = """
     WITH parsed_paths AS (
-      SELECT 
+      SELECT
         id,
         sender,
         path,
@@ -303,7 +303,7 @@ defmodule AprsmeWeb.InfoLive.Show do
         lon,
         location,
         -- Extract the first digipeater from the path
-        CASE 
+        CASE
           WHEN path ~ '[A-Z0-9]+-?[0-9]*\\*' THEN
             substring(path from '([A-Z0-9]+-?[0-9]*)\\*')
           ELSE NULL
@@ -317,25 +317,25 @@ defmodule AprsmeWeb.InfoLive.Show do
         AND path !~ ',TCPIP'
     ),
     digipeater_stats AS (
-      SELECT 
+      SELECT
         first_digipeater as digipeater,
         MIN(received_at) as first_heard,
         MAX(received_at) as last_heard,
         COUNT(*) as packet_count,
         -- Find the packet with maximum distance for this digipeater
-        (SELECT pp2.id 
-         FROM parsed_paths pp2 
-         WHERE pp2.first_digipeater = pp.first_digipeater 
-           AND pp2.lat IS NOT NULL 
+        (SELECT pp2.id
+         FROM parsed_paths pp2
+         WHERE pp2.first_digipeater = pp.first_digipeater
+           AND pp2.lat IS NOT NULL
            AND pp2.lon IS NOT NULL
-         ORDER BY 
+         ORDER BY
            ST_Distance(
              pp2.location::geography,
-             (SELECT location::geography 
-              FROM packets 
-              WHERE sender = pp2.first_digipeater 
-                AND location IS NOT NULL 
-              ORDER BY received_at DESC 
+             (SELECT location::geography
+              FROM packets
+              WHERE sender = pp2.first_digipeater
+                AND location IS NOT NULL
+              ORDER BY received_at DESC
               LIMIT 1)
            ) DESC NULLS LAST
          LIMIT 1
@@ -344,13 +344,13 @@ defmodule AprsmeWeb.InfoLive.Show do
       WHERE first_digipeater IS NOT NULL
       GROUP BY first_digipeater
     )
-    SELECT 
+    SELECT
       ds.digipeater,
       ds.first_heard,
       ds.last_heard,
       ds.packet_count,
       p.received_at as longest_path_time,
-      CASE 
+      CASE
         WHEN p.location IS NOT NULL AND dig_loc.location IS NOT NULL THEN
           ST_Distance(p.location::geography, dig_loc.location::geography) / 1000.0
         ELSE NULL
@@ -358,11 +358,11 @@ defmodule AprsmeWeb.InfoLive.Show do
     FROM digipeater_stats ds
     LEFT JOIN packets p ON p.id = ds.longest_path_packet_id
     LEFT JOIN LATERAL (
-      SELECT location 
-      FROM packets 
-      WHERE sender = ds.digipeater 
-        AND location IS NOT NULL 
-      ORDER BY received_at DESC 
+      SELECT location
+      FROM packets
+      WHERE sender = ds.digipeater
+        AND location IS NOT NULL
+      ORDER BY received_at DESC
       LIMIT 1
     ) dig_loc ON true
     ORDER BY ds.last_heard DESC
@@ -405,11 +405,20 @@ defmodule AprsmeWeb.InfoLive.Show do
     # Get packets from the last month where this callsign heard other stations on RF
     one_month_ago = DateTime.add(DateTime.utc_now(), -30, :day)
 
-    # Query to find stations that were heard directly by this callsign
-    # We look for packets where this callsign appears as the first digipeater with an asterisk
+    # Optimized query to find stations that were heard directly by this callsign
+    # Eliminates correlated subqueries and reduces redundant operations
     query = """
-    WITH parsed_paths AS (
-      SELECT 
+    WITH digipeater_location AS (
+      -- Get the most recent location for the digipeater (cached once)
+      SELECT location::geography as dig_loc
+      FROM packets
+      WHERE sender = $1
+        AND location IS NOT NULL
+      ORDER BY received_at DESC
+      LIMIT 1
+    ),
+    parsed_paths AS (
+      SELECT
         id,
         sender,
         path,
@@ -417,10 +426,10 @@ defmodule AprsmeWeb.InfoLive.Show do
         lat,
         lon,
         location,
-        -- Extract the first digipeater from the path
-        CASE 
-          WHEN path ~ '[A-Z0-9]+-?[0-9]*\\*' THEN
-            substring(path from '([A-Z0-9]+-?[0-9]*)\\*')
+        -- Extract the first digipeater from the path more efficiently
+        CASE
+          WHEN path ~ ($1 || '\\*') THEN
+            substring(path from ($1 || '\\*'))
           ELSE NULL
         END as first_digipeater
       FROM packets
@@ -430,57 +439,46 @@ defmodule AprsmeWeb.InfoLive.Show do
         AND path != ''
         AND path !~ '^TCPIP'
         AND path !~ ',TCPIP'
+        AND lat IS NOT NULL
+        AND lon IS NOT NULL
     ),
     station_stats AS (
-      SELECT 
+      SELECT
         sender as station,
         MIN(received_at) as first_heard,
         MAX(received_at) as last_heard,
-        COUNT(*) as packet_count,
-        -- Find the packet with maximum distance for this station
-        (SELECT pp2.id 
-         FROM parsed_paths pp2 
-         WHERE pp2.sender = pp.sender 
-           AND pp2.first_digipeater = $1
-           AND pp2.lat IS NOT NULL 
-           AND pp2.lon IS NOT NULL
-         ORDER BY 
-           ST_Distance(
-             pp2.location::geography,
-             (SELECT location::geography 
-              FROM packets 
-              WHERE sender = $1 
-                AND location IS NOT NULL 
-              ORDER BY received_at DESC 
-              LIMIT 1)
-           ) DESC NULLS LAST
-         LIMIT 1
-        ) as longest_path_packet_id
-      FROM parsed_paths pp
+        COUNT(*) as packet_count
+      FROM parsed_paths
       WHERE first_digipeater = $1
       GROUP BY sender
+    ),
+    longest_paths AS (
+      -- Find the longest path packet for each station in a single pass
+      SELECT DISTINCT ON (pp.sender)
+        pp.sender,
+        pp.id as longest_path_packet_id,
+        pp.received_at as longest_path_time,
+        CASE
+          WHEN pp.location IS NOT NULL AND dl.dig_loc IS NOT NULL THEN
+            ST_Distance(pp.location::geography, dl.dig_loc) / 1000.0
+          ELSE NULL
+        END as longest_distance_km
+      FROM parsed_paths pp
+      CROSS JOIN digipeater_location dl
+      WHERE pp.first_digipeater = $1
+      ORDER BY pp.sender,
+               ST_Distance(pp.location::geography, dl.dig_loc) DESC NULLS LAST,
+               pp.received_at DESC
     )
-    SELECT 
+    SELECT
       ss.station,
       ss.first_heard,
       ss.last_heard,
       ss.packet_count,
-      p.received_at as longest_path_time,
-      CASE 
-        WHEN p.location IS NOT NULL AND dig_loc.location IS NOT NULL THEN
-          ST_Distance(p.location::geography, dig_loc.location::geography) / 1000.0
-        ELSE NULL
-      END as longest_distance_km
+      lp.longest_path_time,
+      lp.longest_distance_km
     FROM station_stats ss
-    LEFT JOIN packets p ON p.id = ss.longest_path_packet_id
-    LEFT JOIN LATERAL (
-      SELECT location 
-      FROM packets 
-      WHERE sender = $1 
-        AND location IS NOT NULL 
-      ORDER BY received_at DESC 
-      LIMIT 1
-    ) dig_loc ON true
+    LEFT JOIN longest_paths lp ON lp.sender = ss.station
     ORDER BY ss.last_heard DESC
     LIMIT 50
     """

lib/aprsme_web/live/map_live/historical_loader.ex

@@ -13,6 +13,20 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
 
   @max_historical_packets 5000
 
+  # Viewport-based loading limits by zoom level
+  @zoom_packet_limits %{
+    # Very low zoom (world/continent view) - show only major stations
+    (1..5) => 100,
+    # Low zoom (country/state view) - show moderate amount
+    (6..8) => 500,
+    # Medium zoom (city view) - show more detail
+    (9..11) => 1500,
+    # High zoom (neighborhood view) - show most packets
+    (12..14) => 3000,
+    # Very high zoom (street view) - show all available
+    (15..20) => 5000
+  }
+
   @doc """
   Start progressive historical loading based on zoom level.
   """
@@ -94,19 +108,39 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
 
   # Private functions
 
-  defp do_load_historical_batch(socket, batch_offset) do
-    if socket.assigns.map_bounds do
-      bounds = [
-        socket.assigns.map_bounds.west,
-        socket.assigns.map_bounds.south,
-        socket.assigns.map_bounds.east,
-        socket.assigns.map_bounds.north
-      ]
+  defp get_packet_limit_for_zoom(zoom) do
+    @zoom_packet_limits
+    |> Enum.find(fn {range, _} -> zoom in range end)
+    |> case do
+      {_, limit} -> limit
+      nil -> @max_historical_packets
+    end
+  end
 
-      # Calculate zoom-based batch size - higher zoom = smaller batches for faster response
-      zoom = socket.assigns.map_zoom || 5
-      batch_size = calculate_batch_size_for_zoom(zoom)
-      offset = batch_offset * batch_size
+  defp do_load_historical_batch(%{assigns: %{map_bounds: nil}} = socket, _batch_offset), do: socket
+
+  defp do_load_historical_batch(%{assigns: %{map_bounds: bounds}} = socket, batch_offset) do
+    bounds_list = [
+      bounds.west,
+      bounds.south,
+      bounds.east,
+      bounds.north
+    ]
+
+    # Calculate zoom-based batch size - higher zoom = smaller batches for faster response
+    zoom = socket.assigns.map_zoom || 5
+    batch_size = calculate_batch_size_for_zoom(zoom)
+    offset = batch_offset * batch_size
+
+    # Get total packet limit based on zoom level
+    max_packets_for_zoom = get_packet_limit_for_zoom(zoom)
+
+    # Check if we've reached the zoom-based limit
+    if offset >= max_packets_for_zoom do
+      finish_historical_loading(socket)
+    else
+      # Adjust batch size if it would exceed the limit
+      adjusted_batch_size = min(batch_size, max_packets_for_zoom - offset)
 
       packets_module = Application.get_env(:aprsme, :packets_module, Aprsme.Packets)
 
@@ -116,19 +150,14 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
             # Use cached queries for better performance
             # Include zoom level in cache key for better cache efficiency
             params = %{
-              bounds: bounds,
-              limit: batch_size,
+              bounds: bounds_list,
+              limit: adjusted_batch_size,
               offset: offset,
               zoom: zoom
             }
 
             # Add callsign filter if tracking
-            params =
-              if socket.assigns.tracked_callsign == "" do
-                params
-              else
-                Map.put(params, :callsign, socket.assigns.tracked_callsign)
-              end
+            params = add_callsign_filter(params, socket.assigns.tracked_callsign)
 
             # Use the historical_hours setting from the UI dropdown with validation
             historical_hours = SharedPacketUtils.parse_historical_hours(socket.assigns.historical_hours || "1")
@@ -166,74 +195,95 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
               []
           end
 
-        if Enum.any?(packet_data_list) do
-          # Check zoom level to decide between heat map and markers
-          total_batches = socket.assigns.total_batches || 4
-          is_final_batch = batch_offset >= total_batches - 1
-
-          socket =
-            if socket.assigns.map_zoom <= 8 do
-              # For heat maps, store historical packets and update heat map when all batches are loaded
-              # Add packets to historical_packets assign
-              new_historical =
-                Enum.reduce(historical_packets, socket.assigns.historical_packets, fn packet, acc ->
-                  key = if Map.has_key?(packet, :id), do: to_string(packet.id), else: to_string(packet["id"])
-                  Map.put(acc, key, packet)
-                end)
-
-              # Apply memory limit
-              new_historical =
-                if map_size(new_historical) > @max_historical_packets do
-                  SharedPacketUtils.prune_oldest_packets(new_historical, @max_historical_packets)
-                else
-                  new_historical
-                end
-
-              socket = assign(socket, :historical_packets, new_historical)
-
-              # If this is the final batch, update the heat map
-              if is_final_batch do
-                send_heat_map_for_current_bounds(socket)
-              else
-                socket
-              end
-            else
-              # Use LiveView's efficient push_event for incremental updates
-              LiveView.push_event(socket, "add_historical_packets_batch", %{
-                packets: packet_data_list,
-                batch: batch_offset,
-                is_final: is_final_batch
-              })
-            end
-
-          # Update progress for user feedback
-          socket = assign(socket, :loading_batch, batch_offset + 1)
-
-          # Mark loading as complete if this was the final batch
-          socket =
-            if is_final_batch do
-              assign(socket, :historical_loading, false)
-            else
-              socket
-            end
-
-          # Process any pending bounds update if loading is complete
-          if is_final_batch && socket.assigns.pending_bounds do
-            send(self(), {:process_pending_bounds})
-          end
-
-          socket
-        else
-          socket
-        end
+        process_loaded_packets(socket, historical_packets, packet_data_list, batch_offset)
       else
         socket
       end
+    end
+  end
+
+  defp add_callsign_filter(params, ""), do: params
+  defp add_callsign_filter(params, callsign), do: Map.put(params, :callsign, callsign)
+
+  defp process_loaded_packets(socket, _historical_packets, [], _batch_offset), do: socket
+
+  defp process_loaded_packets(socket, historical_packets, packet_data_list, batch_offset) do
+    # Check zoom level to decide between heat map and markers
+    total_batches = socket.assigns.total_batches || 4
+    is_final_batch = batch_offset >= total_batches - 1
+
+    socket = handle_zoom_based_display(socket, historical_packets, packet_data_list, is_final_batch, batch_offset)
+
+    # Update progress for user feedback
+    socket = assign(socket, :loading_batch, batch_offset + 1)
+
+    # Mark loading as complete if this was the final batch
+    socket = update_loading_status(socket, is_final_batch)
+
+    # Process any pending bounds update if loading is complete
+    handle_pending_bounds(socket, is_final_batch)
+  end
+
+  # Handle low zoom (heat map)
+  defp handle_zoom_based_display(
+         %{assigns: %{map_zoom: zoom}} = socket,
+         historical_packets,
+         _packet_data_list,
+         is_final_batch,
+         _batch_offset
+       )
+       when zoom <= 8 do
+    # For heat maps, store historical packets and update heat map when all batches are loaded
+    new_historical = add_packets_to_historical(socket.assigns.historical_packets, historical_packets)
+
+    # Apply memory limit
+    new_historical = apply_memory_limit(new_historical, @max_historical_packets)
+
+    socket = assign(socket, :historical_packets, new_historical)
+
+    # If this is the final batch, update the heat map
+    if is_final_batch do
+      send_heat_map_for_current_bounds(socket)
     else
       socket
     end
   end
 
+  # Handle high zoom (markers)
+  defp handle_zoom_based_display(socket, _historical_packets, packet_data_list, is_final_batch, batch_offset) do
+    # Use LiveView's efficient push_event for incremental updates
+    LiveView.push_event(socket, "add_historical_packets_batch", %{
+      packets: packet_data_list,
+      batch: batch_offset,
+      is_final: is_final_batch
+    })
+  end
+
+  defp add_packets_to_historical(existing_historical, new_packets) do
+    Enum.reduce(new_packets, existing_historical, fn packet, acc ->
+      key = get_packet_key(packet)
+      Map.put(acc, key, packet)
+    end)
+  end
+
+  defp get_packet_key(%{id: id}), do: to_string(id)
+  defp get_packet_key(%{"id" => id}), do: to_string(id)
+
+  defp apply_memory_limit(packets, limit) when map_size(packets) > limit do
+    SharedPacketUtils.prune_oldest_packets(packets, limit)
+  end
+
+  defp apply_memory_limit(packets, _limit), do: packets
+
+  defp update_loading_status(socket, true), do: assign(socket, :historical_loading, false)
+  defp update_loading_status(socket, false), do: socket
+
+  defp handle_pending_bounds(%{assigns: %{pending_bounds: pending}} = _socket, true) when not is_nil(pending) do
+    send(self(), {:process_pending_bounds})
+  end
+
+  defp handle_pending_bounds(socket, _), do: socket
+
   # Consolidated zoom-based loading parameters
   @spec get_loading_params_for_zoom(integer()) :: {batch_size :: integer(), batch_count :: integer()}
   # Very zoomed in - load everything at once, minimal batches
@@ -264,6 +314,19 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
   # All packet data building functions have been moved to AprsmeWeb.MapLive.DataBuilder
 
   # Placeholder for heat map function - this should be moved to DisplayManager
+  defp finish_historical_loading(socket) do
+    socket
+    |> assign(:historical_loading, false)
+    |> assign(:loading_batch, socket.assigns.total_batches)
+    |> then(fn s ->
+      if s.assigns.pending_bounds do
+        send(self(), {:process_pending_bounds})
+      end
+
+      s
+    end)
+  end
+
   defp send_heat_map_for_current_bounds(socket) do
     # This function should be moved to DisplayManager module
     socket

lib/aprsme_web/live/map_live/index.ex

@@ -76,10 +76,25 @@ defmodule AprsmeWeb.MapLive.Index do
   end
 
   defp do_setup_subscriptions(socket, true) do
-    Phoenix.PubSub.subscribe(Aprsme.PubSub, "packets")
+    # Generate a unique client ID for this LiveView instance
+    client_id = "liveview_#{:erlang.phash2(self())}"
+
+    # Register with spatial PubSub (will get viewport later)
+    # Start with a default viewport that will be updated when we get actual bounds
+    default_bounds = %{north: 90.0, south: -90.0, east: 180.0, west: -180.0}
+    {:ok, spatial_topic} = Aprsme.SpatialPubSub.register_viewport(client_id, default_bounds)
+
+    # Subscribe to the spatial topic for this client
+    Phoenix.PubSub.subscribe(Aprsme.PubSub, spatial_topic)
+
+    # Still subscribe to bad packets (they don't have location)
     Phoenix.PubSub.subscribe(Aprsme.PubSub, "bad_packets")
+
     Process.send_after(self(), :cleanup_old_packets, 60_000)
+
     socket
+    |> assign(:spatial_client_id, client_id)
+    |> assign(:spatial_topic, spatial_topic)
   end
 
   defp do_setup_subscriptions(socket, false), do: socket
@@ -575,6 +590,8 @@ defmodule AprsmeWeb.MapLive.Index do
 
   def handle_info({:postgres_packet, packet}, socket), do: handle_info_postgres_packet(packet, socket)
 
+  def handle_info({:spatial_packet, packet}, socket), do: handle_info_postgres_packet(packet, socket)
+
   def handle_info({:process_pending_bounds}, socket) do
     if socket.assigns.pending_bounds && !socket.assigns.historical_loading do
       # Process the pending bounds update
@@ -1310,6 +1327,11 @@ defmodule AprsmeWeb.MapLive.Index do
 
   @impl true
   def terminate(_reason, socket) do
+    # Cleanup spatial registration if we have a client ID
+    if socket.assigns[:spatial_client_id] do
+      Aprsme.SpatialPubSub.unregister_client(socket.assigns.spatial_client_id)
+    end
+
     if socket.assigns.buffer_timer, do: Process.cancel_timer(socket.assigns.buffer_timer)
     # Clean up any pending bounds update timer
     if socket.assigns[:bounds_update_timer] do
@@ -1405,6 +1427,11 @@ 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
+
     # 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)

lib/aprsme_web/live/map_live/packet_store.ex

@@ -0,0 +1,164 @@
+defmodule AprsmeWeb.MapLive.PacketStore do
+  @moduledoc """
+  Efficient packet storage for LiveView that stores only IDs in assigns
+  and fetches full packet data on demand.
+  """
+
+  use GenServer
+
+  require Logger
+
+  @table_name :map_packet_store
+  @ttl_ms to_timeout(hour: 2)
+  @cleanup_interval_ms to_timeout(minute: 5)
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @doc """
+  Store a packet and return its ID.
+  """
+  def store_packet(packet) do
+    packet_id = generate_packet_id(packet)
+    expires_at = System.monotonic_time(:millisecond) + @ttl_ms
+
+    :ets.insert(@table_name, {packet_id, packet, expires_at})
+    packet_id
+  end
+
+  @doc """
+  Store multiple packets and return their IDs.
+  """
+  def store_packets(packets) when is_list(packets) do
+    expires_at = System.monotonic_time(:millisecond) + @ttl_ms
+
+    entries =
+      Enum.map(packets, fn packet ->
+        packet_id = generate_packet_id(packet)
+        {packet_id, packet, expires_at}
+      end)
+
+    :ets.insert(@table_name, entries)
+    Enum.map(entries, fn {id, _, _} -> id end)
+  end
+
+  @doc """
+  Get a packet by ID.
+  """
+  def get_packet(packet_id) do
+    case :ets.lookup(@table_name, packet_id) do
+      [{^packet_id, packet, expires_at}] ->
+        if System.monotonic_time(:millisecond) < expires_at do
+          {:ok, packet}
+        else
+          # Expired, remove it
+          :ets.delete(@table_name, packet_id)
+          {:error, :not_found}
+        end
+
+      [] ->
+        {:error, :not_found}
+    end
+  end
+
+  @doc """
+  Get multiple packets by IDs.
+  """
+  def get_packets(packet_ids) when is_list(packet_ids) do
+    current_time = System.monotonic_time(:millisecond)
+
+    packet_ids
+    |> Enum.map(fn id ->
+      case :ets.lookup(@table_name, id) do
+        [{^id, packet, expires_at}] when current_time < expires_at ->
+          {id, packet}
+
+        _ ->
+          nil
+      end
+    end)
+    |> Enum.reject(&is_nil/1)
+    |> Map.new()
+  end
+
+  @doc """
+  Remove a packet by ID.
+  """
+  def remove_packet(packet_id) do
+    :ets.delete(@table_name, packet_id)
+  end
+
+  @doc """
+  Remove multiple packets by IDs.
+  """
+  def remove_packets(packet_ids) when is_list(packet_ids) do
+    Enum.each(packet_ids, &:ets.delete(@table_name, &1))
+  end
+
+  @doc """
+  Get statistics about the store.
+  """
+  def get_stats do
+    %{
+      total_packets: :ets.info(@table_name, :size),
+      memory_bytes: :ets.info(@table_name, :memory) * :erlang.system_info(:wordsize)
+    }
+  end
+
+  # GenServer callbacks
+
+  @impl true
+  def init(_opts) do
+    # Create ETS table for packet storage
+    :ets.new(@table_name, [
+      :set,
+      :public,
+      :named_table,
+      read_concurrency: true,
+      write_concurrency: true
+    ])
+
+    # Schedule periodic cleanup
+    schedule_cleanup()
+
+    {:ok, %{}}
+  end
+
+  @impl true
+  def handle_info(:cleanup, state) do
+    cleanup_expired_packets()
+    schedule_cleanup()
+    {:noreply, state}
+  end
+
+  # Private functions
+
+  defp generate_packet_id(packet) do
+    # Generate a unique ID based on packet attributes
+    sender = Map.get(packet, :sender, Map.get(packet, "sender", ""))
+    timestamp = Map.get(packet, :received_at, Map.get(packet, "received_at", DateTime.utc_now()))
+
+    # Create a deterministic ID
+    :md5
+    |> :crypto.hash("#{sender}_#{timestamp}")
+    |> Base.encode16(case: :lower)
+    |> binary_part(0, 16)
+  end
+
+  defp schedule_cleanup do
+    Process.send_after(self(), :cleanup, @cleanup_interval_ms)
+  end
+
+  defp cleanup_expired_packets do
+    current_time = System.monotonic_time(:millisecond)
+
+    # Use match_delete for efficient bulk deletion
+    match_spec = [{{:_, :_, :"$1"}, [{:<, :"$1", current_time}], [true]}]
+    deleted = :ets.select_delete(@table_name, match_spec)
+
+    if deleted > 0 do
+      Logger.debug("Cleaned up #{deleted} expired packets from packet store")
+    end
+  end
+end

lib/aprsme_web/live/status_live/index.ex

@@ -4,23 +4,30 @@ defmodule AprsmeWeb.StatusLive.Index do
   """
   use AprsmeWeb, :live_view
 
-  # 30 seconds
-  @refresh_interval 1_000
+  # 5 seconds - reduced from 1 second to improve performance
+  @refresh_interval 5_000
 
   @impl true
   def mount(_params, _session, socket) do
+    # Load cached status on mount for fast initial load
+    aprs_status = get_cached_aprs_status()
+
     socket =
       assign(socket,
         page_title: "System Status",
-        aprs_status: get_aprs_status(),
+        aprs_status: aprs_status,
         version: get_app_version(),
         current_time: DateTime.utc_now(),
-        health_score: calculate_health_score(get_aprs_status())
+        health_score: calculate_health_score(aprs_status),
+        loading: false
       )
 
     if connected?(socket) do
-      # Schedule the first refresh
-      schedule_refresh()
+      # Subscribe to status updates via PubSub
+      Phoenix.PubSub.subscribe(Aprsme.PubSub, "aprs_status")
+
+      # Schedule the first refresh with a slight delay
+      Process.send_after(self(), :refresh_status, 500)
     end
 
     {:ok, socket}
@@ -28,9 +35,38 @@ defmodule AprsmeWeb.StatusLive.Index do
 
   @impl true
   def handle_info(:refresh_status, socket) do
-    socket = refresh_status(socket)
+    # Refresh status asynchronously
+    self_pid = self()
+
+    Task.start(fn ->
+      status = get_aprs_status()
+      send(self_pid, {:status_updated, status})
+    end)
+
     # Schedule next refresh
     schedule_refresh()
+    {:noreply, assign(socket, loading: true)}
+  end
+
+  @impl true
+  def handle_info({:status_updated, status}, socket) do
+    socket =
+      assign(socket,
+        aprs_status: status,
+        current_time: DateTime.utc_now(),
+        health_score: calculate_health_score(status),
+        loading: false
+      )
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_info({:aprs_status_update, status}, socket) do
+    # Handle real-time status updates via PubSub
+    socket =
+      assign(socket, aprs_status: status, current_time: DateTime.utc_now(), health_score: calculate_health_score(status))
+
     {:noreply, socket}
   end
 
@@ -253,37 +289,45 @@ defmodule AprsmeWeb.StatusLive.Index do
 
       Logger.error("Error getting APRS status: #{inspect(error)}")
       # Return a default status when database is unavailable
-      %{
-        connected: false,
-        server: "unknown",
-        port: 0,
-        connected_at: nil,
-        uptime_seconds: 0,
-        login_id: "unknown",
-        filter: "unknown",
-        packet_stats: %{
-          total_packets: 0,
-          packets_per_second: 0,
-          last_packet_at: nil
-        },
-        stored_packet_count: 0
-      }
+      default_status()
+  end
+
+  defp get_cached_aprs_status do
+    # Try to get cached status for instant load
+    case Cachex.get(:query_cache, "aprs_status") do
+      {:ok, status} when not is_nil(status) ->
+        status
+
+      _ ->
+        # Fallback to direct query if cache miss
+        status = get_aprs_status()
+        Cachex.put(:query_cache, "aprs_status", status, ttl: to_timeout(second: 5))
+        status
+    end
+  end
+
+  defp default_status do
+    %{
+      connected: false,
+      server: "unknown",
+      port: 0,
+      connected_at: nil,
+      uptime_seconds: 0,
+      login_id: "unknown",
+      filter: "unknown",
+      packet_stats: %{
+        total_packets: 0,
+        packets_per_second: 0,
+        last_packet_at: nil
+      },
+      stored_packet_count: 0
+    }
   end
 
   defp get_app_version do
     :aprsme |> Application.spec(:vsn) |> List.to_string()
   end
 
-  defp refresh_status(socket) do
-    aprs_status = get_aprs_status()
-
-    assign(socket,
-      aprs_status: aprs_status,
-      current_time: DateTime.utc_now(),
-      health_score: calculate_health_score(aprs_status)
-    )
-  end
-
   defp schedule_refresh do
     Process.send_after(self(), :refresh_status, @refresh_interval)
   end

lib/aprsme_web/telemetry.ex

@@ -88,7 +88,35 @@ defmodule AprsmeWeb.Telemetry do
         description: "Number of successful inserts per batch"
       ),
       summary("aprsme.packet_pipeline.batch.error", unit: :event, description: "Number of errors per batch"),
-      summary("aprsme.packet_pipeline.batch.duration_ms", unit: :millisecond, description: "Batch insert duration (ms)")
+      summary("aprsme.packet_pipeline.batch.duration_ms", unit: :millisecond, description: "Batch insert duration (ms)"),
+
+      # System Monitor Metrics
+      last_value("aprsme.system.memory.total", unit: {:byte, :megabyte}, description: "Total memory usage"),
+      last_value("aprsme.system.memory.process", unit: {:byte, :megabyte}, description: "Process memory usage"),
+      last_value("aprsme.system.memory.binary", unit: {:byte, :megabyte}, description: "Binary memory usage"),
+      last_value("aprsme.system.memory.pressure", unit: :percent, description: "Memory pressure (0-1)"),
+      last_value("aprsme.system.cpu.load1", description: "1-minute load average"),
+      last_value("aprsme.system.cpu.load5", description: "5-minute load average"),
+      last_value("aprsme.system.cpu.load15", description: "15-minute load average"),
+      last_value("aprsme.system.cpu.pressure", unit: :percent, description: "CPU pressure (0-1)"),
+      last_value("aprsme.system.processes.count", description: "Number of Erlang processes"),
+      last_value("aprsme.system.processes.pressure", unit: :percent, description: "Process count pressure (0-1)"),
+      last_value("aprsme.system.db_pool.size", description: "Database pool size"),
+      last_value("aprsme.system.db_pool.available", description: "Available database connections"),
+      last_value("aprsme.system.db_pool.pressure", unit: :percent, description: "Database pool pressure (0-1)"),
+      last_value("aprsme.system.batch_size.current", description: "Current adaptive batch size"),
+      last_value("aprsme.system.batch_size.min", description: "Minimum batch size"),
+      last_value("aprsme.system.batch_size.max", description: "Maximum batch size"),
+
+      # Spatial PubSub Metrics
+      last_value("aprsme.spatial_pubsub.clients.count", description: "Number of connected clients"),
+      last_value("aprsme.spatial_pubsub.clients.grid_cells", description: "Number of active grid cells"),
+      last_value("aprsme.spatial_pubsub.clients.avg_clients_per_cell", description: "Average clients per grid cell"),
+      counter("aprsme.spatial_pubsub.broadcasts.total", description: "Total spatial broadcasts sent"),
+      counter("aprsme.spatial_pubsub.broadcasts.filtered", description: "Broadcasts filtered by viewport"),
+      counter("aprsme.spatial_pubsub.broadcasts.packets", description: "Total packets processed"),
+      last_value("aprsme.spatial_pubsub.efficiency.ratio", unit: :percent, description: "Broadcast efficiency ratio"),
+      counter("aprsme.spatial_pubsub.efficiency.saved_broadcasts", description: "Number of broadcasts saved by filtering")
     ]
   end
 

notes.md

@@ -1,50 +0,0 @@
-# APRS.me Project Exploration Notes
-
-## Overview
-- This is a Phoenix LiveView application for APRS (Automatic Packet Reporting System)
-- APRS is a digital communications system used by amateur radio operators
-- The app tracks and displays APRS packets/messages in real-time on an interactive map
-
-## Project Structure
-- Standard Phoenix 1.7+ structure with LiveView
-- Uses Elixir ~> 1.17
-- Has authentication system (user auth files present)
-- Contains map functionality (map_live modules)
-- Has packet tracking (packets_live, status_live)
-- Database with PostGIS for geographic data
-- Real-time features with PubSub and presence
-
-## Key Features Found
-- **Interactive Map**: Leaflet-based map with OpenStreetMap tiles
-- **Real-time APRS Tracking**: Connected to APRS-IS (dallas.aprs2.net:14580)
-- **Responsive Design**: Slideover panel for controls, mobile-friendly
-- **Search Functionality**: Callsign search capability
-- **Trail Display**: Configurable trail duration (1 hour to 1 week)
-- **Historical Data**: Historical packet loading (1-24 hours)
-- **Packet Display**: Multiple views (map, packets, status, weather, bad packets)
-- **User Authentication**: Complete auth system ready
-- **API Endpoints**: RESTful API for callsign data
-
-## Technical Stack
-- **Backend**: Phoenix 1.7, Elixir 1.17, Ecto with PostgreSQL + PostGIS
-- **Frontend**: LiveView, Tailwind CSS, Leaflet.js for maps
-- **Real-time**: Phoenix PubSub, WebSocket connections
-- **Background Jobs**: Oban for job processing
-- **External Integration**: APRS-IS network connection
-- **Deployment**: Docker-ready with releases
-
-## Current Status
-- ✅ Application runs successfully on localhost:4000
-- ✅ Database migrations completed
-- ✅ APRS-IS connection established
-- ✅ Map interface working with controls
-- ✅ Real-time packet processing active
-- ⚠️  Some asset loading optimizations possible
-- 📡 Currently showing "no internet" (likely due to APRS data feed)
-
-## Next Steps
-- Explore specific LiveView modules
-- Check packet data flow
-- Review real-time update mechanisms
-- Examine API capabilities
-

priv/repo/migrations/20250714161821_add_performance_indexes.exs

@@ -0,0 +1,62 @@
+defmodule Aprsme.Repo.Migrations.AddPerformanceIndexes do
+  use Ecto.Migration
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  def up do
+    # Compound index for callsign history queries
+    create_if_not_exists index(:packets, [:base_callsign, "received_at DESC"],
+                           name: :packets_base_callsign_received_at_idx,
+                           concurrently: true
+                         )
+
+    # Covering index for sender queries to avoid table lookups
+    execute """
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_sender_received_covering_idx 
+    ON packets(sender, received_at DESC) 
+    INCLUDE (lat, lon, data_type, has_position)
+    """
+
+    # Partial index for weather packets (without time constraint for immutability)
+    execute """
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_weather_recent_idx 
+    ON packets(received_at DESC) 
+    WHERE data_type IN ('weather', 'Weather', 'WX', 'wx')
+    """
+
+    # Spatial index for location queries
+    execute """
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_location_recent_idx 
+    ON packets USING GIST (location) 
+    WHERE has_position = true
+    """
+
+    # Index for device lookups
+    create_if_not_exists index(:packets, [:device_identifier, :received_at],
+                           name: :packets_device_received_idx,
+                           concurrently: true
+                         )
+
+    # Index for cleanup worker queries (ascending for old packets)
+    create_if_not_exists index(:packets, [:received_at],
+                           name: :packets_received_at_asc_idx,
+                           concurrently: true
+                         )
+  end
+
+  def down do
+    drop_if_exists index(:packets, [:base_callsign, :received_at],
+                     name: :packets_base_callsign_received_at_idx
+                   )
+
+    execute "DROP INDEX IF EXISTS packets_sender_received_covering_idx"
+    execute "DROP INDEX IF EXISTS packets_weather_recent_idx"
+    execute "DROP INDEX IF EXISTS packets_location_recent_idx"
+
+    drop_if_exists index(:packets, [:device_identifier, :received_at],
+                     name: :packets_device_received_idx
+                   )
+
+    drop_if_exists index(:packets, [:received_at], name: :packets_received_at_asc_idx)
+  end
+end

priv/repo/migrations/20250714200255_add_packet_counter_table.exs

@@ -0,0 +1,93 @@
+defmodule Aprsme.Repo.Migrations.AddPacketCounterTable do
+  use Ecto.Migration
+
+  def up do
+    # Create a dedicated counter table
+    create table(:packet_counters) do
+      add :counter_type, :string, null: false
+      add :count, :bigint, null: false, default: 0
+      timestamps()
+    end
+
+    create unique_index(:packet_counters, [:counter_type])
+
+    # Insert initial counter with current count
+    execute """
+    INSERT INTO packet_counters (counter_type, count, inserted_at, updated_at)
+    SELECT 'total_packets', COUNT(*), NOW(), NOW()
+    FROM packets;
+    """
+
+    # Create function to increment counter
+    execute """
+    CREATE OR REPLACE FUNCTION increment_packet_counter()
+    RETURNS TRIGGER AS $$
+    BEGIN
+      UPDATE packet_counters 
+      SET count = count + 1,
+          updated_at = NOW()
+      WHERE counter_type = 'total_packets';
+      RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+    """
+
+    # Create function to decrement counter
+    execute """
+    CREATE OR REPLACE FUNCTION decrement_packet_counter()
+    RETURNS TRIGGER AS $$
+    BEGIN
+      UPDATE packet_counters 
+      SET count = GREATEST(0, count - 1),
+          updated_at = NOW()
+      WHERE counter_type = 'total_packets';
+      RETURN OLD;
+    END;
+    $$ LANGUAGE plpgsql;
+    """
+
+    # Create triggers
+    execute """
+    CREATE TRIGGER packet_insert_counter
+    AFTER INSERT ON packets
+    FOR EACH ROW
+    EXECUTE FUNCTION increment_packet_counter();
+    """
+
+    execute """
+    CREATE TRIGGER packet_delete_counter
+    AFTER DELETE ON packets
+    FOR EACH ROW
+    EXECUTE FUNCTION decrement_packet_counter();
+    """
+
+    # Create a fast function to get the count
+    execute """
+    CREATE OR REPLACE FUNCTION get_packet_count()
+    RETURNS BIGINT AS $$
+    BEGIN
+      RETURN (SELECT count FROM packet_counters WHERE counter_type = 'total_packets');
+    END;
+    $$ LANGUAGE plpgsql STABLE;
+    """
+
+    # Create index on the function for even faster access
+    execute """
+    CREATE INDEX idx_packet_count ON packet_counters (counter_type) WHERE counter_type = 'total_packets';
+    """
+  end
+
+  def down do
+    # Drop triggers
+    execute "DROP TRIGGER IF EXISTS packet_insert_counter ON packets;"
+    execute "DROP TRIGGER IF EXISTS packet_delete_counter ON packets;"
+
+    # Drop functions
+    execute "DROP FUNCTION IF EXISTS increment_packet_counter();"
+    execute "DROP FUNCTION IF EXISTS decrement_packet_counter();"
+    execute "DROP FUNCTION IF EXISTS get_packet_count();"
+
+    # Drop table
+    drop table(:packet_counters)
+  end
+end

priv/repo/migrations/20250715000000_optimize_stations_heard_by_query.exs

@@ -0,0 +1,52 @@
+defmodule Aprsme.Repo.Migrations.OptimizeStationsHeardByQuery do
+  use Ecto.Migration
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  def up do
+    # Enable the pg_trgm extension for trigram indexing
+    execute "CREATE EXTENSION IF NOT EXISTS pg_trgm;", ""
+
+    # Add index for path pattern matching to optimize the regex operations
+    execute """
+            CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_path_pattern_idx
+            ON packets USING GIN (path gin_trgm_ops)
+            WHERE path IS NOT NULL
+              AND path != ''
+              AND path !~ '^TCPIP'
+              AND path !~ ',TCPIP';
+            """,
+            ""
+
+    # Add composite index for the main filtering conditions
+    execute """
+            CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_stations_heard_idx
+            ON packets (received_at DESC, sender, lat, lon)
+            WHERE path IS NOT NULL
+              AND path != ''
+              AND path !~ '^TCPIP'
+              AND path !~ ',TCPIP'
+              AND lat IS NOT NULL
+              AND lon IS NOT NULL;
+            """,
+            ""
+
+    # Add index for digipeater location lookups
+    execute """
+            CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_digipeater_location_idx
+            ON packets (sender, received_at DESC)
+            WHERE location IS NOT NULL;
+            """,
+            ""
+
+    # Analyze tables to update statistics for query planner
+    execute "ANALYZE packets;", ""
+  end
+
+  def down do
+    execute "DROP INDEX CONCURRENTLY IF EXISTS packets_path_pattern_idx;", ""
+    execute "DROP INDEX CONCURRENTLY IF EXISTS packets_stations_heard_idx;", ""
+    execute "DROP INDEX CONCURRENTLY IF EXISTS packets_digipeater_location_idx;", ""
+    # Note: We don't drop the pg_trgm extension as it might be used by other parts of the system
+  end
+end

test/aprsme/workers/packet_cleanup_worker_test.exs

@@ -1,32 +1,45 @@
 defmodule Aprsme.Workers.PacketCleanupWorkerTest do
   use Aprsme.DataCase, async: true
 
-  import Mox
-
-  alias Aprsme.PacketsMock
+  alias Aprsme.Packet
   alias Aprsme.Workers.PacketCleanupWorker
 
-  # Make sure mocks are verified when the test exits
-  setup :verify_on_exit!
+  defp insert_packet(attrs) do
+    attrs =
+      Map.merge(
+        %{
+          sender: "TEST-#{System.unique_integer([:positive])}",
+          received_at: DateTime.truncate(DateTime.utc_now(), :second),
+          has_position: false
+        },
+        attrs
+      )
+
+    {:ok, packet} = %Packet{} |> Map.merge(attrs) |> Repo.insert()
+    packet
+  end
+
+  defp insert_packets(count, attrs) do
+    for _ <- 1..count, do: insert_packet(attrs)
+  end
 
   describe "perform/1 with cleanup_days" do
     test "cleans up packets older than the specified number of days" do
       days = 30
       job = %Oban.Job{args: %{"cleanup_days" => days}}
 
-      # Mock the Repo.all call for get_packet_ids_for_deletion
-      expect(Aprsme.Repo, :all, fn _query ->
-        # Return some packet IDs
-        [1, 2, 3, 4, 5]
-      end)
+      # Create old packets that should be deleted
+      old_time = DateTime.utc_now() |> DateTime.add(-(days + 1) * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(5, %{received_at: old_time})
 
-      # Mock the Repo.delete_all call
-      expect(Aprsme.Repo, :delete_all, fn _query ->
-        # Return deleted count
-        {5, nil}
-      end)
+      # Create recent packets that should NOT be deleted
+      recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(3, %{received_at: recent_time})
 
       assert :ok = PacketCleanupWorker.perform(job)
+
+      # Verify only recent packets remain
+      assert Repo.aggregate(Packet, :count, :id) == 3
     end
   end
 
@@ -34,40 +47,37 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do
     test "cleans up packets using the default retention period" do
       job = %Oban.Job{args: %{}}
 
-      # Mock the Repo.all call for get_packet_ids_for_deletion
-      expect(Aprsme.Repo, :all, fn _query ->
-        # Return some packet IDs
-        [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
-      end)
+      # Create very old packets that should be deleted (older than 365 days)
+      old_time = DateTime.utc_now() |> DateTime.add(-400 * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(10, %{received_at: old_time})
 
-      # Mock the Repo.delete_all call
-      expect(Aprsme.Repo, :delete_all, fn _query ->
-        # Return deleted count
-        {10, nil}
-      end)
+      # Create recent packets that should NOT be deleted
+      recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(3, %{received_at: recent_time})
 
       assert :ok = PacketCleanupWorker.perform(job)
+
+      # Verify only recent packets remain
+      assert Repo.aggregate(Packet, :count, :id) == 3
     end
   end
 
   describe "cleanup_packets_older_than/1" do
     test "cleans up packets older than the given number of days" do
       days = 60
-      deleted_count = 15
 
-      # Mock the Repo.all call for get_packet_ids_for_deletion
-      expect(Aprsme.Repo, :all, fn _query ->
-        # Return packet IDs
-        Enum.to_list(1..deleted_count)
-      end)
+      # Create old packets that should be deleted
+      old_time = DateTime.utc_now() |> DateTime.add(-(days + 10) * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(15, %{received_at: old_time})
 
-      # Mock the Repo.delete_all call
-      expect(Aprsme.Repo, :delete_all, fn _query ->
-        # Return deleted count
-        {deleted_count, nil}
-      end)
+      # Create recent packets that should NOT be deleted
+      recent_time = DateTime.utc_now() |> DateTime.add(-30 * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(5, %{received_at: recent_time})
 
-      assert ^deleted_count = PacketCleanupWorker.cleanup_packets_older_than(days)
+      deleted_count = PacketCleanupWorker.cleanup_packets_older_than(days)
+
+      assert deleted_count == 15
+      assert Repo.aggregate(Packet, :count, :id) == 5
     end
   end
 
@@ -75,42 +85,45 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do
     test "returns correct tuple format with deleted count and batches" do
       days = 30
 
-      # Mock the Repo.all call for get_packet_ids_for_deletion
-      expect(Aprsme.Repo, :all, fn _query ->
-        # Return some packet IDs
-        [1, 2, 3, 4, 5]
-      end)
+      # Create old packets that should be deleted
+      old_time = DateTime.utc_now() |> DateTime.add(-(days + 1) * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(5, %{received_at: old_time})
 
-      # Mock the Repo.delete_all call
-      expect(Aprsme.Repo, :delete_all, fn _query ->
-        # Return deleted count
-        {5, nil}
-      end)
+      # Create recent packets that should NOT be deleted
+      recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(3, %{received_at: recent_time})
 
       {deleted_count, batches} = PacketCleanupWorker.cleanup_packets_older_than_batched(days)
 
       assert is_integer(deleted_count)
       assert is_integer(batches)
-      assert deleted_count >= 0
-      assert batches >= 0
+      assert deleted_count == 5
+      assert batches >= 1
+      assert Repo.aggregate(Packet, :count, :id) == 3
     end
   end
 
   describe "get_packet_ids_for_deletion/2" do
     test "returns list of packet IDs within batch size limit" do
-      cutoff_time = DateTime.add(DateTime.utc_now(), -30, :day)
-      batch_size = 100
+      cutoff_time = DateTime.utc_now() |> DateTime.add(-30, :day) |> DateTime.truncate(:second)
+      batch_size = 10
 
-      # Mock the Repo.all call
-      expect(Aprsme.Repo, :all, fn _query ->
-        # Return some mock packet IDs
-        [1, 2, 3, 4, 5]
-      end)
+      # Create old packets that should be included
+      old_time = DateTime.utc_now() |> DateTime.add(-40 * 86_400, :second) |> DateTime.truncate(:second)
+      old_packets = insert_packets(15, %{received_at: old_time})
+
+      # Create recent packets that should NOT be included
+      recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second)
+      insert_packets(5, %{received_at: recent_time})
 
       result = PacketCleanupWorker.get_packet_ids_for_deletion(cutoff_time, batch_size)
 
       assert is_list(result)
-      assert length(result) <= batch_size
+      assert length(result) == batch_size
+
+      # Verify that returned IDs are from old packets
+      old_packet_ids = Enum.map(old_packets, & &1.id)
+      assert Enum.all?(result, &(&1 in old_packet_ids))
     end
   end
 end