Fix multiple TODO bugs: drain handler, callsign dedup, packet pipeline

- Fix drain_connections missing {:noreply, socket} wrapper
- Fix get_callsign_key to match atom keys from Ecto structs
- Add SpatialPubSub.broadcast_packet to PacketDistributor
- Fix pid_alive? treating {:badrpc, _} as truthy
- Fix SpatialPubSub ensure_float crash on integer strings
- Fix inverted memory calculation in ConnectionMonitor
- Remove dead SSL handler from Is module, consolidate into handle_socket_data
- Remove duplicate packet transformation in Is.dispatch
- Remove invalid placeholders option from insert_all calls
- Change PacketPipelineSupervisor to :rest_for_one strategy
- Update TODO.md marking fixed items

TODO.md

@@ -2,23 +2,7 @@
 
 ## Critical Bugs
 
-### 1. LiveView: Missing `{:noreply, socket}` in drain handler — crashes LiveView
+### ~~1. LiveView: Missing `{:noreply, socket}` in drain handler — crashes LiveView~~ FIXED
-**File:** `lib/aprsme_web/live/map_live/index.ex:885-896`
-
-The `true` branch returns a bare socket instead of `{:noreply, socket}`, which will crash the LiveView with a `MatchError` when the drain condition is hit.
-
-```elixir
-# BUG: true branch returns bare socket
-def handle_info({:drain_connections, to_drain}, socket) do
-  if :rand.uniform(100) <= to_drain * 10 do
-    socket  # <-- needs {:noreply, socket |> put_flash(...) |> push_event(...)}
-    |> put_flash(:info, "Server load balancing in progress. Reconnecting...")
-    |> push_event("reconnect", %{delay: :rand.uniform(5000)})
-  else
-    {:noreply, socket}
-  end
-end
-```
 
 ### ~~2. ConnectionMonitor: Deadlock via self-RPC every 30 seconds~~ FIXED
 **File:** `lib/aprsme/connection_monitor.ex:126-141`
@@ -27,28 +11,12 @@ end
 
 **Fixed:** Now computes local stats directly from state instead of RPC to self. Remote nodes still use RPC.
 
-### 3. LiveView: `get_callsign_key` generates random keys for Ecto structs — breaks all dedup
+### ~~3. LiveView: `get_callsign_key` generates random keys for Ecto structs — breaks all dedup~~ FIXED
-**File:** `lib/aprsme_web/live/shared/packet_utils.ex:13-14`
+Now matches atom `:id`, string `"id"`, and falls back to `:sender`/`"sender"` before random key.
 
-```elixir
+### ~~4. Cluster: `PacketDistributor` never calls `SpatialPubSub.broadcast_packet`~~ FIXED
-def get_callsign_key(%{"id" => id}), do: to_string(id)
-def get_callsign_key(_packet), do: [:positive] |> System.unique_integer() |> to_string()
-```
 
-Only matches string key `"id"`. Ecto structs with atom key `:id` fall through and get a random integer key every time. This means:
+### ~~5. LeaderElection: `pid_alive?/2` treats `{:badrpc, _}` as truthy~~ FIXED
-- Dedup never works (same packet gets different keys each call)
-- `visible_packets` map accumulates duplicates
-- Cleanup comparison (lines 1756-1766) never finds matches — all packets marked as expired
-
-### 4. Cluster: `PacketDistributor` never calls `SpatialPubSub.broadcast_packet`
-**File:** `lib/aprsme/cluster/packet_distributor.ex:30-37`
-
-`handle_distributed_packet` calls `StreamingPacketsPubSub.broadcast_packet` but never calls `SpatialPubSub.broadcast_packet`. Any SpatialPubSub subscriber receives zero packets when clustering is enabled.
-
-### 5. LeaderElection: `pid_alive?/2` treats `{:badrpc, _}` as truthy
-**File:** `lib/aprsme/cluster/leader_election.ex:257-263`
-
-`:rpc.call/4` returns `{:badrpc, reason}` on failure. Since tuples are truthy in Elixir, the function returns `true` when the RPC fails, preventing cleanup of stale leader registrations.
 
 ### ~~5b. LeaderElection: Loser never detects lost registration after `:global` conflict resolution~~ FIXED
 **File:** `lib/aprsme/cluster/leader_election.ex`
@@ -57,10 +25,8 @@ When two pods start in isolation (before the Erlang cluster forms), both registe
 
 **Fixed:** Added `verify_leadership/1` which checks `:global.whereis_name` on every `check_leadership` tick. If the registration no longer points to `self()`, the node steps down and notifies `ConnectionManager` to stop the APRS-IS connection.
 
-### 6. SpatialPubSub: `ensure_float/1` crashes on integer strings
+### ~~6. SpatialPubSub: `ensure_float/1` crashes on integer strings~~ FIXED
-**File:** `lib/aprsme/spatial_pubsub.ex:239`
+Now uses `Float.parse/1` which handles both `"42"` and `"42.5"`.
-
-`String.to_float("42")` raises `ArgumentError`. If bounds arrive as integer strings, this crashes the singleton SpatialPubSub GenServer.
 
 ---
 
@@ -87,10 +53,8 @@ These are never synchronized:
 
 Every non-weather marker on the map triggers an individual `Repo.exists?` query. With hundreds of visible markers, this is hundreds of DB round-trips per map load. Should batch-query weather callsigns upfront.
 
-### 9. Duplicate packet transformation — every packet processed twice
+### ~~9. Duplicate packet transformation — every packet processed twice~~ FIXED
-**File:** `lib/aprsme/is/is.ex:596-604` and `lib/aprsme/packet_consumer.ex:382-430`
+Removed `extract_additional_data` and `normalize_data_type` from `Is.dispatch/1` — `PacketConsumer` already handles these.
-
-`Is.dispatch/1` runs `struct_to_map`, `extract_additional_data`, and `normalize_data_type`. Then `PacketConsumer.prepare_packet_for_insert/1` runs them all again. Every packet does double work.
 
 ### 10. `batch_length/1` recomputes O(n) list length on every event arrival
 **File:** `lib/aprsme/packet_consumer.ex:64-68, 130`
@@ -175,8 +139,8 @@ Each packet has 100+ columns. 2000 full structs per connected client. Never clea
 
 ## Dead Code
 
-### 25. `handle_info({:ssl, ...})` in Is module — connection uses `:gen_tcp`, not SSL
+### ~~25. `handle_info({:ssl, ...})` in Is module — connection uses `:gen_tcp`, not SSL~~ FIXED
-**File:** `lib/aprsme/is/is.ex:318-352`
+Removed dead SSL handler. Consolidated TCP handler into `handle_socket_data/2` with nil-safe timer cancel.
 
 ### 26. `PacketPipelineSetup` module — not in supervision tree, references unnamed consumers
 **File:** `lib/aprsme/packet_pipeline_setup.ex`
@@ -211,8 +175,8 @@ Both `if cluster_enabled` and `else` return `{Phoenix.PubSub, name: Aprsme.PubSu
 
 Heat map never updates from real-time packets, historical loading at low zoom, or zoom threshold crossing.
 
-### 35. `placeholders` option passed to `Repo.insert_all` — not a valid option
+### ~~35. `placeholders` option passed to `Repo.insert_all` — not a valid option~~ FIXED
-**Files:** `lib/aprsme/packet_consumer.ex:304`, `lib/aprsme/db_optimizer.ex:64`
+Removed from both `packet_consumer.ex` and `db_optimizer.ex`.
 
 ---
 
@@ -223,15 +187,11 @@ Heat map never updates from real-time packets, historical loading at low zoom, o
 
 `calculate_current_state` returns `:half_open` but never writes it back to state. Multiple callers can all enter half-open simultaneously.
 
-### 37. `Process.cancel_timer(nil)` crash risk in Is module
+### ~~37. `Process.cancel_timer(nil)` crash risk in Is module~~ FIXED
-**File:** `lib/aprsme/is/is.ex:321, 356`
+Consolidated into `handle_socket_data/2` with `if state.timer` guard.
 
-If `state.timer` is nil, `Process.cancel_timer(nil)` raises `FunctionClauseError`. Some paths protect with `if state.timer`, but the main data handlers don't.
+### ~~38. `PacketPipelineSupervisor` uses `:one_for_one` — producer restart orphans consumers~~ FIXED
-
+Changed to `:rest_for_one` so consumer pool restarts when producer restarts.
-### 38. `PacketPipelineSupervisor` uses `:one_for_one` — producer restart orphans consumers
-**File:** `lib/aprsme/packet_pipeline_supervisor.ex:23`
-
-If the producer crashes, consumers remain subscribed to the dead PID. Should use `:rest_for_one`.
 
 ### 39. `handle_update_time_display` doesn't trigger re-renders
 **File:** `lib/aprsme_web/live/map_live/index.ex:1774-1781`
@@ -243,10 +203,8 @@ Returns `{:noreply, socket}` with unchanged socket — LiveView diff is empty. T
 
 Different LiveViews can overwrite each other's stored packets. TTL cleanup runs globally.
 
-### 41. ConnectionMonitor memory calculation is inverted
+### ~~41. ConnectionMonitor memory calculation is inverted~~ FIXED
-**File:** `lib/aprsme/connection_monitor.ex:219-232`
+Now computes `processes / total` — process memory as fraction of total VM memory.
-
-`total / system` produces a ratio >1.0 (typically 1.5-3.0), not a percentage.
 
 ### 42. `Callsign.valid?/1` accepts any non-empty string — docstring examples are wrong
 **File:** `lib/aprsme/callsign.ex:40-45`

lib/aprsme/cluster/leader_election.ex

@@ -275,10 +275,9 @@ defmodule Aprsme.Cluster.LeaderElection do
   end
 
   defp pid_alive?(pid, pid_node) do
-    if :rpc.call(pid_node, Process, :alive?, [pid]) do
+    case :rpc.call(pid_node, Process, :alive?, [pid]) do
-      true
+      {:badrpc, _} -> false
-    else
+      result -> result == true
-      false
     end
   end
 

lib/aprsme/cluster/packet_distributor.ex

@@ -28,8 +28,9 @@ defmodule Aprsme.Cluster.PacketDistributor do
   end
 
   def handle_distributed_packet({:distributed_packet, packet}) do
-    # Broadcast to local LiveView clients
+    # Broadcast to local LiveView clients via both PubSub systems
     Aprsme.StreamingPacketsPubSub.broadcast_packet(packet)
+    Aprsme.SpatialPubSub.broadcast_packet(packet)
 
     # Update packet store for LiveView
     PacketStore.store_packet(packet)

lib/aprsme/connection_monitor.ex

@@ -223,13 +223,13 @@ defmodule Aprsme.ConnectionMonitor do
   end
 
   defp get_memory_usage do
-    # Get memory usage as a percentage
     mem_data = :erlang.memory()
     total = Keyword.get(mem_data, :total, 0)
-    system = Keyword.get(mem_data, :system, 0)
+    # processes + system = total; report process memory as fraction of total
+    processes = Keyword.get(mem_data, :processes, 0)
 
-    if system > 0 do
+    if total > 0 do
-      total / system
+      processes / total
     else
       0.0
     end

lib/aprsme/db_optimizer.ex

@@ -60,8 +60,7 @@ defmodule Aprsme.DbOptimizer do
           [
             returning: false,
             on_conflict: :nothing,
-            timeout: 60_000,
+            timeout: 60_000
-            placeholders: length(batch) > 100
           ],
           opts
         )

lib/aprsme/is/is.ex

@@ -315,75 +315,8 @@ defmodule Aprsme.Is do
   end
 
   @impl true
-  @spec handle_info({:ssl, port(), binary()} | any(), state()) :: {:noreply, state()}
-  def handle_info({:ssl, _socket, data}, state) do
-    # Cancel the previous timer
-    Process.cancel_timer(state.timer)
-
-    # Update packet statistics
-    current_time = System.system_time(:second)
-    packet_stats = update_packet_stats(state.packet_stats, current_time)
-
-    # Append new packet data to buffer
-    buffer = state.buffer <> data
-
-    # Process complete lines (ending with \r\n or \n)
-    {complete_lines, remaining_buffer} = extract_complete_lines(buffer)
-
-    # Dispatch each complete line
-    Enum.each(complete_lines, fn line ->
-      trimmed = String.trim(line)
-
-      if trimmed != "" do
-        dispatch(trimmed)
-      end
-    end)
-
-    # Start a new timer
-    timer = Process.send_after(self(), :aprsme_no_message_timeout, @aprs_timeout)
-
-    state =
-      state
-      |> Map.put(:timer, timer)
-      |> Map.put(:packet_stats, packet_stats)
-      |> Map.put(:buffer, remaining_buffer)
-
-    {:noreply, state}
-  end
-
   def handle_info({:tcp, _socket, data}, state) do
-    # Cancel the previous timer
+    handle_socket_data(data, state)
-    Process.cancel_timer(state.timer)
-
-    # Update packet statistics
-    current_time = System.system_time(:second)
-    packet_stats = update_packet_stats(state.packet_stats, current_time)
-
-    # Append new packet data to buffer
-    buffer = state.buffer <> data
-
-    # Process complete lines (ending with \r\n or \n)
-    {complete_lines, remaining_buffer} = extract_complete_lines(buffer)
-
-    # Dispatch each complete line
-    Enum.each(complete_lines, fn line ->
-      trimmed = String.trim(line)
-
-      if trimmed != "" do
-        dispatch(trimmed)
-      end
-    end)
-
-    # Start a new timer
-    timer = Process.send_after(self(), :aprsme_no_message_timeout, @aprs_timeout)
-
-    state =
-      state
-      |> Map.put(:timer, timer)
-      |> Map.put(:packet_stats, packet_stats)
-      |> Map.put(:buffer, remaining_buffer)
-
-    {:noreply, state}
   end
 
   def handle_info({:backpressure, :activate}, %{socket: nil} = state) do
@@ -507,6 +440,25 @@ defmodule Aprsme.Is do
     end
   end
 
+  defp handle_socket_data(data, state) do
+    if state.timer, do: Process.cancel_timer(state.timer)
+
+    current_time = System.system_time(:second)
+    packet_stats = update_packet_stats(state.packet_stats, current_time)
+
+    buffer = state.buffer <> data
+    {complete_lines, remaining_buffer} = extract_complete_lines(buffer)
+
+    Enum.each(complete_lines, fn line ->
+      trimmed = String.trim(line)
+      if trimmed != "", do: dispatch(trimmed)
+    end)
+
+    timer = Process.send_after(self(), :aprsme_no_message_timeout, @aprs_timeout)
+
+    {:noreply, %{state | timer: timer, packet_stats: packet_stats, buffer: remaining_buffer}}
+  end
+
   # Extract complete lines from buffer, returning {complete_lines, remaining_buffer}
   @spec extract_complete_lines(String.t()) :: {[String.t()], String.t()}
   defp extract_complete_lines(buffer) do
@@ -590,20 +542,16 @@ defmodule Aprsme.Is do
         # Store the packet in the database for future replay
         # Use GenStage pipeline for efficient batch processing
         try do
-          # Always set received_at timestamp to ensure consistency
+          # Set received_at and convert struct to map — PacketConsumer.prepare_packet_for_insert
+          # handles extract_additional_data and normalize_data_type, so don't duplicate here.
           current_time = DateTime.truncate(DateTime.utc_now(), :microsecond)
-          packet_data = Map.put(parsed_message, :received_at, current_time)
 
-          # Convert to map before storing to avoid struct conversion issues
+          attrs =
-          attrs = struct_to_map(packet_data)
+            parsed_message
+            |> Map.put(:received_at, current_time)
+            |> Map.put(:raw, message)
+            |> struct_to_map()
 
-          # Extract additional data from the parsed packet including raw packet
-          attrs = Aprsme.Packet.extract_additional_data(attrs, message)
-
-          # Normalize data_type to string if it's an atom
-          attrs = Aprsme.EncodingUtils.normalize_data_type(attrs)
-
-          # Submit to GenStage pipeline for batch processing
           Aprsme.PacketProducer.submit_packet(attrs)
         rescue
           error ->

lib/aprsme/packet_consumer.ex

@@ -299,9 +299,7 @@ defmodule Aprsme.PacketConsumer do
       # Skip conflicts to avoid blocking
       on_conflict: :nothing,
       # Increased timeout for large batches
-      timeout: 60_000,
+      timeout: 60_000
-      # Use placeholders for better performance with large batches
-      placeholders: length(valid_packets) > 100
     ]
 
     try do

lib/aprsme/packet_pipeline_supervisor.ex

@@ -20,6 +20,6 @@ defmodule Aprsme.PacketPipelineSupervisor do
 
     children = [producer_spec, consumer_pool_spec]
 
-    Supervisor.init(children, strategy: :one_for_one)
+    Supervisor.init(children, strategy: :rest_for_one)
   end
 end

lib/aprsme/spatial_pubsub.ex

@@ -236,7 +236,13 @@ defmodule Aprsme.SpatialPubSub do
     }
   end
 
-  defp ensure_float(val) when is_binary(val), do: String.to_float(val)
+  defp ensure_float(val) when is_binary(val) do
+    case Float.parse(val) do
+      {f, _} -> f
+      :error -> 0.0
+    end
+  end
+
   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

lib/aprsme_web/live/map_live/index.ex

@@ -883,13 +883,11 @@ defmodule AprsmeWeb.MapLive.Index do
   end
 
   def handle_info({:drain_connections, to_drain}, socket) do
-    # Check if this connection should be drained
-    # Use a random selection to determine if this connection should disconnect
     if :rand.uniform(100) <= to_drain * 10 do
-      # Gracefully disconnect this client
+      {:noreply,
-      socket
+       socket
-      |> put_flash(:info, "Server load balancing in progress. Reconnecting...")
+       |> put_flash(:info, "Server load balancing in progress. Reconnecting...")
-      |> push_event("reconnect", %{delay: :rand.uniform(5000)})
+       |> push_event("reconnect", %{delay: :rand.uniform(5000)})}
     else
       {:noreply, socket}
     end

lib/aprsme_web/live/shared/packet_utils.ex

@@ -10,7 +10,10 @@ defmodule AprsmeWeb.Live.Shared.PacketUtils do
   Get unique callsign key from packet.
   """
   @spec get_callsign_key(map()) :: binary()
-  def get_callsign_key(%{"id" => id}), do: to_string(id)
+  def get_callsign_key(%{id: id}) when not is_nil(id), do: to_string(id)
+  def get_callsign_key(%{"id" => id}) when not is_nil(id), do: to_string(id)
+  def get_callsign_key(%{sender: sender}) when is_binary(sender), do: sender
+  def get_callsign_key(%{"sender" => sender}) when is_binary(sender), do: sender
   def get_callsign_key(_packet), do: [:positive] |> System.unique_integer() |> to_string()
 
   @doc """

test/aprsme/is_test.exs

@@ -314,20 +314,7 @@ defmodule Aprsme.IsTest do
     end
   end
 
-  describe "handle_info({:ssl, ...}, ...)" do
+  # SSL handler was removed — connection uses :gen_tcp, not SSL (TODO #25)
-    test "processes SSL data the same as TCP" do
-      state = build_state()
-
-      data = "# ssl comment\r\n"
-
-      capture_log(fn ->
-        {:noreply, new_state} = Aprsme.Is.handle_info({:ssl, :fake_port, data}, state)
-
-        assert new_state.buffer == ""
-        assert new_state.packet_stats.total_packets == 1
-      end)
-    end
-  end
 
   describe "handle_info({:tcp_closed, ...}, ...)" do
     test "schedules reconnect and clears socket" do

test/aprsme/packet_pipeline_supervisor_test.exs

@@ -19,10 +19,10 @@ defmodule Aprsme.PacketPipelineSupervisorTest do
       :ok
     end
 
-    test "returns :one_for_one strategy" do
+    test "returns :rest_for_one strategy" do
       {:ok, {sup_flags, _children}} = PacketPipelineSupervisor.init([])
 
-      assert %{strategy: :one_for_one} = sup_flags
+      assert %{strategy: :rest_for_one} = sup_flags
     end
 
     test "returns two child specs" do