defmodule Microwaveprop.Propagation.ScoreCache do @moduledoc """ Node-local ETS cache of propagation scores keyed by `{band_mhz, valid_time}`. Populated eagerly by `Microwaveprop.Workers.PropagationGridWorker` after each hourly compute (via `broadcast_put/3`) and lazily by `Microwaveprop.Propagation.scores_at/3` on a cache miss. `broadcast_put/3` fans the payload out across the cluster on the `"propagation:cache"` PubSub topic so every BEAM node stays in sync with a single compute. """ use GenServer alias Phoenix.PubSub @table :propagation_score_cache @topic "propagation:cache" @pubsub Microwaveprop.PubSub @type score :: %{lat: float(), lon: float(), score: non_neg_integer()} @type bounds :: %{optional(String.t()) => float()} # ---------- Public API ---------- @spec start_link(keyword()) :: GenServer.on_start() def start_link(opts) do GenServer.start_link(__MODULE__, opts, name: __MODULE__) end @spec fetch(non_neg_integer(), DateTime.t()) :: {:ok, [score()]} | :miss def fetch(band_mhz, valid_time) do case :ets.lookup(@table, {band_mhz, valid_time}) do [{_, scores}] -> {:ok, scores} [] -> :miss end end @spec fetch_bounds(non_neg_integer(), DateTime.t(), bounds() | nil) :: {:ok, [score()]} | :miss def fetch_bounds(band_mhz, valid_time, bounds) do case fetch(band_mhz, valid_time) do {:ok, scores} -> {:ok, filter_bounds(scores, bounds)} :miss -> :miss end end @spec put(non_neg_integer(), DateTime.t(), [score()]) :: :ok def put(band_mhz, valid_time, scores) do :ets.insert(@table, {{band_mhz, valid_time}, scores}) :ok end @doc """ Insert locally AND broadcast to peer nodes via PubSub. Call this from the single pod that computed the scores; every cluster member (including the caller) will receive the PubSub message and populate its local ETS. Use `put/3` directly if you already have a cluster-wide fan-out. """ @spec broadcast_put(non_neg_integer(), DateTime.t(), [score()]) :: :ok def broadcast_put(band_mhz, valid_time, scores) do PubSub.broadcast(@pubsub, @topic, {:cache_refresh, band_mhz, valid_time, scores}) :ok end @spec prune_older_than(DateTime.t()) :: non_neg_integer() def prune_older_than(cutoff) do match_spec = [ {{{:_, :"$1"}, :_}, [{:<, :"$1", {:const, cutoff}}], [true]} ] :ets.select_delete(@table, match_spec) end @spec clear() :: :ok def clear do :ets.delete_all_objects(@table) :ok end @doc "Flush the GenServer mailbox so any in-flight cache_refresh messages are applied." @spec sync() :: :ok def sync do GenServer.call(__MODULE__, :sync) end # ---------- GenServer callbacks ---------- @impl true def init(_opts) do :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) PubSub.subscribe(@pubsub, @topic) {:ok, %{}} end @impl true def handle_call(:sync, _from, state), do: {:reply, :ok, state} @impl true def handle_info({:cache_refresh, band_mhz, valid_time, scores}, state) do put(band_mhz, valid_time, scores) {:noreply, state} end def handle_info(_msg, state), do: {:noreply, state} # ---------- Internal ---------- defp filter_bounds(scores, nil), do: scores defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do Enum.filter(scores, fn %{lat: lat, lon: lon} -> lat >= s and lat <= n and lon >= w and lon <= e end) end end