perf: reduce per-packet allocations in ingest pipeline

Drop the pre-insert aprs_messages broadcast in Is.dispatch (subscribers
already get a richer payload via postgres:aprsme_packets after insert).
Switch PacketsLive.CallsignView to the per-callsign packets:<CS> topic
so it only receives relevant packets instead of filtering every one.

In PacketConsumer: reuse the received_at stamped in Is.dispatch instead
of calling DateTime.utc_now/0 per packet; drop the duplicate struct_to_map
pass (Is.dispatch already handled it); fold coordinate validation and
Geo.Point construction into set_lat_lon + create_location_geometry so
coords are normalized once; extract device_identifier from the already-
normalized attrs; pre-build the broadcast payload once (identifier pick,
lat/lon aliases, routing callsign) so the async broadcast task no longer
does Map.drop/Map.merge per packet.

Rewrite PacketSanitizer.sanitize_packet / sanitize_data_map with Map.new/2
instead of Enum.reduce + Map.put — one allocation per packet instead of N.

Also compute has_weather in Packet.changeset/2 so direct-changeset inserts
(tests, backfills) populate it the same way the GenStage pipeline does.

lib/aprsme/is/is.ex

@@ -577,9 +577,6 @@ defmodule Aprsme.Is do
             Logger.debug("Parsed message: #{inspect(parsed_message)}")
         end
 
-        # Broadcast to live clients
-        AprsmeWeb.Endpoint.broadcast("aprs_messages", "packet", parsed_message)
-
       {:error, :invalid_packet} ->
         Logger.debug("PARSE ERROR: invalid packet")
 

lib/aprsme/packet.ex

@@ -147,11 +147,24 @@ defmodule Aprsme.Packet do
     changeset
     |> maybe_create_geometry_from_lat_lon()
     |> maybe_set_has_position()
+    |> maybe_set_has_weather()
     |> normalize_symbols()
     |> normalize_course()
     |> normalize_wind_direction()
   end
 
+  # Mirrors PacketConsumer.set_has_weather/1 so direct changeset inserts
+  # (tests, backfills, anything bypassing the GenStage pipeline) still get
+  # has_weather populated — previously a DB trigger did this.
+  defp maybe_set_has_weather(changeset) do
+    has_weather? =
+      Enum.any?(Aprsme.EncodingUtils.weather_fields(), fn field ->
+        not is_nil(get_field(changeset, field))
+      end)
+
+    put_change(changeset, :has_weather, has_weather?)
+  end
+
   defp normalize_symbols(changeset) do
     # Only normalize if packet has position
     if get_field(changeset, :has_position) do

lib/aprsme/packet_consumer.ex

@@ -5,11 +5,9 @@ defmodule Aprsme.PacketConsumer do
   """
   use GenStage
 
-  alias Aprs.Types.ParseError
   alias Aprsme.Cluster.PacketDistributor
   alias Aprsme.LogSanitizer
   alias Aprsme.Repo
-  alias AprsmeWeb.Live.Shared.CoordinateUtils
 
   require Logger
 
@@ -254,23 +252,25 @@ defmodule Aprsme.PacketConsumer do
   defp process_chunk(packets) do
     # Use Stream for memory-efficient packet preparation
     # Note: truncate_datetimes_to_second is already called inside prepare_packet_for_insert
+    # prepare_packet_for_insert returns {insert_attrs, broadcast_info} or nil
     packet_stream =
       packets
       |> Stream.map(&prepare_packet_for_insert/1)
       |> Stream.reject(&is_nil/1)
 
     # Separate valid and invalid packets using Stream
-    {valid_packets, invalid_count} =
-      Enum.reduce(packet_stream, {[], 0}, fn packet, {valid_acc, invalid_acc} ->
-        if valid_packet?(packet) do
-          {[packet | valid_acc], invalid_acc}
+    {valid_pairs, invalid_count} =
+      Enum.reduce(packet_stream, {[], 0}, fn {insert_attrs, _bcast} = pair, {valid_acc, invalid_acc} ->
+        if valid_packet?(insert_attrs) do
+          {[pair | valid_acc], invalid_acc}
         else
           {valid_acc, invalid_acc + 1}
         end
       end)
 
-    # Reverse to maintain order
-    valid_packets = Enum.reverse(valid_packets)
+    # Reverse to maintain order and split into parallel lists
+    valid_pairs = Enum.reverse(valid_pairs)
+    valid_inserts = Enum.map(valid_pairs, fn {insert, _bcast} -> insert end)
 
     # Insert valid packets in batch
     # Optimized for PostgreSQL with synchronous_commit=off
@@ -285,10 +285,9 @@ defmodule Aprsme.PacketConsumer do
     ]
 
     try do
-      {inserted_count, _} = Repo.insert_all(Aprsme.Packet, valid_packets, insert_opts)
+      {inserted_count, _} = Repo.insert_all(Aprsme.Packet, valid_inserts, insert_opts)
 
-      # Broadcast successfully inserted packets to StreamingPacketsPubSub
-      broadcast_packets_async(valid_packets)
+      broadcast_packets_async(Enum.map(valid_pairs, fn {_insert, bcast} -> bcast end))
 
       {inserted_count, invalid_count}
     rescue
@@ -296,28 +295,28 @@ defmodule Aprsme.PacketConsumer do
         Logger.error("Batch insert failed: #{inspect(error)}, falling back to individual inserts")
 
         # Fall back to individual inserts so partial success is possible
-        {fallback_inserted, fallback_packets} = insert_individually(valid_packets)
+        {fallback_inserted, fallback_bcasts} = insert_individually(valid_pairs)
 
         # Broadcast whatever was successfully inserted
-        if fallback_packets != [] do
-          broadcast_packets_async(fallback_packets)
+        if fallback_bcasts != [] do
+          broadcast_packets_async(fallback_bcasts)
         end
 
-        {fallback_inserted, invalid_count + length(valid_packets) - fallback_inserted}
+        {fallback_inserted, invalid_count + length(valid_pairs) - fallback_inserted}
     end
   end
 
   # Fall back to inserting packets one at a time when batch insert fails.
-  # Returns {inserted_count, list_of_successfully_inserted_packet_attrs}.
-  @spec insert_individually(list(map())) :: {non_neg_integer(), list(map())}
-  defp insert_individually(packets) do
-    Enum.reduce(packets, {0, []}, fn packet_attrs, {count, inserted} ->
+  # Returns {inserted_count, list_of_broadcast_infos_for_successes}.
+  @spec insert_individually(list({map(), map()})) :: {non_neg_integer(), list(map())}
+  defp insert_individually(pairs) do
+    Enum.reduce(pairs, {0, []}, fn {insert_attrs, bcast}, {count, inserted} ->
       try do
-        changeset = Aprsme.Packet.changeset(%Aprsme.Packet{}, packet_attrs)
+        changeset = Aprsme.Packet.changeset(%Aprsme.Packet{}, insert_attrs)
 
         case Repo.insert(changeset, on_conflict: :nothing, conflict_target: [:id, :received_at]) do
           {:ok, _record} ->
-            {count + 1, [packet_attrs | inserted]}
+            {count + 1, [bcast | inserted]}
 
           {:error, _changeset} ->
             {count, inserted}
@@ -331,10 +330,10 @@ defmodule Aprsme.PacketConsumer do
   end
 
   # Broadcast packets asynchronously using supervised task pool
-  defp broadcast_packets_async(packets) do
+  defp broadcast_packets_async(bcasts) do
     fun = fn ->
       cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
-      Enum.each(packets, &broadcast_single_packet(&1, cluster_enabled))
+      Enum.each(bcasts, &broadcast_single_packet(&1, cluster_enabled))
     end
 
     if test_env?() do
@@ -345,28 +344,10 @@ defmodule Aprsme.PacketConsumer do
     end
   end
 
-  defp broadcast_single_packet(packet_attrs, cluster_enabled) do
-    # For objects and items, use object_name/item_name as the identifier
-    # instead of sender (which is the origin station)
-    identifier =
-      cond do
-        packet_attrs[:object_name] -> packet_attrs[:object_name]
-        packet_attrs[:item_name] -> packet_attrs[:item_name]
-        true -> packet_attrs[:sender]
-      end
-
-    # Build a rich packet payload that replaces the JSON produced by the old
-    # packets_notify_insert trigger. Includes latitude/longitude aliases so
-    # downstream consumers that expect either :lat/:lon or :latitude/:longitude work.
-    packet =
-      packet_attrs
-      |> Map.drop([:location, :inserted_at, :updated_at])
-      |> Map.merge(%{
-        sender: identifier,
-        latitude: packet_attrs[:lat],
-        longitude: packet_attrs[:lon]
-      })
-
+  defp broadcast_single_packet(
+         %{payload: packet, routing_callsign: routing_callsign, has_weather: has_weather?},
+         cluster_enabled
+       ) do
     if cluster_enabled do
       PacketDistributor.distribute_packet(packet)
     else
@@ -374,21 +355,18 @@ defmodule Aprsme.PacketConsumer do
       Aprsme.SpatialPubSub.broadcast_packet(packet)
     end
 
-    broadcast_legacy_topics(packet, packet_attrs)
+    broadcast_legacy_topics(packet, routing_callsign, has_weather?)
   end
 
   # Replaces the `aprs_packets` pg_notify path (removed as a DB trigger).
-  defp broadcast_legacy_topics(packet, packet_attrs) do
+  defp broadcast_legacy_topics(packet, routing_callsign, has_weather?) do
     Phoenix.PubSub.broadcast(Aprsme.PubSub, "postgres:aprsme_packets", {:postgres_packet, packet})
 
-    callsign = packet_attrs[:sender] || packet_attrs[:base_callsign]
+    if is_binary(routing_callsign) and routing_callsign != "" do
+      Phoenix.PubSub.broadcast(Aprsme.PubSub, "packets:#{routing_callsign}", {:postgres_packet, packet})
 
-    if is_binary(callsign) and callsign != "" do
-      normalized = callsign |> String.trim() |> String.upcase()
-      Phoenix.PubSub.broadcast(Aprsme.PubSub, "packets:#{normalized}", {:postgres_packet, packet})
-
-      if packet_attrs[:has_weather] do
-        Phoenix.PubSub.broadcast(Aprsme.PubSub, "weather:#{normalized}", {:weather_packet, packet})
+      if has_weather? do
+        Phoenix.PubSub.broadcast(Aprsme.PubSub, "weather:#{routing_callsign}", {:weather_packet, packet})
       end
     end
   end
@@ -398,12 +376,15 @@ defmodule Aprsme.PacketConsumer do
   end
 
   defp prepare_packet_for_insert(packet_data) do
-    # Always set received_at timestamp to ensure consistency
-    current_time = DateTime.truncate(DateTime.utc_now(), :microsecond)
-    packet_data = Map.put(packet_data, :received_at, current_time)
+    # Reuse the received_at stamped in Is.dispatch; fall back if missing.
+    # packet_data is already a plain map — Is.dispatch ran struct_to_map before submitting.
+    current_time =
+      case Map.get(packet_data, :received_at) do
+        %DateTime{} = dt -> DateTime.truncate(dt, :microsecond)
+        _ -> DateTime.truncate(DateTime.utc_now(), :microsecond)
+      end
 
-    # Convert to map before storing to avoid struct conversion issues
-    attrs = struct_to_map(packet_data)
+    attrs = Map.put(packet_data, :received_at, current_time)
 
     # Extract additional data from the parsed packet including raw packet
     attrs = Aprsme.Packet.extract_additional_data(attrs, attrs[:raw_packet] || "")
@@ -417,43 +398,74 @@ defmodule Aprsme.PacketConsumer do
     # Normalize data_type to string if it's an atom
     attrs = normalize_data_type(attrs)
 
-    # Apply the same processing as the original store_packet function
-    attrs
-    |> convert_coordinate_field_names()
-    |> convert_field_names()
-    |> normalize_packet_attrs()
-    |> set_received_at()
-    |> patch_lat_lon_from_data_extended()
-    |> then(fn attrs ->
-      {lat, lon} = extract_position(attrs)
-      set_lat_lon(attrs, lat, lon)
-    end)
-    |> normalize_ssid()
-    |> then(fn attrs ->
-      device_identifier = Aprsme.DeviceParser.extract_device_identifier(packet_data)
-      Map.put(attrs, :device_identifier, device_identifier)
-    end)
-    |> sanitize_packet_strings()
-    |> Map.put(:inserted_at, current_time)
-    |> Map.put(:updated_at, current_time)
-    # Generate id in Elixir so post-insert broadcasts can reference it
-    |> Map.put(:id, Ecto.UUID.generate())
-    |> Map.delete("id")
-    # Remove embedded field for batch insert
-    |> Map.delete(:data_extended)
-    |> normalize_numeric_types()
-    |> truncate_datetimes_to_second()
-    # Create PostGIS geometry for location field BEFORE filtering
-    |> create_location_geometry()
-    |> set_has_weather()
-    # Remove all non-schema fields - do this LAST to ensure all processing is done
-    |> remove_non_schema_fields()
+    # Apply the same processing as the original store_packet function.
+    # `enriched` holds everything needed for both insert and broadcast;
+    # we pre-build the broadcast payload once so the async broadcast path
+    # doesn't have to Map.drop/Map.merge/pick-identifier per packet.
+    enriched =
+      attrs
+      |> convert_coordinate_field_names()
+      |> convert_field_names()
+      |> normalize_packet_attrs()
+      |> set_received_at()
+      |> patch_lat_lon_from_data_extended()
+      |> then(fn attrs ->
+        {lat, lon} = extract_position(attrs)
+        set_lat_lon(attrs, lat, lon)
+      end)
+      |> normalize_ssid()
+      |> then(fn attrs ->
+        Map.put(attrs, :device_identifier, Aprsme.DeviceParser.extract_device_identifier(attrs))
+      end)
+      |> sanitize_packet_strings()
+      |> create_location_geometry()
+      |> Map.put(:inserted_at, current_time)
+      |> Map.put(:updated_at, current_time)
+      |> Map.put(:id, Ecto.UUID.generate())
+      |> Map.delete("id")
+      |> Map.delete(:data_extended)
+      |> normalize_numeric_types()
+      |> truncate_datetimes_to_second()
+      |> set_has_weather()
+
+    insert_attrs = remove_non_schema_fields(enriched)
+    bcast_info = build_broadcast_info(enriched)
+
+    {insert_attrs, bcast_info}
   rescue
     error ->
       Logger.error("Failed to prepare packet for batch insert: #{inspect(error)}")
       nil
   end
 
+  # Pre-build the broadcast payload and routing info once per packet.
+  # `:sender` is replaced with the display identifier (object/item name)
+  # while `routing_callsign` preserves the original sender for topic routing.
+  defp build_broadcast_info(enriched) do
+    identifier =
+      enriched[:object_name] || enriched[:item_name] || enriched[:sender]
+
+    payload =
+      enriched
+      |> Map.drop([:location, :inserted_at, :updated_at])
+      |> Map.put(:sender, identifier)
+      |> Map.put(:latitude, enriched[:lat])
+      |> Map.put(:longitude, enriched[:lon])
+
+    original_sender = enriched[:sender] || enriched[:base_callsign]
+
+    routing_callsign =
+      if is_binary(original_sender) and original_sender != "" do
+        original_sender |> String.trim() |> String.upcase()
+      end
+
+    %{
+      payload: payload,
+      routing_callsign: routing_callsign,
+      has_weather: enriched[:has_weather] == true
+    }
+  end
+
   # Compute has_weather from any populated weather field (replaces DB trigger)
   defp set_has_weather(attrs) do
     has_weather? =
@@ -464,51 +476,12 @@ defmodule Aprsme.PacketConsumer do
     Map.put(attrs, :has_weather, has_weather?)
   end
 
-  # Create PostGIS geometry from lat/lon coordinates
-  defp create_location_geometry(attrs) do
-    lat = attrs[:lat]
-    lon = attrs[:lon]
-
-    if valid_coordinates?(lat, lon) do
-      location = create_point(lat, lon)
-
-      if location do
-        Map.put(attrs, :location, location)
-      else
-        attrs
-      end
-    else
-      attrs
-    end
-  end
-
   # Helper function to remove fields that exist in parser output but not in database schema
   defp remove_non_schema_fields(attrs) do
     # Use whitelist approach - only keep fields that are in our schema
     Aprsme.PacketFieldWhitelist.filter_fields(attrs)
   end
 
-  # Helper functions for coordinate validation and point creation
-  defp valid_coordinates?(lat, lon) do
-    CoordinateUtils.valid_coordinates_any_type?(lat, lon)
-  end
-
-  defp normalize_coordinate(coord) do
-    CoordinateUtils.normalize_coordinate(coord)
-  end
-
-  defp create_point(lat, lon)
-       when (is_number(lat) or is_struct(lat, Decimal)) and (is_number(lon) or is_struct(lon, Decimal)) do
-    lat = normalize_coordinate(lat)
-    lon = normalize_coordinate(lon)
-
-    if valid_coordinates?(lat, lon) do
-      %Geo.Point{coordinates: {lon, lat}, srid: 4326}
-    end
-  end
-
-  defp create_point(_, _), do: nil
-
   defp valid_packet?(nil), do: false
   defp valid_packet?(%{sender: sender}) when is_binary(sender) and byte_size(sender) > 0, do: true
   defp valid_packet?(_), do: false
@@ -704,21 +677,28 @@ defmodule Aprsme.PacketConsumer do
          (ext_map["position"][:longitude] || ext_map["position"]["longitude"]))
   end
 
+  # extract_position already ran to_float; lat/lon are floats or nil here.
+  # Round and store whatever we have; range validation happens in
+  # create_location_geometry/1 where it controls Geo.Point construction.
   defp set_lat_lon(attrs, lat, lon) do
-    round6 = fn
-      nil ->
-        nil
-
-      n when is_float(n) ->
-        Float.round(n, 6)
-    end
-
     attrs
-    |> Map.put(:lat, round6.(lat))
-    |> Map.put(:lon, round6.(lon))
-    |> Map.put(:has_position, not is_nil(lat) and not is_nil(lon))
+    |> Map.put(:lat, round_coord(lat))
+    |> Map.put(:lon, round_coord(lon))
+    |> Map.put(:has_position, is_float(lat) and is_float(lon))
   end
 
+  defp round_coord(n) when is_float(n), do: Float.round(n, 6)
+  defp round_coord(_), do: nil
+
+  # Build the PostGIS point only for in-range coordinates.
+  # Must run AFTER sanitize_packet_strings so the struct survives intact.
+  defp create_location_geometry(%{lat: lat, lon: lon} = attrs)
+       when is_float(lat) and is_float(lon) and lat >= -90.0 and lat <= 90.0 and lon >= -180.0 and lon <= 180.0 do
+    Map.put(attrs, :location, %Geo.Point{coordinates: {lon, lat}, srid: 4326})
+  end
+
+  defp create_location_geometry(attrs), do: attrs
+
   defp normalize_ssid(attrs) do
     case Map.get(attrs, :ssid) do
       nil -> attrs
@@ -730,30 +710,6 @@ defmodule Aprsme.PacketConsumer do
 
   defp normalize_data_type(attrs), do: Aprsme.EncodingUtils.normalize_data_type(attrs)
 
-  defp struct_to_map(%{__struct__: ParseError} = error) do
-    # Handle ParseError specially to avoid Access behavior issues
-    %{
-      error_code: error.error_code,
-      message: error.error_message,
-      __original_struct__: ParseError
-    }
-  end
-
-  defp struct_to_map(%{__struct__: struct_type} = struct) do
-    converted_map =
-      struct
-      |> Map.from_struct()
-      |> Map.new(fn {k, v} -> {k, struct_to_map(v)} end)
-
-    Map.put(converted_map, :__original_struct__, struct_type)
-  end
-
-  defp struct_to_map(value) when is_list(value) do
-    Enum.map(value, &struct_to_map/1)
-  end
-
-  defp struct_to_map(value), do: value
-
   defp truncate_datetimes_to_second(%DateTime{} = dt), do: DateTime.truncate(dt, :second)
   defp truncate_datetimes_to_second({:ok, %DateTime{} = dt}), do: DateTime.truncate(dt, :second)
   defp truncate_datetimes_to_second({:error, _reason}), do: nil

lib/aprsme/packet_sanitizer.ex

@@ -47,10 +47,7 @@ defmodule Aprsme.PacketSanitizer do
   """
   @spec sanitize_packet(map()) :: map()
   def sanitize_packet(packet) when is_map(packet) do
-    Enum.reduce(packet, %{}, fn {key, value}, acc ->
-      sanitized_value = sanitize_field(key, value)
-      Map.put(acc, key, sanitized_value)
-    end)
+    Map.new(packet, fn {key, value} -> {key, sanitize_field(key, value)} end)
   end
 
   defp sanitize_field(:data, value) when is_map(value) do
@@ -71,18 +68,18 @@ defmodule Aprsme.PacketSanitizer do
   defp sanitize_field(_key, value), do: value
 
   defp sanitize_data_map(data) when is_map(data) do
-    Enum.reduce(data, %{}, fn {key, value}, acc ->
-      sanitized =
-        case {is_binary(value), Map.get(@data_string_max_lengths, key)} do
-          {true, nil} -> strip_null_bytes(value)
-          {true, max_length} -> value |> strip_null_bytes() |> truncate_string(max_length)
-          _ -> value
-        end
-
-      Map.put(acc, key, sanitized)
-    end)
+    Map.new(data, fn {key, value} -> {key, sanitize_data_value(key, value)} end)
   end
 
+  defp sanitize_data_value(key, value) when is_binary(value) do
+    case Map.get(@data_string_max_lengths, key) do
+      nil -> strip_null_bytes(value)
+      max_length -> value |> strip_null_bytes() |> truncate_string(max_length)
+    end
+  end
+
+  defp sanitize_data_value(_key, value), do: value
+
   # PostgreSQL JSONB does not support \u0000 (null bytes)
   defp strip_null_bytes(string) when is_binary(string) do
     String.replace(string, <<0x00>>, "")

lib/aprsme_web/live/map_live/index.ex

@@ -13,7 +13,6 @@ defmodule AprsmeWeb.MapLive.Index do
 
   alias Aprsme.Packets
   alias Aprsme.Packets.Clustering
-  alias AprsmeWeb.Endpoint
   alias AprsmeWeb.Live.Shared.BoundsUtils
   alias AprsmeWeb.Live.Shared.CoordinateUtils
   alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils
@@ -156,7 +155,6 @@ defmodule AprsmeWeb.MapLive.Index do
   end
 
   defp do_setup_additional_subscriptions(socket, true) do
-    Endpoint.subscribe("aprs_messages")
     Phoenix.PubSub.subscribe(Aprsme.PubSub, "postgres:aprsme_packets")
     socket
   end
@@ -973,9 +971,6 @@ defmodule AprsmeWeb.MapLive.Index do
     end
   end
 
-  def handle_info(%Broadcast{topic: "aprs_messages", event: "packet", payload: packet}, socket),
-    do: handle_info({:postgres_packet, packet}, socket)
-
   def handle_info({:show_error, message}, socket) do
     socket = put_flash(socket, :error, message)
     {:noreply, socket}
@@ -1878,8 +1873,7 @@ defmodule AprsmeWeb.MapLive.Index do
     # Unsubscribe from StreamingPacketsPubSub
     Aprsme.StreamingPacketsPubSub.unsubscribe(self())
 
-    # Unsubscribe from Endpoint/PubSub topics subscribed in mount
-    Endpoint.unsubscribe("aprs_messages")
+    # Unsubscribe from PubSub topic subscribed in mount
     Phoenix.PubSub.unsubscribe(Aprsme.PubSub, "postgres:aprsme_packets")
 
     if socket.assigns.buffer_timer, do: Process.cancel_timer(socket.assigns.buffer_timer)

lib/aprsme_web/live/packets_live/callsign_view.ex

@@ -9,7 +9,6 @@ defmodule AprsmeWeb.PacketsLive.CallsignView do
   alias Aprsme.EncodingUtils
   alias Aprsme.Packet
   alias Aprsme.Repo
-  alias AprsmeWeb.Endpoint
   alias AprsmeWeb.Live.SharedPacketHandler
 
   @impl true
@@ -18,9 +17,9 @@ defmodule AprsmeWeb.PacketsLive.CallsignView do
     normalized_callsign = Callsign.normalize(callsign)
 
     if Callsign.valid?(normalized_callsign) do
-      # Subscribe to live packet updates if connected
+      # Subscribe to live packet updates for this callsign only
       if connected?(socket) do
-        Endpoint.subscribe("aprs_messages")
+        Phoenix.PubSub.subscribe(Aprsme.PubSub, "packets:#{normalized_callsign}")
       end
 
       # Get stored packets for this callsign (up to 100)
@@ -56,9 +55,9 @@ defmodule AprsmeWeb.PacketsLive.CallsignView do
   end
 
   @impl true
-  def handle_info(%{event: "packet", payload: payload}, socket) do
+  def handle_info({:postgres_packet, payload}, socket) do
     SharedPacketHandler.handle_packet_update(payload, socket,
-      filter_fn: SharedPacketHandler.callsign_filter(socket.assigns.callsign),
+      filter_fn: fn _packet, _socket -> true end,
       process_fn: &process_matching_packet/2
     )
   end

lib/aprsme_web/live/packets_live/index.ex

@@ -4,27 +4,16 @@ defmodule AprsmeWeb.PacketsLive.Index do
   use Gettext, backend: AprsmeWeb.Gettext
 
   alias Aprsme.EncodingUtils
-  alias AprsmeWeb.Endpoint
 
   @impl true
   def mount(_params, _session, socket) do
     if connected?(socket) do
-      Endpoint.subscribe("aprs_messages")
       Phoenix.PubSub.subscribe(Aprsme.PubSub, "postgres:aprsme_packets")
     end
 
     {:ok, assign(socket, :packets, [])}
   end
 
-  @impl true
-  def handle_info(%{event: "packet", payload: payload}, socket) do
-    # Sanitize the packet to prevent JSON encoding errors
-    sanitized_payload = EncodingUtils.sanitize_packet(payload)
-    packets = Enum.take([sanitized_payload | socket.assigns.packets], 100)
-    socket = assign(socket, :packets, packets)
-    {:noreply, socket}
-  end
-
   @impl true
   def handle_info({:postgres_packet, payload}, socket) do
     sanitized_payload = EncodingUtils.sanitize_packet(payload)

test/aprsme_web/live/packets_live/callsign_view_test.exs

@@ -34,7 +34,7 @@ defmodule AprsmeWeb.PacketsLive.CallsignViewTest do
         }
       }
 
-      AprsmeWeb.Endpoint.broadcast!("aprs_messages", "packet", raw_payload)
+      Phoenix.PubSub.broadcast(Aprsme.PubSub, "packets:DB0WUN-13", {:postgres_packet, raw_payload})
 
       # Should render without crashing
       html = render(view)