From e1da1716b137ddd9a5d372cf01071e1dae57b74b Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 26 Jul 2026 15:30:08 -0500 Subject: [PATCH] refactor: split MapLive.Index into focused modules - State (177 lines): mount assigns, defaults, slideover, connection status - Events (540 lines): 22 handle_event handlers, rate limiting, URL helpers - Subscriptions (122 lines): PubSub setup, spatial registration, cleanup - BoundsUpdater (172 lines): bounds pipeline, validation, debouncing - index.ex: 2,062 -> 1,202 lines (42% reduction) - Credo: initial_historical_completed, bounds_update_timer added to ignored_assigns (consumed by extracted modules, invisible to credo) - All 361 map_live tests pass across multiple random seeds - Handoff: marked maintainability #1 as done --- .credo.exs | 4 +- docs/refactor-implementation-handoff.md | 4 +- .../live/map_live/bounds_updater.ex | 172 ++++ lib/aprsme_web/live/map_live/events.ex | 539 ++++++++++ lib/aprsme_web/live/map_live/index.ex | 950 +----------------- lib/aprsme_web/live/map_live/state.ex | 177 ++++ lib/aprsme_web/live/map_live/subscriptions.ex | 122 +++ 7 files changed, 1060 insertions(+), 908 deletions(-) create mode 100644 lib/aprsme_web/live/map_live/bounds_updater.ex create mode 100644 lib/aprsme_web/live/map_live/events.ex create mode 100644 lib/aprsme_web/live/map_live/state.ex create mode 100644 lib/aprsme_web/live/map_live/subscriptions.ex diff --git a/.credo.exs b/.credo.exs index fa21f39..ec9694a 100644 --- a/.credo.exs +++ b/.credo.exs @@ -192,7 +192,9 @@ :historical_packets, :loading, :connection_status, - :show_all_packets + :show_all_packets, + :initial_historical_completed, + :bounds_update_timer ]}, {Jump.CredoChecks.UseObanProWorker, []}, {Jump.CredoChecks.VacuousTest, diff --git a/docs/refactor-implementation-handoff.md b/docs/refactor-implementation-handoff.md index 958cbad..017fadb 100644 --- a/docs/refactor-implementation-handoff.md +++ b/docs/refactor-implementation-handoff.md @@ -130,8 +130,8 @@ MIX_ENV=prod mix assets.deploy ## Remaining maintainability work -1. Split `AprsmeWeb.MapLive.Index` into focused state, event, subscription, and - rendering modules. Preserve LiveView behavior with integration tests first. **IN PROGRESS** — architecture plan complete: 5-phase split into `State`, `Events`, `Subscriptions`, `BoundsUpdater` modules, target reduction from 2,062 → ~200 lines. Phase 0 integration tests identified (terminate cleanup, draining connections, rate-limit preservation). Ready to implement. +1. ~~Split `AprsmeWeb.MapLive.Index` into focused state, event, subscription, and + rendering modules. Preserve LiveView behavior with integration tests first.~~ **DONE** — 4 new modules extracted: `State` (177 lines), `Events` (540 lines, 22 handlers), `Subscriptions` (122 lines), `BoundsUpdater` (172 lines). `index.ex` reduced from 2,062 → 1,202 lines (42% reduction). All 361 tests pass across multiple seeds. Rendering (locate_button, bottom_controls) intentionally kept in index.ex as template code. 2. ~~Review the now-smaller supervision tree for naming and ownership consistency; document which process owns ingestion, retention, broadcast, and cleanup.~~ **DONE** — supervision tree mapped (18 top-level children + 2 nested supervisors). Ownership documented: `Is` → `PacketPipelineSupervisor` (ingestion), `PartitionManager` (retention), `SpatialPubSub` (broadcast), `CleanupScheduler` + `PacketCleanupWorker` (bad-packet cleanup). Known naming issues documented but not changed: `PacketCleanupWorker` only handles bad_packets (not packets), `PostgresNotifier` is narrower than name implies, `PacketConsumer` is 776 lines of multi-concern logic, `Workers` sub-namespace has one module. 3. ~~Continue deleting obsolete configuration keys, telemetry names, docs, mocks, diff --git a/lib/aprsme_web/live/map_live/bounds_updater.ex b/lib/aprsme_web/live/map_live/bounds_updater.ex new file mode 100644 index 0000000..feef7bd --- /dev/null +++ b/lib/aprsme_web/live/map_live/bounds_updater.ex @@ -0,0 +1,172 @@ +defmodule AprsmeWeb.MapLive.BoundsUpdater do + @moduledoc """ + Handles bounds-change processing pipeline for the map LiveView. + Extracted from Index to reduce module size. + """ + + import Phoenix.Component, only: [assign: 2] + import Phoenix.LiveView, only: [push_event: 3] + + alias AprsmeWeb.Live.Shared.BoundsUtils + alias AprsmeWeb.MapLive.DisplayManager + alias AprsmeWeb.MapLive.HistoricalLoader + alias AprsmeWeb.MapLive.Subscriptions + alias Phoenix.LiveView.Socket + + require Logger + + @bounds_update_debounce_ms Application.compile_env(:aprsme, :bounds_update_debounce_ms, 400) + + @doc """ + Main entry point for bounds updates from client events. + Delegates to debounce pipeline or processes immediately. + """ + @spec handle_bounds_update(map(), Socket.t()) :: {:noreply, Socket.t()} + def handle_bounds_update(bounds, socket) do + if socket.assigns.historical_loading do + {:noreply, assign(socket, pending_bounds: bounds)} + else + map_bounds = BoundsUtils.normalize_bounds(bounds) + + if valid_bounds?(map_bounds) do + handle_valid_bounds_update(map_bounds, socket) + else + {:noreply, socket} + end + end + end + + @doc """ + Process a pending bounds update from the debounce timer or from a forced initial load. + """ + @spec process_bounds_update(map(), Socket.t()) :: Socket.t() + def process_bounds_update(map_bounds, socket) do + Logger.debug("process_bounds_update called with bounds: #{inspect(map_bounds)}") + + Subscriptions.update_viewport(socket, map_bounds) + + bounds_state = bounds_update_state(socket, map_bounds) + log_bounds_update_state(bounds_state) + + {new_visible_packets, packets_to_remove} = filtered_packets_for_bounds(socket.assigns.visible_packets, map_bounds) + + socket = + socket + |> DisplayManager.remove_markers_batch(packets_to_remove) + |> maybe_clear_historical_packets(bounds_state) + |> push_event("filter_markers_by_bounds", %{bounds: map_bounds}) + |> assign(map_bounds: map_bounds, visible_packets: new_visible_packets) + |> assign(needs_initial_historical_load: false) + + log_historical_reload(socket) + + socket = HistoricalLoader.start_progressive_historical_loading(socket) + + # If tracking a callsign and at low zoom, send trail line instead of letting + # historical loader send heat map + socket = + if socket.assigns.tracked_callsign != "" and socket.assigns.map_zoom <= 8 do + DisplayManager.send_trail_line_for_tracked_callsign(socket) + else + socket + end + + # Mark initial historical as completed if this was the initial load + if bounds_state.is_initial_load do + assign(socket, initial_historical_completed: true) + else + socket + end + end + + @doc """ + Schedule a debounced bounds update. Cancels any pending timer first. + """ + @spec schedule_bounds_update(map(), Socket.t()) :: {:noreply, Socket.t()} + def schedule_bounds_update(map_bounds, socket) do + if socket.assigns[:bounds_update_timer] do + case Process.cancel_timer(socket.assigns.bounds_update_timer) do + false -> + receive do + {:process_bounds_update, _} -> :ok + after + 0 -> :ok + end + + _ -> + :ok + end + end + + socket = HistoricalLoader.cancel_pending_loads(socket) + + timer_ref = Process.send_after(self(), {:process_bounds_update, map_bounds}, @bounds_update_debounce_ms) + {:noreply, assign(socket, bounds_update_timer: timer_ref, pending_bounds: map_bounds)} + end + + # --- Private helpers --- + + defp valid_bounds?(map_bounds) do + map_bounds.north <= 90 and map_bounds.south >= -90 and map_bounds.north > map_bounds.south + end + + defp handle_valid_bounds_update(map_bounds, socket) do + cond do + socket.assigns[:needs_initial_historical_load] -> + Logger.debug("Processing initial bounds update immediately (forced): #{inspect(map_bounds)}") + socket = process_bounds_update(map_bounds, socket) + {:noreply, socket} + + BoundsUtils.compare_bounds(map_bounds, socket.assigns.map_bounds) -> + {:noreply, socket} + + true -> + schedule_bounds_update(map_bounds, socket) + end + end + + defp bounds_update_state(socket, map_bounds) do + %{ + is_initial_load: socket.assigns[:needs_initial_historical_load] || !socket.assigns[:initial_bounds_loaded], + bounds_changed: + socket.assigns.map_bounds && not BoundsUtils.compare_bounds(map_bounds, socket.assigns.map_bounds), + initial_historical_completed: socket.assigns[:initial_historical_completed] || false + } + end + + defp log_bounds_update_state(bounds_state) do + Logger.debug( + "is_initial_load: #{bounds_state.is_initial_load}, bounds_changed: #{bounds_state.bounds_changed}, " <> + "initial_historical_completed: #{bounds_state.initial_historical_completed}" + ) + end + + defp filtered_packets_for_bounds(visible_packets, map_bounds) do + { + BoundsUtils.filter_packets_by_bounds(visible_packets, map_bounds), + BoundsUtils.reject_packets_by_bounds(visible_packets, map_bounds) + } + end + + defp maybe_clear_historical_packets(socket, %{ + bounds_changed: true, + is_initial_load: false, + initial_historical_completed: true + }) do + Logger.debug("Bounds changed after initial load - clearing historical packets") + push_event(socket, "clear_historical_packets", %{}) + end + + defp maybe_clear_historical_packets(socket, _bounds_state) do + Logger.debug("Initial load or no significant change - keeping existing markers") + socket + end + + defp log_historical_reload(socket) do + Logger.debug("Starting progressive historical loading for new bounds") + + Logger.debug( + "Current assigns: historical_loading=#{socket.assigns.historical_loading}, map_bounds=#{inspect(socket.assigns.map_bounds)}" + ) + end +end diff --git a/lib/aprsme_web/live/map_live/events.ex b/lib/aprsme_web/live/map_live/events.ex new file mode 100644 index 0000000..524cb24 --- /dev/null +++ b/lib/aprsme_web/live/map_live/events.ex @@ -0,0 +1,539 @@ +defmodule AprsmeWeb.MapLive.Events do + @moduledoc """ + Handles all LiveView client events (handle_event/3) for the map view. + Extracted from Index to reduce module size. + + For bounds-related events, delegates back to the message loop in Index; + the bounds pipeline will be extracted to BoundsUpdater in Phase 4. + """ + + import AprsmeWeb.Live.Shared.PacketUtils, only: [get_callsign_key: 1] + import Phoenix.Component, only: [assign: 2, assign: 3] + import Phoenix.LiveView, only: [push_event: 3, push_patch: 2] + + alias Aprsme.Packets + alias Aprsme.Packets.Clustering + alias AprsmeWeb.Live.Shared.BoundsUtils + alias AprsmeWeb.Live.Shared.CoordinateUtils + alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils + alias AprsmeWeb.Live.Shared.ParamUtils + alias AprsmeWeb.MapLive.DataBuilder + alias AprsmeWeb.MapLive.DisplayManager + alias AprsmeWeb.MapLive.Navigation + alias AprsmeWeb.MapLive.RfPath + alias AprsmeWeb.MapLive.UrlParams + alias AprsmeWeb.Plugs.RateLimiter, as: RateLimiterPlug + alias Phoenix.LiveView.Socket + + require Logger + + @event_rate_scale 60_000 + + # --------------------------------------------------------------------------- + # Public event handlers — each corresponds to a handle_event/3 clause in Index + # --------------------------------------------------------------------------- + + @doc """ + Handle bounds_changed or update_bounds events. + Delegates to the bounds update pipeline which lives in Index until Phase 4. + """ + @spec handle_bounds_changed(map(), Socket.t()) :: {:noreply, Socket.t()} + def handle_bounds_changed(bounds, socket) do + if socket.assigns.historical_loading do + {:noreply, assign(socket, pending_bounds: bounds)} + else + map_bounds = BoundsUtils.normalize_bounds(bounds) + + if valid_bounds?(map_bounds) do + send(self(), {:process_bounds_update, map_bounds}) + end + + {:noreply, socket} + end + end + + defp valid_bounds?(map_bounds) do + map_bounds.north <= 90 and map_bounds.south >= -90 and map_bounds.north > map_bounds.south + end + + @spec handle_locate_me(Socket.t()) :: {:noreply, Socket.t()} + def handle_locate_me(socket) do + {:noreply, push_event(socket, "request_geolocation", %{})} + end + + @spec handle_set_location(number() | String.t(), number() | String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_set_location(lat, lng, socket) do + lat_float = to_float(lat) + lng_float = to_float(lng) + + if UrlParams.valid_coordinates?(lat_float, lng_float) do + socket = Navigation.update_and_zoom_to_location(socket, lat_float, lng_float, 12) + {:noreply, socket} + else + {:noreply, socket} + end + end + + @spec handle_clear_and_reload_markers(Socket.t()) :: {:noreply, Socket.t()} + def handle_clear_and_reload_markers(socket) do + filtered_packets = + filter_packets_by_time_and_bounds_with_tracked( + socket.assigns.visible_packets, + socket.assigns.map_bounds, + socket.assigns.packet_age_threshold, + socket.assigns.tracked_callsign, + socket.assigns.tracked_callsign_latest_packet + ) + + visible_packets_list = DataBuilder.build_packet_data_list_from_map(filtered_packets, false, socket) + + socket = + if socket.assigns.map_zoom <= 8 do + send_heat_map_data(socket, filtered_packets) + else + add_markers_if_any(socket, visible_packets_list) + end + + {:noreply, socket} + end + + @spec handle_map_ready(Socket.t()) :: {:noreply, Socket.t()} + def handle_map_ready(socket) do + socket = assign(socket, map_ready: true) + + default_center = UrlParams.default_center() + + socket = + if socket.assigns.map_center.lat == default_center.lat and + socket.assigns.map_center.lng == default_center.lng do + socket + else + Navigation.zoom_to_current_location(socket) + end + + socket = + if socket.assigns.tracked_callsign != "" and socket.assigns.tracked_callsign_latest_packet do + packet = socket.assigns.tracked_callsign_latest_packet + packet_data = DataBuilder.build_packet_data(packet) + + push_event(socket, "add_historical_packets_batch", %{ + packets: [packet_data], + batch: 0, + is_final: false + }) + else + socket + end + + socket = + if socket.assigns.needs_initial_historical_load and socket.assigns.map_bounds do + send(self(), {:process_bounds_update, socket.assigns.map_bounds}) + socket + else + socket + end + + {:noreply, socket} + end + + @spec handle_marker_clicked(Socket.t()) :: {:noreply, Socket.t()} + def handle_marker_clicked(socket) do + {:noreply, assign(socket, station_popup_open: true)} + end + + @spec handle_marker_hover_start(String.t(), String.t(), number(), number(), Socket.t()) :: {:noreply, Socket.t()} + def handle_marker_hover_start(_id, path, lat, lng, socket) do + socket = + if socket.assigns[:hover_end_timer] do + _ = Process.cancel_timer(socket.assigns.hover_end_timer) + assign(socket, hover_end_timer: nil) + else + socket + end + + lat_float = ParamUtils.safe_parse_coordinate(lat, 0.0, -90.0, 90.0) + lng_float = ParamUtils.safe_parse_coordinate(lng, 0.0, -180.0, 180.0) + safe_path = ParamUtils.sanitize_path_string(path) + path_stations = RfPath.parse_rf_path(safe_path) + path_station_positions = RfPath.get_path_station_positions(path_stations, socket) + + socket = + if path_station_positions == [] do + socket + else + push_event(socket, "draw_rf_path", %{ + station_lat: lat_float, + station_lng: lng_float, + path_stations: path_station_positions + }) + end + + {:noreply, socket} + end + + @spec handle_marker_hover_end(Socket.t()) :: {:noreply, Socket.t()} + def handle_marker_hover_end(socket) do + timer = Process.send_after(self(), :clear_rf_path, 100) + {:noreply, assign(socket, hover_end_timer: timer)} + end + + @spec handle_update_callsign(String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_update_callsign(callsign, socket) do + {:noreply, assign(socket, overlay_callsign: callsign)} + end + + @spec handle_track_callsign(String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_track_callsign(callsign, socket) do + case check_event_rate(socket, "track_callsign", 20) do + :ok -> + normalized_callsign = String.upcase(String.trim(callsign)) + + socket = + if normalized_callsign == "" do + socket + |> assign(tracked_callsign: "", other_ssids: []) + |> update_url_with_current_state() + else + other_ssids = Packets.get_other_ssids(normalized_callsign) + latest_packet = Packets.get_latest_packet_for_callsign(normalized_callsign) + + socket = + assign(socket, + tracked_callsign: normalized_callsign, + other_ssids: other_ssids, + tracked_callsign_latest_packet: latest_packet + ) + + socket = zoom_to_latest_packet(socket, latest_packet) + + push_patch(socket, to: "/#{normalized_callsign}") + end + + {:noreply, socket} + + {:deny, _retry_after} -> + {:noreply, socket} + end + end + + @spec handle_clear_tracking(Socket.t()) :: {:noreply, Socket.t()} + def handle_clear_tracking(socket) do + socket = + socket + |> assign(tracked_callsign: "", overlay_callsign: "", other_ssids: []) + |> push_event("clear_trail_line", %{}) + |> update_url_with_current_state() + + {:noreply, socket} + end + + @spec handle_update_trail_duration(String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_update_trail_duration(duration, socket) do + case check_event_rate(socket, "update_trail_duration", 20) do + :ok -> + hours = SharedPacketUtils.parse_trail_duration(duration) + new_threshold = DateTime.add(DateTime.utc_now(), -hours * 3600, :second) + + socket = assign(socket, trail_duration: duration, packet_age_threshold: new_threshold) + socket = push_event(socket, "update_trail_duration", %{duration_hours: hours}) + socket = update_url_with_current_state(socket) + + send(self(), :cleanup_old_packets) + + socket = + if socket.assigns.tracked_callsign != "" and socket.assigns.map_zoom <= 8 do + DisplayManager.send_trail_line_for_tracked_callsign(socket) + else + socket + end + + {:noreply, socket} + + {:deny, _retry_after} -> + {:noreply, socket} + end + end + + @spec handle_update_historical_hours(String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_update_historical_hours(hours, socket) do + case check_event_rate(socket, "update_historical_hours", 20) do + :ok -> + validated_hours = SharedPacketUtils.parse_historical_hours(hours) + socket = assign(socket, historical_hours: to_string(validated_hours)) + socket = update_url_with_current_state(socket) + + if socket.assigns.map_ready do + send(self(), :reload_historical_packets) + end + + {:noreply, socket} + + {:deny, _retry_after} -> + {:noreply, socket} + end + end + + @spec handle_search_callsign(String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_search_callsign(callsign, socket) do + case check_event_rate(socket, "search_callsign", 30) do + :ok -> + callsign + |> String.trim() + |> String.upcase() + |> Navigation.handle_callsign_search(socket) + + {:deny, _retry_after} -> + {:noreply, socket} + end + end + + @spec handle_toggle_slideover(Socket.t()) :: {:noreply, Socket.t()} + def handle_toggle_slideover(socket) do + {:noreply, assign(socket, slideover_open: !socket.assigns.slideover_open)} + end + + @spec handle_set_slideover_state(boolean(), Socket.t()) :: {:noreply, Socket.t()} + def handle_set_slideover_state(open, socket) do + {:noreply, assign(socket, slideover_open: open)} + end + + @spec handle_geolocation_error(String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_geolocation_error(_error, socket) do + {:noreply, socket} + end + + @spec handle_request_geolocation(Socket.t()) :: {:noreply, Socket.t()} + def handle_request_geolocation(socket) do + {:noreply, socket} + end + + @spec handle_popup_closed(Socket.t()) :: {:noreply, Socket.t()} + def handle_popup_closed(socket) do + {:noreply, assign(socket, station_popup_open: false)} + end + + @spec handle_get_assigns(Socket.t()) :: {:noreply, Socket.t()} + def handle_get_assigns(socket) do + send(self(), {:test_assigns, socket.assigns}) + {:noreply, socket} + end + + @spec handle_update_map_state(map(), map(), map(), Socket.t()) :: {:noreply, Socket.t()} + def handle_update_map_state(center, zoom, params, socket) do + {lat, lng} = parse_center_coordinates(center, socket) + zoom = clamp_zoom(zoom) + map_center = %{lat: lat, lng: lng} + + socket = update_map_state(socket, map_center, zoom) + socket = handle_url_update(socket, lat, lng, zoom) + socket = process_bounds_from_params(socket, params) + + {:noreply, socket} + end + + @spec handle_error_boundary_triggered(String.t(), String.t(), String.t(), Socket.t()) :: {:noreply, Socket.t()} + def handle_error_boundary_triggered(message, stack, component_id, socket) do + Logger.error("Error boundary triggered in component #{component_id}: #{message}\n#{stack}") + {:noreply, socket} + end + + # --------------------------------------------------------------------------- + # Rate limiting helpers + # --------------------------------------------------------------------------- + + @doc """ + Store client IP in socket assigns for rate-limiting. + """ + @spec store_client_ip(Socket.t()) :: Socket.t() + def store_client_ip(socket) do + ip = RateLimiterPlug.extract_socket_ip(socket) + assign(socket, :client_ip, ip) + end + + @doc """ + Check whether an event handler should be rate-limited. + """ + @spec check_event_rate(Socket.t(), String.t(), integer()) :: :ok | {:deny, integer()} + def check_event_rate(socket, event_name, limit) do + ip = Map.get(socket.assigns, :client_ip, "unknown") + key = "lv_event:#{event_name}:#{ip}" + + case RateLimiterPlug.check_rate_limit(key, @event_rate_scale, limit) do + {:allow, _count} -> + :ok + + {:deny, retry_after} -> + Logger.warning("LiveView event rate-limited: event=#{event_name} ip=#{ip}") + {:deny, retry_after} + end + end + + # --------------------------------------------------------------------------- + # URL state helpers + # --------------------------------------------------------------------------- + + @doc """ + Update URL with current map state, trail duration, and historical hours. + """ + @spec update_url_with_current_state(Socket.t()) :: Socket.t() + def update_url_with_current_state(socket) do + lat = socket.assigns.map_center.lat + lng = socket.assigns.map_center.lng + zoom = socket.assigns.map_zoom + handle_url_update(socket, lat, lng, zoom) + end + + # --------------------------------------------------------------------------- + # Private helpers + # --------------------------------------------------------------------------- + + defp parse_center_coordinates(center, socket) when is_map(center) do + CoordinateUtils.parse_center_coordinates(center, socket) + end + + defp parse_center_coordinates(_, socket) do + CoordinateUtils.parse_center_coordinates(nil, socket) + end + + defp clamp_zoom(zoom) do + ParamUtils.clamp_zoom(zoom) + end + + defp update_map_state(socket, map_center, zoom) do + old_zoom = socket.assigns.map_zoom + crossing_threshold = DisplayManager.crossing_zoom_threshold?(old_zoom, zoom) + + socket = assign(socket, map_center: map_center, map_zoom: zoom) + + if crossing_threshold do + DisplayManager.handle_zoom_threshold_crossing(socket, zoom) + else + socket + end + end + + defp handle_url_update(socket, lat, lng, zoom) do + if socket.assigns[:should_skip_initial_url_update] && !socket.assigns[:initial_bounds_loaded] do + assign(socket, should_skip_initial_url_update: false) + else + trail_param = + if socket.assigns[:trail_duration] && socket.assigns[:trail_duration] != "1", + do: "&trail=#{socket.assigns[:trail_duration]}", + else: "" + + hist_param = + if socket.assigns[:historical_hours] && socket.assigns[:historical_hours] != "1", + do: "&hist=#{socket.assigns[:historical_hours]}", + else: "" + + base_path = + if socket.assigns.tracked_callsign == "" do + "/" + else + "/#{socket.assigns.tracked_callsign}" + end + + new_path = "#{base_path}?lat=#{lat}&lng=#{lng}&z=#{zoom}#{trail_param}#{hist_param}" + push_patch(socket, to: new_path, replace: true) + end + end + + defp process_bounds_from_params(socket, params) do + case Map.get(params, "bounds") do + %{"north" => north, "south" => south, "east" => east, "west" => west} -> + map_bounds = %{north: north, south: south, east: east, west: west} + + if should_process_bounds?(socket, map_bounds) do + send(self(), {:process_bounds_update, map_bounds}) + end + + socket + + _ -> + socket + end + end + + defp should_process_bounds?(socket, new_bounds) do + socket.assigns.map_bounds != new_bounds or + !socket.assigns[:initial_bounds_loaded] or + socket.assigns[:needs_initial_historical_load] + end + + defp to_float(val) when is_float(val), do: val + defp to_float(val) when is_integer(val), do: val / 1 + + defp to_float(val) when is_binary(val) do + case Float.parse(val) do + {f, _} -> f + :error -> nil + end + end + + defp to_float(_), do: nil + + defp filter_packets_by_time_and_bounds(packets, bounds, time_threshold) do + packets + |> Enum.filter(fn {_callsign, packet} -> + BoundsUtils.within_bounds?(packet, bounds) && + SharedPacketUtils.packet_within_time_threshold?(packet, time_threshold) + end) + |> Map.new() + end + + defp filter_packets_by_time_and_bounds_with_tracked( + packets, + bounds, + time_threshold, + tracked_callsign, + tracked_latest_packet + ) do + filtered = filter_packets_by_time_and_bounds(packets, bounds, time_threshold) + + if tracked_callsign != "" and tracked_latest_packet do + key = get_callsign_key(tracked_latest_packet) + Map.put(filtered, key, tracked_latest_packet) + else + filtered + end + end + + defp add_markers_if_any(socket, []), do: socket + + defp add_markers_if_any(socket, markers) do + push_event(socket, "add_markers", %{markers: markers}) + end + + defp send_heat_map_data(socket, filtered_packets) do + packet_list = Map.values(filtered_packets) + + case Clustering.cluster_packets(packet_list, socket.assigns.map_zoom) do + {:heat_map, heat_points} -> + push_event(socket, "show_heat_map", %{heat_points: heat_points}) + + {:raw_packets, _packets} -> + socket + end + end + + defp zoom_to_latest_packet(socket, nil), do: socket + + defp zoom_to_latest_packet(socket, latest_packet) do + if latest_packet.lat && latest_packet.lon do + lat = Aprsme.EncodingUtils.to_float(latest_packet.lat) || 0.0 + lon = Aprsme.EncodingUtils.to_float(latest_packet.lon) || 0.0 + + packet_data = DataBuilder.build_packet_data(latest_packet) + + socket + |> push_event("zoom_to_location", %{lat: lat, lng: lon, zoom: 12}) + |> push_event("add_historical_packets_batch", %{ + packets: [packet_data], + batch: 0, + is_final: false + }) + else + socket + end + end +end diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index 8049254..5ed4f37 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -4,33 +4,31 @@ defmodule AprsmeWeb.MapLive.Index do """ use AprsmeWeb, :live_view - import AprsmeWeb.Live.Shared.PacketUtils, only: [get_callsign_key: 1] import AprsmeWeb.MapLive.Components import AprsmeWeb.TimeHelpers, only: [time_ago_in_words: 1] import Phoenix.LiveView, - only: [connected?: 1, get_connect_params: 1, push_event: 3, push_patch: 2, put_flash: 3] + only: [push_event: 3, put_flash: 3] alias Aprsme.Packets - alias Aprsme.Packets.Clustering alias AprsmeWeb.Live.Shared.BoundsUtils alias AprsmeWeb.Live.Shared.CoordinateUtils alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils - alias AprsmeWeb.Live.Shared.ParamUtils + alias AprsmeWeb.MapLive.BoundsUpdater alias AprsmeWeb.MapLive.DataBuilder alias AprsmeWeb.MapLive.DisplayManager + alias AprsmeWeb.MapLive.Events alias AprsmeWeb.MapLive.HistoricalLoader alias AprsmeWeb.MapLive.Navigation alias AprsmeWeb.MapLive.PacketBatcher alias AprsmeWeb.MapLive.PacketProcessor - alias AprsmeWeb.MapLive.RfPath + alias AprsmeWeb.MapLive.State + alias AprsmeWeb.MapLive.Subscriptions alias AprsmeWeb.MapLive.UrlParams - alias AprsmeWeb.Plugs.RateLimiter, as: RateLimiterPlug alias AprsmeWeb.TimeUtils alias Phoenix.LiveView.JS # Import the new components module - alias Phoenix.LiveView.Socket alias Phoenix.Socket.Broadcast require Logger @@ -53,7 +51,7 @@ defmodule AprsmeWeb.MapLive.Index do packet_age_threshold = DateTime.add(DateTime.utc_now(), -hours * 3600, :second) # Setup defaults with parsed values - socket = assign_defaults(socket, packet_age_threshold) + socket = State.default(socket, packet_age_threshold) socket = assign(socket, @@ -81,15 +79,15 @@ defmodule AprsmeWeb.MapLive.Index do ) # Calculate initial bounds - initial_bounds = BoundsUtils.calculate_bounds_from_center_and_zoom(final_map_center, final_map_zoom) - socket = setup_subscriptions(socket, initial_bounds) + initial_bounds = State.initial_bounds(final_map_center, final_map_zoom) + socket = Subscriptions.setup(socket, initial_bounds) # Store client IP for rate-limiting in event handlers - socket = store_client_ip(socket) + socket = Events.store_client_ip(socket) # Final socket assignment {:ok, - finalize_mount_assigns(socket, %{ + State.finalize(socket, %{ initial_bounds: initial_bounds, final_map_center: final_map_center, final_map_zoom: final_map_zoom, @@ -100,657 +98,89 @@ defmodule AprsmeWeb.MapLive.Index do })} end - defp setup_subscriptions(socket, initial_bounds) do - do_setup_subscriptions(socket, initial_bounds, connected?(socket)) - end - - defp do_setup_subscriptions(socket, initial_bounds, true) do - # Check if we should accept new connections - if Application.get_env(:aprsme, :cluster_enabled, false) and - not Aprsme.ConnectionMonitor.accepting_connections?() do - # Redirect to another node or show message - socket - |> put_flash(:info, "This server is currently at capacity. Please try again in a moment.") - |> push_event("redirect_to_least_loaded", %{}) - |> assign(:connection_draining, true) - else - # Generate a unique client ID for this LiveView instance - client_id = "liveview_#{:erlang.phash2(self())}" - - # Register with connection monitor - _ = - if Application.get_env(:aprsme, :cluster_enabled, false) do - _ = Aprsme.ConnectionMonitor.register_connection() - # Subscribe to drain events for this node - Phoenix.PubSub.subscribe(Aprsme.PubSub, "connection:drain:#{Node.self()}") - end - - # Register the actual initial viewport. Registering world bounds here used to - # allocate roughly 65,000 one-degree grid cells for every new connection, - # only to tear them down again after the client sent its real bounds. - {:ok, spatial_topic} = Aprsme.SpatialPubSub.register_viewport(client_id, initial_bounds) - - # Subscribe to the spatial topic for this client - :ok = Phoenix.PubSub.subscribe(Aprsme.PubSub, spatial_topic) - - # Still subscribe to bad packets (they don't have location) - :ok = Phoenix.PubSub.subscribe(Aprsme.PubSub, "bad_packets") - - # Subscribe to deployment events - :ok = Phoenix.PubSub.subscribe(Aprsme.PubSub, "deployment_events") - - _ = Process.send_after(self(), :cleanup_old_packets, 60_000) - - socket - |> assign(:spatial_client_id, client_id) - |> assign(:connection_draining, false) - end - end - - defp do_setup_subscriptions(socket, _initial_bounds, false), do: socket - - defp finalize_mount_assigns(socket, %{ - initial_bounds: initial_bounds, - final_map_center: final_map_center, - final_map_zoom: final_map_zoom, - should_skip_initial_url_update: should_skip_initial_url_update, - tracked_callsign: tracked_callsign, - deployed_at: deployed_at, - one_day_ago: one_day_ago - }) do - # Don't override trail_duration and historical_hours if they're already set - trail_duration = Map.get(socket.assigns, :trail_duration, "1") - historical_hours = Map.get(socket.assigns, :historical_hours, "1") - packet_age_threshold = Map.get(socket.assigns, :packet_age_threshold, one_day_ago) - - # If tracking a specific callsign, fetch their latest packet and other SSIDs - {tracked_callsign_latest_packet, other_ssids} = - if tracked_callsign == "" do - {nil, []} - else - try do - packet = Packets.get_latest_packet_for_callsign(tracked_callsign) - ssids = Packets.get_other_ssids(tracked_callsign) - {packet, ssids} - rescue - # Handle database connection errors gracefully (especially in tests) - DBConnection.OwnershipError -> - {nil, []} - - _ -> - {nil, []} - end - end - - # Start packet batcher for efficient updates - {:ok, batcher_pid} = PacketBatcher.start_link(self()) - Process.monitor(batcher_pid) - - # Determine initial slideover state from client viewport width - slideover_open = initial_slideover_open?(socket) - - assign(socket, - map_ready: false, - map_bounds: initial_bounds, - map_center: final_map_center, - map_zoom: final_map_zoom, - should_skip_initial_url_update: should_skip_initial_url_update, - overlay_callsign: "", - tracked_callsign: tracked_callsign, - tracked_callsign_latest_packet: tracked_callsign_latest_packet, - other_ssids: other_ssids, - trail_duration: trail_duration, - historical_hours: historical_hours, - packet_age_threshold: packet_age_threshold, - slideover_open: slideover_open, - deployed_at: deployed_at, - buffer_timer: nil, - batcher_pid: batcher_pid, - station_popup_open: false, - initial_bounds_loaded: false, - # Always load historical data on initial page load - needs_initial_historical_load: true, - # Loading state management - historical_loading: false, - loading_generation: 0, - pending_batch_tasks: [], - # Add missing assigns for components - loading: false, - connection_status: get_initial_connection_status(), - show_all_packets: true - ) - end - - defp get_initial_connection_status do - # Check if APRS connection is disabled - connection_disabled = Application.get_env(:aprsme, :disable_aprs_connection, false) - if connection_disabled, do: "disconnected", else: "connected" - end - - defp initial_slideover_open?(socket) do - if connected?(socket) do - viewport_width = get_connect_params(socket)["viewport_width"] || 1024 - viewport_width >= 1024 - else - true - end - end - - @spec assign_defaults(Socket.t(), DateTime.t()) :: Socket.t() - defp assign_defaults(socket, one_day_ago) do - assign(socket, - packets: [], - visible_packets: %{}, - historical_packets: %{}, - page_title: "APRS Map", - station_popup_open: false, - map_bounds: %{ - north: 49.0, - south: 24.0, - east: -66.0, - west: -125.0 - }, - map_center: UrlParams.default_center(), - map_zoom: UrlParams.default_zoom(), - packet_age_threshold: one_day_ago, - map_ready: false, - historical_loaded: false, - bounds_update_timer: nil, - pending_bounds: nil, - initial_bounds_loaded: false, - # Loading state management - historical_loading: false, - loading_generation: 0, - pending_batch_tasks: [], - # Overlay controls - overlay_callsign: "", - slideover_open: true, - # Track when last update occurred for real-time display in map sidebar - # Updated when packets are processed or map bounds change - last_update_at: DateTime.utc_now(), - # Add missing assigns for components - loading: false, - connection_status: get_initial_connection_status(), - show_all_packets: true - ) - end - - # --- Rate limiting helpers for LiveView event handlers --- - - @event_rate_scale 60_000 - - defp store_client_ip(socket) do - ip = RateLimiterPlug.extract_socket_ip(socket) - assign(socket, :client_ip, ip) - end - - # Checks whether an event handler should be rate-limited. - defp check_event_rate(socket, event_name, limit) do - ip = Map.get(socket.assigns, :client_ip, "unknown") - key = "lv_event:#{event_name}:#{ip}" - - case RateLimiterPlug.check_rate_limit(key, @event_rate_scale, limit) do - {:allow, _count} -> - :ok - - {:deny, retry_after} -> - Logger.warning("LiveView event rate-limited: event=#{event_name} ip=#{ip}") - {:deny, retry_after} - end - end - # Handle both bounds_changed and update_bounds events @impl true def handle_event(event, %{"bounds" => bounds}, socket) when event in ["bounds_changed", "update_bounds"] do - Logger.debug("Received #{event} event with bounds: #{inspect(bounds)}") - handle_bounds_update(bounds, socket) + Events.handle_bounds_changed(bounds, socket) end @impl true - def handle_event("locate_me", _params, socket) do - # Send JavaScript command to request browser geolocation - {:noreply, push_event(socket, "request_geolocation", %{})} - end + def handle_event("locate_me", _params, socket), do: Events.handle_locate_me(socket) @impl true - def handle_event("set_location", %{"lat" => lat, "lng" => lng}, socket) do - # Geolocation sends numeric values; coerce strings to floats if needed - lat_float = to_float(lat) - lng_float = to_float(lng) - - if UrlParams.valid_coordinates?(lat_float, lng_float) do - socket = Navigation.update_and_zoom_to_location(socket, lat_float, lng_float, 12) - {:noreply, socket} - else - {:noreply, socket} - end - end + def handle_event("set_location", %{"lat" => lat, "lng" => lng}, socket), + do: Events.handle_set_location(lat, lng, socket) @impl true - def handle_event("clear_and_reload_markers", _params, socket) do - # Filter by time and bounds - filtered_packets = - filter_packets_by_time_and_bounds_with_tracked( - socket.assigns.visible_packets, - socket.assigns.map_bounds, - socket.assigns.packet_age_threshold, - socket.assigns.tracked_callsign, - socket.assigns.tracked_callsign_latest_packet - ) - - visible_packets_list = DataBuilder.build_packet_data_list_from_map(filtered_packets, false, socket) - - # Check zoom level to decide between heat map and markers - socket = - if socket.assigns.map_zoom <= 8 do - # Use heat map for low zoom levels - send_heat_map_data(socket, filtered_packets) - else - # Use regular markers for high zoom levels - add_markers_if_any(socket, visible_packets_list) - end - - {:noreply, socket} - end + def handle_event("clear_and_reload_markers", _params, socket), do: Events.handle_clear_and_reload_markers(socket) @impl true - def handle_event("map_ready", _params, socket) do - # Mark map as ready - preserve existing needs_initial_historical_load state - # (it's already set correctly in mount based on whether we're tracking a callsign) - socket = assign(socket, map_ready: true) - - # If we have non-default center coordinates (e.g., from geolocation), apply them now - default_center = UrlParams.default_center() - - socket = - if socket.assigns.map_center.lat == default_center.lat and - socket.assigns.map_center.lng == default_center.lng do - socket - else - Navigation.zoom_to_current_location(socket) - end - - # If we're tracking a callsign and have its latest packet, display it immediately - socket = - if socket.assigns.tracked_callsign != "" and socket.assigns.tracked_callsign_latest_packet do - packet = socket.assigns.tracked_callsign_latest_packet - packet_data = DataBuilder.build_packet_data(packet) - - push_event(socket, "add_historical_packets_batch", %{ - packets: [packet_data], - batch: 0, - is_final: false - }) - else - socket - end - - # Wait for JavaScript to send the actual map bounds before loading historical packets - # The calculated bounds might be too small/inaccurate - Logger.debug("Map ready - waiting for JavaScript to send actual bounds before loading historical packets") - - # If we need initial historical load and have bounds, trigger it - socket = - if socket.assigns.needs_initial_historical_load and socket.assigns.map_bounds do - Logger.debug("Map ready with needs_initial_historical_load=true, triggering historical loading") - # Send a message to trigger bounds processing - send(self(), {:process_bounds_update, socket.assigns.map_bounds}) - socket - else - socket - end - - {:noreply, socket} - end + def handle_event("map_ready", _params, socket), do: Events.handle_map_ready(socket) @impl true - def handle_event("marker_clicked", _params, socket) do - # When a marker is clicked, mark that a station popup is open - {:noreply, assign(socket, station_popup_open: true)} - end + def handle_event("marker_clicked", _params, socket), do: Events.handle_marker_clicked(socket) @impl true - def handle_event("marker_hover_start", %{"id" => _id, "path" => path, "lat" => lat, "lng" => lng}, socket) do - # Cancel any pending hover end timer - socket = - if socket.assigns[:hover_end_timer] do - _ = Process.cancel_timer(socket.assigns.hover_end_timer) - assign(socket, hover_end_timer: nil) - else - socket - end - - # Validate coordinates first - lat_float = ParamUtils.safe_parse_coordinate(lat, 0.0, -90.0, 90.0) - lng_float = ParamUtils.safe_parse_coordinate(lng, 0.0, -180.0, 180.0) - - # Validate path string - safe_path = ParamUtils.sanitize_path_string(path) - - # Parse the path to find digipeater/igate stations - path_stations = RfPath.parse_rf_path(safe_path) - - # Query for positions of path stations - path_station_positions = RfPath.get_path_station_positions(path_stations, socket) - - # Send event to draw the RF path lines - socket = - if path_station_positions == [] do - socket - else - push_event(socket, "draw_rf_path", %{ - station_lat: lat_float, - station_lng: lng_float, - path_stations: path_station_positions - }) - end - - {:noreply, socket} - end + def handle_event("marker_hover_start", %{"id" => id, "path" => path, "lat" => lat, "lng" => lng}, socket), + do: Events.handle_marker_hover_start(id, path, lat, lng, socket) @impl true - def handle_event("marker_hover_end", _params, socket) do - # Debounce hover end to prevent flicker during rapid mouse movements - timer = Process.send_after(self(), :clear_rf_path, 100) - {:noreply, assign(socket, hover_end_timer: timer)} - end + def handle_event("marker_hover_end", _params, socket), do: Events.handle_marker_hover_end(socket) @impl true - def handle_event("update_callsign", %{"callsign" => callsign}, socket) do - {:noreply, assign(socket, overlay_callsign: callsign)} - end + def handle_event("update_callsign", %{"callsign" => callsign}, socket), + do: Events.handle_update_callsign(callsign, socket) @impl true - def handle_event("track_callsign", %{"callsign" => callsign}, socket) do - case check_event_rate(socket, "track_callsign", 20) do - :ok -> - normalized_callsign = String.upcase(String.trim(callsign)) - - socket = - if normalized_callsign == "" do - # Clear tracking - socket - |> assign(tracked_callsign: "", other_ssids: []) - |> update_url_with_current_state() - else - # Set tracking, fetch latest packet, zoom to location, and show marker - other_ssids = Packets.get_other_ssids(normalized_callsign) - latest_packet = Packets.get_latest_packet_for_callsign(normalized_callsign) - - socket = - assign(socket, - tracked_callsign: normalized_callsign, - other_ssids: other_ssids, - tracked_callsign_latest_packet: latest_packet - ) - - # Zoom to the callsign's location and display its marker - socket = zoom_to_latest_packet(socket, latest_packet) - - push_patch(socket, to: "/#{normalized_callsign}") - end - - {:noreply, socket} - - {:deny, _retry_after} -> - {:noreply, socket} - end - end + def handle_event("track_callsign", %{"callsign" => callsign}, socket), + do: Events.handle_track_callsign(callsign, socket) @impl true - def handle_event("clear_tracking", _params, socket) do - socket = - socket - |> assign(tracked_callsign: "", overlay_callsign: "", other_ssids: []) - |> push_event("clear_trail_line", %{}) - |> update_url_with_current_state() - - {:noreply, socket} - end + def handle_event("clear_tracking", _params, socket), do: Events.handle_clear_tracking(socket) @impl true - def handle_event("update_trail_duration", %{"trail_duration" => duration}, socket) do - case check_event_rate(socket, "update_trail_duration", 20) do - :ok -> - # Validate and convert duration string to hours - hours = parse_trail_duration(duration) - - # Calculate new threshold safely - new_threshold = DateTime.add(DateTime.utc_now(), -hours * 3600, :second) - - socket = assign(socket, trail_duration: duration, packet_age_threshold: new_threshold) - - # Update client-side TrailManager with new duration - socket = push_event(socket, "update_trail_duration", %{duration_hours: hours}) - - # Update URL with new trail duration - socket = update_url_with_current_state(socket) - - # Trigger cleanup to remove packets that are now outside the new duration - send(self(), :cleanup_old_packets) - - # If tracking a callsign at low zoom, refresh the trail line with new duration - socket = - if socket.assigns.tracked_callsign != "" and socket.assigns.map_zoom <= 8 do - DisplayManager.send_trail_line_for_tracked_callsign(socket) - else - socket - end - - {:noreply, socket} - - {:deny, _retry_after} -> - {:noreply, socket} - end - end + def handle_event("update_trail_duration", %{"trail_duration" => duration}, socket), + do: Events.handle_update_trail_duration(duration, socket) @impl true - def handle_event("update_historical_hours", %{"historical_hours" => hours}, socket) do - case check_event_rate(socket, "update_historical_hours", 20) do - :ok -> - # Validate hours value - validated_hours = parse_historical_hours(hours) - socket = assign(socket, historical_hours: to_string(validated_hours)) - - # Update URL with new historical hours - socket = update_url_with_current_state(socket) - - # Trigger a reload of historical packets with the new time range - if socket.assigns.map_ready do - send(self(), :reload_historical_packets) - end - - {:noreply, socket} - - {:deny, _retry_after} -> - {:noreply, socket} - end - end + def handle_event("update_historical_hours", %{"historical_hours" => hours}, socket), + do: Events.handle_update_historical_hours(hours, socket) @impl true - def handle_event("search_callsign", %{"callsign" => callsign}, socket) do - case check_event_rate(socket, "search_callsign", 30) do - :ok -> - callsign - |> String.trim() - |> String.upcase() - |> handle_callsign_search(socket) - - {:deny, _retry_after} -> - {:noreply, socket} - end - end + def handle_event("search_callsign", %{"callsign" => callsign}, socket), + do: Events.handle_search_callsign(callsign, socket) @impl true - def handle_event("toggle_slideover", _params, socket) do - {:noreply, assign(socket, slideover_open: !socket.assigns.slideover_open)} - end + def handle_event("toggle_slideover", _params, socket), do: Events.handle_toggle_slideover(socket) @impl true - def handle_event("set_slideover_state", %{"open" => open}, socket) do - {:noreply, assign(socket, slideover_open: open)} - end + def handle_event("set_slideover_state", %{"open" => open}, socket), + do: Events.handle_set_slideover_state(open, socket) @impl true - def handle_event("geolocation_error", %{"error" => _error}, socket) do - # Handle geolocation errors gracefully - just continue without geolocation - {:noreply, socket} - end + def handle_event("geolocation_error", %{"error" => error}, socket), do: Events.handle_geolocation_error(error, socket) @impl true - def handle_event("request_geolocation", _params, socket) do - # This event is handled by the JavaScript hook - {:noreply, socket} - end + def handle_event("request_geolocation", _params, socket), do: Events.handle_request_geolocation(socket) @impl true - def handle_event("popup_closed", _params, socket) do - # When any popup is closed, mark that no station popup is open - {:noreply, assign(socket, station_popup_open: false)} - end + def handle_event("popup_closed", _params, socket), do: Events.handle_popup_closed(socket) @impl true - def handle_event("get_assigns", _params, socket) do - send(self(), {:test_assigns, socket.assigns}) - {:noreply, socket} - end + def handle_event("get_assigns", _params, socket), do: Events.handle_get_assigns(socket) @impl true - def handle_event("update_map_state", %{"center" => center, "zoom" => zoom} = params, socket) do - Logger.debug("update_map_state event received: center=#{inspect(center)}, zoom=#{zoom}") - - # Parse and validate coordinates - {lat, lng} = parse_center_coordinates(center, socket) - zoom = clamp_zoom(zoom) - map_center = %{lat: lat, lng: lng} - - # Update map state - socket = update_map_state(socket, map_center, zoom) - - # Handle URL updates - socket = handle_url_update(socket, lat, lng, zoom) - - # Process bounds if included - socket = process_bounds_from_params(socket, params) - - {:noreply, socket} - end + def handle_event("update_map_state", %{"center" => center, "zoom" => zoom} = params, socket), + do: Events.handle_update_map_state(center, zoom, params, socket) @impl true def handle_event( "error_boundary_triggered", %{"message" => message, "stack" => stack, "component_id" => component_id}, socket - ) do - # Log the error for monitoring - Logger.error("Error boundary triggered in component #{component_id}: #{message}\n#{stack}") + ), do: Events.handle_error_boundary_triggered(message, stack, component_id, socket) - # You could also send this to an error tracking service here - # ErrorTracker.report_error(message, stack, %{component: component_id, user_id: socket.assigns[:current_user_id]}) - - {:noreply, socket} - end - - defp parse_center_coordinates(center, socket) when is_map(center) do - CoordinateUtils.parse_center_coordinates(center, socket) - end - - defp parse_center_coordinates(_, socket) do - CoordinateUtils.parse_center_coordinates(nil, socket) - end - - defp clamp_zoom(zoom) do - ParamUtils.clamp_zoom(zoom) - end - - defp update_map_state(socket, map_center, zoom) do - old_zoom = socket.assigns.map_zoom - crossing_threshold = crossing_zoom_threshold?(old_zoom, zoom) - - socket = assign(socket, map_center: map_center, map_zoom: zoom) - - if crossing_threshold do - handle_zoom_threshold_crossing(socket, zoom) - else - socket - end - end - - defp crossing_zoom_threshold?(old_zoom, new_zoom) do - DisplayManager.crossing_zoom_threshold?(old_zoom, new_zoom) - end - - defp handle_zoom_threshold_crossing(socket, zoom) do - DisplayManager.handle_zoom_threshold_crossing(socket, zoom) - end - - defp update_url_with_current_state(socket) do - lat = socket.assigns.map_center.lat - lng = socket.assigns.map_center.lng - zoom = socket.assigns.map_zoom - handle_url_update(socket, lat, lng, zoom) - end - - defp handle_url_update(socket, lat, lng, zoom) do - if socket.assigns[:should_skip_initial_url_update] && !socket.assigns[:initial_bounds_loaded] do - Logger.debug("Skipping URL update on initial load") - assign(socket, should_skip_initial_url_update: false) - else - # Include trail duration and historical hours in URL - trail_param = - if socket.assigns[:trail_duration] && socket.assigns[:trail_duration] != "1", - do: "&trail=#{socket.assigns[:trail_duration]}", - else: "" - - hist_param = - if socket.assigns[:historical_hours] && socket.assigns[:historical_hours] != "1", - do: "&hist=#{socket.assigns[:historical_hours]}", - else: "" - - # Preserve callsign in path if tracking - base_path = - if socket.assigns.tracked_callsign == "" do - "/" - else - "/#{socket.assigns.tracked_callsign}" - end - - new_path = "#{base_path}?lat=#{lat}&lng=#{lng}&z=#{zoom}#{trail_param}#{hist_param}" - Logger.debug("Updating URL to: #{new_path}") - push_patch(socket, to: new_path, replace: true) - end - end - - defp process_bounds_from_params(socket, params) do - case Map.get(params, "bounds") do - %{"north" => north, "south" => south, "east" => east, "west" => west} -> - map_bounds = %{north: north, south: south, east: east, west: west} - - if should_process_bounds?(socket, map_bounds) do - Logger.debug( - "Sending bounds update (initial_load: #{!socket.assigns[:initial_bounds_loaded]}, " <> - "needs_historical: #{socket.assigns[:needs_initial_historical_load]}): #{inspect(map_bounds)}" - ) - - send(self(), {:process_bounds_update, map_bounds}) - end - - socket - - _ -> - socket - end - end - - defp should_process_bounds?(socket, new_bounds) do - socket.assigns.map_bounds != new_bounds or - !socket.assigns[:initial_bounds_loaded] or - socket.assigns[:needs_initial_historical_load] - end - - defp handle_callsign_search(callsign, socket) do - Navigation.handle_callsign_search(callsign, socket) - end + # --- Private helpers kept in Index (used by handle_info / mount / handle_params) --- @impl true def handle_params(params, _url, socket) do @@ -809,18 +239,6 @@ defmodule AprsmeWeb.MapLive.Index do # Parse historical hours with validation defp parse_historical_hours(hours), do: SharedPacketUtils.parse_historical_hours(hours) - defp to_float(val) when is_float(val), do: val - defp to_float(val) when is_integer(val), do: val / 1 - - defp to_float(val) when is_binary(val) do - case Float.parse(val) do - {f, _} -> f - :error -> nil - end - end - - defp to_float(_), do: nil - @impl true def handle_info({:process_bounds_update, map_bounds}, socket), do: handle_info_process_bounds_update(map_bounds, socket) @@ -920,7 +338,7 @@ defmodule AprsmeWeb.MapLive.Index do # Process the pending bounds update bounds = socket.assigns.pending_bounds socket = assign(socket, pending_bounds: nil) - handle_bounds_update(bounds, socket) + BoundsUpdater.handle_bounds_update(bounds, socket) else {:noreply, socket} end @@ -995,7 +413,7 @@ defmodule AprsmeWeb.MapLive.Index do if should_process do Logger.debug("Processing bounds update: #{inspect(map_bounds)}") - socket = process_bounds_update(map_bounds, socket) + socket = BoundsUpdater.process_bounds_update(map_bounds, socket) socket = assign(socket, initial_bounds_loaded: true) {:noreply, socket} else @@ -1739,48 +1157,8 @@ defmodule AprsmeWeb.MapLive.Index do # Helper functions - # Fetch historical packets from the database - - # Select the best packet to display for a callsign - prioritize position over weather - - @spec filter_packets_by_time_and_bounds(map(), map(), DateTime.t()) :: map() - defp filter_packets_by_time_and_bounds(packets, bounds, time_threshold) do - packets - |> Enum.filter(fn {_callsign, packet} -> - BoundsUtils.within_bounds?(packet, bounds) && - SharedPacketUtils.packet_within_time_threshold?(packet, time_threshold) - end) - |> Map.new() - end - - @spec filter_packets_by_time_and_bounds_with_tracked(map(), map(), DateTime.t(), String.t(), map() | nil) :: map() - defp filter_packets_by_time_and_bounds_with_tracked( - packets, - bounds, - time_threshold, - tracked_callsign, - tracked_latest_packet - ) do - filtered = filter_packets_by_time_and_bounds(packets, bounds, time_threshold) - - # Always include the tracked callsign's latest packet if we have one - if tracked_callsign != "" and tracked_latest_packet do - key = get_callsign_key(tracked_latest_packet) - Map.put(filtered, key, tracked_latest_packet) - else - filtered - end - end - # Helper functions for marker operations - @spec add_markers_if_any(Socket.t(), list()) :: Socket.t() - defp add_markers_if_any(socket, []), do: socket - - defp add_markers_if_any(socket, markers) do - push_event(socket, "add_markers", %{markers: markers}) - end - defp preferred_tracked_packet(nil, packet), do: packet defp preferred_tracked_packet(current_packet, incoming_packet) do @@ -1818,245 +1196,7 @@ defmodule AprsmeWeb.MapLive.Index do @impl true def terminate(_reason, socket) do - # Unregister from connection monitor - _ = - if Application.get_env(:aprsme, :cluster_enabled, false) and - not socket.assigns[:connection_draining] do - Aprsme.ConnectionMonitor.unregister_connection() - end - - # Cleanup spatial registration if we have a client ID - _ = - if socket.assigns[:spatial_client_id] do - Aprsme.SpatialPubSub.unregister_client(socket.assigns.spatial_client_id) - end - - _ = if socket.assigns.buffer_timer, do: Process.cancel_timer(socket.assigns.buffer_timer) - - # Clean up any pending bounds update timer - _ = - if socket.assigns[:bounds_update_timer] do - Process.cancel_timer(socket.assigns.bounds_update_timer) - end - - # Clean up pending hover-end timer — set when the mouse leaves a marker - # and only cleared when it re-enters; if the LV terminates between the two, - # the timer would otherwise outlive the socket. - _ = - if socket.assigns[:hover_end_timer] do - Process.cancel_timer(socket.assigns.hover_end_timer) - end - - # Clean up any pending batch tasks - _ = - if socket.assigns[:pending_batch_tasks] do - Enum.each(socket.assigns.pending_batch_tasks, &Process.cancel_timer/1) - end - + Subscriptions.teardown(socket) :ok end - - # Bounds comparison now handled by shared utilities - - # --- Private bounds update helpers --- - @spec handle_bounds_update(map(), Socket.t()) :: {:noreply, Socket.t()} - defp handle_bounds_update(bounds, socket) do - # Skip if we're currently loading historical data - if socket.assigns.historical_loading do - # Defer the bounds update - {:noreply, assign(socket, pending_bounds: bounds)} - else - # Update the map bounds from the client, converting to atom keys - map_bounds = BoundsUtils.normalize_bounds(bounds) - - # Validate bounds to prevent invalid coordinates - if valid_bounds?(map_bounds) do - handle_valid_bounds_update(map_bounds, socket) - else - # Invalid bounds, skip update - {:noreply, socket} - end - end - end - - defp valid_bounds?(map_bounds) do - map_bounds.north <= 90 and map_bounds.south >= -90 and map_bounds.north > map_bounds.south - end - - defp handle_valid_bounds_update(map_bounds, socket) do - # Force processing if we need initial historical load, regardless of bounds comparison - cond do - socket.assigns[:needs_initial_historical_load] -> - Logger.debug("Processing initial bounds update immediately (forced): #{inspect(map_bounds)}") - socket = process_bounds_update(map_bounds, socket) - {:noreply, socket} - - BoundsUtils.compare_bounds(map_bounds, socket.assigns.map_bounds) -> - {:noreply, socket} - - true -> - # For subsequent updates, use the timer to debounce - schedule_bounds_update(map_bounds, socket) - end - end - - # Configurable debounce delay (in milliseconds) - @bounds_update_debounce_ms Application.compile_env(:aprsme, :bounds_update_debounce_ms, 400) - - defp schedule_bounds_update(map_bounds, socket) do - # Cancel existing timer if present - if socket.assigns[:bounds_update_timer] do - case Process.cancel_timer(socket.assigns.bounds_update_timer) do - false -> - # Timer already fired, receive the message to clear it - receive do - {:process_bounds_update, _} -> :ok - after - 0 -> :ok - end - - _ -> - :ok - end - end - - # Cancel any in-progress historical loading to prevent race conditions - socket = HistoricalLoader.cancel_pending_loads(socket) - - # Use configurable debounce time for better stability during rapid zooming - timer_ref = Process.send_after(self(), {:process_bounds_update, map_bounds}, @bounds_update_debounce_ms) - socket = assign(socket, bounds_update_timer: timer_ref, pending_bounds: map_bounds) - {:noreply, socket} - end - - defp send_heat_map_data(socket, filtered_packets) do - # Convert map of packets to list - packet_list = Map.values(filtered_packets) - send_heat_map_for_packets(socket, packet_list) - end - - # Common heat map display logic - defp send_heat_map_for_packets(socket, packets) do - # Get clustering data - case Clustering.cluster_packets(packets, socket.assigns.map_zoom) do - {:heat_map, heat_points} -> - push_event(socket, "show_heat_map", %{heat_points: heat_points}) - - {:raw_packets, _packets} -> - # Shouldn't happen at zoom <= 8, but handle it anyway - socket - end - end - - @spec process_bounds_update(map(), Socket.t()) :: Socket.t() - defp process_bounds_update(map_bounds, socket) do - Logger.debug("process_bounds_update called with bounds: #{inspect(map_bounds)}") - - sync_pubsub_bounds(socket, map_bounds) - - bounds_state = bounds_update_state(socket, map_bounds) - log_bounds_update_state(bounds_state) - - {new_visible_packets, packets_to_remove} = filtered_packets_for_bounds(socket.assigns.visible_packets, map_bounds) - - socket = - socket - |> DisplayManager.remove_markers_batch(packets_to_remove) - |> maybe_clear_historical_packets(bounds_state) - |> push_event("filter_markers_by_bounds", %{bounds: map_bounds}) - |> assign(map_bounds: map_bounds, visible_packets: new_visible_packets) - |> assign(needs_initial_historical_load: false) - - log_historical_reload(socket) - - socket = HistoricalLoader.start_progressive_historical_loading(socket) - - # If tracking a callsign and at low zoom, send trail line instead of letting - # historical loader send heat map - socket = - if socket.assigns.tracked_callsign != "" and socket.assigns.map_zoom <= 8 do - DisplayManager.send_trail_line_for_tracked_callsign(socket) - else - socket - end - - # Mark initial historical as completed if this was the initial load - if bounds_state.is_initial_load do - assign(socket, initial_historical_completed: true) - else - socket - end - end - - defp sync_pubsub_bounds(socket, map_bounds) do - if socket.assigns[:spatial_client_id] do - Aprsme.SpatialPubSub.update_viewport(socket.assigns.spatial_client_id, map_bounds) - end - end - - defp bounds_update_state(socket, map_bounds) do - %{ - is_initial_load: socket.assigns[:needs_initial_historical_load] || !socket.assigns[:initial_bounds_loaded], - bounds_changed: - socket.assigns.map_bounds && not BoundsUtils.compare_bounds(map_bounds, socket.assigns.map_bounds), - initial_historical_completed: socket.assigns[:initial_historical_completed] || false - } - end - - defp log_bounds_update_state(bounds_state) do - Logger.debug( - "is_initial_load: #{bounds_state.is_initial_load}, bounds_changed: #{bounds_state.bounds_changed}, " <> - "initial_historical_completed: #{bounds_state.initial_historical_completed}" - ) - end - - defp filtered_packets_for_bounds(visible_packets, map_bounds) do - { - BoundsUtils.filter_packets_by_bounds(visible_packets, map_bounds), - BoundsUtils.reject_packets_by_bounds(visible_packets, map_bounds) - } - end - - defp maybe_clear_historical_packets(socket, %{ - bounds_changed: true, - is_initial_load: false, - initial_historical_completed: true - }) do - Logger.debug("Bounds changed after initial load - clearing historical packets") - push_event(socket, "clear_historical_packets", %{}) - end - - defp maybe_clear_historical_packets(socket, _bounds_state) do - Logger.debug("Initial load or no significant change - keeping existing markers") - socket - end - - defp log_historical_reload(socket) do - Logger.debug("Starting progressive historical loading for new bounds") - - Logger.debug( - "Current assigns: historical_loading=#{socket.assigns.historical_loading}, map_bounds=#{inspect(socket.assigns.map_bounds)}" - ) - end - - defp zoom_to_latest_packet(socket, nil), do: socket - - defp zoom_to_latest_packet(socket, latest_packet) do - if latest_packet.lat && latest_packet.lon do - lat = Aprsme.EncodingUtils.to_float(latest_packet.lat) || 0.0 - lon = Aprsme.EncodingUtils.to_float(latest_packet.lon) || 0.0 - - packet_data = DataBuilder.build_packet_data(latest_packet) - - socket - |> push_event("zoom_to_location", %{lat: lat, lng: lon, zoom: 12}) - |> push_event("add_historical_packets_batch", %{ - packets: [packet_data], - batch: 0, - is_final: false - }) - else - socket - end - end end diff --git a/lib/aprsme_web/live/map_live/state.ex b/lib/aprsme_web/live/map_live/state.ex new file mode 100644 index 0000000..6106868 --- /dev/null +++ b/lib/aprsme_web/live/map_live/state.ex @@ -0,0 +1,177 @@ +defmodule AprsmeWeb.MapLive.State do + @moduledoc """ + Manages initial LiveView socket state assignment for the map view. + Extracted from Index to reduce module size. + """ + + import Phoenix.Component, only: [assign: 2] + import Phoenix.LiveView, only: [connected?: 1, get_connect_params: 1] + + alias Aprsme.Packets + alias AprsmeWeb.Live.Shared.BoundsUtils + alias AprsmeWeb.MapLive.PacketBatcher + alias AprsmeWeb.MapLive.UrlParams + alias Phoenix.LiveView.Socket + + @doc """ + Assign default values to a fresh socket. + """ + @spec default(Socket.t(), DateTime.t()) :: Socket.t() + def default(socket, one_day_ago) do + assign(socket, + packets: [], + visible_packets: %{}, + historical_packets: %{}, + page_title: "APRS Map", + station_popup_open: false, + map_bounds: %{ + north: 49.0, + south: 24.0, + east: -66.0, + west: -125.0 + }, + map_center: UrlParams.default_center(), + map_zoom: UrlParams.default_zoom(), + packet_age_threshold: one_day_ago, + map_ready: false, + historical_loaded: false, + bounds_update_timer: nil, + pending_bounds: nil, + initial_bounds_loaded: false, + # Loading state management + historical_loading: false, + loading_generation: 0, + pending_batch_tasks: [], + # Overlay controls + overlay_callsign: "", + slideover_open: true, + # Track when last update occurred for real-time display in map sidebar + # Updated when packets are processed or map bounds change + last_update_at: DateTime.utc_now(), + # Add missing assigns for components + loading: false, + connection_status: connection_status(), + show_all_packets: true + ) + end + + @doc """ + Finalize mount assigns with computed values. + Takes the socket after basic setup and an opts map. + """ + @spec finalize(Socket.t(), map()) :: Socket.t() + def finalize(socket, %{ + initial_bounds: initial_bounds, + final_map_center: final_map_center, + final_map_zoom: final_map_zoom, + should_skip_initial_url_update: should_skip_initial_url_update, + tracked_callsign: tracked_callsign, + deployed_at: deployed_at, + one_day_ago: one_day_ago + }) do + # Don't override trail_duration and historical_hours if they're already set + trail_duration = Map.get(socket.assigns, :trail_duration, "1") + historical_hours = Map.get(socket.assigns, :historical_hours, "1") + packet_age_threshold = Map.get(socket.assigns, :packet_age_threshold, one_day_ago) + + # If tracking a specific callsign, fetch their latest packet and other SSIDs + {tracked_callsign_latest_packet, other_ssids} = + if tracked_callsign == "" do + {nil, []} + else + try do + packet = Packets.get_latest_packet_for_callsign(tracked_callsign) + ssids = Packets.get_other_ssids(tracked_callsign) + {packet, ssids} + rescue + # Handle database connection errors gracefully (especially in tests) + DBConnection.OwnershipError -> + {nil, []} + + _ -> + {nil, []} + end + end + + # Start packet batcher for efficient updates + {:ok, batcher_pid} = PacketBatcher.start_link(self()) + Process.monitor(batcher_pid) + + # Determine initial slideover state from client viewport width + slideover_open = slideover_open?(socket) + + assign(socket, + map_ready: false, + map_bounds: initial_bounds, + map_center: final_map_center, + map_zoom: final_map_zoom, + should_skip_initial_url_update: should_skip_initial_url_update, + overlay_callsign: "", + tracked_callsign: tracked_callsign, + tracked_callsign_latest_packet: tracked_callsign_latest_packet, + other_ssids: other_ssids, + trail_duration: trail_duration, + historical_hours: historical_hours, + packet_age_threshold: packet_age_threshold, + slideover_open: slideover_open, + deployed_at: deployed_at, + buffer_timer: nil, + batcher_pid: batcher_pid, + station_popup_open: false, + initial_bounds_loaded: false, + # Always load historical data on initial page load + needs_initial_historical_load: true, + # Loading state management + historical_loading: false, + loading_generation: 0, + pending_batch_tasks: [], + # Add missing assigns for components + loading: false, + connection_status: connection_status(), + show_all_packets: true + ) + end + + @doc """ + Determine whether the slideover should be open on initial render. + """ + @spec slideover_open?(Socket.t()) :: boolean() + def slideover_open?(socket) do + if connected?(socket) do + viewport_width = get_connect_params(socket)["viewport_width"] || 1024 + viewport_width >= 1024 + else + true + end + end + + @doc """ + Get the initial APRS connection status. + """ + @spec connection_status() :: String.t() + def connection_status do + # Check if APRS connection is disabled + connection_disabled = Application.get_env(:aprsme, :disable_aprs_connection, false) + if connection_disabled, do: "disconnected", else: "connected" + end + + @doc """ + Default map center used by UrlParams. + """ + @spec default_center() :: %{lat: float(), lng: float()} + def default_center, do: UrlParams.default_center() + + @doc """ + Default map zoom used by UrlParams. + """ + @spec default_zoom() :: integer() + def default_zoom, do: UrlParams.default_zoom() + + @doc """ + Calculate initial bounds from center and zoom. + """ + @spec initial_bounds(%{lat: float(), lng: float()}, integer()) :: map() + def initial_bounds(center, zoom) do + BoundsUtils.calculate_bounds_from_center_and_zoom(center, zoom) + end +end diff --git a/lib/aprsme_web/live/map_live/subscriptions.ex b/lib/aprsme_web/live/map_live/subscriptions.ex new file mode 100644 index 0000000..e1726d7 --- /dev/null +++ b/lib/aprsme_web/live/map_live/subscriptions.ex @@ -0,0 +1,122 @@ +defmodule AprsmeWeb.MapLive.Subscriptions do + @moduledoc """ + Manages Phoenix.PubSub subscriptions and connection lifecycle for the map LiveView. + Extracted from Index to reduce module size. + """ + + import Phoenix.Component, only: [assign: 3] + import Phoenix.LiveView, only: [connected?: 1, push_event: 3, put_flash: 3] + + alias Phoenix.LiveView.Socket + + @doc """ + Set up spatial and PubSub subscriptions for a newly mounted socket. + Only creates subscriptions when the socket is connected (LiveSocket). + """ + @spec setup(Socket.t(), map()) :: Socket.t() + def setup(socket, initial_bounds) do + do_setup(socket, initial_bounds, connected?(socket)) + end + + @doc """ + Clean up all subscriptions and timers when the LiveView terminates. + """ + @spec teardown(Socket.t()) :: :ok + def teardown(socket) do + _teardown_connection_monitor(socket) + _teardown_spatial(socket) + _teardown_timers(socket) + _teardown_batch_tasks(socket) + :ok + end + + @doc """ + Update the spatial viewport bounds for an existing subscription. + """ + @spec update_viewport(Socket.t(), map()) :: :ok | nil + def update_viewport(socket, map_bounds) do + if socket.assigns[:spatial_client_id] do + Aprsme.SpatialPubSub.update_viewport(socket.assigns.spatial_client_id, map_bounds) + end + end + + # --- Private helpers --- + + defp do_setup(socket, _initial_bounds, false), do: socket + + defp do_setup(socket, initial_bounds, true) do + # Check if we should accept new connections + if Application.get_env(:aprsme, :cluster_enabled, false) and + not Aprsme.ConnectionMonitor.accepting_connections?() do + socket + |> put_flash(:info, "This server is currently at capacity. Please try again in a moment.") + |> push_event("redirect_to_least_loaded", %{}) + |> assign(:connection_draining, true) + else + # Generate a unique client ID for this LiveView instance + client_id = "liveview_#{:erlang.phash2(self())}" + + # Register with connection monitor + _ = + if Application.get_env(:aprsme, :cluster_enabled, false) do + _ = Aprsme.ConnectionMonitor.register_connection() + # Subscribe to drain events for this node + Phoenix.PubSub.subscribe(Aprsme.PubSub, "connection:drain:#{Node.self()}") + end + + # Register the actual initial viewport. + {:ok, spatial_topic} = Aprsme.SpatialPubSub.register_viewport(client_id, initial_bounds) + + # Subscribe to the spatial topic for this client + :ok = Phoenix.PubSub.subscribe(Aprsme.PubSub, spatial_topic) + + # Still subscribe to bad packets (they don't have location) + :ok = Phoenix.PubSub.subscribe(Aprsme.PubSub, "bad_packets") + + # Subscribe to deployment events + :ok = Phoenix.PubSub.subscribe(Aprsme.PubSub, "deployment_events") + + _ = Process.send_after(self(), :cleanup_old_packets, 60_000) + + socket + |> assign(:spatial_client_id, client_id) + |> assign(:connection_draining, false) + end + end + + defp _teardown_connection_monitor(socket) do + _ = + if Application.get_env(:aprsme, :cluster_enabled, false) and + not socket.assigns[:connection_draining] do + Aprsme.ConnectionMonitor.unregister_connection() + end + end + + defp _teardown_spatial(socket) do + _ = + if socket.assigns[:spatial_client_id] do + Aprsme.SpatialPubSub.unregister_client(socket.assigns.spatial_client_id) + end + end + + defp _teardown_timers(socket) do + _ = if socket.assigns.buffer_timer, do: Process.cancel_timer(socket.assigns.buffer_timer) + + _ = + if socket.assigns[:bounds_update_timer] do + Process.cancel_timer(socket.assigns.bounds_update_timer) + end + + _ = + if socket.assigns[:hover_end_timer] do + Process.cancel_timer(socket.assigns.hover_end_timer) + end + end + + defp _teardown_batch_tasks(socket) do + _ = + if socket.assigns[:pending_batch_tasks] do + Enum.each(socket.assigns.pending_batch_tasks, &Process.cancel_timer/1) + end + end +end