prop/lib/microwaveprop/weather/era5_client.ex

defmodule Microwaveprop.Weather.Era5Client do
  @moduledoc """
  Client for the Copernicus Climate Data Store (CDS) API.
  Fetches ERA5 reanalysis data (pressure-level profiles + single-level surface fields)
  for historical contact enrichment where HRRR is unavailable (pre-2014).

  The CDS API is asynchronous: submit a request, poll for completion, download result.
  Results are GRIB2 format, decoded with the existing GRIB2 infrastructure.
  """

  alias Microwaveprop.Weather.Grib2.Extractor
  alias Microwaveprop.Weather.SoundingParams

  require Logger

  @cds_url "https://cds.climate.copernicus.eu/api"
  @poll_interval_ms 5_000
  # CDS routinely takes 30+ min to return month-tile jobs under load. The prior
  # 120-attempt (10 min) ceiling was burning through Era5MonthBatchWorker
  # max_attempts before CDS even produced a result. 720 attempts × 5s = 1 hour
  # bounds the worker but still gives CDS enough room.
  @max_poll_attempts 720

  @pressure_levels ~w(
    1000 975 950 925 900 875 850 825 800 775 750 725 700
    650 600 550 500 450 400 350 300 250 200 150 100
  )

  @single_level_vars ~w(
    2m_temperature
    2m_dewpoint_temperature
    surface_pressure
    10m_u_component_of_wind
    10m_v_component_of_wind
    total_column_water_vapour
    boundary_layer_height
  )

  # `dewpoint_temperature` is only a single-level 2m variable in ERA5 — CDS
  # rejects pressure-level requests that include it. Use specific_humidity on
  # pressure levels and derive Td in the profile builder via ThetaE.
  @pressure_level_vars ~w(temperature specific_humidity geopotential)

  @doc false
  @spec pressure_levels() :: [String.t()]
  def pressure_levels, do: @pressure_levels
  @doc false
  @spec single_level_vars() :: [String.t()]
  def single_level_vars, do: @single_level_vars
  @doc false
  @spec pressure_level_vars() :: [String.t()]
  def pressure_level_vars, do: @pressure_level_vars
  @doc false
  @spec cds_url() :: String.t()
  def cds_url, do: @cds_url
  @doc false
  @spec poll_interval_ms() :: pos_integer()
  def poll_interval_ms, do: @poll_interval_ms
  @doc false
  @spec max_poll_attempts() :: pos_integer()
  def max_poll_attempts, do: @max_poll_attempts

  @doc """
  Fetch ERA5 profile for a single point and time.
  Returns {:ok, profile_attrs} or {:error, reason}.

  The profile_attrs map has the same shape as HRRR profiles for interoperability.
  """
  @spec fetch_profile(float(), float(), DateTime.t()) :: {:ok, map()} | {:error, term()}
  def fetch_profile(lat, lon, timestamp) do
    # Round to ERA5 grid (0.25°)
    rlat = Float.round(lat * 4) / 4
    rlon = Float.round(lon * 4) / 4

    # ERA5 hourly — round to nearest hour
    valid_time = DateTime.truncate(timestamp, :second)
    valid_time = %{valid_time | minute: 0, second: 0}

    # Build a small bounding box (0.25° around the point)
    area = [rlat + 0.25, rlon - 0.25, rlat - 0.25, rlon + 0.25]

    with {:ok, single_data} <- fetch_single_levels(valid_time, area),
         {:ok, pressure_data} <- fetch_pressure_levels(valid_time, area) do
      build_profile(rlat, rlon, valid_time, single_data, pressure_data)
    end
  end

  @doc """
  Fetch ERA5 single-level (surface) data for a time and area.
  """
  @spec fetch_single_levels(DateTime.t(), [number()]) :: {:ok, binary()} | {:error, term()}
  def fetch_single_levels(valid_time, area) do
    request = %{
      "product_type" => ["reanalysis"],
      "variable" => @single_level_vars,
      "year" => [Calendar.strftime(valid_time, "%Y")],
      "month" => [Calendar.strftime(valid_time, "%m")],
      "day" => [Calendar.strftime(valid_time, "%d")],
      "time" => [Calendar.strftime(valid_time, "%H:00")],
      "area" => area,
      "data_format" => "grib"
    }

    do_submit_and_download("reanalysis-era5-single-levels", request)
  end

  @doc """
  Fetch ERA5 pressure-level data for a time and area.
  """
  @spec fetch_pressure_levels(DateTime.t(), [number()]) :: {:ok, binary()} | {:error, term()}
  def fetch_pressure_levels(valid_time, area) do
    request = %{
      "product_type" => ["reanalysis"],
      "variable" => @pressure_level_vars,
      "pressure_level" => @pressure_levels,
      "year" => [Calendar.strftime(valid_time, "%Y")],
      "month" => [Calendar.strftime(valid_time, "%m")],
      "day" => [Calendar.strftime(valid_time, "%d")],
      "time" => [Calendar.strftime(valid_time, "%H:00")],
      "area" => area,
      "data_format" => "grib"
    }

    do_submit_and_download("reanalysis-era5-pressure-levels", request)
  end

  @doc false
  @spec submit_and_download(String.t(), map()) :: {:ok, binary()} | {:error, term()}
  def submit_and_download(dataset, request) do
    do_submit_and_download(dataset, request)
  end

  @doc """
  Like `submit_and_download/2` but streams the GRIB2 response directly to
  `path` on disk rather than loading the whole body (50-200 MB for month
  tiles) into the BEAM heap.
  """
  @spec submit_and_download_to_file(String.t(), map(), Path.t()) :: :ok | {:error, term()}
  def submit_and_download_to_file(dataset, request, path) do
    with {:ok, source, api_key} <- submit_and_poll(dataset, request) do
      download_source_to_file(source, api_key, path)
    end
  end

  @doc """
  Submits a CDS job and returns `{:ok, job_id}` without waiting for it to
  complete. Used by the split `Era5SubmitWorker` / `Era5PollWorker` flow so
  Oban worker slots aren't pinned to the 30+ minute CDS queue time.
  """
  @spec submit_job(String.t(), map()) :: {:ok, String.t()} | {:error, String.t()}
  def submit_job(dataset, request) do
    api_key = api_key()

    if is_nil(api_key) do
      {:error, "ERA5_CDS_API_KEY not configured"}
    else
      submit_request(dataset, request, api_key)
    end
  end

  @doc """
  Checks the status of a previously-submitted CDS job and, if complete,
  follows the results pointer to return the download source.

  Returns:
  - `:running` — job is queued or in progress; caller should retry later.
  - `{:done, {:url, href} | {:body, binary}}` — ready; pass the source to
    `download_source_to_file/3`.
  - `:not_found` — CDS returned 404 for this job id; terminal (either the
    job was reaped, deleted manually, or never existed). Callers should
    stop polling and clean up; a new submit is required.
  - `{:rejected, reason}` — CDS accepted the submit (gave us a job id) but
    immediately rejected it, typically because our in-flight queue was
    over the 150-job cap. Terminal; caller should clean up and re-submit
    once cap space is available.
  - `{:failed, reason}` — CDS marked the job failed; don't retry.
  - `{:error, reason}` — transient HTTP/network error; caller should retry.
  """
  @spec check_status(String.t()) ::
          :running
          | :not_found
          | {:done, {:url, String.t()} | {:body, binary()}}
          | {:rejected, String.t()}
          | {:failed, String.t()}
          | {:error, String.t()}
  def check_status(job_id) do
    api_key = api_key()

    if is_nil(api_key) do
      {:error, "ERA5_CDS_API_KEY not configured"}
    else
      do_check_status(job_id, api_key)
    end
  end

  @doc """
  Deletes a completed CDS job from the server so it doesn't count against
  our per-user job quota. Called by `Era5PollWorker` after a successful
  download. Treats 200/202/204/404 as success — 404 is fine because the
  job may already have been reaped by CDS's own cleanup.
  """
  @spec delete_job(String.t()) :: :ok | {:error, String.t()}
  def delete_job(job_id) do
    api_key = api_key()

    if is_nil(api_key) do
      {:error, "ERA5_CDS_API_KEY not configured"}
    else
      url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"

      case Req.delete(
             url,
             [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 30_000] ++ req_options()
           ) do
        {:ok, %{status: status}} when status in [200, 202, 204, 404] -> :ok
        {:ok, %{status: status}} -> {:error, "ERA5 delete_job HTTP #{status}"}
        {:error, reason} -> {:error, "ERA5 delete_job failed: #{inspect(reason)}"}
      end
    end
  end

  @doc """
  Writes a CDS result source to `path`. `{:body, binary}` is written as a
  plain file; `{:url, href}` is streamed through Req's `into: File.stream!`
  so peak heap stays O(chunk) rather than O(full response).
  """
  @spec download_source_to_file(
          {:body, binary()} | {:url, String.t()},
          String.t() | nil,
          Path.t()
        ) :: :ok | {:error, String.t()}
  def download_source_to_file({:body, body}, _api_key, path), do: File.write(path, body)

  def download_source_to_file({:url, href}, api_key, path) do
    download_href_to_file(href, api_key, path)
  end

  defp do_submit_and_download(dataset, request) do
    with {:ok, source, api_key} <- submit_and_poll(dataset, request) do
      case source do
        {:body, body} -> {:ok, body}
        {:url, href} -> download_file(href, api_key)
      end
    end
  end

  defp submit_and_poll(dataset, request) do
    api_key = api_key()

    if is_nil(api_key) do
      {:error, "ERA5_CDS_API_KEY not configured"}
    else
      with {:ok, job_id} <- submit_request(dataset, request, api_key),
           {:ok, source} <- poll_until_complete(job_id, api_key) do
        {:ok, source, api_key}
      end
    end
  end

  defp do_check_status(job_id, api_key) do
    url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"

    url
    |> Req.get([headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000] ++ req_options())
    |> interpret_status_response(url, api_key)
  end

  defp interpret_status_response({:ok, %{status: 200, body: body}}, url, api_key) do
    interpret_status_body(body, url, api_key)
  end

  defp interpret_status_response({:ok, %{status: 404}}, _url, _api_key), do: :not_found

  defp interpret_status_response({:ok, %{status: status, body: body}}, _url, _api_key) do
    {:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"}
  end

  defp interpret_status_response({:error, reason}, _url, _api_key) do
    {:error, "ERA5 poll failed: #{inspect(reason)}"}
  end

  defp interpret_status_body(%{"status" => "successful"}, url, api_key) do
    case fetch_results_source("#{url}/results", api_key) do
      {:ok, source} -> {:done, source}
      {:error, reason} -> {:error, reason}
    end
  end

  defp interpret_status_body(%{"status" => status}, _url, _api_key) when status in ["accepted", "running"] do
    :running
  end

  defp interpret_status_body(%{"status" => "failed"} = body, _url, _api_key) do
    reason = body["message"] || body["error"] || inspect(body)
    {:failed, reason}
  end

  # CDS returns this when the submit was accepted into the user's session
  # but immediately rejected server-side — usually because we were over the
  # 150-job per-user cap. The worker should abandon the job and re-submit
  # once there's room.
  defp interpret_status_body(%{"status" => "rejected"} = body, _url, _api_key) do
    reason = body["message"] || body["detail"] || "CDS job rejected (likely over cap)"
    {:rejected, reason}
  end

  # `dismissed` is what a manually deleted job transitions to — treat it
  # the same as rejected so we drop the row and re-submit.
  defp interpret_status_body(%{"status" => "dismissed"} = body, _url, _api_key) do
    reason = body["message"] || body["detail"] || "CDS job dismissed"
    {:rejected, reason}
  end

  defp interpret_status_body(body, _url, _api_key) do
    {:error, "ERA5 poll unknown status body: #{inspect(body)}"}
  end

  defp submit_request(dataset, request, api_key) do
    url = "#{@cds_url}/retrieve/v1/processes/#{dataset}/execution"

    case Req.post(
           url,
           [
             json: %{"inputs" => request},
             headers: [{"PRIVATE-TOKEN", api_key}],
             receive_timeout: 30_000
           ] ++ req_options()
         ) do
      {:ok, %{status: status, body: body}} when status in [200, 201, 202] ->
        job_id = body["jobID"] || body["request_id"] || body["jobId"]

        if job_id do
          Logger.info("ERA5: submitted job #{job_id} for #{dataset}")
          {:ok, job_id}
        else
          {:error, "ERA5: no job_id in response: #{inspect(body)}"}
        end

      {:ok, %{status: status, body: body}} ->
        {:error, "ERA5 CDS HTTP #{status}: #{inspect(body)}"}

      {:error, reason} ->
        {:error, "ERA5 CDS request failed: #{inspect(reason)}"}
    end
  end

  # Returns {:ok, {:url, href}} when CDS returns a JSON asset pointer (new CDS)
  # or {:ok, {:body, binary}} for direct binary responses (legacy CDS).
  defp poll_until_complete(job_id, api_key, attempt \\ 0)

  defp poll_until_complete(_job_id, _api_key, attempt) when attempt >= @max_poll_attempts do
    {:error, "ERA5: poll timeout after #{@max_poll_attempts} attempts"}
  end

  defp poll_until_complete(job_id, api_key, attempt) do
    url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"

    case Req.get(
           url,
           [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000] ++ req_options()
         ) do
      {:ok, %{status: 200, body: %{"status" => "successful"}}} ->
        results_url = "#{url}/results"
        fetch_results_source(results_url, api_key)

      {:ok, %{status: 200, body: %{"status" => status}}}
      when status in ["accepted", "running"] ->
        Process.sleep(@poll_interval_ms)
        poll_until_complete(job_id, api_key, attempt + 1)

      {:ok, %{status: 200, body: %{"status" => "failed"} = body}} ->
        # CDS often omits "message" — fall back to dumping the whole body so the
        # failure isn't summarised as `nil` in oban_jobs.errors.
        reason = body["message"] || body["error"] || inspect(body)
        {:error, "ERA5 job #{job_id} failed: #{reason}"}

      {:ok, %{status: status, body: body}} ->
        {:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"}

      {:error, reason} ->
        {:error, "ERA5 poll failed: #{inspect(reason)}"}
    end
  end

  defp fetch_results_source(url, api_key) do
    case Req.get(
           url,
           [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000] ++ req_options()
         ) do
      {:ok, %{status: 200, body: %{"asset" => %{"value" => %{"href" => href}}}}} ->
        # New CDS returns a JSON with download link
        {:ok, {:url, href}}

      {:ok, %{status: 200, headers: headers, body: body}} ->
        # Direct binary download
        content_type =
          %Req.Response{headers: headers}
          |> Req.Response.get_header("content-type")
          |> List.first("")

        if String.contains?(content_type, "grib") or is_binary(body) do
          {:ok, {:body, body}}
        else
          {:error, "ERA5: unexpected response format"}
        end

      {:ok, %{status: status, body: body}} ->
        {:error, "ERA5 download HTTP #{status}: #{inspect(body)}"}

      {:error, reason} ->
        {:error, "ERA5 download failed: #{inspect(reason)}"}
    end
  end

  defp download_file(href, api_key) do
    case Req.get(
           href,
           [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000] ++ req_options()
         ) do
      {:ok, %{status: 200, body: body}} -> {:ok, body}
      {:ok, %{status: status}} -> {:error, "ERA5 file download HTTP #{status}"}
      {:error, reason} -> {:error, "ERA5 file download failed: #{inspect(reason)}"}
    end
  end

  # Streams the GRIB2 response directly to `path`. Uses Req's `into:` with a
  # `File.stream!` collectable, so chunks land on disk as they arrive and peak
  # heap stays O(chunk size) instead of O(full response). Target path is
  # removed on HTTP error so a failed download never leaves a partial tmp file.
  defp download_href_to_file(href, api_key, path) do
    result =
      Req.get(
        href,
        [
          headers: [{"PRIVATE-TOKEN", api_key}],
          receive_timeout: 300_000,
          into: File.stream!(path)
        ] ++ req_options()
      )

    case result do
      {:ok, %{status: 200}} ->
        :ok

      {:ok, %{status: status}} ->
        File.rm(path)
        {:error, "ERA5 file download HTTP #{status}"}

      {:error, reason} ->
        File.rm(path)
        {:error, "ERA5 file download failed: #{inspect(reason)}"}
    end
  end

  defp req_options do
    Application.get_env(:microwaveprop, :era5_req_options, [])
  end

  defp build_profile(lat, lon, valid_time, single_grib, pressure_grib) do
    # Decode GRIB2 data using existing infrastructure
    with {:ok, single_points} <- extract_point(single_grib, lat, lon),
         {:ok, pressure_points} <- extract_point(pressure_grib, lat, lon) do
      # ERA5 temperatures are in Kelvin, convert to Celsius
      temp_c = kelvin_to_celsius(single_points["TMP:2 m above ground"])
      dewpoint_c = kelvin_to_celsius(single_points["DPT:2 m above ground"])
      # ERA5 surface pressure is in Pa, convert to mb
      pressure_mb = pa_to_mb(single_points["PRES:surface"])
      hpbl_m = single_points["HPBL:surface"]
      # ERA5 TCWV is in kg/m² which equals mm
      pwat_mm = single_points["TCWV:entire atmosphere"]

      # Build pressure-level profile
      profile =
        @pressure_levels
        |> Enum.map(fn level_str ->
          level = String.to_integer(level_str)
          pres_key = "#{level} mb"

          %{
            "pres" => level * 1.0,
            "tmpc" => kelvin_to_celsius(pressure_points["TMP:#{pres_key}"]),
            "dwpc" => kelvin_to_celsius(pressure_points["DPT:#{pres_key}"]),
            "hght" => geopotential_to_height(pressure_points["HGT:#{pres_key}"])
          }
        end)
        |> Enum.reject(fn p -> is_nil(p["tmpc"]) end)

      # Derive refractivity and ducting from profile
      params = SoundingParams.derive(profile)

      attrs = %{
        valid_time: valid_time,
        lat: lat,
        lon: lon,
        profile: profile,
        hpbl_m: hpbl_m,
        pwat_mm: pwat_mm,
        surface_temp_c: temp_c,
        surface_dewpoint_c: dewpoint_c,
        surface_pressure_mb: pressure_mb
      }

      attrs =
        if params do
          Map.merge(attrs, %{
            surface_refractivity: params.surface_refractivity,
            min_refractivity_gradient: params.min_refractivity_gradient,
            ducting_detected: params.ducting_detected || false,
            duct_characteristics: params.duct_characteristics
          })
        else
          attrs
        end

      {:ok, attrs}
    end
  rescue
    e -> {:error, "ERA5 GRIB2 decode failed: #{Exception.message(e)}"}
  end

  defp extract_point(grib_data, lat, lon) do
    case Extractor.extract_points(grib_data, lat, lon) do
      {:ok, fields} when map_size(fields) > 0 -> {:ok, fields}
      {:ok, _} -> {:error, "no data at #{lat},#{lon}"}
      {:error, reason} -> {:error, reason}
    end
  end

  defp kelvin_to_celsius(nil), do: nil
  defp kelvin_to_celsius(k), do: k - 273.15

  defp pa_to_mb(nil), do: nil
  defp pa_to_mb(pa), do: pa / 100.0

  defp geopotential_to_height(nil), do: nil
  defp geopotential_to_height(gp), do: gp / 9.80665

  defp api_key do
    System.get_env("ERA5_CDS_API_KEY")
  end
end