prop/lib/microwaveprop/propagation/score_cache_reconciler.ex

defmodule Microwaveprop.Propagation.ScoreCacheReconciler do
  @moduledoc """
  Periodic safety net that reconciles `ScoreCache` with the on-disk
  `/data/scores` tree written by the Rust `prop-grid-rs` worker.

  The primary refresh path is `NOTIFY propagation_ready` → `NotifyListener`
  → `Propagation.warm_cache_and_broadcast/2` → PubSub fan-out. That path
  fails silently in three ways this reconciler is designed to cover:

    1. **NOTIFY dropped on reconnect.** Postgres queues LISTEN notifies
       per-connection and drops anything unread on disconnect. When
       `Postgrex.Notifications` reconnects (DB restart, network blip)
       a full run of score files may already be on disk.
    2. **No listener running locally.** Orphan pods, readiness misfires,
       or early startup mean the LISTEN handler hasn't subscribed yet.
    3. **Rust writes completed pre-boot.** Pod starts, cache is empty,
       and we don't hit the miss path until the first LiveView click.

  Every `interval_ms` the reconciler lists `/data/scores/<band>/*.prop`
  per band, compares against `ScoreCache.valid_times/1`, and fills in
  any {band, valid_time} pairs on disk but absent in the local ETS.
  Warming is node-local (`ScoreCache.put/3`) — no PubSub fan-out, since
  every pod runs its own reconciler and each one converges independently.
  """
  use GenServer

  alias Microwaveprop.Propagation
  alias Microwaveprop.Propagation.BandConfig
  alias Microwaveprop.Propagation.ScoreCache
  alias Microwaveprop.Propagation.ScoresFile

  require Logger

  @default_interval_ms 60_000
  @default_jitter_max_ms 20_000

  @spec start_link(keyword()) :: GenServer.on_start() | :ignore
  def start_link(opts) do
    if Application.get_env(:microwaveprop, :start_score_cache_reconciler, true) do
      GenServer.start_link(__MODULE__, opts, name: __MODULE__)
    else
      :ignore
    end
  end

  @doc """
  Runs one reconciliation pass synchronously. Returns the count of
  {band, valid_time} pairs warmed into the cache.
  """
  @spec sweep_once() :: non_neg_integer()
  def sweep_once do
    Enum.reduce(BandConfig.all_bands(), 0, fn %{freq_mhz: band_mhz}, acc ->
      acc + warm_missing_for_band(band_mhz)
    end)
  end

  @doc """
  Returns the next sweep delay in milliseconds: the configured base
  interval plus a uniformly random jitter in `[1, jitter_max_ms]`.

  Scheduling each pod's next tick with a randomized offset prevents the
  cluster-wide "every replica sweeps at :00" thundering herd against the
  shared NFS scores mount and Postgres.
  """
  @spec next_sweep_interval() :: non_neg_integer()
  def next_sweep_interval(base_ms \\ @default_interval_ms, jitter_max_ms \\ @default_jitter_max_ms)

  def next_sweep_interval(base_ms, jitter_max_ms) when jitter_max_ms > 0 do
    base_ms + :rand.uniform(jitter_max_ms)
  end

  def next_sweep_interval(base_ms, _jitter_max_ms), do: base_ms

  @impl true
  def init(opts) do
    interval = Keyword.get(opts, :interval_ms, @default_interval_ms)
    jitter_max = Keyword.get(opts, :jitter_max_ms, @default_jitter_max_ms)

    _ =
      if Keyword.get(opts, :run_on_start, true) do
        _ = Process.send_after(self(), :sweep, 500)
      end

    {:ok, %{interval_ms: interval, jitter_max_ms: jitter_max}}
  end

  @impl true
  def handle_info(:sweep, state) do
    warmed = sweep_once()

    if warmed > 0 do
      Logger.info("ScoreCacheReconciler: warmed #{warmed} {band, valid_time} pairs from disk")
    end

    delay = next_sweep_interval(state.interval_ms, state.jitter_max_ms)
    Process.send_after(self(), :sweep, delay)
    {:noreply, state}
  end

  def handle_info(_msg, state), do: {:noreply, state}

  defp warm_missing_for_band(band_mhz) do
    {past_cutoff, future_cutoff} = Propagation.hot_cache_window()

    on_disk =
      band_mhz
      |> ScoresFile.list_valid_times()
      |> Enum.filter(fn vt ->
        DateTime.compare(vt, past_cutoff) != :lt and
          DateTime.compare(vt, future_cutoff) != :gt
      end)
      |> MapSet.new()

    in_cache = MapSet.new(ScoreCache.valid_times(band_mhz))
    missing = MapSet.difference(on_disk, in_cache)

    Enum.reduce(missing, 0, fn valid_time, acc ->
      case warm_one(band_mhz, valid_time) do
        :ok -> acc + 1
        :error -> acc
      end
    end)
  end

  # File may vanish between list and read if a prune runs concurrently,
  # or be half-written during an atomic-rename race. `ScoresFile.read/2`
  # reports both as `{:error, _}`; we skip and self-heal on the next
  # sweep rather than caching an empty grid under that key.
  defp warm_one(band_mhz, valid_time) do
    case ScoresFile.read(band_mhz, valid_time) do
      {:ok, payload} ->
        scores = ScoresFile.extract_points(payload, nil)
        ScoreCache.put(band_mhz, valid_time, scores)
        :ok

      {:error, reason} ->
        Logger.debug("ScoreCacheReconciler: skipped #{band_mhz}/#{valid_time}: #{inspect(reason)}")

        :error
    end
  end
end