diff --git a/assets/js/app.js b/assets/js/app.js index a318235..05b7cf3 100644 --- a/assets/js/app.js +++ b/assets/js/app.js @@ -33,9 +33,45 @@ Hooks.APRSMap = { const initialCenter = JSON.parse(this.el.dataset.center); const initialZoom = parseInt(this.el.dataset.zoom); + console.log("Initializing map with center:", initialCenter, "and zoom:", initialZoom); + console.log("Map container element:", this.el); + console.log("Container dimensions:", this.el.offsetWidth, "x", this.el.offsetHeight); + console.log("Parsed coordinates:", initialCenter.lat, initialCenter.lng, "zoom:", initialZoom); + // Initialize the map with the server-provided location - const map = L.map(this.el).setView([initialCenter.lat, initialCenter.lng], initialZoom); - console.log("Map initialized:", map); + const map = L.map(this.el, { + zoomControl: true, + attributionControl: true, + closePopupOnClick: true, + }).setView([initialCenter.lat, initialCenter.lng], initialZoom); + + console.log("Map setView called with:", [initialCenter.lat, initialCenter.lng], initialZoom); + console.log("Map object created:", map); + + // Validate container size and force refresh if needed + if (this.el.offsetWidth === 0 || this.el.offsetHeight === 0) { + console.warn("Map container has zero dimensions, forcing size refresh"); + setTimeout(() => { + map.invalidateSize(); + map.setView([initialCenter.lat, initialCenter.lng], initialZoom); + console.log("Map size refreshed and view reset"); + }, 100); + } + + // Force initial size calculation + map.invalidateSize(); + + // Track when map is ready + map.whenReady(() => { + console.log("Map is fully ready and rendered"); + console.log("Map center after ready:", map.getCenter()); + console.log("Map zoom after ready:", map.getZoom()); + console.log("Map bounds after ready:", map.getBounds()); + this.mapReady = true; + this.pushEvent("map_ready", {}); + }); + + console.log("Map initialized"); // Handle geolocation requests from server this.handleEvent("request_geolocation", () => { @@ -112,25 +148,107 @@ Hooks.APRSMap = { // Store markers to avoid duplicates this.markers = new Map(); + this.historicalMarkers = new Map(); this.packetCount = 0; - // Store map instance + // Store map instance and initialize state this.map = map; + this.mapReady = false; + this.userLocationMarker = null; // Send initial bounds to server this.sendBoundsToServer(); // Listen for new packets from the server this.handleEvent("new_packet", (packet) => { - console.log("Received new packet:", packet); this.addPacketMarker(packet); }); + // Listen for historical packets + this.handleEvent("historical_packet", (packet) => { + this.addPacketMarker(packet, true); + }); + + // Listen for zoom to location event + this.handleEvent("zoom_to_location", (data) => { + console.log("Received zoom_to_location event:", data); + + if (data.lat && data.lng) { + // Convert to numbers to ensure proper handling + const lat = parseFloat(data.lat); + const lng = parseFloat(data.lng); + const zoom = parseInt(data.zoom || 12); + + console.log(`Setting map view to [${lat}, ${lng}] with zoom ${zoom}`); + + // Force map invalidation before setting view + this.map.invalidateSize(); + + // Check container dimensions before zoom + if (this.el.offsetWidth === 0 || this.el.offsetHeight === 0) { + console.warn( + "Container has zero dimensions during zoom, width:", + this.el.offsetWidth, + "height:", + this.el.offsetHeight, + ); + } + + // Small delay to ensure map is ready + setTimeout(() => { + // Force another size refresh right before setView + this.map.invalidateSize(); + console.log("About to call setView with:", [lat, lng], zoom); + this.map.setView([lat, lng], zoom, { + animate: true, + duration: 1, + }); + + // Force final size refresh after setView + setTimeout(() => { + this.map.invalidateSize(); + }, 100); + console.log("Map view updated after delay"); + console.log("New map center:", this.map.getCenter()); + console.log("New map zoom:", this.map.getZoom()); + + // Add a marker at the user's location + if (this.userLocationMarker) { + this.map.removeLayer(this.userLocationMarker); + } + + this.userLocationMarker = L.marker([lat, lng], { + icon: L.divIcon({ + html: '
', + className: "user-location-marker", + iconSize: [16, 16], + iconAnchor: [8, 8], + }), + }) + .addTo(this.map) + .bindPopup("Your location") + .openPopup(); + }, 300); + } else { + console.error("Invalid coordinates in zoom_to_location event:", data); + } + }); + + // Listen for clearing historical packets + this.handleEvent("clear_historical_packets", () => { + this.clearHistoricalMarkers(); + }); + // Listen for clear markers event this.handleEvent("clear_markers", () => { this.clearAllMarkers(); }); + // Listen for refresh markers event + this.handleEvent("refresh_markers", () => { + this.refreshMarkers(); + }); + // Handle geolocation button clicks this.handleEvent("request_geolocation", () => { if ("geolocation" in navigator) { @@ -151,7 +269,11 @@ Hooks.APRSMap = { // Update bounds when map moves or zooms map.on("moveend", () => { - this.sendBoundsToServer(); + // Debounce the bounds update to avoid too many server calls + if (this.boundsTimer) clearTimeout(this.boundsTimer); + this.boundsTimer = setTimeout(() => { + this.sendBoundsToServer(); + }, 300); }); // Handle map resize when window is resized @@ -184,8 +306,12 @@ Hooks.APRSMap = { this.markers.forEach((marker) => { this.map.removeLayer(marker); }); + this.historicalMarkers.forEach((marker) => { + this.map.removeLayer(marker); + }); } this.markers.clear(); + this.historicalMarkers.clear(); this.packetCount = 0; const counterElement = document.getElementById("packet-count"); if (counterElement) { @@ -193,36 +319,70 @@ Hooks.APRSMap = { } }, - removeMarkersOutsideBounds(bounds) { - // With clustering, the cluster group handles visibility automatically - // We'll just track the count of markers within bounds - let visibleCount = 0; + refreshMarkers() { + // More efficient refresh that doesn't require clearing all markers + // Filter markers to keep only those added in the last hour + const now = new Date(); + const oneHourAgo = new Date(now.getTime() - 60 * 60 * 1000); + // First, collect markers to remove + const markersToRemove = []; this.markers.forEach((marker, callsign) => { - const latLng = marker.getLatLng(); - if (bounds.contains(latLng)) { - visibleCount++; + if (marker.addedAt && marker.addedAt < oneHourAgo) { + markersToRemove.push(callsign); } }); - // Update counter to show visible markers + // Then remove them from the map and collection + markersToRemove.forEach((callsign) => { + const marker = this.markers.get(callsign); + if (marker) { + if (this.markerClusterGroup) { + this.markerClusterGroup.removeLayer(marker); + } else { + this.map.removeLayer(marker); + } + this.markers.delete(callsign); + } + }); + + // Update counter + this.packetCount = this.markers.size; const counterElement = document.getElementById("packet-count"); if (counterElement) { - counterElement.textContent = visibleCount; + counterElement.textContent = this.packetCount; } + }, - // Notify server of the updated packet count + clearHistoricalMarkers() { + // Remove only historical markers + if (this.markerClusterGroup) { + this.historicalMarkers.forEach((marker) => { + this.markerClusterGroup.removeLayer(marker); + }); + } else { + // Fallback: remove markers directly from map + this.historicalMarkers.forEach((marker) => { + this.map.removeLayer(marker); + }); + } + this.historicalMarkers.clear(); + }, + + removeMarkersOutsideBounds(bounds) { + // Let LiveView handle the packet count + // The cluster handles marker visibility automatically + const visibleCount = this.markers.size; this.pushEvent("update_packet_count", { count: visibleCount }); }, - addPacketMarker(packet) { - console.log("addPacketMarker called with:", packet); + addPacketMarker(packet, isHistorical = false) { + // Skip packets without required data if ( !packet["data_extended"] || !packet["data_extended"]["latitude"] || !packet["data_extended"]["longitude"] ) { - console.warn("Packet missing required location data:", packet); return; } @@ -231,76 +391,112 @@ Hooks.APRSMap = { // Validate coordinates are within valid ranges if (isNaN(lat) || isNaN(lng) || lat < -90 || lat > 90 || lng < -180 || lng > 180) { - console.error("Invalid coordinates:", { lat, lng, packet }); return; } - console.log("Valid coordinates:", { lat, lng, callsign: packet["base_callsign"] }); - const callsign = packet["base_callsign"] + (packet["ssid"] ? "-" + packet["ssid"] : ""); - // Create popup content + // Generate a unique ID for the marker + const callsign = packet["base_callsign"] + (packet["ssid"] ? "-" + packet["ssid"] : ""); + // For historical packets, add a timestamp or unique ID to distinguish them + const markerId = isHistorical ? `hist_${callsign}_${Date.now()}` : callsign; + + // Create popup content with historical indicator if needed + const timestamp = + isHistorical && packet["timestamp"] + ? new Date(packet["timestamp"]).toLocaleString() + : new Date().toLocaleTimeString(); + const popupContent = `
-

${callsign}

+

${callsign} ${isHistorical ? "(Historical)" : ""}

Position: ${lat.toFixed(4)}°, ${lng.toFixed(4)}°
Type: ${packet["data_type"]}
${packet["data_extended"]["comment"] ? `Comment: ${packet["data_extended"]["comment"]}
` : ""} Path: ${packet["path"]}
- Time: ${new Date().toLocaleTimeString()} + Time: ${timestamp}

`; - // Check if marker already exists - if (this.markers.has(callsign)) { - // Update existing marker - const existingMarker = this.markers.get(callsign); - if (this.markerClusterGroup) { - // Remove and re-add to cluster group - this.markerClusterGroup.removeLayer(existingMarker); - existingMarker.setLatLng([lat, lng]); - existingMarker.setPopupContent(popupContent); - this.markerClusterGroup.addLayer(existingMarker); - } else { - // Fallback: just update position - existingMarker.setLatLng([lat, lng]); - existingMarker.setPopupContent(popupContent); + // Determine which marker collection to use + const markerCollection = isHistorical ? this.historicalMarkers : this.markers; + + // Check if marker already exists (only for non-historical markers) + if (!isHistorical && this.markers.has(markerId)) { + try { + // Update existing marker + const existingMarker = this.markers.get(markerId); + if (this.markerClusterGroup) { + // Remove and re-add to cluster group + this.markerClusterGroup.removeLayer(existingMarker); + existingMarker.setLatLng([lat, lng]); + existingMarker.setPopupContent(popupContent); + this.markerClusterGroup.addLayer(existingMarker); + } else { + // Fallback: just update position + existingMarker.setLatLng([lat, lng]); + existingMarker.setPopupContent(popupContent); + } + } catch (e) { + // If there's an error updating, proceed with creating a new marker + this.markers.delete(markerId); } - } else { + } + + // Create new marker if it doesn't exist or if we're handling a historical packet + if (isHistorical || !this.markers.has(markerId)) { // Create new marker const icon = this.createAPRSIcon( packet["data_extended"]["symbol_table_id"] || "/", packet["data_extended"]["symbol_code"] || ">", + isHistorical, ); const marker = L.marker([lat, lng], { icon: icon }).bindPopup(popupContent); + marker.addedAt = new Date(); // Track when the marker was added - // Add to cluster group or directly to map - if (this.markerClusterGroup) { - this.markerClusterGroup.addLayer(marker); - } else { - marker.addTo(this.map); + // Add CSS class for historical markers + if (isHistorical) { + marker.options.className = "historical-marker"; } - this.markers.set(callsign, marker); - console.log("New marker added for:", callsign, "Total markers:", this.markers.size); + // Add to cluster group or directly to map + try { + if (this.markerClusterGroup) { + this.markerClusterGroup.addLayer(marker); + } else { + marker.addTo(this.map); + } + } catch (e) { + // Silently ignore errors when adding markers + } - // Update packet counter - this.packetCount++; - const counterElement = document.getElementById("packet-count"); - if (counterElement) { - counterElement.textContent = this.packetCount; + // Store the marker + markerCollection.set(markerId, marker); + + // Update packet counter (only for live packets) + if (!isHistorical) { + this.packetCount++; + const counterElement = document.getElementById("packet-count"); + if (counterElement) { + counterElement.textContent = this.packetCount; + } } } }, - createAPRSIcon(symbolTable, symbolCode) { + createAPRSIcon(symbolTable, symbolCode, isHistorical = false) { // Default icon color based on symbol table - const color = symbolTable === "/" ? "#2563eb" : "#dc2626"; + let color = symbolTable === "/" ? "#2563eb" : "#dc2626"; + + // Use a different color for historical markers + if (isHistorical) { + color = symbolTable === "/" ? "#90b4ed" : "#e98a84"; + } return L.divIcon({ html: `
`, - className: "aprs-marker", + className: isHistorical ? "aprs-marker historical-marker" : "aprs-marker", iconSize: [16, 16], iconAnchor: [8, 8], popupAnchor: [0, -8], @@ -308,16 +504,18 @@ Hooks.APRSMap = { }, destroyed() { - console.log("APRSMap hook destroyed"); + if (this.boundsTimer) { + clearTimeout(this.boundsTimer); + } if (this.markerClusterGroup) { this.markerClusterGroup.clearLayers(); } if (this.map) { - console.log("Removing map instance"); this.map.remove(); this.map = null; } this.markers.clear(); + this.historicalMarkers.clear(); }, }; diff --git a/config/config.exs b/config/config.exs index f6daef8..2b26663 100644 --- a/config/config.exs +++ b/config/config.exs @@ -26,6 +26,18 @@ config :aprs, AprsWeb.Endpoint, pubsub_server: Aprs.PubSub, live_view: [signing_salt: "ees098qG"] +# Configure Oban for background jobs +config :aprs, Oban, + repo: Aprs.Repo, + plugins: [ + {Oban.Plugins.Pruner, max_age: 60 * 60 * 24 * 7}, + {Oban.Plugins.Cron, + crontab: [ + {"0 0 * * *", Aprs.Workers.PacketCleanupWorker} + ]} + ], + queues: [default: 10, maintenance: 2] + config :aprs, ecto_repos: [Aprs.Repo] diff --git a/lib/aprs/application.ex b/lib/aprs/application.ex index e6fca08..92a7189 100644 --- a/lib/aprs/application.ex +++ b/lib/aprs/application.ex @@ -5,6 +5,8 @@ defmodule Aprs.Application do use Application + # Configure Oban for background jobs + @impl true def start(_type, _args) do topologies = Application.get_env(:libcluster, :topologies) || [] @@ -24,6 +26,10 @@ defmodule Aprs.Application do # {Aprs.Worker, arg} {Registry, keys: :duplicate, name: Registry.PubSub, partitions: System.schedulers_online()}, {Cluster.Supervisor, [topologies, [name: Aprs.ClusterSupervisor]]}, + # Start Oban for background jobs + {Oban, :aprs |> Application.get_env(Oban, []) |> Keyword.put(:queues, default: 10, maintenance: 2)}, + # Start the scheduler for recurring Oban jobs + Aprs.Workers.Scheduler, Aprs.Presence ] diff --git a/lib/aprs/db_test.ex b/lib/aprs/db_test.ex new file mode 100644 index 0000000..a202745 --- /dev/null +++ b/lib/aprs/db_test.ex @@ -0,0 +1,139 @@ +defmodule Aprs.DbTest do + @moduledoc """ + Helper module for testing database connections and operations from IEx. + Run with: + iex> Aprs.DbTest.test_packet_storage() + """ + + alias Aprs.Packet + alias Aprs.Packets + alias Aprs.Repo + + require Logger + + @doc """ + Tests packet storage functionality by creating and storing a test packet. + """ + def test_packet_storage do + IO.puts("Starting database test...") + + # Test database connection + case check_db_connection() do + :ok -> + # Create a test packet with minimum required fields + test_packet = %{ + base_callsign: "TEST", + ssid: "1", + sender: "TEST-1", + destination: "APRS", + data_type: "position", + path: "TCPIP*", + information_field: "Test packet", + received_at: DateTime.utc_now(), + lat: 33.5, + lon: -97.5, + region: "test", + has_position: true, + data_extended: %{ + latitude: 33.5, + longitude: -97.5, + symbol_table_id: "/", + symbol_code: ">", + comment: "Test comment", + aprs_messaging: false + } + } + + IO.puts("Created test packet: #{inspect(test_packet, pretty: true)}") + + # Attempt to store the packet + case Packets.store_packet(test_packet) do + {:ok, stored_packet} -> + IO.puts("Successfully stored packet!") + IO.puts("Stored packet ID: #{stored_packet.id}") + {:ok, stored_packet} + + {:error, changeset} -> + IO.puts("Failed to store packet!") + IO.puts("Errors: #{inspect(changeset.errors, pretty: true)}") + {:error, changeset} + end + + error -> + IO.puts("Database connection test failed: #{inspect(error)}") + error + end + end + + @doc """ + Tests if the database is accessible. + """ + def check_db_connection do + IO.puts("Checking database connection...") + + try do + # Try to get a connection from the pool + Repo.checkout(fn conn -> + IO.puts("Got database connection!") + Postgrex.query!(conn, "SELECT 1", []) + end) + + # Try a simple query through Ecto + count = Repo.aggregate(Packet, :count, :id) + IO.puts("Current packet count in database: #{count}") + + :ok + rescue + e -> + IO.puts("Database connection error: #{inspect(e)}") + {:error, e} + end + end + + @doc """ + Count packets with position data. + """ + def count_packets_with_position do + import Ecto.Query + + IO.puts("Counting packets with position data...") + + try do + count = Repo.aggregate(from(p in Packet, where: p.has_position == true), :count, :id) + + IO.puts("Found #{count} packets with position data") + {:ok, count} + rescue + e -> + IO.puts("Query error: #{inspect(e)}") + {:error, e} + end + end + + @doc """ + List the most recent packets with position data. + """ + def list_recent_packets(limit \\ 10) do + import Ecto.Query + + IO.puts("Fetching #{limit} most recent packets...") + + try do + packets = Repo.all(from(p in Packet, where: p.has_position == true, order_by: [desc: p.received_at], limit: ^limit)) + + IO.puts("Found #{length(packets)} recent packets") + + if length(packets) > 0 do + Enum.each(packets, fn packet -> + IO.puts("#{packet.sender} at #{packet.lat}, #{packet.lon} - #{packet.received_at}") + end) + end + + {:ok, packets} + rescue + e -> + IO.puts("Query error: #{inspect(e)}") + {:error, e} + end + end +end diff --git a/lib/aprs/is/is.ex b/lib/aprs/is/is.ex index 2d89da4..b3658ff 100644 --- a/lib/aprs/is/is.ex +++ b/lib/aprs/is/is.ex @@ -279,37 +279,45 @@ defmodule Aprs.Is do def dispatch(message) do case Parser.parse(message) do {:ok, parsed_message} -> - # IO.inspect(parsed_message) - # time_message_received = - # :calendar.universal_time() |> :calendar.datetime_to_gregorian_seconds() + # Store the packet in the database for future replay + # Use Task to avoid slowing down the main process + Task.start(fn -> + require Logger + # Store the packet if it has position data + if has_position_data?(parsed_message) do + Logger.info("Storing packet with position data: #{inspect(parsed_message.sender)}") + # Always set received_at timestamp to ensure consistency + current_time = DateTime.truncate(DateTime.utc_now(), :microsecond) + packet_data = Map.put(parsed_message, :received_at, current_time) - # This doesnt work if dispatch if spawned as a task, since it does not own the table - # :ets.insert(:aprs, {parsed_message.sender, message, time_message_received}) + # Convert to map before storing to avoid struct conversion issues + attrs = Map.from_struct(packet_data) - # Get messages since last time the callsign was seen - # TODO: Get last seen timestamp and use that - # _message_count = - # case :ets.lookup(:aprs_messages, parsed_message.sender) do - # [{_callsign, _message, timestamp}] -> - # recent_messages_for(parsed_message.sender, timestamp) + # Normalize data_type to string if it's an atom + attrs = normalize_data_type(attrs) - # [] -> - # 0 + # Ensure SSID is never nil + attrs = + if Map.has_key?(attrs, :ssid) and is_nil(attrs.ssid) do + Map.put(attrs, :ssid, "0") + else + attrs + end - # _ -> - # 0 - # end + # Store in database through the Packets context + case Aprs.Packets.store_packet(attrs) do + {:ok, packet} -> + Logger.info("Successfully stored packet from #{packet.sender}") - # Logger.debug("#{message_count} recent messages for #{parsed_message.sender}") - - # Do something interesting with the message - # process(parsed_message, parsed_message.data_type) - - # Publish the parsed message to all interested parties - # Registry.dispatch(Registry.PubSub, "aprs_messages", fn entries -> - # for {pid, _} <- entries, do: send(pid, {:broadcast, parsed_message}) - # end) + {:error, changeset} -> + Logger.error("Failed to store packet: #{inspect(changeset.errors)}") + end + else + Logger.debug("Skipping packet without position data: #{inspect(parsed_message.sender)}") + end + end) + # Broadcast to live clients AprsWeb.Endpoint.broadcast("aprs_messages", "packet", parsed_message) # Logger.debug("BROADCAST: " <> inspect(parsed_message)) @@ -333,21 +341,98 @@ defmodule Aprs.Is do end end - # def process(message, message_type) when message_type == :message do - # # time_message_received = - # # :calendar.universal_time() |> :calendar.datetime_to_gregorian_seconds() + # Helper to check if a packet has position data worth storing + defp has_position_data?(packet) do + require Logger - # # :ets.insert( - # # :aprs_messages, - # # {message.data_extended.to, message.data_extended, time_message_received} - # # ) - # end + result = + case packet do + %{data_extended: nil} -> + Logger.debug("Packet has nil data_extended: #{inspect(packet.sender)}") + false - # def process(_, _), do: nil + %{data_extended: %{latitude: lat, longitude: lon}} when not is_nil(lat) and not is_nil(lon) -> + # Check if coordinates are valid numbers + valid = are_valid_coords?(lat, lon) + + if !valid do + Logger.debug("Invalid coordinates: lat=#{inspect(lat)}, lon=#{inspect(lon)}") + end + + valid + + %{data_extended: %Parser.Types.MicE{} = mic_e} -> + # MicE packets have lat/lon in separate components + valid = + is_number(mic_e.lat_degrees) and is_number(mic_e.lat_minutes) and + is_number(mic_e.lon_degrees) and is_number(mic_e.lon_minutes) + + Logger.debug("MicE packet position check: #{valid} for #{inspect(packet.sender)}") + valid + + %{lat: lat, lon: lon} when not is_nil(lat) and not is_nil(lon) -> + # Handle case where coordinates are at top level + are_valid_coords?(lat, lon) + + _other -> + Logger.debug("Unrecognized packet format: #{inspect(Map.keys(packet))}") + false + end + + Logger.debug("Position data check for #{inspect(packet.sender)}: #{result}") + result + end + + # Helper to validate coordinate values + defp are_valid_coords?(lat, lon) do + require Logger + + # Convert to float if possible + lat_float = to_float(lat) + lon_float = to_float(lon) + + # Log coordinate conversion + if !is_number(lat_float) do + Logger.debug("Could not convert latitude to float: #{inspect(lat)}") + end + + if !is_number(lon_float) do + Logger.debug("Could not convert longitude to float: #{inspect(lon)}") + end + + # Check if conversion succeeded and values are in valid ranges + result = + is_number(lat_float) and is_number(lon_float) and + lat_float >= -90 and lat_float <= 90 and lon_float >= -180 and lon_float <= 180 + + if !result do + Logger.debug("Invalid coordinates: lat=#{inspect(lat_float)}, lon=#{inspect(lon_float)}") + end + + result + end + + # Helper to convert various types to float + defp to_float(value) when is_float(value), do: value + defp to_float(value) when is_integer(value), do: value * 1.0 + defp to_float(%Decimal{} = value), do: Decimal.to_float(value) + + defp to_float(value) when is_binary(value) do + case Float.parse(value) do + {float, _} -> float + :error -> nil + end + end + + defp to_float(_), do: nil + + # Normalize data_type to ensure proper storage + defp normalize_data_type(%{data_type: data_type} = attrs) when is_atom(data_type) do + %{attrs | data_type: to_string(data_type)} + end + + defp normalize_data_type(attrs), do: attrs - # def recent_messages_for(callsign, since_time) do - # callsign_guard = {:==, :"$1", {:const, callsign}} - # timestamp_guard = {:>=, :"$2", {:const, since_time}} # total_spec = [{{:"$1", :_, :"$2"}, [{:andalso, callsign_guard, timestamp_guard}], [true]}] # :ets.select_count(:aprs_messages, total_spec) # end diff --git a/lib/aprs/packet.ex b/lib/aprs/packet.ex index a4c2241..f3f7abd 100644 --- a/lib/aprs/packet.ex +++ b/lib/aprs/packet.ex @@ -14,6 +14,11 @@ defmodule Aprs.Packet do field(:path, :string) field(:sender, :string) field(:ssid, :string) + field(:received_at, :utc_datetime_usec) + field(:region, :string) + field(:lat, :float) + field(:lon, :float) + field(:has_position, :boolean, default: false) embeds_one(:data_extended, DataExtended) timestamps() @@ -21,6 +26,9 @@ defmodule Aprs.Packet do @doc false def changeset(packet, attrs) do + # Convert atom data_type to string + attrs = normalize_data_type(attrs) + packet |> cast(attrs, [ :base_callsign, @@ -29,7 +37,12 @@ defmodule Aprs.Packet do :information_field, :path, :sender, - :ssid + :ssid, + :received_at, + :region, + :lat, + :lon, + :has_position ]) |> validate_required([ :base_callsign, @@ -38,7 +51,40 @@ defmodule Aprs.Packet do :information_field, :path, :sender, - :ssid + :ssid, + :received_at ]) + |> maybe_set_has_position() end + + defp maybe_set_has_position(changeset) do + if (get_field(changeset, :lat) && get_field(changeset, :lon)) || + (get_change(changeset, :data_extended) && + get_change(changeset, :data_extended).latitude && + get_change(changeset, :data_extended).longitude) do + put_change(changeset, :has_position, true) + else + changeset + end + end + + # Convert atom data_type to string for storage + defp normalize_data_type(%{data_type: data_type} = attrs) when is_atom(data_type) do + %{attrs | data_type: to_string(data_type)} + end + + defp normalize_data_type(%{"data_type" => data_type} = attrs) when is_atom(data_type) do + %{attrs | "data_type" => to_string(data_type)} + end + + # Handle :data_type key access format + defp normalize_data_type(attrs) when is_map(attrs) do + if Map.has_key?(attrs, :data_type) and is_atom(attrs.data_type) do + %{attrs | data_type: to_string(attrs.data_type)} + else + attrs + end + end + + defp normalize_data_type(attrs), do: attrs end diff --git a/lib/aprs/packet_replay.ex b/lib/aprs/packet_replay.ex new file mode 100644 index 0000000..4e872a3 --- /dev/null +++ b/lib/aprs/packet_replay.ex @@ -0,0 +1,401 @@ +defmodule Aprs.PacketReplay do + @moduledoc """ + Handles replaying of historical APRS packets alongside live packets. + + This module provides functionality to: + 1. Start a replay session for historical packets + 2. Stream historical packets to a client with timing similar to original reception + 3. Merge historical packet streams with live packets + 4. Control replay parameters (speed, filters, etc.) + """ + + use GenServer + + alias Aprs.Packets + alias AprsWeb.Endpoint + + require Logger + + @default_replay_window_minutes 60 + @default_replay_speed 5.0 + + # Client API + + @doc """ + Starts a packet replay session for a specific user/connection. + + ## Options + * `:user_id` - Unique identifier for the user/session (required) + * `:bounds` - Bounding box for visible map area [min_lon, min_lat, max_lon, max_lat] (required) + * `:callsign` - Filter packets by callsign (optional) + * `:start_time` - Start time for replay (default: 60 minutes ago) + * `:end_time` - End time for replay (default: now) + * `:replay_speed` - Speed multiplier for replay (default: 5.0) + * `:limit` - Maximum number of packets to replay (default: 5000) + * `:with_position` - Only include packets with position data (default: true) + """ + def start_replay(opts) do + user_id = Keyword.fetch!(opts, :user_id) + + # Ensure bounds are provided for map area filtering + bounds = Keyword.get(opts, :bounds) + + if !bounds do + raise ArgumentError, "Map bounds are required for packet replay" + end + + # Convert user_id to process name + name = via_tuple(user_id) + + GenServer.start_link(__MODULE__, opts, name: name) + end + + @doc """ + Stops an active replay session for a user. + """ + def stop_replay(user_id) do + name = via_tuple(user_id) + + if GenServer.whereis(name) do + GenServer.stop(name) + else + {:error, :not_found} + end + end + + @doc """ + Pauses an active replay session. + """ + def pause_replay(user_id) do + name = via_tuple(user_id) + GenServer.call(name, :pause) + end + + @doc """ + Resumes a paused replay session. + """ + def resume_replay(user_id) do + name = via_tuple(user_id) + GenServer.call(name, :resume) + end + + @doc """ + Changes the replay speed. + """ + def set_replay_speed(user_id, speed) when is_number(speed) and speed > 0 do + name = via_tuple(user_id) + GenServer.call(name, {:set_speed, speed}) + end + + @doc """ + Updates replay filters (region, bounds, callsign, etc.) + """ + def update_filters(user_id, filters) when is_list(filters) do + name = via_tuple(user_id) + GenServer.call(name, {:update_filters, filters}) + end + + @doc """ + Gets information about the current replay session. + """ + def get_replay_info(user_id) do + name = via_tuple(user_id) + GenServer.call(name, :get_info) + end + + # Server implementation + + @impl true + def init(opts) do + user_id = Keyword.fetch!(opts, :user_id) + + # Default time range - always use last hour at most + now = DateTime.utc_now() + window_minutes = @default_replay_window_minutes + + # Force start time to be at most 1 hour ago + default_start_time = DateTime.add(now, -window_minutes * 60, :second) + user_start_time = Keyword.get(opts, :start_time) + + # If user provided a start time, make sure it's not older than 1 hour + effective_start_time = + if user_start_time && DateTime.before?(user_start_time, default_start_time) do + default_start_time + else + user_start_time || default_start_time + end + + # Set up initial state + state = %{ + user_id: user_id, + replay_topic: "replay:#{user_id}", + replay_speed: Keyword.get(opts, :replay_speed, @default_replay_speed), + start_time: effective_start_time, + end_time: Keyword.get(opts, :end_time, now), + # Not using region - using bounds instead + region: nil, + bounds: Keyword.fetch!(opts, :bounds), + callsign: Keyword.get(opts, :callsign), + with_position: Keyword.get(opts, :with_position, true), + limit: Keyword.get(opts, :limit, 5000), + paused: false, + packets_sent: 0, + replay_started_at: now, + replay_timer: nil, + last_packet_time: nil + } + + # Start the replay immediately + send(self(), :start_replay) + + {:ok, state} + end + + @impl true + def handle_info(:start_replay, state) do + # Fetch historical packets based on filters + # Always filter by bounds (visible map area) and limit to packets with position + replay_opts = [ + start_time: state.start_time, + end_time: state.end_time, + limit: state.limit, + callsign: state.callsign, + with_position: true, + bounds: state.bounds + ] + + # Log the start of replay with map bounds + Logger.info("Starting packet replay for user #{state.user_id} in map area #{inspect(state.bounds)} for the last hour") + + # Send notification to client that replay is starting + Endpoint.broadcast(state.replay_topic, "replay_started", %{ + total_packets: Packets.get_historical_packet_count(replay_opts), + start_time: state.start_time, + end_time: state.end_time, + replay_speed: state.replay_speed, + bounds: state.bounds + }) + + # Get packets and start streaming + stream = Packets.stream_packets_for_replay(Keyword.put(replay_opts, :playback_speed, state.replay_speed)) + + # Schedule the first packet + case stream |> Stream.take(1) |> Enum.to_list() do + [{delay, packet}] -> + # Convert delay to milliseconds + delay_ms = trunc(delay * 1000) + timer = Process.send_after(self(), {:send_packet, packet, stream}, delay_ms) + + {:noreply, %{state | replay_timer: timer, last_packet_time: packet.received_at}} + + [] -> + # No packets found, end replay immediately + Endpoint.broadcast(state.replay_topic, "replay_complete", %{ + packets_sent: 0, + message: "No matching packets found for replay" + }) + + {:stop, :normal, state} + end + end + + @impl true + def handle_info({:send_packet, packet, stream}, %{paused: true} = state) do + # If paused, reschedule the current packet + timer = Process.send_after(self(), {:send_packet, packet, stream}, 1000) + {:noreply, %{state | replay_timer: timer}} + end + + @impl true + def handle_info({:send_packet, packet, stream}, state) do + # Send the packet to the client + Endpoint.broadcast(state.replay_topic, "historical_packet", %{ + packet: sanitize_packet_for_transport(packet), + timestamp: packet.received_at, + is_historical: true + }) + + # Update counter + new_packets_sent = state.packets_sent + 1 + + # Schedule the next packet + case stream |> Stream.take(1) |> Enum.to_list() do + [{delay, next_packet}] -> + # Convert delay to milliseconds + delay_ms = trunc(delay * 1000) + timer = Process.send_after(self(), {:send_packet, next_packet, stream}, delay_ms) + + {:noreply, + %{state | replay_timer: timer, packets_sent: new_packets_sent, last_packet_time: next_packet.received_at}} + + [] -> + # No more packets, end replay + Endpoint.broadcast(state.replay_topic, "replay_complete", %{ + packets_sent: new_packets_sent, + message: "Replay complete" + }) + + {:stop, :normal, state} + end + end + + @impl true + def handle_call(:pause, _from, state) do + if state.replay_timer do + Process.cancel_timer(state.replay_timer) + end + + Endpoint.broadcast(state.replay_topic, "replay_paused", %{ + packets_sent: state.packets_sent, + last_packet_time: state.last_packet_time + }) + + {:reply, :ok, %{state | paused: true, replay_timer: nil}} + end + + @impl true + def handle_call(:resume, _from, %{paused: true} = state) do + # Force the next packet to be sent soon + send(self(), {:continue_replay}) + + Endpoint.broadcast(state.replay_topic, "replay_resumed", %{ + packets_sent: state.packets_sent, + last_packet_time: state.last_packet_time + }) + + {:reply, :ok, %{state | paused: false}} + end + + @impl true + def handle_call(:resume, _from, state) do + # Already running, just acknowledge + {:reply, :ok, state} + end + + @impl true + def handle_call({:set_speed, speed}, _from, state) do + # Update speed and notify client + Endpoint.broadcast(state.replay_topic, "replay_speed_changed", %{ + replay_speed: speed + }) + + {:reply, :ok, %{state | replay_speed: speed}} + end + + @impl true + def handle_call({:update_filters, filters}, _from, state) do + # Update filters - this requires restarting the replay + if state.replay_timer do + Process.cancel_timer(state.replay_timer) + end + + # Update state with new filters + new_state = + Enum.reduce(filters, state, fn {key, value}, acc -> + Map.put(acc, key, value) + end) + + # Always ensure we're filtering by map bounds + new_state = + if not Map.has_key?(new_state, :bounds) or is_nil(new_state.bounds) do + # Keep existing bounds if none provided + new_state + else + # Validate new bounds if provided + case new_state.bounds do + [min_lon, min_lat, max_lon, max_lat] + when is_number(min_lon) and is_number(min_lat) and + is_number(max_lon) and is_number(max_lat) -> + new_state + + _ -> + # Invalid bounds format, keep old bounds + %{new_state | bounds: state.bounds} + end + end + + # Restart replay with new filters + send(self(), :start_replay) + + {:reply, :ok, %{new_state | replay_timer: nil, packets_sent: 0}} + end + + @impl true + def handle_call(:get_info, _from, state) do + info = %{ + user_id: state.user_id, + replay_speed: state.replay_speed, + start_time: state.start_time, + end_time: state.end_time, + bounds: state.bounds, + callsign: state.callsign, + with_position: state.with_position, + packets_sent: state.packets_sent, + paused: state.paused, + replay_started_at: state.replay_started_at, + last_packet_time: state.last_packet_time + } + + {:reply, info, state} + end + + @impl true + def terminate(_reason, state) do + # Clean up any timers + if state.replay_timer do + Process.cancel_timer(state.replay_timer) + end + + # Notify client that replay has ended + Endpoint.broadcast(state.replay_topic, "replay_stopped", %{ + packets_sent: state.packets_sent, + message: "Replay stopped" + }) + + :ok + end + + # Helper functions + + defp via_tuple(user_id) do + {:via, Registry, {Aprs.ReplayRegistry, "replay:#{user_id}"}} + end + + defp sanitize_packet_for_transport(packet) do + # Convert to map and ensure all fields are JSON-safe + packet + |> Map.from_struct() + |> Map.drop([:__meta__, :__struct__]) + |> sanitize_map_values() + end + + defp sanitize_map_values(map) when is_map(map) do + Enum.reduce(map, %{}, fn {k, v}, acc -> + Map.put(acc, k, sanitize_value(v)) + end) + end + + defp sanitize_value(value) when is_map(value) do + sanitize_map_values(value) + end + + defp sanitize_value(value) when is_list(value) do + Enum.map(value, &sanitize_value/1) + end + + defp sanitize_value(%DateTime{} = dt) do + DateTime.to_iso8601(dt) + end + + defp sanitize_value(%NaiveDateTime{} = dt) do + NaiveDateTime.to_iso8601(dt) + end + + defp sanitize_value(%Decimal{} = d) do + Decimal.to_float(d) + end + + defp sanitize_value(value) do + value + end +end diff --git a/lib/aprs/packets.ex b/lib/aprs/packets.ex index cd6aee0..fffc861 100644 --- a/lib/aprs/packets.ex +++ b/lib/aprs/packets.ex @@ -4,4 +4,300 @@ defmodule Aprs.Packets do """ import Ecto.Query, warn: false + + alias Aprs.Packet + alias Aprs.Repo + + @doc """ + Stores a packet in the database. + + ## Parameters + * `packet_data` - Map containing packet data to be stored + """ + def store_packet(packet_data) do + require Logger + + # Convert to map if it's a struct, or use as is if already a map + packet_attrs = + case packet_data do + %Packet{} = packet -> + packet + |> Map.from_struct() + |> Map.delete(:__meta__) + + %{} -> + packet_data + end + + # Convert data_type to string if it's an atom + packet_attrs = + case packet_attrs do + %{data_type: data_type} when is_atom(data_type) -> + Map.put(packet_attrs, :data_type, to_string(data_type)) + + _ -> + packet_attrs + end + + # Make sure received_at is set with explicit UTC DateTime + current_time = DateTime.truncate(DateTime.utc_now(), :microsecond) + packet_attrs = Map.put(packet_attrs, :received_at, current_time) + + # Extract position data + {lat, lon} = extract_position(packet_attrs) + + # Set position fields if found + packet_attrs = + if lat && lon do + packet_attrs + |> Map.put(:lat, lat) + |> Map.put(:lon, lon) + |> Map.put(:has_position, true) + |> Map.put(:region, "#{Float.round(lat, 1)},#{Float.round(lon, 1)}") + else + # Set region based on callsign if no position + sender_region = if packet_attrs[:sender], do: String.slice(packet_attrs.sender || "", 0, 3), else: "unknown" + Map.put(packet_attrs, :region, "call:#{sender_region}") + end + + # Insert the packet + %Packet{} + |> Packet.changeset(packet_attrs) + |> Repo.insert() + end + + # Extracts position data from packet, checking various possible locations + defp extract_position(packet_data) do + cond do + # Check for lat/lon at top level + not is_nil(packet_data[:lat]) and not is_nil(packet_data[:lon]) -> + {to_float(packet_data.lat), to_float(packet_data.lon)} + + # Check data_extended struct or map + not is_nil(packet_data[:data_extended]) -> + data_extended = packet_data.data_extended + + cond do + # Standard position format + is_map(data_extended) and not is_nil(data_extended[:latitude]) and not is_nil(data_extended[:longitude]) -> + {to_float(data_extended.latitude), to_float(data_extended.longitude)} + + # MicE packet format with components + is_map(data_extended) and data_extended.__struct__ == Parser.Types.MicE -> + mic_e = data_extended + + if is_number(mic_e.lat_degrees) and is_number(mic_e.lat_minutes) and + is_number(mic_e.lon_degrees) and is_number(mic_e.lon_minutes) do + # Convert MicE components to decimal degrees + lat = mic_e.lat_degrees + mic_e.lat_minutes / 60.0 + lat = if mic_e.lat_direction == :south, do: -lat, else: lat + + lon = mic_e.lon_degrees + mic_e.lon_minutes / 60.0 + lon = if mic_e.lon_direction == :west, do: -lon, else: lon + + {lat, lon} + else + {nil, nil} + end + + true -> + {nil, nil} + end + + true -> + {nil, nil} + end + end + + @doc """ + Gets packets for replay. + + ## Parameters + * `opts` - Map of options for filtering and pagination: + * `:lat` - Latitude for center point filtering + * `:lon` - Longitude for center point filtering + * `:radius` - Radius in kilometers for filtering + * `:callsign` - Filter by callsign + * `:region` - Filter by region + * `:start_time` - Start time for replay (DateTime) + * `:end_time` - End time for replay (DateTime) + * `:limit` - Maximum number of packets to return + * `:page` - Page number for pagination + """ + def get_packets_for_replay(opts \\ %{}) do + base_query = from(p in Packet, order_by: [asc: p.received_at], where: p.has_position == true) + + query = + base_query + |> filter_by_time(opts) + |> filter_by_region(opts) + |> filter_by_callsign(opts) + |> filter_by_map_bounds(opts) + |> limit_results(opts) + + Repo.all(query) + end + + @doc """ + Gets historical packet count for a map area. + """ + def get_historical_packet_count(opts \\ %{}) do + base_query = from(p in Packet, select: count(p.id), where: p.has_position == true) + + query = + base_query + |> filter_by_time(opts) + |> filter_by_region(opts) + |> filter_by_callsign(opts) + |> filter_by_map_bounds(opts) + + Repo.one(query) + end + + # Query building helpers + # Handle both start_time and end_time + defp filter_by_time(query, %{start_time: start_time, end_time: end_time}) do + # Ensure we prioritize packets from the last hour + one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) + effective_start_time = if DateTime.before?(start_time, one_hour_ago), do: one_hour_ago, else: start_time + + from p in query, + where: p.received_at >= ^effective_start_time and p.received_at <= ^end_time + end + + # Handle only start_time + defp filter_by_time(query, %{start_time: start_time}) do + from p in query, where: p.received_at >= ^start_time + end + + # Handle only end_time + defp filter_by_time(query, %{end_time: end_time}) do + from p in query, where: p.received_at <= ^end_time + end + + # Default case + defp filter_by_time(query, _), do: query + + @doc """ + Get packets only from the last hour for current display. + This is used for initial map loading to show only recent packets. + """ + def get_recent_packets(opts \\ %{}) do + # Always limit to the last hour + one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) + + # Merge the one-hour limit with any other filters + opts_with_time = Map.put(opts, :start_time, one_hour_ago) + + get_packets_for_replay(opts_with_time) + end + + @doc """ + Retrieves a continuous stream of stored packets for replay in chronological order. + + This function returns a Stream that can be used to process packets in chronological + order, preserving the timing between packets. + + ## Parameters + * `opts` - The same options as `get_packets_for_replay/1` + * `:playback_speed` - Speed multiplier (1.0 = real-time, 2.0 = 2x speed, etc.) + + ## Returns + * Stream of packets with timing information + """ + def stream_packets_for_replay(opts \\ %{}) do + packets = get_packets_for_replay(opts) + playback_speed = Map.get(opts, :playback_speed, 1.0) + + # Return a stream that emits packets with their original timing + Stream.unfold({packets, nil}, fn + {[], _} -> + nil + + {[packet | rest], nil} -> + {{0, packet}, {rest, packet}} + + {[next | rest], prev} -> + # Calculate delay between packets in milliseconds, then convert to seconds + delay_ms = DateTime.diff(next.received_at, prev.received_at, :millisecond) + adjusted_delay = delay_ms / (playback_speed * 1000) + {{adjusted_delay, next}, {rest, next}} + end) + end + + defp filter_by_region(query, %{region: region}) do + from p in query, where: p.region == ^region + end + + defp filter_by_region(query, _), do: query + + defp filter_by_callsign(query, %{callsign: callsign}) do + pattern = "%#{callsign}%" + from p in query, where: ilike(p.sender, ^pattern) or ilike(p.base_callsign, ^pattern) + end + + defp filter_by_callsign(query, _), do: query + + defp filter_by_map_bounds(query, %{bounds: [min_lon, min_lat, max_lon, max_lat]}) + when not is_nil(min_lon) and not is_nil(min_lat) and not is_nil(max_lon) and not is_nil(max_lat) do + from p in query, + where: p.has_position == true, + where: p.lat >= ^min_lat and p.lat <= ^max_lat, + where: p.lon >= ^min_lon and p.lon <= ^max_lon + end + + defp filter_by_map_bounds(query, _), do: query + + defp limit_results(query, %{limit: limit, page: page}) when not is_nil(limit) and not is_nil(page) do + offset = (page - 1) * limit + from p in query, limit: ^limit, offset: ^offset + end + + defp limit_results(query, %{limit: limit}) when not is_nil(limit) do + from p in query, limit: ^limit + end + + defp limit_results(query, _), do: query + + @doc """ + Configure packet retention policy. + + Packets are retained based on these rules: + - Default retention is 30 days (configurable via :packet_retention_days) + - Returns the number of packets deleted + """ + def clean_old_packets do + retention_days = Application.get_env(:aprs, :packet_retention_days, 30) + cutoff_time = DateTime.add(DateTime.utc_now(), -retention_days * 86_400, :second) + + {deleted_count, _} = Repo.delete_all(from(p in Packet, where: p.received_at < ^cutoff_time)) + + deleted_count + end + + # Helper to convert various types to float + defp to_float(value) when is_float(value), do: value + defp to_float(value) when is_integer(value), do: value * 1.0 + defp to_float(%Decimal{} = value), do: Decimal.to_float(value) + + defp to_float(value) when is_binary(value) do + case Float.parse(value) do + {float, _} -> float + :error -> nil + end + end + + defp to_float(_), do: nil + + # Get packets from last hour only - used to initialize the map + def get_last_hour_packets do + one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) + + Packet + |> where([p], p.has_position == true) + |> where([p], p.received_at >= ^one_hour_ago) + |> order_by([p], asc: p.received_at) + |> limit(500) + |> Repo.all() + end end diff --git a/lib/aprs/scheduled/packet_cleanup.ex b/lib/aprs/scheduled/packet_cleanup.ex new file mode 100644 index 0000000..e8a02d7 --- /dev/null +++ b/lib/aprs/scheduled/packet_cleanup.ex @@ -0,0 +1,68 @@ +defmodule Aprs.Scheduled.PacketCleanup do + @moduledoc """ + Scheduled task to clean up old APRS packet data. + + This module is responsible for: + 1. Removing packets older than the retention period (30 days by default) + 2. Logging statistics about the cleanup operation + """ + + use GenServer + + alias Aprs.Packets + + require Logger + + # Run cleanup once per day + @cleanup_interval_ms 24 * 60 * 60 * 1000 + + def start_link(_opts) do + GenServer.start_link(__MODULE__, %{}) + end + + @impl true + def init(state) do + # Schedule first cleanup + schedule_cleanup() + {:ok, state} + end + + @impl true + def handle_info(:cleanup, state) do + # Run the cleanup + _cleanup_count = perform_cleanup() + + # Schedule next cleanup + schedule_cleanup() + + {:noreply, state} + end + + defp perform_cleanup do + Logger.info("Starting scheduled APRS packet cleanup") + + # Count packets before cleanup for statistics + _total_before = count_total_packets() + + # Perform the cleanup + deleted_count = Packets.clean_old_packets() + + # Log results + retention_days = Application.get_env(:aprs, :packet_retention_days, 30) + Logger.info("APRS packet cleanup complete: removed #{deleted_count} packets older than #{retention_days} days") + + # Return deleted count + deleted_count + end + + defp count_total_packets do + # Import modules needed for database operations + + Aprs.Repo.aggregate(Aprs.Packet, :count, :id) + end + + defp schedule_cleanup do + # Schedule next cleanup + Process.send_after(self(), :cleanup, @cleanup_interval_ms) + end +end diff --git a/lib/aprs/workers/packet_cleanup_worker.ex b/lib/aprs/workers/packet_cleanup_worker.ex new file mode 100644 index 0000000..e6d93e4 --- /dev/null +++ b/lib/aprs/workers/packet_cleanup_worker.ex @@ -0,0 +1,77 @@ +defmodule Aprs.Workers.PacketCleanupWorker do + @moduledoc """ + Oban worker for cleaning up old APRS packet data. + + This worker is responsible for: + 1. Removing packets older than the retention period (30 days by default) + 2. Removing old packets from the displayed map (older than 1 hour) + 3. Logging statistics about the cleanup operation + """ + + use Oban.Worker, queue: :maintenance, max_attempts: 3 + + # Import modules needed for database operations + + alias Aprs.Packet + alias Aprs.Packets + alias Aprs.Repo + + require Logger + + @impl Oban.Worker + def perform(%Oban.Job{args: _args}) do + Logger.info("Starting scheduled APRS packet cleanup") + + # Count packets before cleanup for statistics + _total_before = count_total_packets() + + # Perform the cleanup of old packets (older than 30 days) + deleted_count = cleanup_old_packets() + + # Log results + retention_days = Application.get_env(:aprs, :packet_retention_days, 30) + Logger.info("APRS packet cleanup complete: removed #{deleted_count} packets older than #{retention_days} days") + + # Return success + :ok + end + + # Schedule this job to run daily + def schedule_cleanup do + %{} + |> Oban.Job.new(worker: __MODULE__, queue: :maintenance) + |> Oban.insert() + end + + # Schedule this job to run daily at midnight UTC + def schedule_daily_cleanup do + # Calculate when the next midnight UTC is + now = DateTime.utc_now() + tomorrow = Date.add(now, 1) + + next_midnight = %{ + now + | year: tomorrow.year, + month: tomorrow.month, + day: tomorrow.day, + hour: 0, + minute: 0, + second: 0, + microsecond: {0, 6} + } + + # Schedule the job + %{} + |> Oban.Job.new(worker: __MODULE__, queue: :maintenance) + |> Oban.insert(scheduled_at: next_midnight) + end + + defp count_total_packets do + Repo.aggregate(Packet, :count, :id) + end + + defp cleanup_old_packets do + # Use the existing function from Packets context + Packets.clean_old_packets() + end +end diff --git a/lib/aprs/workers/scheduler.ex b/lib/aprs/workers/scheduler.ex new file mode 100644 index 0000000..651666e --- /dev/null +++ b/lib/aprs/workers/scheduler.ex @@ -0,0 +1,81 @@ +defmodule Aprs.Workers.Scheduler do + @moduledoc """ + Scheduler for Oban jobs. + + This module is responsible for scheduling recurring jobs in the application. + It should be started as part of the application supervision tree. + """ + + use GenServer + + alias Aprs.Workers.PacketCleanupWorker + + require Logger + + # Start the scheduler on application boot + def start_link(_opts) do + GenServer.start_link(__MODULE__, %{}) + end + + @impl true + def init(state) do + # Schedule initial jobs + schedule_initial_jobs() + + # Return with state + {:ok, state} + end + + @impl true + def handle_info(:schedule_daily_jobs, state) do + # Schedule daily cleanup job + schedule_daily_jobs() + + # Schedule this function to run again tomorrow + schedule_next_day_scheduler() + + {:noreply, state} + end + + # Schedule jobs when the application starts + defp schedule_initial_jobs do + Logger.info("Scheduling initial Oban jobs") + + # Schedule the packet cleanup job + {:ok, _job} = PacketCleanupWorker.schedule_cleanup() + + # Schedule the daily scheduler to run at midnight + schedule_next_day_scheduler() + end + + # Schedule jobs that should run daily + defp schedule_daily_jobs do + Logger.info("Scheduling daily Oban jobs") + + # Schedule the packet cleanup job + {:ok, _job} = PacketCleanupWorker.schedule_cleanup() + end + + # Schedule the next day's job scheduler to run at midnight UTC + defp schedule_next_day_scheduler do + # Calculate when the next midnight UTC is + now = DateTime.utc_now() + tomorrow = Date.add(now, 1) + + next_midnight = %{ + now + | year: tomorrow.year, + month: tomorrow.month, + day: tomorrow.day, + hour: 0, + minute: 0, + second: 0, + microsecond: {0, 6} + } + + seconds_until_midnight = DateTime.diff(next_midnight, now) + + # Schedule this function to run at midnight + Process.send_after(self(), :schedule_daily_jobs, seconds_until_midnight * 1000) + end +end diff --git a/lib/aprs_web/components/layouts/app.html.heex b/lib/aprs_web/components/layouts/app.html.heex index 114cf0d..916071b 100644 --- a/lib/aprs_web/components/layouts/app.html.heex +++ b/lib/aprs_web/components/layouts/app.html.heex @@ -1,40 +1,3 @@ -
-
-
- - - -

- v1.7 -

-
- -
-
<.flash kind={:info} title="Success!" flash={@flash} /> diff --git a/lib/aprs_web/components/layouts/root.html.heex b/lib/aprs_web/components/layouts/root.html.heex index f86de4a..28a36c1 100644 --- a/lib/aprs_web/components/layouts/root.html.heex +++ b/lib/aprs_web/components/layouts/root.html.heex @@ -12,26 +12,6 @@ - {@inner_content} diff --git a/lib/aprs_web/live/map_live/index.ex b/lib/aprs_web/live/map_live/index.ex index 6dc0112..cbeab74 100644 --- a/lib/aprs_web/live/map_live/index.ex +++ b/lib/aprs_web/live/map_live/index.ex @@ -10,11 +10,15 @@ defmodule AprsWeb.MapLive.Index do @default_center %{lat: 39.8283, lng: -98.5795} @default_zoom 5 - @ip_api_url "http://ip-api.com/json/" + @ip_api_url "https://ip-api.com/json/" @finch_name Aprs.Finch + @default_replay_speed 1000 @impl true def mount(_params, _session, socket) do + # Calculate one hour ago for packet age filtering + one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) + socket = assign(socket, packets: [], @@ -30,12 +34,35 @@ defmodule AprsWeb.MapLive.Index do west: -125.0 }, map_center: @default_center, - map_zoom: @default_zoom + map_zoom: @default_zoom, + # Replay controls + replay_active: false, + replay_speed: @default_replay_speed, + replay_paused: false, + replay_packets: [], + replay_index: 0, + replay_timer_ref: nil, + replay_start_time: nil, + replay_end_time: nil, + # Map of packet IDs to packet data for historical packets + historical_packets: %{}, + # Timestamp for filtering out old packets + packet_age_threshold: one_hour_ago, + # Flag to indicate if map is ready for replay + map_ready: false, + # Flag to prevent multiple replay starts + replay_started: false, + # Pending geolocation to zoom to after map is ready + pending_geolocation: nil ) if connected?(socket) do + IO.puts("Socket is connected, attempting to get IP location") Endpoint.subscribe("aprs_messages") # Get IP-based location on initial load + IO.puts("Connect info: #{inspect(socket.private[:connect_info])}") + IO.puts("Peer data: #{inspect(socket.private[:connect_info][:peer_data])}") + ip = case socket.private[:connect_info][:peer_data][:address] do {a, b, c, d} -> "#{a}.#{b}.#{c}.#{d}" @@ -43,7 +70,29 @@ defmodule AprsWeb.MapLive.Index do _ -> nil end - if ip, do: Task.async(fn -> get_ip_location(ip) end) + IO.puts("Extracted IP address: #{inspect(ip)}") + + if ip do + IO.puts("Starting IP geolocation task for IP: #{ip}") + # Start as a separate task and await the result + Task.start(fn -> + try do + get_ip_location(ip) + rescue + error -> + IO.puts("Error in IP geolocation task: #{inspect(error)}") + IO.puts("Stacktrace: #{inspect(__STACKTRACE__)}") + send(self(), {:ip_location, @default_center}) + end + end) + else + IO.puts("No IP address found, skipping geolocation") + end + + # Schedule regular cleanup of old packets from the map + if connected?(socket), do: Process.send_after(self(), :cleanup_old_packets, 60_000) + # Schedule initialization of replay after a short delay + if connected?(socket), do: Process.send_after(self(), :initialize_replay, 2000) end {:ok, socket} @@ -70,6 +119,25 @@ defmodule AprsWeb.MapLive.Index do packet_count = map_size(visible_packets) + # If replay is not active, update the replay packets based on the new bounds + socket = + if socket.assigns.replay_active do + socket + else + # Clear any existing replay data when bounds change + socket = + assign(socket, + replay_packets: [], + replay_index: 0, + historical_packets: %{}, + map_ready: true + ) + + # Don't automatically start replay on every bounds change + # We'll handle this in initialize_replay to avoid too many restarts + socket + end + {:noreply, assign(socket, map_bounds: map_bounds, visible_packets: visible_packets, packet_count: packet_count)} end @@ -82,38 +150,182 @@ defmodule AprsWeb.MapLive.Index do @impl true def handle_event("locate_me", _params, socket) do # Send JavaScript command to request browser geolocation + IO.puts("locate_me event received, requesting geolocation") {:noreply, push_event(socket, "request_geolocation", %{})} end @impl true def handle_event("set_location", %{"lat" => lat, "lng" => lng}, socket) do # Update map center and zoom when location is received - {:noreply, assign(socket, map_center: %{lat: lat, lng: lng}, map_zoom: 12)} + IO.puts("set_location event received with lat=#{lat}, lng=#{lng}") + + # Ensure coordinates are floats + lat_float = + cond do + is_binary(lat) -> String.to_float(lat) + is_integer(lat) -> lat / 1.0 + true -> lat + end + + lng_float = + cond do + is_binary(lng) -> String.to_float(lng) + is_integer(lng) -> lng / 1.0 + true -> lng + end + + IO.puts("Sending zoom_to_location event from set_location with lat=#{lat_float}, lng=#{lng_float}") + + socket = + socket + |> assign(map_center: %{lat: lat_float, lng: lng_float}, map_zoom: 12) + |> push_event("zoom_to_location", %{lat: lat_float, lng: lng_float, zoom: 12}) + + {:noreply, socket} + end + + @impl true + def handle_event("toggle_replay", _params, socket) do + if socket.assigns.replay_active do + # Stop replay + if socket.assigns.replay_timer_ref, do: Process.cancel_timer(socket.assigns.replay_timer_ref) + + # Clear historical packets from the map + socket = + socket + |> push_event("clear_historical_packets", %{}) + |> assign( + replay_active: false, + replay_timer_ref: nil, + replay_paused: false, + replay_packets: [], + replay_index: 0, + historical_packets: %{}, + replay_started: false + ) + + # Restart replay after a short delay + Process.send_after(self(), :initialize_replay, 1000) + + {:noreply, socket} + else + # If not active, the user manually requested a replay restart + {:noreply, start_historical_replay(socket)} + end + end + + @impl true + def handle_event("pause_replay", _params, socket) do + if socket.assigns.replay_active do + if socket.assigns.replay_paused do + # Resume replay + timer_ref = Process.send_after(self(), :replay_next_packet, 1000) + {:noreply, assign(socket, replay_paused: false, replay_timer_ref: timer_ref)} + else + # Pause replay + if socket.assigns.replay_timer_ref, do: Process.cancel_timer(socket.assigns.replay_timer_ref) + {:noreply, assign(socket, replay_paused: true, replay_timer_ref: nil)} + end + else + {:noreply, socket} + end + end + + @impl true + def handle_event("adjust_replay_speed", %{"speed" => speed}, socket) do + speed_float = String.to_float(speed) + {:noreply, assign(socket, replay_speed: speed_float)} + end + + @impl true + def handle_event("map_ready", _params, socket) do + IO.puts("Map ready event received") + socket = assign(socket, map_ready: true) + + # If we have pending geolocation, zoom to it now + socket = + if socket.assigns.pending_geolocation do + %{lat: lat, lng: lng} = socket.assigns.pending_geolocation + IO.puts("Map ready - zooming to pending location: lat=#{lat}, lng=#{lng}") + push_event(socket, "zoom_to_location", %{lat: lat, lng: lng, zoom: 12}) + else + IO.puts("Map ready - no pending geolocation") + socket + end + + {:noreply, socket} end @impl true def handle_info(msg, socket) do case msg do + {:delayed_zoom, %{lat: lat, lng: lng}} -> + IO.puts("Processing delayed zoom to lat=#{lat}, lng=#{lng}") + socket = push_event(socket, "zoom_to_location", %{lat: lat, lng: lng, zoom: 12}) + {:noreply, socket} + {:ip_location, %{lat: lat, lng: lng}} -> - {:noreply, assign(socket, map_center: %{lat: lat, lng: lng}, map_zoom: 12)} + # Log IP location received + IO.puts("IP location received: lat=#{lat}, lng=#{lng}") + + # Ensure we're using numeric values for coordinates + lat_float = + cond do + is_binary(lat) -> String.to_float(lat) + is_integer(lat) -> lat / 1.0 + true -> lat + end + + lng_float = + cond do + is_binary(lng) -> String.to_float(lng) + is_integer(lng) -> lng / 1.0 + true -> lng + end + + # Update map center first + socket = assign(socket, map_center: %{lat: lat_float, lng: lng_float}, map_zoom: 12) + + # If map is ready, zoom to location immediately, otherwise store for later + socket = + if socket.assigns.map_ready do + IO.puts("Map is ready, zooming to location immediately to lat=#{lat_float}, lng=#{lng_float}") + push_event(socket, "zoom_to_location", %{lat: lat_float, lng: lng_float, zoom: 12}) + else + IO.puts("Map not ready yet, storing location lat=#{lat_float}, lng=#{lng_float} for later") + assign(socket, pending_geolocation: %{lat: lat_float, lng: lng_float}) + end + + {:noreply, socket} + + :initialize_replay -> + # Only start replay if it hasn't been started yet + if not socket.assigns.replay_started and socket.assigns.map_ready do + socket = start_historical_replay(socket) + {:noreply, assign(socket, replay_started: true)} + else + {:noreply, socket} + end + + :replay_next_packet -> + handle_replay_next_packet(socket) + + :cleanup_old_packets -> + # Clean up packets older than 1 hour from the map display + handle_cleanup_old_packets(socket) %{event: "packet", payload: payload} -> # Sanitize the packet to prevent encoding errors sanitized_packet = EncodingUtils.sanitize_packet(payload) - # Log packet type for debugging - IO.inspect(sanitized_packet.data_type, label: "Packet type") - - if sanitized_packet.data_extended do - # Check if data_extended is a struct before accessing __struct__ - case sanitized_packet.data_extended do - %{__struct__: module} -> IO.inspect(module, label: "Data extended type") - _ -> IO.inspect("Plain map", label: "Data extended type") - end - end + # Add received timestamp if not present + sanitized_packet = Map.put_new(sanitized_packet, :received_at, DateTime.utc_now()) # Only process packets with position data that are within current map bounds - if has_position_data?(sanitized_packet) && within_bounds?(sanitized_packet, socket.assigns.map_bounds) do + # AND are not older than 1 hour + if has_position_data?(sanitized_packet) && + within_bounds?(sanitized_packet, socket.assigns.map_bounds) && + packet_within_time_threshold?(sanitized_packet, socket.assigns.packet_age_threshold) do # Convert to a simple map structure for JSON encoding packet_data = build_packet_data(sanitized_packet) @@ -139,12 +351,75 @@ defmodule AprsWeb.MapLive.Index do {:noreply, socket} end else - # Ignore packets without position data or outside bounds + # Ignore packets without position data, outside bounds, or too old {:noreply, socket} end end end + # Handle replaying the next historical packet + defp handle_replay_next_packet(socket) do + %{ + replay_packets: packets, + replay_index: index, + replay_speed: speed, + historical_packets: historical_packets + } = socket.assigns + + if index < length(packets) do + # Get the next packet and advance the index + packet = Enum.at(packets, index) + next_index = index + 1 + + # Convert to a simple map structure for JSON encoding + packet_data = build_packet_data(packet) + + # Add is_historical flag and timestamp + packet_data = + Map.merge(packet_data, %{ + "is_historical" => true, + "timestamp" => if(Map.has_key?(packet, :received_at), do: DateTime.to_iso8601(packet.received_at)) + }) + + # Generate a unique key for this packet + packet_id = "hist_#{if Map.has_key?(packet, :id), do: packet.id, else: System.unique_integer([:positive])}" + + # Update historical packets tracking + new_historical_packets = Map.put(historical_packets, packet_id, packet) + + # Calculate delay for next packet (faster based on replay speed) + delay_ms = trunc(1000 / speed) + + # Schedule the next packet + timer_ref = Process.send_after(self(), :replay_next_packet, delay_ms) + + # Push the packet to the client-side JavaScript + socket = + socket + |> push_event("historical_packet", packet_data) + |> assign( + replay_index: next_index, + replay_timer_ref: timer_ref, + historical_packets: new_historical_packets + ) + + {:noreply, socket} + else + # All packets replayed, start over with a short delay + Process.send_after(self(), :initialize_replay, 10_000) + + # Set active to false but don't clear packets - will auto-restart + socket = + assign(socket, + replay_active: false, + replay_timer_ref: nil, + replay_started: false + ) + + {:noreply, socket} + end + end + @impl true def render(assigns) do ~H""" @@ -192,6 +467,7 @@ defmodule AprsWeb.MapLive.Index do border-radius: 5px; box-shadow: 0 2px 5px rgba(0,0,0,0.2); z-index: 1000; + max-width: 300px; } .locate-button { @@ -214,12 +490,116 @@ defmodule AprsWeb.MapLive.Index do font-size: 14px; font-weight: 600; color: #333; + margin-bottom: 10px; + display: flex; + justify-content: space-between; + align-items: center; + } + + .replay-badge { + background-color: #3498db; + color: white; + padding: 2px 6px; + border-radius: 10px; + font-size: 10px; + font-weight: bold; + animation: pulse 2s infinite; + } + + @keyframes pulse { + 0% { opacity: 0.7; } + 50% { opacity: 1; } + 100% { opacity: 0.7; } + } + + .replay-controls { + margin-top: 10px; + display: flex; + flex-direction: column; + gap: 8px; + } + + .control-buttons { + display: flex; + gap: 8px; + } + + .replay-controls button { + padding: 5px 10px; + border-radius: 4px; + font-size: 12px; + cursor: pointer; + font-weight: 600; + border: none; + flex: 1; + display: flex; + align-items: center; + justify-content: center; + gap: 4px; + } + + .btn-action { + background-color: #3498db; + color: white; + } + + .btn-danger { + background-color: #e74c3c; + color: white; + } + + .btn-secondary { + background-color: #7f8c8d; + color: white; + } + + .icon { + font-weight: bold; + } + + .speed-control { + font-size: 12px; + margin-top: 5px; + } + + .speed-control input { + width: 100%; + margin-top: 5px; + } + + .replay-progress { + font-size: 12px; + margin-top: 5px; + text-align: center; + } + + .progress-bar { + height: 6px; + background-color: #eee; + border-radius: 3px; + overflow: hidden; + margin-bottom: 5px; + } + + .progress-fill { + height: 100%; + background-color: #3498db; + transition: width 0.5s ease-in-out; + } + + .progress-text { + font-size: 10px; + color: #666; } .aprs-marker { background: transparent !important; border: none !important; } + + .historical-marker { + opacity: 0.7; + }
- {@packet_count} packets in view + {@packet_count} +  packets in view + <%= if @replay_active do %> +
+ Replaying historical packets +
+ <% end %> +
+
+ <%= if @replay_active do %> +
+ + +
+
+ + +
+
+
+
0, do: @replay_index * 100 / length(@replay_packets), else: 0}%"} + > +
+
+
+ {@replay_index} of {length(@replay_packets)} packets +
+
+ <% end %>
@@ -259,8 +682,120 @@ defmodule AprsWeb.MapLive.Index do """ end + # Handle cleanup of old packets + defp handle_cleanup_old_packets(socket) do + # Update the packet age threshold to current time minus one hour + one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) + + # Filter out packets older than one hour + visible_packets = + socket.assigns.visible_packets + |> Enum.filter(fn {_key, packet} -> + packet_within_time_threshold?(packet, one_hour_ago) + end) + |> Map.new() + + # Get updated packet count + packet_count = map_size(visible_packets) + + # Push event to remove old markers from the map + socket = + socket + |> push_event("refresh_markers", %{}) + |> assign( + visible_packets: visible_packets, + packet_count: packet_count, + packet_age_threshold: one_hour_ago + ) + + # Schedule the next cleanup in 1 minute + if connected?(socket), do: Process.send_after(self(), :cleanup_old_packets, 60_000) + + {:noreply, socket} + end + + # Check if a packet is within the time threshold (not too old) + defp packet_within_time_threshold?(packet, threshold) do + case packet do + %{received_at: received_at} when not is_nil(received_at) -> + DateTime.compare(received_at, threshold) in [:gt, :eq] + + _ -> + # If no timestamp, treat as current + true + end + end + # Helper functions + # Fetch historical packets from the database + # Helper function to start historical replay + defp start_historical_replay(socket) do + # Get time range for historical data + now = DateTime.utc_now() + one_hour_ago = DateTime.add(now, -60 * 60, :second) + + # Convert map bounds to the format expected by the database query + bounds = [ + socket.assigns.map_bounds.west, + socket.assigns.map_bounds.south, + socket.assigns.map_bounds.east, + socket.assigns.map_bounds.north + ] + + # Fetch historical packets with position data within the current map bounds + historical_packets = fetch_historical_packets(bounds, one_hour_ago, now) + + if Enum.empty?(historical_packets) do + # No historical packets found - silently continue without replay + socket + else + # Clear any previous historical packets from the map + socket = push_event(socket, "clear_historical_packets", %{}) + + # Start replay + timer_ref = Process.send_after(self(), :replay_next_packet, 1000) + + assign(socket, + replay_active: true, + replay_packets: historical_packets, + replay_index: 0, + replay_timer_ref: timer_ref, + replay_start_time: one_hour_ago, + replay_end_time: now, + historical_packets: %{}, + replay_started: true + ) + end + end + + # Fetch historical packets from the database + defp fetch_historical_packets(bounds, start_time, end_time) do + # Force start_time to be at most 1 hour ago + one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) + + effective_start_time = + if DateTime.before?(start_time, one_hour_ago), + do: one_hour_ago, + else: start_time + + # Use the Packets context to retrieve historical packets + packets_params = %{ + bounds: bounds, + start_time: effective_start_time, + end_time: end_time, + with_position: true, + # Reasonable limit to prevent overwhelming the client + limit: 1000 + } + + # Call the database through the Packets context + packets = Aprs.Packets.get_packets_for_replay(packets_params) + + # Sort packets by received_at timestamp to ensure chronological replay + Enum.sort_by(packets, fn packet -> packet.received_at end) + end + defp has_position_data?(packet) do case packet.data_extended do %MicE{} = mic_e -> @@ -314,15 +849,14 @@ defmodule AprsWeb.MapLive.Index do defp build_packet_data(packet) do data_extended = build_data_extended(packet.data_extended) - if data_extended do - %{ - "base_callsign" => packet.base_callsign || "", - "ssid" => packet.ssid || "", - "data_type" => to_string(packet.data_type || "unknown"), - "path" => packet.path || "", - "data_extended" => data_extended - } - end + # Always return a map, even when data_extended is nil + %{ + "base_callsign" => packet.base_callsign || "", + "ssid" => packet.ssid || "", + "data_type" => to_string(packet.data_type || "unknown"), + "path" => packet.path || "", + "data_extended" => data_extended || %{} + } end # Get IP location from external service @@ -332,26 +866,66 @@ defmodule AprsWeb.MapLive.Index do defp get_ip_location(ip) do url = "#{@ip_api_url}#{ip}" - request = Finch.build(:get, url) + IO.puts("Fetching location for IP: #{ip} from URL: #{url}") - case Finch.request(request, @finch_name) do + # Add headers to make the request more likely to succeed + request = + Finch.build(:get, url, [ + {"User-Agent", "APRS.me/1.0"}, + {"Accept", "application/json"} + ]) + + # Add a small delay to ensure the client is connected + Process.sleep(2000) + + IO.puts("Making HTTP request to IP API...") + + case Finch.request(request, @finch_name, receive_timeout: 10_000) do {:ok, %{status: 200, body: body}} -> + IO.puts("IP API response received successfully") + IO.puts("Response body: #{String.slice(body, 0, 200)}...") + case Jason.decode(body) do - {:ok, %{"lat" => lat, "lon" => lng}} when is_number(lat) and is_number(lng) -> + {:ok, %{"status" => "success", "lat" => lat, "lon" => lng}} when is_number(lat) and is_number(lng) -> + IO.puts("Valid coordinates found: lat=#{lat}, lng=#{lng}") + if lat >= -90 and lat <= 90 and lng >= -180 and lng <= 180 do - send(self(), {:ip_location, %{lat: lat, lng: lng}}) + IO.puts("Sending IP location message with coordinates") + # Force numeric values + lat_float = lat / 1.0 + lng_float = lng / 1.0 + send(self(), {:ip_location, %{lat: lat_float, lng: lng_float}}) else + IO.puts("Coordinates out of range: lat=#{lat}, lng=#{lng}, using default center") send(self(), {:ip_location, @default_center}) end - _ -> + {:ok, %{"status" => status}} -> + IO.puts("IP API returned non-success status: #{status}") + send(self(), {:ip_location, @default_center}) + + {:ok, data} -> + IO.puts("IP API returned unexpected format: #{inspect(data)}") + send(self(), {:ip_location, @default_center}) + + {:error, decode_error} -> + IO.puts("Failed to decode JSON response: #{inspect(decode_error)}") + IO.puts("Raw response: #{body}") send(self(), {:ip_location, @default_center}) end - {:ok, _} -> + {:ok, response} -> + IO.puts("IP API request failed with status: #{response.status}") + IO.puts("Response headers: #{inspect(response.headers)}") + IO.puts("Response body: #{String.slice(response.body || "", 0, 200)}...") send(self(), {:ip_location, @default_center}) - {:error, _} -> + {:error, %{reason: :timeout}} -> + IO.puts("IP API request timed out after 10 seconds") + send(self(), {:ip_location, @default_center}) + + {:error, error} -> + IO.puts("IP API request error: #{inspect(error)}") send(self(), {:ip_location, @default_center}) end end diff --git a/lib/parser.ex b/lib/parser.ex index 214bc70..2476689 100644 --- a/lib/parser.ex +++ b/lib/parser.ex @@ -29,8 +29,11 @@ defmodule Parser do information_field: data, data_type: data_type, base_callsign: base_callsign, - ssid: ssid, - data_extended: data_extended + # Ensure ssid is never nil + ssid: ssid || "0", + data_extended: data_extended, + # Set received_at when creating packet + received_at: DateTime.truncate(DateTime.utc_now(), :microsecond) }} else true -> @@ -49,7 +52,8 @@ defmodule Parser do if String.contains?(callsign, "-") do String.split(callsign, "-") else - [callsign, nil] + # Default SSID to "0" instead of nil + [callsign, "0"] end end diff --git a/lib/types/mic_e.ex b/lib/types/mic_e.ex index 5b1946a..64a3cf8 100644 --- a/lib/types/mic_e.ex +++ b/lib/types/mic_e.ex @@ -2,6 +2,8 @@ defmodule Parser.Types.MicE do @moduledoc """ Type struct for MicE """ + @behaviour Access + defstruct lat_degrees: 0, lat_minutes: 0, lat_fractional: 0, @@ -18,4 +20,62 @@ defmodule Parser.Types.MicE do speed: 0, manufacturer: :unknown, message: "" + + @doc """ + Implements the Access behaviour for MicE struct. + This allows us to access fields using the access syntax: struct[:field] + Additionally, it provides support for dynamic access via get_in/update_in/etc. + + Fetch a key from the MicE struct. + Special handling for :latitude and :longitude which are calculated from components. + """ + def fetch(mic_e, :latitude) do + # Calculate decimal latitude from components + lat = mic_e.lat_degrees + mic_e.lat_minutes / 60.0 + lat = if mic_e.lat_direction == :south, do: -lat, else: lat + {:ok, lat} + end + + def fetch(mic_e, :longitude) do + # Calculate decimal longitude from components + lon = mic_e.lon_degrees + mic_e.lon_minutes / 60.0 + lon = if mic_e.lon_direction == :west, do: -lon, else: lon + {:ok, lon} + end + + def fetch(mic_e, key) when is_atom(key) do + # Fall back to struct field access for other keys + Map.fetch(mic_e, key) + end + + @doc """ + Gets a value and updates it with the given function. + """ + def get_and_update(mic_e, key, fun) do + value = + case key do + :latitude -> + lat = mic_e.lat_degrees + mic_e.lat_minutes / 60.0 + if mic_e.lat_direction == :south, do: -lat, else: lat + + :longitude -> + lon = mic_e.lon_degrees + mic_e.lon_minutes / 60.0 + if mic_e.lon_direction == :west, do: -lon, else: lon + + _ -> + Map.get(mic_e, key) + end + + case fun.(value) do + {get, update} -> {get, Map.put(mic_e, key, update)} + :pop -> {value, Map.put(mic_e, key, nil)} + end + end + + @doc """ + Removes the given key from the struct with the default implementation. + """ + def pop(mic_e, key) do + {Map.get(mic_e, key), Map.put(mic_e, key, nil)} + end end diff --git a/priv/repo/migrations/20230600000000_add_oban_jobs_table.exs b/priv/repo/migrations/20230600000000_add_oban_jobs_table.exs new file mode 100644 index 0000000..08a7766 --- /dev/null +++ b/priv/repo/migrations/20230600000000_add_oban_jobs_table.exs @@ -0,0 +1,11 @@ +defmodule Aprs.Repo.Migrations.AddObanJobsTable do + use Ecto.Migration + + def up do + Oban.Migrations.up(version: 11) + end + + def down do + Oban.Migrations.down(version: 11) + end +end diff --git a/priv/repo/migrations/20230606000001_add_packet_replay_fields.exs b/priv/repo/migrations/20230606000001_add_packet_replay_fields.exs new file mode 100644 index 0000000..ef06e65 --- /dev/null +++ b/priv/repo/migrations/20230606000001_add_packet_replay_fields.exs @@ -0,0 +1,36 @@ +defmodule Aprs.Repo.Migrations.AddPacketReplayFields do + use Ecto.Migration + + def change do + alter table(:packets) do + # Timestamp when the packet was received (can be different from inserted_at) + add :received_at, :utc_datetime_usec + + # Geographic region information to help with filtering/querying + add :region, :string + + # Store lat/lon as separate fields for simpler queries + add :lat, :float + add :lon, :float + + # Add indices for efficient querying + add :has_position, :boolean, default: false + end + + # Index for time-based queries (useful for replay) + create index(:packets, [:received_at]) + + # Index for region-based filtering + create index(:packets, [:region]) + + # Index for filtering packets with position data + create index(:packets, [:has_position]) + + # Indices for geographic filtering + create index(:packets, [:lat]) + create index(:packets, [:lon]) + + # Compound index for efficient replay queries (region + time) + create index(:packets, [:region, :received_at]) + end +end diff --git a/priv/repo/migrations/20250616014248_updateoban.exs b/priv/repo/migrations/20250616014248_updateoban.exs new file mode 100644 index 0000000..96a473a --- /dev/null +++ b/priv/repo/migrations/20250616014248_updateoban.exs @@ -0,0 +1,7 @@ +defmodule Aprs.Repo.Migrations.Updateoban do + use Ecto.Migration + + def up, do: Oban.Migrations.up() + + def down, do: Oban.Migrations.down() +end