prop/lib/microwaveprop/propagation/profiles_file.ex

defmodule Microwaveprop.Propagation.ProfilesFile do
  @moduledoc """
  On-disk store for the raw, enriched HRRR grid_data that
  `PropagationGridWorker` produces for the f00 analysis hour. One
  compressed ETF file per `valid_time` lands at
  `{base_dir}/profiles/{iso}.etf.gz`. Used by `Propagation.point_detail/4`
  to rebuild the factor breakdown for a clicked cell without a database
  round trip.

  Only f00 is persisted — forecast hours intentionally skip the factor
  breakdown on the map to keep the scoring+upsert phase under a minute
  (see `Propagation.compute_scores/3`).

  ## Atomic writes

  Files are written through `rename(2)`: `path.tmp.<uniq>` → rename to
  the final path. On the shared NFS mount the rename is atomic, so a
  concurrent reader sees either the old file, the new file, or nothing —
  never a partial write.
  """

  alias Microwaveprop.Propagation.Grid

  require Logger

  @doc "Base directory the profile store lives under."
  @spec base_dir() :: String.t()
  def base_dir do
    Path.join(
      Application.get_env(:microwaveprop, :propagation_scores_dir, "/data/scores"),
      "profiles"
    )
  end

  @doc "Absolute path for the profile file covering `valid_time`."
  @spec path_for(DateTime.t()) :: String.t()
  def path_for(%DateTime{} = valid_time) do
    iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601()
    Path.join(base_dir(), "#{iso}.etf.gz")
  end

  @doc """
  Persist the enriched grid_data (`%{{lat, lon} => profile}`) for
  `valid_time`. Writes compressed ETF; lat/lon keys are rounded to the
  grid step so later point lookups can snap user-provided coordinates.
  """
  @spec write!(DateTime.t(), %{{float(), float()} => map()}) :: :ok
  def write!(%DateTime{} = valid_time, grid_data) when is_map(grid_data) do
    path = path_for(valid_time)
    File.mkdir_p!(Path.dirname(path))

    snapped =
      Map.new(grid_data, fn {{lat, lon}, profile} ->
        {{Float.round(lat, 3), Float.round(lon, 3)}, profile}
      end)

    binary = :erlang.term_to_binary(snapped, [:compressed])

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

  @doc """
  Read a single profile for `(valid_time, lat, lon)`. Returns `nil` if
  the file is missing or the point has no profile. Input lat/lon are
  snapped to the nearest grid cell before lookup.
  """
  @spec read_point(DateTime.t(), float(), float()) :: map() | nil
  def read_point(%DateTime{} = valid_time, lat, lon) do
    case File.read(path_for(valid_time)) do
      {:ok, binary} ->
        {snapped_lat, snapped_lon} = snap(lat, lon)

        binary
        |> :erlang.binary_to_term([:safe])
        |> Map.get({snapped_lat, snapped_lon})

      {:error, _} ->
        nil
    end
  end

  @doc """
  Delete profile files whose valid_time is strictly before `cutoff`.
  Returns the number of files 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_profile_files()
    |> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
      if valid_time_unix < cutoff_unix do
        _ = File.rm(path)
        acc + 1
      else
        acc
      end
    end)
  end

  @doc """
  Keep only profile files whose valid_time is inside the closed window
  `[run_time, run_time + max_forecast_hour * 3600]`. Mirrors
  `ScoresFile.retain_window/2` and is called at the end of a
  `PropagationGridWorker` chain.
  """
  @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_profile_files()
    |> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
      if valid_time_unix < lo or valid_time_unix > hi do
        _ = File.rm(path)
        acc + 1
      else
        acc
      end
    end)
  end

  @doc """
  Latest valid_time of any persisted profile, or nil if the store is
  empty.
  """
  @spec latest_valid_time() :: DateTime.t() | nil
  def latest_valid_time do
    case list_profile_files(base_dir()) do
      [] ->
        nil

      files ->
        files
        |> Enum.map(fn {_path, unix} -> unix end)
        |> Enum.max()
        |> DateTime.from_unix!()
    end
  end

  defp snap(lat, lon) do
    step = Grid.step()
    snapped_lat = Float.round(Float.round(lat / step) * step, 3)
    snapped_lon = Float.round(Float.round(lon / step) * step, 3)
    {snapped_lat, snapped_lon}
  end

  defp list_profile_files(dir) do
    case File.ls(dir) do
      {:ok, entries} ->
        for entry <- entries,
            valid_time_unix = parse_valid_time(entry),
            valid_time_unix != nil do
          {Path.join(dir, entry), valid_time_unix}
        end

      _ ->
        []
    end
  end

  defp parse_valid_time(filename) do
    with [_, iso] <- Regex.run(~r/^(.+)\.etf\.gz$/, filename),
         {:ok, dt, _} <- DateTime.from_iso8601(iso) do
      DateTime.to_unix(dt)
    else
      _ -> nil
    end
  end

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