aprs.me/lib/aprsme/packet_consumer.ex

defmodule Aprsme.PacketConsumer do
  @moduledoc """
  GenStage consumer that batches APRS packets and inserts them into the database
  efficiently to reduce database load.
  """
  use GenStage

  alias Aprs.Types.ParseError
  alias Aprsme.Cluster.PacketDistributor
  alias Aprsme.LogSanitizer
  alias Aprsme.Repo

  require Logger

  @type state :: %{
          batch: list(map()),
          batch_size: integer(),
          batch_timeout: integer(),
          max_batch_size: integer(),
          timer: reference() | nil
        }

  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts \\ []) do
    # Allow unnamed consumers for pool usage
    name = opts[:name]

    if name do
      GenStage.start_link(__MODULE__, opts, name: name)
    else
      GenStage.start_link(__MODULE__, opts)
    end
  end

  @impl true
  def init(opts) do
    batch_size = opts[:batch_size] || 100
    batch_timeout = opts[:batch_timeout] || 1000
    # Maximum batch size to prevent unbounded memory growth
    max_batch_size = opts[:max_batch_size] || 1000

    # Start a timer for batch processing
    timer = Process.send_after(self(), :process_batch, batch_timeout)

    # Extract subscription options if provided
    subscribe_to = opts[:subscribe_to] || [{Aprsme.PacketProducer, max_demand: opts[:max_demand] || 250}]

    {:consumer,
     %{
       batch: [],
       batch_size: batch_size,
       batch_timeout: batch_timeout,
       max_batch_size: max_batch_size,
       timer: timer
     }, subscribe_to: subscribe_to}
  end

  @impl true
  def handle_events(events, _from, state) do
    handle_batch_update(events, state)
  end

  # Pattern matching for batch handling with optimized list operations
  defp handle_batch_update(events, %{batch: batch} = state) do
    # Optimize: Keep batch in reverse order for O(1) prepending
    # Only reverse when processing
    new_batch = Enum.reverse(events, batch)
    new_batch_length = batch_length(batch) + length(events)

    handle_batch_by_size(new_batch, new_batch_length, state)
  end

  # Pattern matching for different batch size scenarios
  defp handle_batch_by_size(batch, length, %{max_batch_size: max} = state) when length >= max do
    handle_oversized_batch(batch, max, state)
  end

  defp handle_batch_by_size(batch, length, %{batch_size: size} = state) when length >= size do
    handle_full_batch(batch, state)
  end

  defp handle_batch_by_size(batch, _length, state) do
    handle_partial_batch(batch, state)
  end

  # Handle oversized batch with pattern matching
  defp handle_oversized_batch(batch, max_size, state) do
    # Reverse once for processing
    reversed_batch = Enum.reverse(batch)
    {process_batch, drop_batch} = Enum.split(reversed_batch, max_size)

    process_batch(process_batch)
    log_dropped_packets(drop_batch, process_batch)

    new_state = reset_batch_timer(state)
    {:noreply, [], new_state}
  end

  # Handle full batch
  defp handle_full_batch(batch, state) do
    # Reverse once for processing
    process_batch(Enum.reverse(batch))

    new_state = reset_batch_timer(state)
    {:noreply, [], new_state}
  end

  # Handle partial batch
  defp handle_partial_batch(batch, state) do
    # Keep batch in reverse order
    {:noreply, [], %{state | batch: batch}}
  end

  # Helper functions with pattern matching
  @spec reset_batch_timer(state()) :: state()
  defp reset_batch_timer(%{timer: nil} = state) do
    new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
    %{state | batch: [], timer: new_timer}
  end

  defp reset_batch_timer(%{timer: timer} = state) do
    Process.cancel_timer(timer)
    new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
    %{state | batch: [], timer: new_timer}
  end

  # Efficient batch length calculation (cached if possible)
  @spec batch_length(list()) :: non_neg_integer()
  defp batch_length([]), do: 0
  defp batch_length(batch), do: length(batch)

  # Pattern matching for logging
  @spec log_dropped_packets(list(), list()) :: :ok
  defp log_dropped_packets([], _), do: :ok

  defp log_dropped_packets(dropped, processed) do
    Logger.warning("Dropped #{length(dropped)} packets due to batch size limit",
      batch_info:
        LogSanitizer.log_data(
          dropped_count: length(dropped),
          processed_count: length(processed),
          reason: "batch_size_limit_exceeded"
        )
    )
  end

  @impl true
  # Pattern matching for empty batch
  def handle_info(:process_batch, %{batch: []} = state) do
    # Just restart the timer for empty batch
    new_state = start_batch_timer(state)
    {:noreply, [], new_state}
  end

  # Pattern matching for non-empty batch
  def handle_info(:process_batch, %{batch: batch} = state) when is_list(batch) do
    check_batch_utilization(batch, state.max_batch_size)
    # Remember to reverse the batch since we keep it in reverse order
    process_batch(Enum.reverse(batch))

    new_state = start_batch_timer(%{state | batch: []})
    {:noreply, [], new_state}
  end

  # Helper to start batch timer
  @spec start_batch_timer(state()) :: state()
  defp start_batch_timer(%{batch_timeout: timeout} = state) do
    timer = Process.send_after(self(), :process_batch, timeout)
    %{state | timer: timer}
  end

  # Pattern matching for batch utilization check
  @spec check_batch_utilization(list(), integer()) :: :ok
  defp check_batch_utilization(batch, max_size) when length(batch) > max_size * 0.8 do
    Logger.warning("Batch size approaching limit",
      batch_status:
        LogSanitizer.log_data(
          current_size: length(batch),
          max_size: max_size,
          utilization_percent: trunc(length(batch) / max_size * 100)
        )
    )
  end

  defp check_batch_utilization(_, _), do: :ok

  @spec process_batch(list(map())) :: :ok
  defp process_batch(packets) do
    require Logger

    # Monitor memory usage before processing (only this process)
    {:memory, memory_before} = Process.info(self(), :memory)
    start_time = System.monotonic_time(:millisecond)

    # Chunk size optimized for PostgreSQL work_mem=16MB
    # Using smaller batches to reduce memory pressure
    chunk_size = Application.get_env(:aprsme, :packet_pipeline)[:batch_size] || 100

    # Use Stream for memory-efficient processing
    # Process and reduce in one pass to avoid materializing the entire list
    {success_count, error_count} =
      packets
      |> Stream.chunk_every(chunk_size)
      |> Stream.map(&process_chunk/1)
      |> Enum.reduce({0, 0}, fn {success, error}, {total_success, total_error} ->
        {total_success + success, total_error + error}
      end)

    end_time = System.monotonic_time(:millisecond)
    duration = end_time - start_time

    # Monitor memory usage after processing (only this process)
    {:memory, memory_after} = Process.info(self(), :memory)
    memory_diff = memory_after - memory_before

    # Get current process memory info
    process_info = Process.info(self(), [:memory, :heap_size, :total_heap_size])

    # Force garbage collection if memory usage is high
    # 50MB threshold for memory diff (per process)
    # 100MB threshold for process memory
    if memory_diff > 52_428_800 or process_info[:memory] > 104_857_600 do
      :erlang.garbage_collect()

      Logger.warning("High memory usage detected, forced garbage collection",
        memory_info:
          LogSanitizer.log_data(
            memory_diff_bytes: memory_diff,
            process_memory: process_info[:memory],
            heap_size: process_info[:heap_size],
            total_heap_size: process_info[:total_heap_size],
            batch_size: length(packets),
            gc_forced: true
          )
      )
    end

    # Only do minor GC after very large batches to prevent memory accumulation
    # This should rarely trigger with batch_size of 100
    if length(packets) > 1000 do
      :erlang.garbage_collect(self(), type: :minor)
    end

    :telemetry.execute(
      [
        :aprsme,
        :packet_pipeline,
        :batch
      ],
      %{
        count: length(packets),
        success: success_count,
        error: error_count,
        duration_ms: duration,
        memory_diff: memory_diff
      },
      %{}
    )

    # Logger.info("Batch processing completed",
    #   batch_result:
    #     LogSanitizer.log_data(
    #       packet_count: length(packets),
    #       duration_ms: duration,
    #       success_count: success_count,
    #       error_count: error_count,
    #       memory_diff_bytes: memory_diff
    #     )
    # )
  end

  @spec process_chunk(list(map())) :: {non_neg_integer(), non_neg_integer()}
  defp process_chunk(packets) do
    # Use Stream for memory-efficient packet preparation
    packet_stream =
      packets
      |> Stream.map(&prepare_packet_for_insert/1)
      |> Stream.map(&truncate_datetimes_to_second/1)
      |> Stream.reject(&is_nil/1)

    # Separate valid and invalid packets using Stream
    {valid_packets, invalid_count} =
      Enum.reduce(packet_stream, {[], 0}, fn packet, {valid_acc, invalid_acc} ->
        if valid_packet?(packet) do
          {[packet | valid_acc], invalid_acc}
        else
          {valid_acc, invalid_acc + 1}
        end
      end)

    # Reverse to maintain order
    valid_packets = Enum.reverse(valid_packets)

    # Insert valid packets in batch
    # Optimized for PostgreSQL with synchronous_commit=off
    insert_opts = [
      # Don't return IDs for better performance
      returning: false,
      # Skip conflicts to avoid blocking
      on_conflict: :nothing,
      # Increased timeout for large batches
      timeout: 60_000,
      # Use placeholders for better performance with large batches
      placeholders: length(valid_packets) > 100
    ]

    case Repo.insert_all(Aprsme.Packet, valid_packets, insert_opts) do
      {:error, error} ->
        Logger.error("Batch insert failed: #{inspect(error)}")

        # Log sample packet to help debug field issues
        if valid_packets != [] do
          sample_packet = List.first(valid_packets)
          Logger.error("Sample packet fields: #{inspect(Map.keys(sample_packet))}")
          Logger.error("Sample packet data: #{inspect(sample_packet)}")
        end

        {0, length(packets)}

      {inserted_count, _} ->
        # Broadcast successfully inserted packets to StreamingPacketsPubSub
        broadcast_packets_async(valid_packets)

        {inserted_count, invalid_count}
    end
  end

  # Broadcast packets asynchronously to avoid blocking
  defp broadcast_packets_async(packets) do
    Task.start(fn ->
      try do
        cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)

        Enum.each(packets, fn packet_attrs ->
          # Convert back to a format suitable for broadcasting
          packet = %{
            sender: packet_attrs[:sender],
            latitude: packet_attrs[:lat],
            longitude: packet_attrs[:lon],
            received_at: packet_attrs[:received_at],
            data_type: packet_attrs[:data_type],
            altitude: packet_attrs[:altitude],
            speed: packet_attrs[:speed],
            course: packet_attrs[:course],
            comment: packet_attrs[:comment]
          }

          if cluster_enabled do
            # Use cluster distributor to broadcast to all nodes
            PacketDistributor.distribute_packet(packet)
          else
            # Normal single-node broadcasting
            Aprsme.StreamingPacketsPubSub.broadcast_packet(packet)
            # Also broadcast to SpatialPubSub for viewport-based filtering
            Aprsme.SpatialPubSub.broadcast_packet(packet)
          end
        end)
      rescue
        error ->
          Logger.error("Failed to broadcast packets asynchronously: #{inspect(error)}")
      end
    end)
  end

  defp prepare_packet_for_insert(packet_data) do
    # Always set received_at timestamp to ensure consistency
    current_time = DateTime.truncate(DateTime.utc_now(), :microsecond)
    packet_data = Map.put(packet_data, :received_at, current_time)

    # Convert to map before storing to avoid struct conversion issues
    attrs = struct_to_map(packet_data)

    # Extract additional data from the parsed packet including raw packet
    attrs = Aprsme.Packet.extract_additional_data(attrs, attrs[:raw_packet] || "")

    # 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)

    # Normalize data_type to string if it's an atom
    attrs = normalize_data_type(attrs)

    # Apply the same processing as the original store_packet function
    attrs
    |> 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)
      set_lat_lon(attrs, lat, lon)
    end)
    |> normalize_ssid()
    |> then(fn attrs ->
      device_identifier = Aprsme.DeviceParser.extract_device_identifier(packet_data)
      Map.put(attrs, :device_identifier, device_identifier)
    end)
    |> sanitize_packet_strings()
    |> Map.put(:inserted_at, current_time)
    |> Map.put(:updated_at, current_time)
    |> Map.delete(:id)
    |> Map.delete("id")
    # Remove embedded field for batch insert
    |> Map.delete(:data_extended)
    |> normalize_numeric_types()
    |> truncate_datetimes_to_second()
    # Create PostGIS geometry for location field BEFORE filtering
    |> create_location_geometry()
    # Remove all non-schema fields - do this LAST to ensure all processing is done
    |> remove_non_schema_fields()
  rescue
    error ->
      Logger.error("Failed to prepare packet for batch insert: #{inspect(error)}")
      nil
  end

  # Create PostGIS geometry from lat/lon coordinates
  defp create_location_geometry(attrs) do
    lat = attrs[:lat]
    lon = attrs[:lon]

    if valid_coordinates?(lat, lon) do
      location = create_point(lat, lon)

      if location do
        Map.put(attrs, :location, location)
      else
        attrs
      end
    else
      attrs
    end
  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
    Aprsme.PacketFieldWhitelist.filter_fields(attrs)
  end

  # Helper functions for coordinate validation and point creation
  defp valid_coordinates?(lat, lon) do
    lat = normalize_coordinate(lat)
    lon = normalize_coordinate(lon)

    is_number(lat) && is_number(lon) &&
      lat >= -90 && lat <= 90 &&
      lon >= -180 && lon <= 180
  end

  defp normalize_coordinate(%Decimal{} = decimal), do: Decimal.to_float(decimal)
  defp normalize_coordinate(coord), do: coord

  defp create_point(lat, lon)
       when (is_number(lat) or is_struct(lat, Decimal)) and (is_number(lon) or is_struct(lon, Decimal)) do
    lat = normalize_coordinate(lat)
    lon = normalize_coordinate(lon)

    if valid_coordinates?(lat, lon) do
      %Geo.Point{coordinates: {lon, lat}, srid: 4326}
    end
  end

  defp create_point(_, _), do: nil

  defp valid_packet?(nil), do: false
  defp valid_packet?(%{sender: sender}) when is_binary(sender) and byte_size(sender) > 0, do: true
  defp valid_packet?(_), do: false

  # Detect if packet is an item or object and set appropriate fields
  defp detect_item_or_object(attrs) do
    cond do
      # Check for itemname field from parser
      Map.has_key?(attrs, :itemname) or Map.has_key?(attrs, "itemname") ->
        item_name = Map.get(attrs, :itemname) || Map.get(attrs, "itemname")

        attrs
        |> Map.put(:item_name, item_name)
        |> Map.put(:is_item, true)
        |> Map.delete(:itemname)
        |> Map.delete("itemname")

      # Check for object data type and extract object name from data_extended
      attrs[:data_type] == "object" or attrs["data_type"] == "object" ->
        object_name = extract_object_name(attrs)

        attrs
        |> Map.put(:object_name, object_name)
        |> Map.put(:is_object, true)

      # Check information field for object format (starts with ;)
      is_binary(attrs[:information_field]) and String.starts_with?(attrs[:information_field], ";") ->
        # Extract object name from information field
        object_name = extract_object_name_from_info_field(attrs[:information_field])

        attrs
        |> Map.put(:object_name, object_name)
        |> Map.put(:is_object, true)

      true ->
        attrs
    end
  end

  defp extract_object_name(attrs) do
    case attrs[:data_extended] do
      %{name: name} when is_binary(name) -> name
      %{"name" => name} when is_binary(name) -> name
      _ -> nil
    end
  end

  defp extract_object_name_from_info_field(info_field) do
    # Object format: ;OBJECTNAM*DDHHMMz...
    case Regex.run(~r/^;([A-Z0-9 ]{9})\*/, info_field) do
      [_, name] -> String.trim(name)
      _ -> nil
    end
  end

  # Convert field names that don't match our schema
  defp convert_field_names(attrs) do
    attrs
    |> then(fn a ->
      # Convert aprs_messaging? to aprs_messaging
      case Map.get(a, :aprs_messaging?) || Map.get(a, "aprs_messaging?") do
        nil ->
          a

        value ->
          a
          |> Map.put(:aprs_messaging, value)
          |> Map.delete(:aprs_messaging?)
          |> Map.delete("aprs_messaging?")
      end
    end)
    |> then(fn a ->
      # Convert timestamp from integer to string if needed
      case Map.get(a, :timestamp) do
        timestamp when is_integer(timestamp) ->
          Map.put(a, :timestamp, Integer.to_string(timestamp))

        _ ->
          a
      end
    end)
  end

  # Convert latitude/longitude to lat/lon if present at top level
  defp convert_coordinate_field_names(attrs) do
    attrs
    |> then(fn a ->
      case Map.get(a, :latitude) do
        nil -> a
        lat -> a |> Map.put(:lat, lat) |> Map.delete(:latitude)
      end
    end)
    |> then(fn a ->
      case Map.get(a, :longitude) do
        nil -> a
        lon -> a |> Map.put(:lon, lon) |> Map.delete(:longitude)
      end
    end)
  end

  # Helper functions copied from Packets module for consistency
  defp normalize_packet_attrs(attrs) do
    attrs
    |> Map.put_new(:base_callsign, attrs[:sender])
    |> Map.put_new(:data_type, "unknown")
    |> Map.put_new(:destination, "")
    |> Map.put_new(:information_field, "")
    |> Map.put_new(:path, "")
    |> Map.put_new(:ssid, "")
    |> 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) ->
        attrs
        |> Map.put(:lat, lat)
        |> Map.put(:lon, lon)
        |> Map.put(:has_position, true)

      _ ->
        attrs
    end
  end

  defp extract_position(packet_data) do
    if not is_nil(packet_data[:lat]) and not is_nil(packet_data[:lon]) do
      {to_float(packet_data.lat), to_float(packet_data.lon)}
    else
      extract_position_from_data_extended(packet_data[:data_extended])
    end
  end

  defp extract_position_from_data_extended(nil), do: {nil, nil}

  defp extract_position_from_data_extended(data_extended) when is_map(data_extended) do
    if has_standard_position?(data_extended) do
      extract_standard_position(data_extended)
    else
      extract_position_from_data_extended_case(data_extended)
    end
  end

  defp extract_position_from_data_extended(_), do: {nil, nil}

  defp has_standard_position?(data_extended) when is_map(data_extended) and not is_struct(data_extended) do
    not is_nil(data_extended[:latitude]) and not is_nil(data_extended[:longitude])
  end

  defp has_standard_position?(_), do: false

  defp extract_standard_position(data_extended) when is_map(data_extended) and not is_struct(data_extended) do
    {to_float(data_extended[:latitude]), to_float(data_extended[:longitude])}
  end

  defp extract_standard_position(_), do: {nil, nil}

  defp extract_position_from_data_extended_case(data_extended) do
    lat = extract_lat_from_ext_map(data_extended)
    lon = extract_lon_from_ext_map(data_extended)
    {to_float(lat), to_float(lon)}
  end

  defp extract_lat_from_ext_map(ext_map) do
    ext_map[:latitude] || ext_map["latitude"] ||
      (Map.has_key?(ext_map, :position) &&
         (ext_map[:position][:latitude] || ext_map[:position]["latitude"])) ||
      (Map.has_key?(ext_map, "position") &&
         (ext_map["position"][:latitude] || ext_map["position"]["latitude"]))
  end

  defp extract_lon_from_ext_map(ext_map) do
    ext_map[:longitude] || ext_map["longitude"] ||
      (Map.has_key?(ext_map, :position) &&
         (ext_map[:position][:longitude] || ext_map[:position]["longitude"])) ||
      (Map.has_key?(ext_map, "position") &&
         (ext_map["position"][:longitude] || ext_map["position"]["longitude"]))
  end

  defp set_lat_lon(attrs, lat, lon) do
    round6 = fn
      nil ->
        nil

      n when is_float(n) ->
        Float.round(n, 6)
    end

    attrs
    |> Map.put(:lat, round6.(lat))
    |> Map.put(:lon, round6.(lon))
    |> Map.put(:has_position, not is_nil(lat) and not is_nil(lon))
  end

  defp normalize_ssid(attrs) do
    case Map.get(attrs, :ssid) do
      nil -> attrs
      ssid -> Map.put(attrs, :ssid, to_string(ssid))
    end
  end

  defp sanitize_packet_strings(value), do: Aprsme.EncodingUtils.sanitize_packet_strings(value)

  defp to_float(value), do: Aprsme.EncodingUtils.to_float(value)

  defp normalize_data_type(attrs), do: Aprsme.EncodingUtils.normalize_data_type(attrs)

  defp struct_to_map(%{__struct__: ParseError} = error) do
    # Handle ParseError specially to avoid Access behavior issues
    %{
      error_code: error.error_code,
      message: error.message,
      __original_struct__: ParseError
    }
  end

  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)

    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 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 = [
      :temperature,
      :humidity,
      :wind_speed,
      :wind_gust,
      :pressure,
      :rain_1h,
      :rain_24h,
      :rain_since_midnight,
      :snow,
      :speed,
      :altitude,
      :posresolution,
      :rain_midnight
    ]

    Enum.reduce(float_fields, attrs, fn field, acc ->
      case Map.get(acc, field) do
        value when is_integer(value) -> Map.put(acc, field, value * 1.0)
        _ -> acc
      end
    end)
  end
end