prop/lib/microwaveprop/weather/grid.ex

defmodule Microwaveprop.Weather.Grid do
  @moduledoc false

  import Ecto.Query

  alias Microwaveprop.Propagation.ProfilesFile
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.GridCache
  alias Microwaveprop.Weather.HrrrProfile
  alias Microwaveprop.Weather.ScalarFile
  alias Microwaveprop.Weather.SoundingParams
  alias Microwaveprop.Weather.WeatherLayers

  require Logger

  @hrdps_nearest_window_seconds 6 * 3600

  # Cap on cached valid_times. Each entry is a chunked map of the full
  # ~92k-cell CONUS grid with 22 fields per cell, ~32 MiB compressed in
  # ETS. The previous cap of 24 meant the cache could hoard up to
  # ~768 MiB on a single pod against a 6 GiB memory limit (the headroom
  # that disappeared in the 2026-05-03 OOM cascade).
  #
  # 8 covers the current hour plus the next ~7 forecast hours — the
  # typical user scrub window. Hours beyond 8 fall back to a disk read
  # (`ScalarFile.read_bounds`) which is sub-100 ms per file, so rare
  # deep-forecast scrubs pay a one-shot latency hit instead of every
  # pod permanently parking ~500 MiB on data the user almost never
  # looks at. Worst case ETS spend drops from 768 MiB → 256 MiB.
  @grid_cache_valid_time_cap 8

  @spec latest_grid_valid_time() :: DateTime.t() | nil
  def latest_grid_valid_time do
    cond do
      vt = GridCache.latest_valid_time() -> vt
      vt = ProfilesFile.latest_valid_time() -> vt
      true -> latest_grid_valid_time_from_db()
    end
  end

  # Last-resort fallback for historical data sitting in the legacy
  # hrrr_profiles table. PropagationGridWorker no longer writes
  # grid-point rows there, so in steady state this returns nil and
  # the ProfilesFile fallback above is the real source of truth.
  defp latest_grid_valid_time_from_db do
    Repo.one(
      from(h in HrrrProfile,
        where: h.is_grid_point == true,
        select: max(h.valid_time)
      )
    )
  end

  @doc """
  Cache-only read for the /weather LiveView mount hot path. Returns whatever
  is in `GridCache` for the latest valid_time and fires a deduped background
  fill task on a miss instead of blocking. The async task broadcasts
  `weather:updated` when done, triggering every connected LiveView to refresh.

  Callers that genuinely need synchronous data (tests, scripts) should use
  `load_weather_grid/1` instead.
  """
  @spec latest_weather_grid(%{optional(String.t()) => float()} | nil) :: [map()]
  def latest_weather_grid(bounds) do
    case latest_grid_valid_time() do
      nil ->
        []

      latest_vt ->
        case GridCache.fetch_bounds(latest_vt, bounds) do
          {:ok, rows} ->
            rows

          :miss ->
            kickoff_async_grid_fill(latest_vt)
            []
        end
    end
  end

  @doc """
  Synchronous cache-or-disk read. Blocks for ~1s on a cold cache while
  `ProfilesFile.read/1` loads the latest grid from `/data/profiles`.
  Used by tests and by `weather_point_detail/3` fallbacks. LiveView
  callers should prefer `latest_weather_grid/1`.
  """
  @spec load_weather_grid(%{optional(String.t()) => float()} | nil) :: [map()]
  def load_weather_grid(bounds) do
    case latest_grid_valid_time() do
      nil ->
        []

      latest_vt ->
        case GridCache.fetch_bounds(latest_vt, bounds) do
          {:ok, rows} ->
            rows

          :miss ->
            full = load_grid_rows_for(latest_vt)
            GridCache.put(latest_vt, full)
            filter_weather_bounds(full, bounds)
        end
    end
  end

  @doc """
  All persisted weather valid_times sorted ascending. The grid worker
  writes a ProfilesFile for every forecast hour (f00..f18 hourly, f21..f48 3-hourly), and the
  derived `ScalarFile` mirrors that for any hour that has been
  materialized. We union both so the timeline survives an aggressive
  retention sweep on either side as long as one artifact remains.
  """
  @spec available_weather_valid_times() :: [DateTime.t()]
  def available_weather_valid_times do
    scalar_times = ScalarFile.list_valid_times()
    profile_times = ProfilesFile.list_valid_times()

    (scalar_times ++ profile_times)
    |> Enum.uniq()
    |> Enum.sort(DateTime)
  end

  @doc """
  All persisted HRDPS valid_times (i.e. those with a `<iso>.hrdps`
  scalar dir on disk) sorted ascending. Backs the `/weather-ca`
  timeline.
  """
  @spec available_hrdps_valid_times() :: [DateTime.t()]
  def available_hrdps_valid_times do
    ScalarFile.list_valid_times_hrdps()
  end

  @doc """
  HRDPS-only counterpart to `weather_grid_at/2`. Reads from the
  `<vt>.hrdps` scalar dir and skips HRRR completely. Bypasses GridCache
  because the cache mixes HRRR + HRDPS rows by design — caching this
  separately would double the per-pod memory budget for marginal benefit
  (Canadian viewport reads are infrequent compared to CONUS).

  The /weather (dual-source) timeline picks valid_times from HRRR's
  hourly cadence, but HRDPS publishes 4×/day with multi-hour latency,
  so the requested time will frequently miss any on-disk HRDPS dir.
  Snapping to the nearest available HRDPS time within a 6 h window
  keeps the Canadian overlay rendering slightly stale instead of
  disappearing entirely. /weather-ca picks times from HRDPS-only listings
  so the exact time always matches and the snap is a no-op.
  """
  @spec weather_grid_hrdps_at(DateTime.t(), %{optional(String.t()) => float()} | nil) :: [map()]
  def weather_grid_hrdps_at(%DateTime{} = valid_time, bounds) do
    ScalarFile.read_bounds_hrdps_nearest(valid_time, bounds, @hrdps_nearest_window_seconds)
  end

  @doc """
  Read the weather grid for a specific `valid_time` and bounds. Like
  `load_weather_grid/1` but takes the valid_time explicitly so the
  timeline can scrub to any forecast hour, not just the analysis hour.
  Returns `[]` if no profile file exists for that valid_time.

  Deliberately does NOT write forecast-hour grids back into `GridCache`
  on a miss: caching 48 forecast hours × 92k points would add ~300 MB
  per pod. The ProfilesFile read is a single ~2 MB ETF decode per scrub,
  which is fast enough for a user click.
  """
  @spec weather_grid_at(DateTime.t(), %{optional(String.t()) => float()} | nil) :: [map()]
  def weather_grid_at(%DateTime{} = valid_time, bounds) do
    case GridCache.fetch_bounds(valid_time, bounds) do
      {:ok, rows} ->
        rows

      :miss ->
        # Once a scalar file exists for `valid_time` it's the source of
        # truth — even an empty viewport read (e.g. over the ocean) is
        # authoritative, so don't fall back to a full ProfilesFile decode
        # in that case.
        if ScalarFile.exists?(valid_time) do
          read_via_scalar(valid_time, bounds)
        else
          read_and_derive_grid(valid_time, bounds)
        end
    end
  end

  defp read_via_scalar(valid_time, bounds) do
    fill_grid_cache_from_scalar(valid_time)

    case GridCache.fetch_bounds(valid_time, bounds) do
      {:ok, rows} -> rows
      :miss -> ScalarFile.read_bounds(valid_time, bounds)
    end
  end

  # Hydrate GridCache from the on-disk ScalarFile so concurrent viewport
  # reads for the same valid_time don't each gunzip+msgpack-decode the
  # same chunk files. Only the caller that wins `claim_fill` does the
  # work; losers wait briefly for the ETS write to land.
  defp fill_grid_cache_from_scalar(valid_time) do
    lock_key = {:scalar_to_grid_cache, valid_time}

    if GridCache.claim_fill(lock_key) do
      try do
        rows = ScalarFile.read_bounds(valid_time, nil)

        if rows != [] do
          GridCache.put(valid_time, rows)
          _ = GridCache.prune_keep_latest(@grid_cache_valid_time_cap)
        end
      after
        GridCache.release_fill(lock_key)
      end
    else
      wait_for_grid_cache(valid_time)
    end
  end

  defp wait_for_grid_cache(valid_time) do
    Enum.reduce_while(1..50, :miss, fn _, _ ->
      case GridCache.fetch(valid_time) do
        {:ok, _} ->
          {:halt, :ok}

        :miss ->
          Process.sleep(20)
          {:cont, :miss}
      end
    end)
  end

  # Cold path: ScalarFile didn't have anything for this valid_time, so
  # decode the raw ProfilesFile and derive the requested viewport. Kicks
  # off a background materialization of the full scalar file so the
  # next read hits the cheap path.
  defp read_and_derive_grid(valid_time, bounds) do
    case ProfilesFile.read(valid_time) do
      {:ok, grid_data} ->
        # Filter-before-derive: only derive the points inside the
        # viewport instead of all 92k CONUS grid points. On a DFW
        # viewport that's ~20× less `SoundingParams.derive` work
        # per timeline scrub.
        rows = build_grid_cache_rows(grid_data, valid_time, bounds)
        kickoff_async_scalar_materialize(valid_time, grid_data)
        rows

      {:error, _} ->
        []
    end
  end

  # Materialize the full ScalarFile for `valid_time` once per node,
  # asynchronously. Reuses GridCache.claim_fill so concurrent
  # `weather_grid_at` callers don't trigger N derivations of the same
  # 92k-cell grid. Lock key namespaced so it doesn't collide with the
  # GridCache cold-fill claim for the same valid_time.
  defp kickoff_async_scalar_materialize(valid_time, grid_data) do
    if ScalarFile.exists?(valid_time) do
      :ok
    else
      lock_key = {:scalar_materialize, valid_time}

      _ =
        if GridCache.claim_fill(lock_key) do
          {:ok, _pid} =
            Task.start(fn ->
              try do
                rows = build_grid_cache_rows(grid_data, valid_time)
                ScalarFile.write!(valid_time, rows)

                Logger.info("Weather.scalar_file materialized valid_time=#{valid_time} rows=#{length(rows)}")
              rescue
                e ->
                  Logger.error("Weather.scalar_file materialize failed valid_time=#{valid_time} #{inspect(e)}")
              after
                GridCache.release_fill(lock_key)
              end
            end)
        end

      :ok
    end
  end

  # Build derived GridCache rows for a valid_time from whichever
  # source has data: the persisted ProfilesFile first (hot path in
  # steady state), then the legacy hrrr_profiles table (historical
  # data only).
  defp load_grid_rows_for(valid_time) do
    case ProfilesFile.read(valid_time) do
      {:ok, grid_data} -> build_grid_cache_rows(grid_data, valid_time)
      {:error, _} -> []
    end
  end

  defp filter_weather_bounds(rows, nil), do: rows

  defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
    Enum.filter(rows, fn %{lat: lat, lon: lon} ->
      lat >= s and lat <= n and lon >= w and lon <= e
    end)
  end

  defp kickoff_async_grid_fill(valid_time) do
    _ =
      if GridCache.claim_fill(valid_time) do
        {:ok, _pid} =
          Task.start(fn ->
            try do
              Logger.info("Weather.grid_cache async fill starting for #{valid_time}")
              warm_grid_cache_and_broadcast(valid_time)

              _ =
                Phoenix.PubSub.broadcast(
                  Microwaveprop.PubSub,
                  "weather:updated",
                  {:weather_updated, valid_time}
                )

              Logger.info("Weather.grid_cache async fill complete for #{valid_time}")
            rescue
              e ->
                Logger.error("Weather.grid_cache async fill failed: #{inspect(e)}")
            after
              GridCache.release_fill(valid_time)
            end
          end)
      end

    :ok
  end

  @doc """
  Eagerly populate the `GridCache` with the full CONUS weather grid for
  `valid_time` and broadcast it to every node in the cluster. Used by the cold
  cache fill path (`kickoff_async_grid_fill/1`) — prefer
  `build_grid_cache_rows/2` inside `PropagationGridWorker`, which already has
  the in-memory grid data and avoids the ~20s JSONB round trip.
  """
  @spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok
  def warm_grid_cache_and_broadcast(valid_time) do
    rows = load_grid_rows_for(valid_time)
    GridCache.broadcast_put(valid_time, rows)
    persist_scalar_file(valid_time, rows)
    :ok
  end

  @doc """
  Warm the local `GridCache` from the latest persisted `ProfilesFile`
  on pod startup. Makes `/weather` usable immediately after a deploy
  instead of waiting for the next hourly PropagationGridWorker run.
  Local put only — every node reads the same NFS mount so no need to
  broadcast.
  """
  @spec warm_grid_cache_from_latest_profile() :: :ok
  def warm_grid_cache_from_latest_profile do
    case ProfilesFile.latest_valid_time() do
      nil ->
        :ok

      valid_time ->
        try do
          rows = load_grid_rows_for(valid_time)
          GridCache.put(valid_time, rows)
          persist_scalar_file(valid_time, rows)
          Logger.info("Weather: warmed GridCache from ProfilesFile for #{valid_time} (#{length(rows)} rows)")
        rescue
          e ->
            Logger.warning("Weather: ProfilesFile warm failed: #{inspect(e)}")
        end

        :ok
    end
  end

  # Persist a derived grid as a ScalarFile so subsequent `/weather` reads
  # don't have to re-decode the raw ProfilesFile or re-run
  # `SoundingParams.derive` + `WeatherLayers.derive`. Best-effort: an NFS
  # write failure is logged but never fatal — the cold-derive path keeps
  # working.
  defp persist_scalar_file(_valid_time, []), do: :ok

  defp persist_scalar_file(valid_time, rows) do
    ScalarFile.write!(valid_time, rows)
    :ok
  rescue
    e ->
      Logger.warning("Weather: ScalarFile.write failed valid_time=#{valid_time} #{inspect(e)}")
      :ok
  end

  @doc """
  Materialize the `ScalarFile` for `valid_time` from the on-disk
  `ProfilesFile`. Idempotent — if a scalar file already exists, returns
  `:ok` without re-deriving. Used by `NotifyListener` to pre-warm scalar
  artifacts the moment the Rust propagation pipeline fires
  `propagation_ready`, so the first `/weather` reader of a new forecast
  hour never falls back to the slow `ProfilesFile.read/1` + per-cell
  derive path.

  Synchronous; callers should run this inside a `Task` if they need
  not to block (the listener does).
  """
  @spec materialize_scalar_file(DateTime.t()) :: :ok
  def materialize_scalar_file(%DateTime{} = valid_time) do
    if ScalarFile.exists?(valid_time) do
      :ok
    else
      try do
        rows = load_grid_rows_for(valid_time)
        persist_scalar_file(valid_time, rows)
        Logger.info("Weather.materialize_scalar_file vt=#{valid_time} rows=#{length(rows)}")
        :ok
      rescue
        e ->
          Logger.warning("Weather.materialize_scalar_file failed vt=#{valid_time} #{inspect(e)}")
          :ok
      end
    end
  end

  @doc """
  Build derived weather grid cache rows directly from an in-memory HRRR
  `grid_data` map. Used by the cold-cache fill path after `ProfilesFile.read/1`
  returns the grid written by the Rust worker.

  `grid_data` is `%{{lat, lon} => profile_map}` as produced by
  `ProfilesFile.read/1`. Each row is pushed through `derive_and_clean/1`
  to compute the derived fields consumed by the weather map LiveView.
  """
  @spec build_grid_cache_rows(
          %{{float(), float()} => map()},
          DateTime.t(),
          %{optional(String.t()) => float()} | nil
        ) :: [map()]
  def build_grid_cache_rows(grid_data, valid_time, bounds \\ nil) do
    grid_data
    |> filter_grid_data_bounds(bounds)
    |> Enum.flat_map(fn {{lat, lon}, profile} ->
      build_grid_cache_row(lat, lon, profile, valid_time)
    end)
  end

  defp filter_grid_data_bounds(grid_data, nil), do: grid_data

  defp filter_grid_data_bounds(grid_data, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
    :maps.filter(fn {lat, lon}, _ -> lat >= s and lat <= n and lon >= w and lon <= e end, grid_data)
  end

  defp build_grid_cache_row(lat, lon, profile, valid_time) do
    temp_c = profile[:surface_temp_c]

    if is_nil(temp_c) or temp_c < -80 or temp_c > 60 do
      []
    else
      sounding = derive_sounding(profile[:profile])

      row = %{
        lat: lat,
        lon: lon,
        valid_time: valid_time,
        temperature: temp_c,
        dewpoint_depression: depression(temp_c, profile[:surface_dewpoint_c]),
        bl_height: profile[:hpbl_m],
        pwat: profile[:pwat_mm],
        refractivity_gradient:
          prefer(profile, :min_refractivity_gradient, sounding[:min_refractivity_gradient]) ||
            profile[:native_min_gradient],
        ducting: prefer(profile, :ducting_detected, sounding[:ducting_detected]),
        surface_pressure_mb: profile[:surface_pressure_mb],
        surface_temp_c: temp_c,
        surface_dewpoint_c: profile[:surface_dewpoint_c],
        surface_refractivity: prefer(profile, :surface_refractivity, sounding[:surface_refractivity]),
        profile: profile[:profile] || [],
        duct_characteristics: prefer(profile, :duct_characteristics, sounding[:duct_characteristics])
      }

      [derive_and_clean(row)]
    end
  end

  # Fetch `key` from `profile` verbatim when present (including `false`,
  # `0`, or `[]`); only fall through to the derived value when the key is
  # absent. `profile[key] || default` would discard legitimate `false` /
  # `0` as if they weren't set.
  defp prefer(profile, key, default) do
    case Map.fetch(profile, key) do
      {:ok, value} -> value
      :error -> default
    end
  end

  defp derive_sounding([_, _, _ | _] = profile) do
    SoundingParams.derive(profile) || %{}
  end

  defp derive_sounding(_), do: %{}

  defp depression(_, nil), do: nil
  defp depression(t, d), do: t - d

  @spec weather_point_detail(float(), float(), DateTime.t()) :: map() | nil
  def weather_point_detail(lat, lon, valid_time) do
    step = 0.125
    snapped_lat = Float.round(Float.round(lat / step) * step, 3)
    snapped_lon = Float.round(Float.round(lon / step) * step, 3)

    case GridCache.fetch_point(valid_time, snapped_lat, snapped_lon) do
      {:ok, row} -> row
      :miss -> point_detail_off_cache(valid_time, snapped_lat, snapped_lon)
    end
  end

  defp point_detail_off_cache(valid_time, snapped_lat, snapped_lon) do
    case weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
      nil -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
      row -> row
    end
  end

  # Derive a single GridCache-shaped row from a persisted ProfilesFile
  # entry for `(valid_time, lat, lon)`. Returns nil when the file
  # doesn't exist or the point has no profile.
  defp weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
    case ProfilesFile.read_point(valid_time, snapped_lat, snapped_lon) do
      nil ->
        nil

      profile ->
        case build_grid_cache_rows(%{{snapped_lat, snapped_lon} => profile}, valid_time) do
          [row] -> row
          _ -> nil
        end
    end
  end

  defp weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon) do
    from(h in HrrrProfile,
      where: h.lat == ^snapped_lat and h.lon == ^snapped_lon and h.valid_time == ^valid_time,
      select: %{
        lat: h.lat,
        lon: h.lon,
        valid_time: h.valid_time,
        temperature: h.surface_temp_c,
        dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c),
        bl_height: h.hpbl_m,
        pwat: h.pwat_mm,
        refractivity_gradient: h.min_refractivity_gradient,
        ducting: h.ducting_detected,
        surface_pressure_mb: h.surface_pressure_mb,
        surface_temp_c: h.surface_temp_c,
        surface_dewpoint_c: h.surface_dewpoint_c,
        surface_refractivity: h.surface_refractivity,
        profile: h.profile,
        duct_characteristics: h.duct_characteristics
      }
    )
    |> Repo.one()
    |> then(fn
      nil -> nil
      row -> derive_and_clean(row)
    end)
  end

  defp derive_and_clean(row) do
    derived = WeatherLayers.derive(row)

    row
    |> Map.merge(derived)
    |> Map.drop([:profile, :duct_characteristics, :surface_temp_c, :surface_dewpoint_c])
  end
end