diff --git a/lib/aprsme/cache.ex b/lib/aprsme/cache.ex index 3914d74..da00d41 100644 --- a/lib/aprsme/cache.ex +++ b/lib/aprsme/cache.ex @@ -1,26 +1,55 @@ defmodule Aprsme.Cache do @moduledoc """ Cache abstraction layer using ETS (Erlang Term Storage). - Provides a unified API for in-memory caching. + Provides a unified API for in-memory caching with optional TTL support. + + Entries are stored as `{key, value, expires_at}` tuples where `expires_at` + is a monotonic time in milliseconds, or `:infinity` for no expiration. + Expired entries are lazily evicted on read. """ @doc """ - Get a value from cache + Get a value from cache. Returns `{:ok, nil}` for expired entries. """ def get(cache_name, key) do case :ets.lookup(cache_name, key) do - [{^key, value}] -> {:ok, value} - [] -> {:ok, nil} + [{^key, value, :infinity}] -> + {:ok, value} + + [{^key, value, expires_at}] -> + if System.monotonic_time(:millisecond) < expires_at do + {:ok, value} + else + :ets.delete(cache_name, key) + {:ok, nil} + end + + # Support legacy {key, value} tuples during transition + [{^key, value}] -> + {:ok, value} + + [] -> + {:ok, nil} end rescue ArgumentError -> {:error, :no_cache} end @doc """ - Put a value in cache with optional TTL (TTL not implemented for ETS) + Put a value in cache with optional TTL. + + ## Options + * `:ttl` - Time to live in milliseconds. Use `Cache.to_timeout/1` for convenience. """ - def put(cache_name, key, value, _opts \\ []) do - :ets.insert(cache_name, {key, value}) + def put(cache_name, key, value, opts \\ []) do + expires_at = + case Keyword.get(opts, :ttl) do + nil -> :infinity + ttl when is_integer(ttl) and ttl > 0 -> System.monotonic_time(:millisecond) + ttl + _ -> :infinity + end + + :ets.insert(cache_name, {key, value, expires_at}) {:ok, true} rescue ArgumentError -> {:error, :no_cache} @@ -57,28 +86,34 @@ defmodule Aprsme.Cache do end @doc """ - Check if key exists + Check if key exists and is not expired. """ def exists?(cache_name, key) do - :ets.member(cache_name, key) - rescue - ArgumentError -> false + case get(cache_name, key) do + {:ok, nil} -> false + {:ok, _} -> true + _ -> false + end end @doc """ - Get TTL for a key (not supported in ETS, always returns nil) + Get TTL for a key. Returns remaining milliseconds or nil if no TTL. """ - def ttl(_cache_name, _key) do - {:ok, nil} + def ttl(cache_name, key) do + case :ets.lookup(cache_name, key) do + [{^key, _value, :infinity}] -> {:ok, nil} + [{^key, _value, expires_at}] -> {:ok, max(0, expires_at - System.monotonic_time(:millisecond))} + [{^key, _value}] -> {:ok, nil} + [] -> {:ok, nil} + end + rescue + ArgumentError -> {:ok, nil} end - # Helper functions - no longer needed as we only use Cachex - @doc """ - Convert timeout keyword list to milliseconds + Convert timeout keyword list to milliseconds. """ def to_timeout(opts) do - # Cachex's to_timeout is private, so we implement our own Enum.reduce(opts, 0, fn {:second, n}, acc -> acc + n * 1000 {:seconds, n}, acc -> acc + n * 1000 diff --git a/lib/aprsme/packets.ex b/lib/aprsme/packets.ex index 4e91b5a..29a3f07 100644 --- a/lib/aprsme/packets.ex +++ b/lib/aprsme/packets.ex @@ -848,6 +848,15 @@ defmodule Aprsme.Packets do PreparedQueries.has_weather_packets?(callsign) end + @doc """ + Check which callsigns from a list have weather packets. + Returns a MapSet of uppercased callsigns with weather data. + """ + @spec weather_callsigns(list(String.t())) :: MapSet.t(String.t()) + def weather_callsigns(callsigns) when is_list(callsigns) do + PreparedQueries.weather_callsigns(callsigns) + end + @doc """ Gets other SSIDs for a given callsign's base callsign. Returns a list of maps with callsign, ssid, received_at, and packet info. diff --git a/lib/aprsme/packets/prepared_queries.ex b/lib/aprsme/packets/prepared_queries.ex index 73db4ec..051f3b2 100644 --- a/lib/aprsme/packets/prepared_queries.ex +++ b/lib/aprsme/packets/prepared_queries.ex @@ -69,6 +69,29 @@ defmodule Aprsme.Packets.PreparedQueries do Repo.exists?(query) end + @doc """ + Check which callsigns from a list have weather packets. + Returns a MapSet of callsigns (uppercased) that have weather data. + Single query replaces N individual has_weather_packets? calls. + """ + @spec weather_callsigns(list(String.t())) :: MapSet.t(String.t()) + def weather_callsigns(callsigns) when is_list(callsigns) do + if callsigns == [] do + MapSet.new() + else + normalized = Enum.map(callsigns, &String.upcase(String.trim(&1))) + + from(p in Packet, + where: fragment("upper(?)", p.sender) in ^normalized, + where: p.has_weather == true, + distinct: fragment("upper(?)", p.sender), + select: fragment("upper(?)", p.sender) + ) + |> Repo.all() + |> MapSet.new() + end + end + @doc """ Get nearby stations using KNN (K-nearest neighbors) search. Uses the <-> operator for efficient spatial queries. diff --git a/lib/aprsme/spatial_pubsub.ex b/lib/aprsme/spatial_pubsub.ex index d35b379..fe59d24 100644 --- a/lib/aprsme/spatial_pubsub.ex +++ b/lib/aprsme/spatial_pubsub.ex @@ -5,8 +5,6 @@ defmodule Aprsme.SpatialPubSub do """ use GenServer - alias Phoenix.PubSub - require Logger # Grid size in degrees for spatial indexing @@ -62,9 +60,6 @@ defmodule Aprsme.SpatialPubSub do @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) @@ -187,12 +182,6 @@ defmodule Aprsme.SpatialPubSub do 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 diff --git a/lib/aprsme/streaming_packets_pubsub.ex b/lib/aprsme/streaming_packets_pubsub.ex index 397064c..27fbccb 100644 --- a/lib/aprsme/streaming_packets_pubsub.ex +++ b/lib/aprsme/streaming_packets_pubsub.ex @@ -109,15 +109,8 @@ defmodule Aprsme.StreamingPacketsPubSub do lon = packet[:longitude] || packet[:lon] || packet[:lng] if lat && lon do - # Find all subscribers whose bounds contain this packet - subscribers = - :ets.select(@table_name, [ - { - {:"$1", :"$2"}, - [], - [{{:"$1", :"$2"}}] - } - ]) + # Get all subscribers — table is small (one entry per LiveView client) + subscribers = :ets.tab2list(@table_name) # Send to matching subscribers using BroadcastTaskSupervisor # Collect dead pids to clean up diff --git a/lib/aprsme_web/live/map_live/data_builder.ex b/lib/aprsme_web/live/map_live/data_builder.ex index 2255091..d027f17 100644 --- a/lib/aprsme_web/live/map_live/data_builder.ex +++ b/lib/aprsme_web/live/map_live/data_builder.ex @@ -33,11 +33,25 @@ defmodule AprsmeWeb.MapLive.DataBuilder do def build_packet_data_list_from_map(packets_map, is_most_recent, socket) do locale = get_locale(socket) - packets_map - |> Enum.map(fn {_callsign, packet} -> - build_packet_data(packet, is_most_recent, locale) - end) - |> Enum.filter(& &1) + # Preload weather callsigns in one batch query instead of N+1 individual queries + callsigns = + packets_map + |> Map.values() + |> Enum.map(&display_name/1) + |> Enum.uniq() + + weather_set = Aprsme.Packets.weather_callsigns(callsigns) + Process.put(:weather_callsigns_cache, weather_set) + + result = + packets_map + |> Enum.map(fn {_callsign, packet} -> + build_packet_data(packet, is_most_recent, locale) + end) + |> Enum.filter(& &1) + + Process.delete(:weather_callsigns_cache) + result end @doc """ @@ -474,7 +488,10 @@ defmodule AprsmeWeb.MapLive.DataBuilder do @spec has_weather_packets?(String.t()) :: boolean() defp has_weather_packets?(callsign) when is_binary(callsign) do - Aprsme.Packets.has_weather_packets?(callsign) + case Process.get(:weather_callsigns_cache) do + nil -> Aprsme.Packets.has_weather_packets?(callsign) + weather_set -> MapSet.member?(weather_set, String.upcase(String.trim(callsign))) + end rescue _ -> false end diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index 52764d3..0e15d66 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -1762,12 +1762,12 @@ defmodule AprsmeWeb.MapLive.Index do current_packets = PacketManager.get_visible_packets(socket.assigns.packet_state) remaining_packets = PacketManager.get_visible_packets(updated_packet_state) + remaining_keys = MapSet.new(remaining_packets, &get_callsign_key/1) + expired_keys = current_packets |> Enum.reject(fn current_packet -> - Enum.any?(remaining_packets, fn remaining_packet -> - get_callsign_key(current_packet) == get_callsign_key(remaining_packet) - end) + MapSet.member?(remaining_keys, get_callsign_key(current_packet)) end) |> Enum.map(&get_callsign_key/1) diff --git a/lib/aprsme_web/router.ex b/lib/aprsme_web/router.ex index e086c42..f9890d3 100644 --- a/lib/aprsme_web/router.ex +++ b/lib/aprsme_web/router.ex @@ -42,9 +42,13 @@ defmodule AprsmeWeb.Router do end scope "/", AprsmeWeb do - pipe_through [:browser] + pipe_through [:browser, :require_authenticated_user] live_dashboard "/dashboard", metrics: AprsmeWeb.Telemetry error_tracker_dashboard("/errors") + end + + scope "/", AprsmeWeb do + pipe_through [:browser] delete "/users/log_out", UserSessionController, :delete diff --git a/test/aprsme/cache_test.exs b/test/aprsme/cache_test.exs new file mode 100644 index 0000000..b753ee9 --- /dev/null +++ b/test/aprsme/cache_test.exs @@ -0,0 +1,142 @@ +defmodule Aprsme.CacheTest do + use ExUnit.Case, async: true + + alias Aprsme.Cache + + @table_name :cache_test_table + + setup do + :ets.new(@table_name, [:set, :public, :named_table]) + + on_exit(fn -> + try do + :ets.delete(@table_name) + rescue + ArgumentError -> :ok + end + end) + + :ok + end + + describe "put/4 and get/2" do + test "stores and retrieves a value" do + assert {:ok, true} = Cache.put(@table_name, :key, "value") + assert {:ok, "value"} = Cache.get(@table_name, :key) + end + + test "returns nil for missing keys" do + assert {:ok, nil} = Cache.get(@table_name, :missing) + end + + test "overwrites existing values" do + Cache.put(@table_name, :key, "first") + Cache.put(@table_name, :key, "second") + assert {:ok, "second"} = Cache.get(@table_name, :key) + end + end + + describe "TTL support" do + test "entry with TTL is returned before expiry" do + Cache.put(@table_name, :key, "value", ttl: 60_000) + assert {:ok, "value"} = Cache.get(@table_name, :key) + end + + test "entry with expired TTL returns nil" do + # Use a TTL of 1ms and sleep briefly + Cache.put(@table_name, :key, "value", ttl: 1) + Process.sleep(5) + assert {:ok, nil} = Cache.get(@table_name, :key) + end + + test "entry without TTL never expires" do + Cache.put(@table_name, :key, "value") + assert {:ok, "value"} = Cache.get(@table_name, :key) + end + + test "expired entry is lazily deleted from ETS" do + Cache.put(@table_name, :key, "value", ttl: 1) + Process.sleep(5) + + # First get triggers lazy deletion + Cache.get(@table_name, :key) + + # Verify entry is gone from ETS + assert [] = :ets.lookup(@table_name, :key) + end + end + + describe "exists?/2" do + test "returns true for existing key" do + Cache.put(@table_name, :key, "value") + assert Cache.exists?(@table_name, :key) + end + + test "returns false for missing key" do + refute Cache.exists?(@table_name, :missing) + end + + test "returns false for expired key" do + Cache.put(@table_name, :key, "value", ttl: 1) + Process.sleep(5) + refute Cache.exists?(@table_name, :key) + end + end + + describe "ttl/2" do + test "returns nil for entries without TTL" do + Cache.put(@table_name, :key, "value") + assert {:ok, nil} = Cache.ttl(@table_name, :key) + end + + test "returns remaining time for entries with TTL" do + Cache.put(@table_name, :key, "value", ttl: 60_000) + assert {:ok, remaining} = Cache.ttl(@table_name, :key) + assert is_integer(remaining) + assert remaining > 0 + assert remaining <= 60_000 + end + + test "returns nil for missing keys" do + assert {:ok, nil} = Cache.ttl(@table_name, :missing) + end + end + + describe "del/2" do + test "deletes an existing key" do + Cache.put(@table_name, :key, "value") + assert {:ok, true} = Cache.del(@table_name, :key) + assert {:ok, nil} = Cache.get(@table_name, :key) + end + end + + describe "clear/1" do + test "removes all entries" do + Cache.put(@table_name, :a, 1) + Cache.put(@table_name, :b, 2) + assert {:ok, true} = Cache.clear(@table_name) + assert {:ok, nil} = Cache.get(@table_name, :a) + assert {:ok, nil} = Cache.get(@table_name, :b) + end + end + + describe "to_timeout/1" do + test "converts time units to milliseconds" do + assert Cache.to_timeout(hour: 1) == 3_600_000 + assert Cache.to_timeout(minute: 30) == 1_800_000 + assert Cache.to_timeout(day: 1) == 86_400_000 + assert Cache.to_timeout(second: 5) == 5_000 + assert Cache.to_timeout(hours: 2, minutes: 30) == 9_000_000 + end + end + + describe "error handling" do + test "get returns error for non-existent table" do + assert {:error, :no_cache} = Cache.get(:nonexistent_table, :key) + end + + test "put returns error for non-existent table" do + assert {:error, :no_cache} = Cache.put(:nonexistent_table, :key, "value") + end + end +end diff --git a/test/aprsme/packets_test.exs b/test/aprsme/packets_test.exs index e132b12..68200e7 100644 --- a/test/aprsme/packets_test.exs +++ b/test/aprsme/packets_test.exs @@ -602,6 +602,30 @@ defmodule Aprsme.PacketsTest do end end + describe "weather_callsigns/1" do + test "returns callsigns that have weather packets" do + PacketsFixtures.packet_fixture(%{sender: "WX-BATCH1", temperature: 72.0}) + PacketsFixtures.packet_fixture(%{sender: "WX-BATCH2", wind_speed: 10.0}) + PacketsFixtures.packet_fixture(%{sender: "NOWX-BATCH", comment: "No weather"}) + + result = Packets.weather_callsigns(["WX-BATCH1", "WX-BATCH2", "NOWX-BATCH"]) + assert MapSet.member?(result, "WX-BATCH1") + assert MapSet.member?(result, "WX-BATCH2") + refute MapSet.member?(result, "NOWX-BATCH") + end + + test "returns empty set for empty input" do + assert MapSet.new() == Packets.weather_callsigns([]) + end + + test "is case-insensitive" do + PacketsFixtures.packet_fixture(%{sender: "WX-CASE", temperature: 72.0}) + + result = Packets.weather_callsigns(["wx-case"]) + assert MapSet.member?(result, "WX-CASE") + end + end + describe "get_packets_for_replay/1" do setup do now = DateTime.utc_now() diff --git a/test/aprsme/spatial_pubsub_test.exs b/test/aprsme/spatial_pubsub_test.exs index aea4ebc..b6633f2 100644 --- a/test/aprsme/spatial_pubsub_test.exs +++ b/test/aprsme/spatial_pubsub_test.exs @@ -36,6 +36,23 @@ defmodule Aprsme.SpatialPubSubTest do end end + describe "duplicate packet prevention" do + test "does not subscribe to postgres PubSub topic" do + # SpatialPubSub used to subscribe to "postgres:aprsme_packets" causing + # duplicate broadcasts (once from PacketConsumer direct cast, once from PubSub). + # Verify it no longer subscribes. + # The SpatialPubSub process should NOT be in the subscriber list for postgres topic + spatial_pid = Process.whereis(SpatialPubSub) + + subscribed? = + Aprsme.PubSub + |> Registry.lookup("postgres:aprsme_packets") + |> Enum.any?(fn {pid, _} -> pid == spatial_pid end) + + refute subscribed?, "SpatialPubSub should not subscribe to postgres:aprsme_packets" + end + end + describe "stats" do test "get_stats returns expected keys" do stats = SpatialPubSub.get_stats()