prop/lib/microwaveprop/pskr/aggregator.ex

defmodule Microwaveprop.Pskr.Aggregator do
  @moduledoc """
  In-memory accumulator for incoming PSK Reporter spots.

  The MQTT firehose can push 500+ messages/second when subscribed
  broadly. Writing each spot as a row would (a) saturate the
  Postgres write path and (b) bloat the dataset 50× because every
  QSO is heard by many reporters. This GenServer keeps a small
  state map keyed by `Pskr.path_key/1`, folds each spot into the
  matching row with `Pskr.merge_spot/2`, and periodically batches
  the whole map into a single `Repo.insert_all/3` upsert.

  Conflict handling is additive: when a row with the same
  `(hour_utc, band, sender_grid, receiver_grid)` already exists in
  the DB, the upsert merges counts, expands the SNR envelope, and
  unions the mode set rather than overwriting. That makes flushes
  idempotent in the face of overlap (e.g. a flush firing a few
  seconds before the hour boundary).

  Memory profile: each pending row is small (~200 bytes), and the
  unique key space is bounded by `bands × paths_per_hour`. With our
  CONUS-anchored topic filter that's typically a few thousand keys
  per hour — flushed every minute, the working set never exceeds a
  handful of MiB.
  """
  use GenServer

  import Ecto.Query

  alias Microwaveprop.Pskr
  alias Microwaveprop.Pskr.SpotHourly
  alias Microwaveprop.Repo

  require Logger

  @default_flush_ms 60_000

  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts \\ []) do
    {name, opts} = Keyword.pop(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, opts, name: name)
  end

  @doc """
  Push one raw MQTT payload (binary JSON) into the accumulator.

  `opts` is forwarded to `Pskr.parse_spot/2` for topic-derived
  fallback values (e.g. callsigns when the JSON payload omits them).
  """
  @spec ingest(GenServer.server(), binary(), keyword()) :: :ok
  def ingest(server \\ __MODULE__, payload, opts \\ []) when is_binary(payload) do
    GenServer.cast(server, {:ingest, payload, opts})
  end

  @doc "Force a synchronous flush. Returns the number of rows written."
  @spec flush(GenServer.server()) :: non_neg_integer()
  def flush(server \\ __MODULE__) do
    GenServer.call(server, :flush, 30_000)
  end

  @doc "Inspect the current pending row count without flushing."
  @spec pending_count(GenServer.server()) :: non_neg_integer()
  def pending_count(server \\ __MODULE__), do: GenServer.call(server, :pending_count)

  @impl true
  def init(opts) do
    flush_ms = Keyword.get(opts, :flush_ms, @default_flush_ms)
    _ = if flush_ms > 0, do: schedule_flush(flush_ms)
    {:ok, %{rows: %{}, parsed: 0, dropped: 0, flush_ms: flush_ms}}
  end

  @impl true
  # Accept both old (2-tuple) and new (3-tuple with opts) ingest messages.
  def handle_cast({:ingest, payload}, state) do
    handle_cast({:ingest, payload, []}, state)
  end

  def handle_cast({:ingest, payload, opts}, state) do
    case Pskr.parse_spot(payload, opts) do
      {:ok, spot} ->
        key = Pskr.path_key(spot)
        rows = Map.update(state.rows, key, Pskr.merge_spot(nil, spot), &Pskr.merge_spot(&1, spot))
        {:noreply, %{state | rows: rows, parsed: state.parsed + 1}}

      {:error, reason} ->
        Logger.debug("Pskr.Aggregator dropped spot: #{inspect(reason)}")
        {:noreply, %{state | dropped: state.dropped + 1}}
    end
  end

  @impl true
  def handle_call(:flush, _from, state) do
    written = do_flush(state.rows)
    {:reply, written, %{state | rows: %{}}}
  end

  def handle_call(:pending_count, _from, state) do
    {:reply, map_size(state.rows), state}
  end

  @impl true
  def handle_info(:flush_tick, state) do
    written = do_flush(state.rows)

    if written > 0 or state.dropped > 0 do
      Logger.info("Pskr.Aggregator flushed #{written} rows (parsed #{state.parsed}, dropped #{state.dropped})")
    end

    _ = schedule_flush(state.flush_ms)
    {:noreply, %{state | rows: %{}, parsed: 0, dropped: 0}}
  end

  defp schedule_flush(ms) when ms > 0, do: Process.send_after(self(), :flush_tick, ms)
  defp schedule_flush(_), do: :ok

  defp do_flush(rows) when map_size(rows) == 0, do: 0

  defp do_flush(rows) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    entries =
      rows
      |> Map.values()
      |> Enum.map(&Map.merge(&1, %{inserted_at: now, updated_at: now}))

    {count, _} =
      Repo.insert_all(SpotHourly, entries,
        on_conflict:
          from(s in SpotHourly,
            update: [
              set: [
                spot_count: fragment("? + EXCLUDED.spot_count", s.spot_count),
                max_snr_db: fragment("GREATEST(?, EXCLUDED.max_snr_db)", s.max_snr_db),
                min_snr_db: fragment("LEAST(?, EXCLUDED.min_snr_db)", s.min_snr_db),
                modes: fragment("ARRAY(SELECT DISTINCT unnest(? || EXCLUDED.modes))", s.modes),
                sender_callsigns:
                  fragment("ARRAY(SELECT DISTINCT unnest(? || EXCLUDED.sender_callsigns))", s.sender_callsigns),
                receiver_callsigns:
                  fragment("ARRAY(SELECT DISTINCT unnest(? || EXCLUDED.receiver_callsigns))", s.receiver_callsigns),
                first_spot_at: fragment("LEAST(?, EXCLUDED.first_spot_at)", s.first_spot_at),
                last_spot_at: fragment("GREATEST(?, EXCLUDED.last_spot_at)", s.last_spot_at),
                updated_at: fragment("EXCLUDED.updated_at")
              ]
            ]
          ),
        conflict_target: [:hour_utc, :band, :sender_grid, :receiver_grid]
      )

    count
  end
end