Fix potential crashes and unsafe operations throughout the app

- Fixed File.read crash in devices_seeder.ex with proper error handling
- Added timeout to TCP connect in aprs_is_connection.ex (10s timeout)
- Fixed unsafe socket send operations with error handling
- Added Process.alive? check before sending to PIDs in streaming_packets_pubsub.ex
- Fixed stream materialization memory issue in packet_consumer.ex
- Added error handling for datetime parsing errors
- Prevented crashes from dead processes and closed sockets

These fixes ensure the application is more resilient to:
- Missing files
- Network timeouts
- Dead processes
- Closed sockets
- Invalid data parsing

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

lib/aprsme/aprs_is_connection.ex

@@ -114,8 +114,16 @@ defmodule Aprsme.AprsIsConnection do
 
   @impl true
   def handle_call({:send, packet}, _from, %{socket: socket} = state) when is_port(socket) do
-    :ok = :gen_tcp.send(socket, packet <> "\r\n")
-    {:reply, :ok, state}
+    # Safely send to socket with error handling
+    case :gen_tcp.send(socket, packet <> "\r\n") do
+      :ok ->
+        {:reply, :ok, state}
+
+      {:error, reason} = error ->
+        Logger.error("Failed to send packet: #{inspect(reason)}")
+        # Socket is likely closed, reset state
+        {:reply, error, %{state | socket: nil}}
+    end
   end
 
   def handle_call({:send, _packet}, _from, state) do
@@ -141,9 +149,12 @@ defmodule Aprsme.AprsIsConnection do
     passcode = Application.get_env(:aprsme, :aprsme_is_passcode, "00000")
     filter = Application.get_env(:aprsme, :aprsme_is_filter, "")
 
-    opts = [:binary, active: true, packet: :line, keepalive: true]
+    # Add timeout to prevent indefinite hanging
+    opts = [:binary, active: true, packet: :line, keepalive: true, send_timeout: 5000]
+    # 10 seconds
+    connect_timeout = 10_000
 
-    case :gen_tcp.connect(host, port, opts) do
+    case :gen_tcp.connect(host, port, opts, connect_timeout) do
       {:ok, socket} ->
         login = "user #{callsign} pass #{passcode} vers aprs.me 0.1 #{filter}\r\n"
         :ok = :gen_tcp.send(socket, login)

lib/aprsme/devices_seeder.ex

@@ -5,8 +5,22 @@ defmodule Aprsme.DevicesSeeder do
   alias Aprsme.Repo
 
   def seed_from_json(path \\ "test/support/test_devices.json") do
-    {:ok, body} = File.read(path)
-    {:ok, json} = Jason.decode(body)
+    case File.read(path) do
+      {:ok, body} ->
+        case Jason.decode(body) do
+          {:ok, json} ->
+            seed_from_decoded_json(json)
+
+          {:error, reason} ->
+            {:error, "Failed to decode JSON: #{inspect(reason)}"}
+        end
+
+      {:error, reason} ->
+        {:error, "Failed to read file #{path}: #{inspect(reason)}"}
+    end
+  end
+
+  defp seed_from_decoded_json(json) do
     tocalls = Map.get(json, "tocalls", %{})
     mice = Map.get(json, "mice", %{})
     micelegacy = Map.get(json, "micelegacy", %{})
@@ -17,6 +31,8 @@ defmodule Aprsme.DevicesSeeder do
     Enum.each([tocalls, mice, micelegacy], fn group ->
       seed_device_group(group, now)
     end)
+
+    {:ok, :seeded}
   end
 
   defp seed_device_group(group, now) do

lib/aprsme/packet_consumer.ex

@@ -127,14 +127,12 @@ defmodule Aprsme.PacketConsumer do
     chunk_size = Application.get_env(:aprsme, :packet_pipeline)[:batch_size] || 100
 
     # Use Stream for memory-efficient processing
-    results =
+    # Process and reduce in one pass to avoid materializing the entire list
+    {success_count, error_count} =
       packets
       |> Stream.chunk_every(chunk_size)
       |> Stream.map(&process_chunk/1)
-      |> Enum.to_list()
-
-    {success_count, error_count} =
-      Enum.reduce(results, {0, 0}, fn {success, error}, {total_success, total_error} ->
+      |> Enum.reduce({0, 0}, fn {success, error}, {total_success, total_error} ->
         {total_success + success, total_error + error}
       end)
 
@@ -511,6 +509,7 @@ defmodule Aprsme.PacketConsumer do
 
   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
 
   defp truncate_datetimes_to_second(term) when is_map(term) and not is_struct(term) do
     Map.new(term, fn {k, v} -> {k, truncate_datetimes_to_second(v)} end)

lib/aprsme/streaming_packets_pubsub.ex

@@ -118,7 +118,15 @@ defmodule Aprsme.StreamingPacketsPubSub do
       Aprsme.BroadcastTaskSupervisor.async_execute(fn ->
         subscribers
         |> Stream.filter(fn {_pid, bounds} -> packet_in_bounds?(lat, lon, bounds) end)
-        |> Enum.each(fn {pid, _bounds} -> send(pid, {:streaming_packet, packet}) end)
+        |> Enum.each(fn {pid, _bounds} ->
+          # Only send if process is alive
+          if Process.alive?(pid) do
+            send(pid, {:streaming_packet, packet})
+          else
+            # Clean up dead subscriber
+            :ets.delete(@table_name, pid)
+          end
+        end)
       end)
     end