diff --git a/lib/aprsme/is/is.ex b/lib/aprsme/is/is.ex index ee72db4..ce55505 100644 --- a/lib/aprsme/is/is.ex +++ b/lib/aprsme/is/is.ex @@ -602,13 +602,12 @@ defmodule Aprsme.Is do try do # Set received_at and convert struct to map — PacketConsumer.prepare_packet_for_insert # handles extract_additional_data and normalize_data_type, so don't duplicate here. - current_time = DateTime.truncate(DateTime.utc_now(), :microsecond) + current_time = DateTime.truncate(DateTime.utc_now(), :second) attrs = parsed_message |> Map.put(:received_at, current_time) - |> Map.put(:raw, message) - |> struct_to_map() + |> Map.put(:raw_packet, message) Aprsme.PacketProducer.submit_packet(attrs) rescue @@ -668,25 +667,6 @@ defmodule Aprsme.Is do %PacketStats{last_second_timestamp: System.system_time(:second)} end - # Helper function to recursively convert structs to maps - # This handles nested structs that Map.from_struct/1 cannot handle - @spec struct_to_map(any()) :: any() - defp struct_to_map(%{__struct__: struct_type} = struct) do - converted_map = - struct - |> Map.from_struct() - |> Map.new(fn {k, v} -> {k, struct_to_map(v)} end) - - # Add type information to help with later processing - Map.put(converted_map, :__original_struct__, struct_type) - end - - defp struct_to_map(value) when is_list(value) do - Enum.map(value, &struct_to_map/1) - end - - defp struct_to_map(value), do: value - defp cancel_safety_valve(%{safety_valve_timer: nil}), do: :ok defp cancel_safety_valve(%{safety_valve_timer: timer}) do diff --git a/lib/aprsme/packet.ex b/lib/aprsme/packet.ex index 37565a0..85d0737 100644 --- a/lib/aprsme/packet.ex +++ b/lib/aprsme/packet.ex @@ -349,9 +349,6 @@ defmodule Aprsme.Packet do %MicE{} = mic_e -> extract_from_mic_e(mic_e) - %{__original_struct__: MicE} = mic_e_map -> - extract_from_mic_e_map(mic_e_map) - # Handle ParseError structs gracefully %{__struct__: Aprs.Types.ParseError} -> %{} @@ -627,20 +624,6 @@ defmodule Aprsme.Packet do |> maybe_put(:symbol_table_id, mic_e[:symbol_table_id]) end - # Extract data from converted MicE map (from struct_to_map conversion) - defp extract_from_mic_e_map(mic_e_map) do - %{} - |> maybe_put(:lat, mic_e_map[:latitude]) - |> maybe_put(:lon, mic_e_map[:longitude]) - |> maybe_put(:comment, mic_e_map[:message]) - |> maybe_put(:manufacturer, mic_e_map[:manufacturer]) - |> maybe_put(:course, mic_e_map[:heading]) - |> maybe_put(:speed, mic_e_map[:speed]) - # Use symbol data from MicE if available, otherwise use default car symbol - |> maybe_put(:symbol_code, mic_e_map[:symbol_code] || ">") - |> maybe_put(:symbol_table_id, mic_e_map[:symbol_table_id] || "/") - end - # Extract weather data from various formats, including new wx field defp extract_weather_data(attrs, data_extended) do weather_data = find_weather_data(data_extended) @@ -660,8 +643,13 @@ defmodule Aprsme.Packet do defp process_weather_data(attrs, weather_data) do case weather_data do - weather when is_map(weather) -> process_map_weather_data(attrs, weather) - _ -> attrs + weather when is_map(weather) and map_size(weather) > 0 -> + attrs + |> process_map_weather_data(weather) + |> Map.put(:has_weather, true) + + _ -> + attrs end end diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 1f0391d..c7f35a5 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -259,27 +259,25 @@ defmodule Aprsme.PacketConsumer do @spec process_chunk(list(map())) :: {non_neg_integer(), non_neg_integer()} defp process_chunk(packets) do - # Use Stream for memory-efficient packet preparation - # Note: truncate_datetimes_to_second is already called inside prepare_packet_for_insert + # Use Stream for memory-efficient packet preparation. # prepare_packet_for_insert returns {insert_attrs, broadcast_info} or nil packet_stream = packets |> Stream.map(&prepare_packet_for_insert/1) |> Stream.reject(&is_nil/1) - # Separate valid and invalid packets using Stream - {valid_pairs, invalid_count} = - Enum.reduce(packet_stream, {[], 0}, fn {insert_attrs, _bcast} = pair, {valid_acc, invalid_acc} -> + # Single pass: classify valid/invalid and split insert/bcasts + {valid_inserts_rev, valid_bcasts_rev, invalid_count} = + Enum.reduce(packet_stream, {[], [], 0}, fn {insert_attrs, bcast}, {inserts, bcasts, invalid} -> if valid_packet?(insert_attrs) do - {[pair | valid_acc], invalid_acc} + {[insert_attrs | inserts], [bcast | bcasts], invalid} else - {valid_acc, invalid_acc + 1} + {inserts, bcasts, invalid + 1} end end) - # Reverse to maintain order and split into parallel lists - valid_pairs = Enum.reverse(valid_pairs) - valid_inserts = Enum.map(valid_pairs, fn {insert, _bcast} -> insert end) + valid_inserts = Enum.reverse(valid_inserts_rev) + valid_bcasts = Enum.reverse(valid_bcasts_rev) # Insert valid packets in batch # Optimized for PostgreSQL with synchronous_commit=off @@ -296,7 +294,7 @@ defmodule Aprsme.PacketConsumer do try do {inserted_count, _} = Repo.insert_all(Aprsme.Packet, valid_inserts, insert_opts) - _ = broadcast_packets_async(Enum.map(valid_pairs, fn {_insert, bcast} -> bcast end)) + _ = broadcast_packets_async(valid_bcasts) {inserted_count, invalid_count} rescue @@ -304,6 +302,7 @@ defmodule Aprsme.PacketConsumer do Logger.error("Batch insert failed: #{inspect(error)}, falling back to individual inserts") # Fall back to individual inserts so partial success is possible + valid_pairs = Enum.zip(valid_inserts, valid_bcasts) {fallback_inserted, fallback_bcasts} = insert_individually(valid_pairs) # Broadcast whatever was successfully inserted @@ -312,7 +311,7 @@ defmodule Aprsme.PacketConsumer do broadcast_packets_async(fallback_bcasts) end - {fallback_inserted, invalid_count + length(valid_pairs) - fallback_inserted} + {fallback_inserted, invalid_count + length(valid_inserts) - fallback_inserted} end end @@ -406,24 +405,18 @@ defmodule Aprsme.PacketConsumer do end defp prepare_packet_for_insert(packet_data) do - # Reuse the received_at stamped in Is.dispatch; fall back if missing. - # packet_data is already a plain map — Is.dispatch ran struct_to_map before submitting. - current_time = - case Map.get(packet_data, :received_at) do - %DateTime{} = dt -> DateTime.truncate(dt, :microsecond) - _ -> DateTime.truncate(DateTime.utc_now(), :microsecond) - end - - attrs = Map.put(packet_data, :received_at, current_time) + # received_at is stamped at second precision in Is.dispatch. + current_time = Map.get(packet_data, :received_at) || DateTime.truncate(DateTime.utc_now(), :second) # Extract additional data from the parsed packet including raw packet - attrs = Aprsme.Packet.extract_additional_data(attrs, attrs[:raw_packet] || "") + attrs = Aprsme.Packet.extract_additional_data(packet_data, packet_data[:raw_packet] || "") + attrs = Map.put_new(attrs, :received_at, current_time) # Detect and set item/object fields attrs = detect_item_or_object(attrs) - # Sanitize packet data to prevent database field overflow - attrs = Aprsme.PacketSanitizer.sanitize_packet(attrs) + # Sanitize packet data (truncate + encoding) in one pass + attrs = Aprsme.PacketSanitizer.sanitize_packet_with_encoding(attrs) # Normalize data_type to string if it's an atom attrs = normalize_data_type(attrs) @@ -437,7 +430,6 @@ defmodule Aprsme.PacketConsumer do |> convert_coordinate_field_names() |> convert_field_names() |> normalize_packet_attrs() - |> set_received_at() |> patch_lat_lon_from_data_extended() |> then(fn attrs -> {lat, lon} = extract_position(attrs) @@ -447,16 +439,13 @@ defmodule Aprsme.PacketConsumer do |> then(fn attrs -> Map.put(attrs, :device_identifier, Aprsme.DeviceParser.extract_device_identifier(attrs)) end) - |> sanitize_packet_strings() + |> Map.delete(:data_extended) |> create_location_geometry() |> Map.put(:inserted_at, current_time) |> Map.put(:updated_at, current_time) |> Map.put(:id, Ecto.UUID.generate()) |> Map.delete("id") - |> Map.delete(:data_extended) |> normalize_numeric_types() - |> truncate_datetimes_to_second() - |> set_has_weather() insert_attrs = remove_non_schema_fields(enriched) bcast_info = build_broadcast_info(enriched) @@ -496,16 +485,6 @@ defmodule Aprsme.PacketConsumer do } end - # Compute has_weather from any populated weather field (replaces DB trigger) - defp set_has_weather(attrs) do - has_weather? = - Enum.any?(Aprsme.EncodingUtils.weather_fields(), fn field -> - not is_nil(Map.get(attrs, field)) - end) - - Map.put(attrs, :has_weather, has_weather?) - end - # Helper function to remove fields that exist in parser output but not in database schema defp remove_non_schema_fields(attrs) do # Use whitelist approach - only keep fields that are in our schema @@ -635,11 +614,6 @@ defmodule Aprsme.PacketConsumer do |> Map.put_new(:data_extended, %{}) end - defp set_received_at(attrs) do - received_at = attrs[:received_at] || DateTime.utc_now() - Map.put(attrs, :received_at, received_at) - end - defp patch_lat_lon_from_data_extended(attrs) do case attrs[:data_extended] do %{latitude: lat, longitude: lon} when not is_nil(lat) and not is_nil(lon) -> @@ -714,7 +688,7 @@ defmodule Aprsme.PacketConsumer do defp round_coord(n) when is_float(n), do: Float.round(n, 6) defp round_coord(_), do: nil - # Must run AFTER sanitize_packet_strings so the struct survives intact. + # Must run AFTER encoding sanitization so the struct survives intact. defp create_location_geometry(%{lat: lat, lon: lon} = attrs) when is_float(lat) and is_float(lon) and lat >= -90.0 and lat <= 90.0 and lon >= -180.0 and lon <= 180.0 do Map.put(attrs, :location, %Geo.Point{coordinates: {lon, lat}, srid: 4326}) @@ -729,21 +703,8 @@ defmodule Aprsme.PacketConsumer do end end - defp sanitize_packet_strings(value), do: Aprsme.EncodingUtils.sanitize_packet_strings(value) - defp normalize_data_type(attrs), do: Aprsme.EncodingUtils.normalize_data_type(attrs) - defp truncate_datetimes_to_second(%DateTime{} = dt), do: DateTime.truncate(dt, :second) - defp truncate_datetimes_to_second({:ok, %DateTime{} = dt}), do: DateTime.truncate(dt, :second) - defp truncate_datetimes_to_second({:error, _reason}), do: nil - - defp truncate_datetimes_to_second(term) when is_map(term) and not is_struct(term) do - Map.new(term, fn {k, v} -> {k, truncate_datetimes_to_second(v)} end) - end - - defp truncate_datetimes_to_second(list) when is_list(list), do: Enum.map(list, &truncate_datetimes_to_second/1) - defp truncate_datetimes_to_second(other), do: other - defp normalize_numeric_types(attrs) do # Convert integer values to floats for float fields float_fields = [ diff --git a/lib/aprsme/packet_field_whitelist.ex b/lib/aprsme/packet_field_whitelist.ex index 8b9210e..1e0cb75 100644 --- a/lib/aprsme/packet_field_whitelist.ex +++ b/lib/aprsme/packet_field_whitelist.ex @@ -61,17 +61,14 @@ defmodule Aprsme.PacketFieldWhitelist do """ def allowed_fields, do: @allowed_fields + @allowed_set MapSet.new(@allowed_fields) + @doc """ Filters a map to only include allowed fields. Handles both atom and string keys. """ def filter_fields(attrs) when is_map(attrs) do - attrs - |> Enum.filter(fn {key, _value} -> - key_str = to_string(key) - key_str in @allowed_fields - end) - |> Map.new() + Map.new(Enum.filter(attrs, fn {key, _value} -> MapSet.member?(@allowed_set, to_string(key)) end)) end @doc """ diff --git a/lib/aprsme/packet_sanitizer.ex b/lib/aprsme/packet_sanitizer.ex index b494c68..c50915a 100644 --- a/lib/aprsme/packet_sanitizer.ex +++ b/lib/aprsme/packet_sanitizer.ex @@ -50,6 +50,38 @@ defmodule Aprsme.PacketSanitizer do Map.new(packet, fn {key, value} -> {key, sanitize_field(key, value)} end) end + @doc """ + Sanitizes packet data by truncating strings AND encoding-sanitizing them + in a single pass. Intended for the hot path after `data_extended` is + deleted (no recursive map walking needed). + """ + @spec sanitize_packet_with_encoding(map()) :: map() + def sanitize_packet_with_encoding(packet) when is_map(packet) do + Map.new(packet, fn {key, value} -> {key, sanitize_field_with_encoding(key, value)} end) + end + + defp sanitize_field_with_encoding(:data, value) when is_map(value) do + sanitize_data_map_with_encoding(value) + end + + defp sanitize_field_with_encoding(key, value) when is_binary(value) do + truncated = + case Map.get(@max_lengths, key) do + nil -> value + max_length -> truncate_string(value, max_length) + end + + Aprsme.EncodingUtils.sanitize_string(truncated) || "" + end + + defp sanitize_field_with_encoding(_key, value), do: value + + defp sanitize_data_map_with_encoding(data) when is_map(data) do + Map.new(data, fn {key, value} -> + {key, sanitize_data_value(key, value)} + end) + end + defp sanitize_field(:data, value) when is_map(value) do sanitize_data_map(value) end diff --git a/test/aprsme/packet_extras_test.exs b/test/aprsme/packet_extras_test.exs index 625ce0f..495e662 100644 --- a/test/aprsme/packet_extras_test.exs +++ b/test/aprsme/packet_extras_test.exs @@ -29,24 +29,7 @@ defmodule Aprsme.PacketExtrasTest do end end - describe "extract_additional_data/2 with mic_e_map symbol defaults" do - test "uses default '>' symbol_code and '/' symbol_table_id when MicE map omits them" do - attrs = %{ - sender: "MICE-1", - data_type: "mic_e", - data_extended: %{ - __original_struct__: Aprs.Types.MicE, - latitude: 33.0, - longitude: -96.0, - message: "MicE comment" - } - } - - result = Packet.extract_additional_data(attrs, "MICE-1>APRS:`abc") - assert result[:symbol_code] == ">" - assert result[:symbol_table_id] == "/" - end - + describe "extract_from_map struct handling" do test "extract_from_map struct branch handles a non-MicE struct via Map.from_struct" do # URI is a stdlib struct that survives Map.from_struct without raising. ext = %URI{scheme: "https", host: "example.test"} diff --git a/test/aprsme/packets_test.exs b/test/aprsme/packets_test.exs index 1c122a3..483169a 100644 --- a/test/aprsme/packets_test.exs +++ b/test/aprsme/packets_test.exs @@ -522,9 +522,10 @@ defmodule Aprsme.PacketsTest do "lon" => -96.0 } - # Ecto changesets reject mixed-key maps — we only care that the - # get_raw_packet string-key clause was exercised without crashing. - assert {:error, :storage_exception} = Packets.store_packet(packet_data) + # normalize_packet_keys converts all string keys to atoms, so the + # changeset sees a clean map. Verify raw_packet was stored correctly. + assert {:ok, packet} = Packets.store_packet(packet_data) + assert packet.raw_packet == "STRRAW1>APRS:test packet" end end