diff --git a/docs/mobile-api.md b/docs/mobile-api.md index 26b7a30..0db4b52 100644 --- a/docs/mobile-api.md +++ b/docs/mobile-api.md @@ -189,6 +189,46 @@ Once subscribed, you'll receive packets within your bounds. Historical packets a **Note:** Optional fields are omitted from the payload entirely when their value is `nil`. +### Receiving Batched Packets (Historical Loads) + +When the server delivers historical packets in bulk (for example, the replay +that follows `subscribe_bounds` or `subscribe_callsign`), they arrive as a +batched `packets` event instead of many individual `packet` events. Live +streaming packets continue to use the single `packet` event described above. + +**Event:** `packets` + +**Payload:** +```json +{ + "packets": [ + { "id": "...", "callsign": "...", "lat": 33.1, "lng": -96.1, "timestamp": "2025-10-25T16:17:20Z", "symbol_table_id": "/", "symbol_code": ">" }, + { "id": "...", "callsign": "...", "lat": 33.2, "lng": -96.2, "timestamp": "2025-10-25T16:17:25Z", "symbol_table_id": "/", "symbol_code": "#" } + ], + "count": 2 +} +``` + +Each entry inside `packets` has the same schema as the single-packet `packet` +event above. Batch sizes are capped at 100 entries per frame, so a 1,000-packet +history load arrives as ~10 `packets` events. + +**Migration guidance:** existing clients that only listen for the `packet` +event will still receive live updates correctly — they simply won't populate +the historical backlog. To show history, add a handler for `packets` that +iterates over `payload.packets` and applies the same rendering logic. + +### Rate Limiting + +- **Connect:** up to 30 new socket connections per IP per minute. Exceeding the + limit returns an immediate `:error` from the socket handshake; retry after + the minute window elapses. +- **Channel requests:** the expensive operations (`subscribe_bounds`, + `subscribe_callsign`, `search_callsign`) are rate-limited to 30 calls per + socket per minute. Exceeding the limit returns + `{ "status": "error", "response": { "message": "Rate limit exceeded; retry later", "retry_after_ms": } }`. +- Streaming packet delivery is *not* rate-limited. + ### Unsubscribing Stop receiving packets. diff --git a/lib/aprsme_web/channels/mobile_channel.ex b/lib/aprsme_web/channels/mobile_channel.ex index b360ad5..58ddd8f 100644 --- a/lib/aprsme_web/channels/mobile_channel.ex +++ b/lib/aprsme_web/channels/mobile_channel.ex @@ -57,6 +57,11 @@ defmodule AprsmeWeb.MobileChannel do require Logger + # Per-socket rate limits on the expensive operations. Streaming packet + # delivery is unaffected — these only guard client-initiated queries. + @expensive_op_limit 30 + @expensive_op_scale 60_000 + @impl true def join("mobile:packets", _payload, socket) do Logger.info("Mobile client joined packets channel") @@ -71,38 +76,40 @@ defmodule AprsmeWeb.MobileChannel do ) do Logger.info("Mobile websocket received subscribe_bounds: #{inspect(payload)}") - bounds = %{ - north: ensure_float(north), - south: ensure_float(south), - east: ensure_float(east), - west: ensure_float(west) - } + with :ok <- check_rate_limit(socket, "subscribe_bounds") do + bounds = %{ + north: ensure_float(north), + south: ensure_float(south), + east: ensure_float(east), + west: ensure_float(west) + } - # Validate bounds - case validate_bounds(bounds) do - :ok -> - # Generate unique client ID - client_id = "mobile_#{:erlang.phash2(self())}" + # Validate bounds + case validate_bounds(bounds) do + :ok -> + # Generate unique client ID + client_id = "mobile_#{:erlang.phash2(self())}" - # Subscribe to StreamingPacketsPubSub with bounds - Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), bounds) + # Subscribe to StreamingPacketsPubSub with bounds + Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), bounds) - # Store client info in socket - socket = - socket - |> assign(:client_id, client_id) - |> assign(:bounds, bounds) - |> assign(:subscribed, true) + # Store client info in socket + socket = + socket + |> assign(:client_id, client_id) + |> assign(:bounds, bounds) + |> assign(:subscribed, true) - Logger.info("Mobile client #{client_id} subscribed to bounds: #{inspect(bounds)}") + Logger.info("Mobile client #{client_id} subscribed to bounds: #{inspect(bounds)}") - # Load and send historical packets - socket = load_historical_packets(socket, bounds, payload) + # Load and send historical packets + socket = load_historical_packets(socket, bounds, payload) - {:reply, {:ok, %{bounds: bounds, message: "Subscribed to packet stream"}}, socket} + {:reply, {:ok, %{bounds: bounds, message: "Subscribed to packet stream"}}, socket} - {:error, reason} -> - {:reply, {:error, %{message: reason}}, socket} + {:error, reason} -> + {:reply, {:error, %{message: reason}}, socket} + end end end @@ -160,37 +167,41 @@ defmodule AprsmeWeb.MobileChannel do def handle_in("search_callsign", %{"query" => query} = payload, socket) do Logger.info("Mobile websocket received search_callsign: #{inspect(payload)}") - # Trim whitespace and control characters from query - query = String.trim(query) + with :ok <- check_rate_limit(socket, "search_callsign") do + # Trim whitespace and control characters from query + query = String.trim(query) - limit = payload |> Map.get("limit", 50) |> ensure_integer(50) - limit = min(limit, 500) + limit = payload |> Map.get("limit", 50) |> ensure_integer(50) + limit = min(limit, 500) - results = search_callsign(query, limit) + results = search_callsign(query, limit) - {:reply, {:ok, %{results: results, count: length(results)}}, socket} + {:reply, {:ok, %{results: results, count: length(results)}}, socket} + end end @impl true def handle_in("subscribe_callsign", %{"callsign" => callsign} = payload, socket) do Logger.info("Mobile websocket received subscribe_callsign: #{inspect(payload)}") - hours_back = payload |> Map.get("hours_back", 24) |> ensure_integer(24) - # Max 1 week - hours_back = min(hours_back, 168) + with :ok <- check_rate_limit(socket, "subscribe_callsign") do + hours_back = payload |> Map.get("hours_back", 24) |> ensure_integer(24) + # Max 1 week + hours_back = min(hours_back, 168) - # Normalize callsign - trim whitespace and convert to uppercase - callsign = callsign |> String.trim() |> String.upcase() + # Normalize callsign - trim whitespace and convert to uppercase + callsign = callsign |> String.trim() |> String.upcase() - # Load historical packets for this callsign - socket = load_callsign_history(socket, callsign, hours_back) + # Load historical packets for this callsign + socket = load_callsign_history(socket, callsign, hours_back) - socket = ensure_callsign_subscription(socket) + socket = ensure_callsign_subscription(socket) - # Store tracked callsign in socket - socket = assign(socket, :tracked_callsign, callsign) + # Store tracked callsign in socket + socket = assign(socket, :tracked_callsign, callsign) - {:reply, {:ok, %{callsign: callsign, message: "Subscribed to callsign updates"}}, socket} + {:reply, {:ok, %{callsign: callsign, message: "Subscribed to callsign updates"}}, socket} + end end @impl true @@ -249,6 +260,40 @@ defmodule AprsmeWeb.MobileChannel do {:noreply, socket} end + # Per-socket limiter for the expensive client-initiated handlers + # (subscribe_bounds, subscribe_callsign, search_callsign). Returns :ok on + # allow or a {:reply, ..., socket} tuple that propagates straight out of the + # with block to the channel reply path on deny. + defp check_rate_limit(socket, op) do + key = "mobile_channel:#{op}:#{:erlang.phash2(self())}" + + case Aprsme.RateLimiter.hit(key, @expensive_op_scale, @expensive_op_limit) do + {:allow, _count} -> + :ok + + {:deny, retry_after} -> + Logger.warning("Mobile channel rate-limited op=#{op} client=#{inspect(self())}") + + {:reply, {:error, %{message: "Rate limit exceeded; retry later", retry_after_ms: retry_after}}, socket} + end + end + + # Push packets to the client in moderately sized batches under the new + # `packets` event. Keeps the per-packet `packet` event format intact for + # streaming deliveries; bulk historical loads move through this path so + # clients don't get hit with 5000 individual WebSocket frames. + @packets_batch_size 100 + defp push_packets_batch(_socket, []), do: :ok + + defp push_packets_batch(socket, packets) do + packets + |> Stream.map(&build_mobile_packet/1) + |> Stream.chunk_every(@packets_batch_size) + |> Enum.each(fn chunk -> + push(socket, "packets", %{packets: chunk, count: length(chunk)}) + end) + end + # Clean up on terminate @impl true def terminate(_reason, socket) do @@ -411,14 +456,10 @@ defmodule AprsmeWeb.MobileChannel do Logger.info("Loaded #{length(packets)} historical packets for mobile client") - # Send historical packets to client - if packets != [] do - # Convert packets to mobile format and send them - Enum.each(packets, fn packet -> - packet_data = build_mobile_packet(packet) - push(socket, "packet", packet_data) - end) - end + # Send historical packets as batched "packets" events rather than one + # "packet" frame per row — previously a single subscription could queue + # thousands of WebSocket frames. + push_packets_batch(socket, packets) socket end @@ -508,13 +549,8 @@ defmodule AprsmeWeb.MobileChannel do Logger.info("Loaded #{length(packets)} historical packets for callsign #{callsign}") - # Send historical packets to client - if packets != [] do - Enum.each(packets, fn packet -> - packet_data = build_mobile_packet(packet) - push(socket, "packet", packet_data) - end) - end + # Send batched "packets" events — see push_packets_batch/2. + push_packets_batch(socket, packets) socket end diff --git a/lib/aprsme_web/channels/mobile_user_socket.ex b/lib/aprsme_web/channels/mobile_user_socket.ex index fbbf192..1ae7d5b 100644 --- a/lib/aprsme_web/channels/mobile_user_socket.ex +++ b/lib/aprsme_web/channels/mobile_user_socket.ex @@ -1,36 +1,63 @@ defmodule AprsmeWeb.MobileUserSocket do @moduledoc """ Socket for mobile applications (iOS/Android) to receive real-time APRS packets. + + Connections are anonymous but rate-limited per peer IP to prevent a single + client from opening unbounded connections and exhausting server resources. """ use Phoenix.Socket + require Logger + # Channels channel "mobile:packets", AprsmeWeb.MobileChannel - # Socket params are passed from the client and can be used to verify and authenticate a user. - # After verification, you can put default assigns into the socket that will be set for all channels. - @impl true - def connect(_params, socket, _connect_info) do - # For now, allow anonymous connections - # In the future, you can add authentication here: - # case verify_token(params["token"]) do - # {:ok, user_id} -> {:ok, assign(socket, :user_id, user_id)} - # {:error, _} -> :error - # end + # 30 new connections per IP per minute — well above normal reconnect churn + # but tight enough to stop a script from floodting the socket layer. + @connect_limit 30 + @connect_scale 60_000 - {:ok, socket} + @impl true + def connect(_params, socket, connect_info) do + ip = peer_ip(connect_info) + + if test_env?() or allow_connect?(ip) do + {:ok, assign(socket, :peer_ip, ip)} + else + Logger.warning("Mobile socket connect rate-limited for ip=#{ip}") + :error + end end - # Socket id's are topics that allow you to identify all sockets for a given user: - # - # def id(socket), do: "mobile_user_socket:#{socket.assigns.user_id}" - # - # Would allow you to broadcast a "disconnect" event and terminate - # all active sockets and channels for a given user: - # - # Elixir.AprsmeWeb.Endpoint.broadcast("mobile_user_socket:#{user.id}", "disconnect", %{}) - # - # Returning `nil` makes this socket anonymous. + defp allow_connect?(ip) do + case Aprsme.RateLimiter.hit("mobile_socket_connect:#{ip}", @connect_scale, @connect_limit) do + {:allow, _count} -> true + {:deny, _retry_after} -> false + end + end + + defp test_env?, do: Application.get_env(:aprsme, :env) == :test + @impl true def id(_socket), do: nil + + defp peer_ip(%{x_headers: headers}) when is_list(headers) do + headers + |> Enum.find_value(fn + {"cf-connecting-ip", v} -> v + {"x-forwarded-for", v} -> v |> String.split(",") |> List.first() |> String.trim() + {"x-real-ip", v} -> v + _ -> nil + end) + |> case do + nil -> "unknown" + ip -> ip + end + end + + defp peer_ip(%{peer_data: %{address: address}}) when not is_nil(address) do + address |> :inet.ntoa() |> to_string() + end + + defp peer_ip(_), do: "unknown" end diff --git a/test/aprsme_web/channels/mobile_channel_test.exs b/test/aprsme_web/channels/mobile_channel_test.exs index 69a9de1..4b53a89 100644 --- a/test/aprsme_web/channels/mobile_channel_test.exs +++ b/test/aprsme_web/channels/mobile_channel_test.exs @@ -52,8 +52,12 @@ defmodule AprsmeWeb.MobileChannelTest do end describe "join" do - test "successful join returns welcome message", %{socket: socket} do - assert socket.assigns == %{} + test "successful join stamps the peer ip and has no other assigns", %{socket: socket} do + # Socket test client has no real peer_data, so the IP helper falls back + # to "unknown"; the important part is that the assign is present and + # no extra per-subscription state has been set yet. + assert socket.assigns.peer_ip == "unknown" + assert Map.delete(socket.assigns, :peer_ip) == %{} end end