140 lines
5 KiB
Elixir
140 lines
5 KiB
Elixir
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."
|
||
@spec ingest(GenServer.server(), binary()) :: :ok
|
||
def ingest(server \\ __MODULE__, payload) when is_binary(payload) do
|
||
GenServer.cast(server, {:ingest, payload})
|
||
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
|
||
def handle_cast({:ingest, payload}, state) do
|
||
case Pskr.parse_spot(payload) 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} ->
|
||
{: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
|