prop/lib/microwaveprop/weather/scalar_file.ex

defmodule Microwaveprop.Weather.ScalarFile do
  @moduledoc """
  On-disk store for the per-cell *derived* weather rows that feed `/weather`.
  One directory per `valid_time`, with rows bucketed into 5°×5° chunk files
  so viewport reads only decode the chunks that overlap the requested bounds.

  This is the cheap-read sibling of `Microwaveprop.Propagation.ProfilesFile`:

  * `ProfilesFile` keeps the raw HRRR profile per cell (~10 MB decoded) and
    is the source of truth for advanced diagnostics, terrain analysis, and
    point-detail breakdowns.
  * `ScalarFile` keeps only the scalar fields the weather map renders
    (temperature, dewpoint depression, refractivity, lapse rates, duct
    summary, etc.), already pushed through `WeatherLayers.derive/1`.

  Because the scalar shape is small and the derivation has already happened,
  a tile request can read just the chunks it needs and skip the
  `SoundingParams.derive/1` + `WeatherLayers.derive/1` work entirely.

  ## Layout

      <base_dir>/weather_scalars/
        <iso>/                # e.g. 2026-04-28T12:00:00Z/
          <lat_band>_<lon_band>.etf.gz

  `lat_band = floor(lat / 5)`, `lon_band = floor(lon / 5)`. Each chunk is
  the compressed ETF of `[row_map, ...]`. Writes go through
  `rename(2)` → atomic on NFS.

  ## Chunk size

  5°×5° was picked so the typical /weather viewport (a US state at z=6-7)
  overlaps 1-4 chunks. Smaller chunks would balloon directory entries on
  NFS without measurable read benefit.
  """

  require Logger

  @chunk_step 5
  @subdir "weather_scalars"

  @type row :: %{required(:lat) => float(), required(:lon) => float(), optional(atom()) => term()}
  @type bounds :: %{optional(String.t()) => number()}

  @spec base_dir() :: String.t()
  def base_dir do
    Path.join(
      Application.get_env(:microwaveprop, :propagation_scores_dir, "/data/scores"),
      @subdir
    )
  end

  @spec dir_for(DateTime.t()) :: String.t()
  def dir_for(%DateTime{} = valid_time) do
    Path.join(base_dir(), iso_key(valid_time))
  end

  @spec exists?(DateTime.t()) :: boolean()
  def exists?(%DateTime{} = valid_time) do
    case File.ls(dir_for(valid_time)) do
      {:ok, [_ | _]} -> true
      _ -> false
    end
  end

  @doc """
  Persist `rows` for `valid_time`. Rows are bucketed by 5°×5° chunk; each
  chunk is written atomically via tmp + rename. Existing chunks for the
  same `valid_time` are removed first so a smaller follow-up write doesn't
  leave stale chunks behind.
  """
  @spec write!(DateTime.t(), [row()]) :: :ok
  def write!(%DateTime{} = valid_time, rows) when is_list(rows) do
    dir = dir_for(valid_time)
    File.mkdir_p!(dir)

    # Drop any pre-existing chunks for this valid_time so this write is
    # the canonical state — otherwise a smaller follow-up write would
    # leave stale chunks behind.
    case File.ls(dir) do
      {:ok, entries} ->
        for entry <- entries do
          _ = File.rm(Path.join(dir, entry))
        end

      _ ->
        :ok
    end

    rows
    |> Enum.group_by(&chunk_key/1)
    |> Enum.each(fn {{lat_band, lon_band}, chunk_rows} ->
      path = Path.join(dir, "#{lat_band}_#{lon_band}.etf.gz")
      binary = :erlang.term_to_binary(chunk_rows, [:compressed])

      tmp = path <> ".tmp." <> unique_suffix()
      File.write!(tmp, binary, [:binary])
      File.rename!(tmp, path)
    end)

    :ok
  end

  @doc """
  Read every persisted row for `valid_time` whose lat/lon falls within
  `bounds`. Pass `nil` to read every chunk. Returns `[]` if no scalar
  file exists for `valid_time`.
  """
  @spec read_bounds(DateTime.t(), bounds() | nil) :: [row()]
  def read_bounds(%DateTime{} = valid_time, bounds) do
    case list_chunk_files(valid_time) do
      [] ->
        []

      files ->
        files
        |> Enum.filter(fn {key, _path} -> chunk_intersects_bounds?(key, bounds) end)
        |> Enum.flat_map(fn {_key, path} -> decode_chunk(path) end)
        |> filter_bounds(bounds)
    end
  end

  @doc """
  Look up a single derived row at `(valid_time, lat, lon)`. Returns
  `{:ok, row}` or `:miss` when either the chunk file is absent or the
  cell isn't present.
  """
  @spec read_point(DateTime.t(), float(), float()) :: {:ok, row()} | :miss
  def read_point(%DateTime{} = valid_time, lat, lon) when is_number(lat) and is_number(lon) do
    key = {chunk_band(lat * 1.0), chunk_band(lon * 1.0)}
    path = Path.join(dir_for(valid_time), chunk_filename(key))

    if File.exists?(path) do
      find_point_in_chunk(path, snap(lat), snap(lon))
    else
      :miss
    end
  end

  defp find_point_in_chunk(path, snapped_lat, snapped_lon) do
    case Enum.find(decode_chunk(path), fn r ->
           r.lat == snapped_lat and r.lon == snapped_lon
         end) do
      nil -> :miss
      row -> {:ok, row}
    end
  end

  @doc """
  Every persisted valid_time, sorted ascending. Used to back the /weather
  forecast timeline once scalars take over from raw profiles as the
  primary read.
  """
  @spec list_valid_times() :: [DateTime.t()]
  def list_valid_times do
    case File.ls(base_dir()) do
      {:ok, entries} ->
        times =
          for entry <- entries,
              {:ok, dt, _} <- [DateTime.from_iso8601(entry)],
              do: dt

        Enum.sort(times, DateTime)

      _ ->
        []
    end
  end

  @doc "Delete scalar dirs whose valid_time is strictly before `cutoff`. Returns count removed."
  @spec prune_older_than(DateTime.t()) :: non_neg_integer()
  def prune_older_than(%DateTime{} = cutoff) do
    cutoff_unix = DateTime.to_unix(cutoff)

    base_dir()
    |> list_valid_time_dirs()
    |> Enum.reduce(0, fn {path, dt}, acc ->
      if DateTime.to_unix(dt) < cutoff_unix do
        _ = File.rm_rf(path)
        acc + 1
      else
        acc
      end
    end)
  end

  @doc """
  Keep only scalar dirs whose valid_time is inside the closed window
  `[run_time, run_time + max_forecast_hour * 3600]`. Mirrors
  `ProfilesFile.retain_window/2`.
  """
  @spec retain_window(DateTime.t(), non_neg_integer()) :: non_neg_integer()
  def retain_window(%DateTime{} = run_time, max_forecast_hour) when max_forecast_hour >= 0 do
    lo = DateTime.to_unix(run_time)
    hi = lo + max_forecast_hour * 3600

    base_dir()
    |> list_valid_time_dirs()
    |> Enum.reduce(0, fn {path, dt}, acc ->
      unix = DateTime.to_unix(dt)

      if unix < lo or unix > hi do
        _ = File.rm_rf(path)
        acc + 1
      else
        acc
      end
    end)
  end

  # ---------- Internal ----------

  defp iso_key(%DateTime{} = valid_time) do
    valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601()
  end

  defp chunk_key(%{lat: lat, lon: 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_filename({lat_band, lon_band}), do: "#{lat_band}_#{lon_band}.etf.gz"

  defp chunk_intersects_bounds?(_key, nil), do: true

  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

  defp filter_bounds(rows, nil), do: rows

  defp filter_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 decode_chunk(path) do
    case File.read(path) do
      {:ok, binary} ->
        :erlang.binary_to_term(binary)

      {:error, reason} ->
        Logger.warning("ScalarFile chunk read failed path=#{path} reason=#{inspect(reason)}")
        []
    end
  end

  defp list_chunk_files(%DateTime{} = valid_time) do
    dir = dir_for(valid_time)

    case File.ls(dir) do
      {:ok, entries} ->
        for entry <- entries,
            key = parse_chunk_filename(entry),
            key != nil,
            do: {key, Path.join(dir, entry)}

      _ ->
        []
    end
  end

  defp parse_chunk_filename(filename) do
    with [_, lat_str, lon_str] <- Regex.run(~r/^(-?\d+)_(-?\d+)\.etf\.gz$/, filename),
         {lat_band, ""} <- Integer.parse(lat_str),
         {lon_band, ""} <- Integer.parse(lon_str) do
      {lat_band, lon_band}
    else
      _ -> nil
    end
  end

  defp list_valid_time_dirs(base) do
    case File.ls(base) do
      {:ok, entries} ->
        for entry <- entries,
            {:ok, dt, _} <- [DateTime.from_iso8601(entry)],
            do: {Path.join(base, entry), dt}

      _ ->
        []
    end
  end

  # Match ProfilesFile's snap step (0.125°) so a click at any lat/lon
  # rounds to the same key the writer used.
  defp snap(value) do
    step = 0.125
    Float.round(Float.round(value / step) * step, 3)
  end

  defp unique_suffix do
    "#{System.system_time(:nanosecond)}.#{:erlang.unique_integer([:positive])}"
  end
end