prop/lib/microwaveprop/propagation/score_cache.ex

defmodule Microwaveprop.Propagation.ScoreCache do
  @moduledoc """
  Node-local ETS cache of propagation scores keyed by `{band_mhz, valid_time}`.

  Populated lazily by `Microwaveprop.Propagation.scores_at/3` on the first
  request for a given `{band_mhz, valid_time}`; subsequent requests hit ETS
  until the entry is evicted. The cache is bounded to `max_entries/0` total
  entries via LRU-by-`valid_time` eviction; on overflow the oldest
  `valid_time` is dropped first so the freshly-arrived hour stays warm.

  `broadcast_put/3` and the matching `"propagation:cache"` PubSub topic are
  retained for the rare cluster-wide pre-warm path; the hot pipeline (Rust
  Phase 2 cutover) does not call them — `NotifyListener` only emits the
  lighter-weight `"propagation:updated"` PubSub so LiveView clients refresh
  while the actual scores load on demand from `/data/scores/.../*.prop`.

  ## Storage layout

  Each cache entry is bucketed into 5°×5° spatial chunks matching the layout
  used by `Microwaveprop.Weather.GridCache`. The ETS value for
  `{band_mhz, valid_time}` is `%{{lat_band, lon_band} => %{{lat, lon} => score}}`.
  That way `fetch_bounds/3` only walks the chunks that intersect the
  requested viewport instead of the full 92k-cell CONUS map; `fetch_point/4`
  reads exactly one chunk.
  """
  use GenServer

  alias Phoenix.PubSub

  @table :propagation_score_cache
  @topic "propagation:cache"
  @pubsub Microwaveprop.PubSub
  @chunk_step 5

  # Hard cap on cached `{band_mhz, valid_time}` entries. Each value is a
  # 92k-cell map (now bucketed into 5°×5° chunks) on the order of ~2 MiB;
  # without a cap the eager warm path materialises 23 bands × 19 forecast
  # hours = 437 entries per pod (~850 MiB) and OOMs the BEAM. The cap
  # keeps the LiveView hot path warm (one band × the hour or two the
  # user is actively scrubbing) and lets the rest fall back to direct
  # `.prop` reads via `read_from_disk_and_cache/3`, which is sub-100 ms
  # per file.
  @max_entries 32

  @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
      [{_, chunked}] -> {:ok, chunks_to_list(chunked)}
      [] -> :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 :ets.lookup(@table, {band_mhz, valid_time}) do
      [{_, chunked}] -> {:ok, chunks_filtered_to_list(chunked, bounds)}
      [] -> :miss
    end
  end

  @doc """
  Look up the score for a single `{lat, lon}` point at `{band_mhz, valid_time}`.
  Returns `{:ok, score}` on hit or `:miss` on cache miss. The coordinates are
  used as-is (no snap-to-grid) — callers that need the nearest grid cell should
  snap first.
  """
  @spec fetch_point(non_neg_integer(), DateTime.t(), float(), float()) ::
          {:ok, non_neg_integer()} | :miss
  def fetch_point(band_mhz, valid_time, lat, lon) do
    case :ets.lookup(@table, {band_mhz, valid_time}) do
      [{_, chunked}] ->
        chunk_key = chunk_key_for(lat, lon)

        case chunked |> Map.get(chunk_key, %{}) |> Map.get({lat, lon}) do
          nil -> :miss
          score -> {:ok, score}
        end

      [] ->
        :miss
    end
  end

  @doc "Maximum number of `{band_mhz, valid_time}` entries kept in ETS."
  @spec max_entries() :: pos_integer()
  def max_entries, do: @max_entries

  @spec put(non_neg_integer(), DateTime.t(), [score()]) :: :ok
  def put(band_mhz, valid_time, scores) do
    chunked = list_to_chunks(scores)
    :ets.insert(@table, {{band_mhz, valid_time}, chunked})
    enforce_capacity()
    :ok
  end

  # When the table is over the cap, evict whichever entry has the oldest
  # `valid_time`. Picking the oldest matches the LiveView access pattern:
  # the timeline scrubs forward, the hourly chain replaces the trailing
  # edge, and the freshest hours are by far the most-requested. Loops
  # until the table is back at or below the cap so that any historical
  # over-shoot self-heals after this commit deploys.
  defp enforce_capacity do
    if :ets.info(@table, :size) > @max_entries do
      case oldest_key() do
        nil -> :ok
        key -> :ets.delete(@table, key)
      end

      enforce_capacity()
    end
  end

  defp oldest_key do
    fun = fn
      {{band, vt}, _}, nil ->
        {band, vt}

      {{band, vt}, _}, {_, oldest} = acc ->
        if DateTime.before?(vt, oldest), do: {band, vt}, else: acc
    end

    :ets.foldl(fun, nil, @table)
  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

  @doc "Returns the sorted list of valid_times cached for `band_mhz`."
  @spec valid_times(non_neg_integer()) :: [DateTime.t()]
  def valid_times(band_mhz) do
    fn
      {{^band_mhz, vt}, _}, acc -> [vt | acc]
      _, acc -> acc
    end
    |> :ets.foldl(
      [],
      @table
    )
    |> Enum.sort({:asc, DateTime})
  end

  @spec prune_older_than(DateTime.t()) :: non_neg_integer()
  def prune_older_than(cutoff) do
    # Use DateTime.compare in Elixir rather than ETS match-spec guards —
    # DateTime structs are maps in Elixir ≥ 1.15 and ETS can't compare
    # maps with :< / :> operators.
    :ets.foldl(
      fn {{band, vt}, _val}, acc ->
        if DateTime.before?(vt, cutoff) do
          :ets.delete(@table, {band, vt})
          acc + 1
        else
          acc
        end
      end,
      0,
      @table
    )
  end

  @doc """
  Delete every entry whose `valid_time` falls outside the inclusive
  window `[past_cutoff, future_cutoff]`. Used to bound the cache to
  the active forecast horizon (past ~1 h + HRRR's 48 h forward) so
  long-horizon GEFS valid_times never accumulate in ETS.
  """
  @spec prune_outside_window(DateTime.t(), DateTime.t()) :: non_neg_integer()
  def prune_outside_window(past_cutoff, future_cutoff) do
    :ets.foldl(
      fn {{band, vt}, _val}, acc ->
        if DateTime.before?(vt, past_cutoff) or DateTime.after?(vt, future_cutoff) do
          :ets.delete(@table, {band, vt})
          acc + 1
        else
          acc
        end
      end,
      0,
      @table
    )
  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, :compressed, read_concurrency: true])
    :ok = 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 list_to_chunks(scores) do
    Enum.reduce(scores, %{}, fn %{lat: lat, lon: lon, score: score}, acc ->
      key = chunk_key_for(lat, lon)
      chunk = Map.get(acc, key, %{})
      Map.put(acc, key, Map.put(chunk, {lat, lon}, score))
    end)
  end

  defp chunks_to_list(chunked) do
    Enum.flat_map(chunked, fn {_chunk_key, cells} ->
      Enum.map(cells, fn {{lat, lon}, score} ->
        %{lat: lat, lon: lon, score: score}
      end)
    end)
  end

  defp chunks_filtered_to_list(chunked, nil), do: chunks_to_list(chunked)

  defp chunks_filtered_to_list(chunked, %{"south" => s, "north" => n, "west" => w, "east" => e} = bounds) do
    chunked
    |> Enum.filter(fn {chunk_key, _} -> chunk_intersects_bounds?(chunk_key, bounds) end)
    |> Enum.flat_map(fn {_, cells} ->
      for {{lat, lon}, score} <- cells,
          lat >= s,
          lat <= n,
          lon >= w,
          lon <= e,
          do: %{lat: lat, lon: lon, score: score}
    end)
  end

  defp chunk_key_for(lat, lon) do
    {chunk_band(lat * 1.0), chunk_band(lon * 1.0)}
  end

  defp chunk_band(value) when is_float(value) do
    (value / @chunk_step) |> Float.floor() |> trunc()
  end

  defp chunk_intersects_bounds?({lat_band, lon_band}, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
    chunk_south = lat_band * @chunk_step
    chunk_north = chunk_south + @chunk_step
    chunk_west = lon_band * @chunk_step
    chunk_east = chunk_west + @chunk_step

    chunk_north >= s and chunk_south <= n and chunk_east >= w and chunk_west <= e
  end
end