From 3730b8c8486cb72852248b680a5efe24daa4976a Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 14 Jul 2025 16:35:58 -0500 Subject: [PATCH] performance improvements --- lib/aprsme/application.ex | 6 + lib/aprsme/cached_queries.ex | 7 +- lib/aprsme/packet_consumer.ex | 41 +- lib/aprsme/packet_counter.ex | 43 ++ lib/aprsme/packets.ex | 53 ++- lib/aprsme/spatial_pubsub.ex | 396 ++++++++++++++++++ lib/aprsme/system_monitor.ex | 298 +++++++++++++ lib/aprsme/workers/packet_cleanup_worker.ex | 2 - lib/aprsme_web/live/info_live/show.ex | 132 +++--- .../live/map_live/historical_loader.ex | 223 ++++++---- lib/aprsme_web/live/map_live/index.ex | 29 +- lib/aprsme_web/live/map_live/packet_store.ex | 164 ++++++++ lib/aprsme_web/live/status_live/index.ex | 108 +++-- lib/aprsme_web/telemetry.ex | 30 +- notes.md | 50 --- ...20250714161821_add_performance_indexes.exs | 62 +++ ...0250714200255_add_packet_counter_table.exs | 93 ++++ ...00000_optimize_stations_heard_by_query.exs | 52 +++ .../workers/packet_cleanup_worker_test.exs | 127 +++--- 19 files changed, 1607 insertions(+), 309 deletions(-) create mode 100644 lib/aprsme/packet_counter.ex create mode 100644 lib/aprsme/spatial_pubsub.ex create mode 100644 lib/aprsme/system_monitor.ex create mode 100644 lib/aprsme_web/live/map_live/packet_store.ex delete mode 100644 notes.md create mode 100644 priv/repo/migrations/20250714161821_add_performance_indexes.exs create mode 100644 priv/repo/migrations/20250714200255_add_packet_counter_table.exs create mode 100644 priv/repo/migrations/20250715000000_optimize_stations_heard_by_query.exs diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index 546b8ad..d39e08a 100644 --- a/lib/aprsme/application.ex +++ b/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, diff --git a/lib/aprsme/cached_queries.ex b/lib/aprsme/cached_queries.ex index 8075539..b4c88a3 100644 --- a/lib/aprsme/cached_queries.ex +++ b/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 diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 857a13f..607997a 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/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 diff --git a/lib/aprsme/packet_counter.ex b/lib/aprsme/packet_counter.ex new file mode 100644 index 0000000..a0dbe49 --- /dev/null +++ b/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 diff --git a/lib/aprsme/packets.ex b/lib/aprsme/packets.ex index d92544d..fdbf3b9 100644 --- a/lib/aprsme/packets.ex +++ b/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 diff --git a/lib/aprsme/spatial_pubsub.ex b/lib/aprsme/spatial_pubsub.ex new file mode 100644 index 0000000..6a0689b --- /dev/null +++ b/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 diff --git a/lib/aprsme/system_monitor.ex b/lib/aprsme/system_monitor.ex new file mode 100644 index 0000000..d1cfad0 --- /dev/null +++ b/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 diff --git a/lib/aprsme/workers/packet_cleanup_worker.ex b/lib/aprsme/workers/packet_cleanup_worker.ex index 72f3983..d028299 100644 --- a/lib/aprsme/workers/packet_cleanup_worker.ex +++ b/lib/aprsme/workers/packet_cleanup_worker.ex @@ -200,6 +200,4 @@ defmodule Aprsme.Workers.PacketCleanupWorker do ) ) end - - end diff --git a/lib/aprsme_web/live/info_live/show.ex b/lib/aprsme_web/live/info_live/show.ex index a8a37fb..95b17a7 100644 --- a/lib/aprsme_web/live/info_live/show.ex +++ b/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 """ diff --git a/lib/aprsme_web/live/map_live/historical_loader.ex b/lib/aprsme_web/live/map_live/historical_loader.ex index 5fe52d4..56f888b 100644 --- a/lib/aprsme_web/live/map_live/historical_loader.ex +++ b/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 diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index f043ba9..5008a7d 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/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) diff --git a/lib/aprsme_web/live/map_live/packet_store.ex b/lib/aprsme_web/live/map_live/packet_store.ex new file mode 100644 index 0000000..55bfafa --- /dev/null +++ b/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 diff --git a/lib/aprsme_web/live/status_live/index.ex b/lib/aprsme_web/live/status_live/index.ex index ce1338d..ce47430 100644 --- a/lib/aprsme_web/live/status_live/index.ex +++ b/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 diff --git a/lib/aprsme_web/telemetry.ex b/lib/aprsme_web/telemetry.ex index 408ed9d..48649e7 100644 --- a/lib/aprsme_web/telemetry.ex +++ b/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 diff --git a/notes.md b/notes.md deleted file mode 100644 index 479900c..0000000 --- a/notes.md +++ /dev/null @@ -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 - diff --git a/priv/repo/migrations/20250714161821_add_performance_indexes.exs b/priv/repo/migrations/20250714161821_add_performance_indexes.exs new file mode 100644 index 0000000..1f56792 --- /dev/null +++ b/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 diff --git a/priv/repo/migrations/20250714200255_add_packet_counter_table.exs b/priv/repo/migrations/20250714200255_add_packet_counter_table.exs new file mode 100644 index 0000000..2b00ced --- /dev/null +++ b/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 diff --git a/priv/repo/migrations/20250715000000_optimize_stations_heard_by_query.exs b/priv/repo/migrations/20250715000000_optimize_stations_heard_by_query.exs new file mode 100644 index 0000000..6932820 --- /dev/null +++ b/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 diff --git a/test/aprsme/workers/packet_cleanup_worker_test.exs b/test/aprsme/workers/packet_cleanup_worker_test.exs index 0edb9ad..31a2d66 100644 --- a/test/aprsme/workers/packet_cleanup_worker_test.exs +++ b/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