Fix code review batch 3: auth, performance, dedup, cache TTL

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

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.

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.

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

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

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

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)
 

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
 

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

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()

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()