prop/lib/microwaveprop/workers/era5_poll_worker.ex

defmodule Microwaveprop.Workers.Era5PollWorker do
  @moduledoc """
  Second half of the split ERA5 backfill pipeline. Polls CDS for the two
  jobs submitted by `Era5SubmitWorker`; on completion, streams both GRIB2
  files to disk, decodes them, bulk-inserts the profiles, deletes the
  `era5_cds_jobs` row, and asks CDS to delete the completed jobs so they
  don't count against our per-user quota.

  While either CDS job is still running the worker returns
  `{:snooze, N}` — Oban re-schedules the same job without counting an
  attempt and releases the worker slot immediately, so one pod can keep
  dozens of tile-months in flight without pinning workers to
  `Process.sleep`.

  `max_attempts` is high because each attempt models a human-scale
  failure (e.g. CDS reports the job failed), not a poll retry — snoozes
  don't consume attempts.
  """
  use Oban.Worker,
    queue: :era5_poll,
    max_attempts: 10

  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.Era5BatchClient
  alias Microwaveprop.Weather.Era5CdsJob
  alias Microwaveprop.Weather.Era5Client
  alias Microwaveprop.Workers.Era5SubmitWorker

  require Logger

  # Re-check every 5 minutes. CDS month-tile jobs typically take 30-45 min,
  # so 5 minutes strikes a balance between responsiveness and noise.
  @snooze_seconds 5 * 60

  # A CDS job that's been in 'accepted' state for more than this is stuck
  # in the server's queue and almost certainly won't run. Observed in
  # prod: pressure-level completes in ~90 min, but single-level
  # routinely sits at 'accepted' for 12+ hours under CDS load. The
  # previous 4h threshold was firing on normal slow runs, forcing a
  # resubmit which then got rejected by the per-user cap — an infinite
  # reject→resubmit spiral that burned CDS quota without producing any
  # decoded data. 18h is comfortably above the worst observed queue
  # times while still self-healing within a day.
  @stuck_after_seconds 18 * 3600
  @stuck_after_hours div(@stuck_after_seconds, 3600)

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"era5_cds_job_id" => id}}) do
    case Repo.get(Era5CdsJob, id) do
      nil ->
        Logger.debug("Era5Poll: row #{id} no longer exists — no-op")
        :ok

      %Era5CdsJob{} = row ->
        handle_row(row)
    end
  end

  defp handle_row(row) do
    single = Era5Client.check_status(row.single_job_id)
    pressure = Era5Client.check_status(row.pressure_job_id)

    case {single, pressure} do
      {{:done, single_source}, {:done, pressure_source}} ->
        handle_both_done(row, single_source, pressure_source)

      _other ->
        if stuck?(row) do
          resubmit_vanished(row, "both", "stuck in CDS queue > #{@stuck_after_hours}h")
        else
          handle_non_both_done(row, single, pressure)
        end
    end
  end

  # True when the row is older than @stuck_after_seconds AND hasn't
  # already landed both legs. CDS jobs that sit in 'accepted' state past
  # this threshold are effectively dead and we should give up rather
  # than polling forever.
  defp stuck?(row) do
    age_seconds = DateTime.diff(DateTime.utc_now(), row.submitted_at, :second)
    age_seconds > @stuck_after_seconds
  end

  # The non-both-done dispatch is split out to keep handle_row/1 under
  # credo's cyclomatic complexity limit. `leg_outcome/1` collapses each
  # leg's status into a simple tag (:ok | :terminal | :failed | :error |
  # :running) plus a payload so the outer dispatch only has to match on
  # the *pair* of tags, not the cross-product of individual statuses.
  defp handle_non_both_done(row, single, pressure) do
    case {leg_outcome(single), leg_outcome(pressure)} do
      {{:terminal, reason}, _} ->
        resubmit_vanished(row, "single-level", reason)

      {_, {:terminal, reason}} ->
        resubmit_vanished(row, "pressure-level", reason)

      {{:failed, reason}, _} ->
        fail_row(row, "single-level job failed: #{reason}")

      {_, {:failed, reason}} ->
        fail_row(row, "pressure-level job failed: #{reason}")

      {{:error, reason}, _} ->
        {:error, "single-level check_status error: #{reason}"}

      {_, {:error, reason}} ->
        {:error, "pressure-level check_status error: #{reason}"}

      _both_running_or_one_done ->
        bump_poll_count(row)
        {:snooze, @snooze_seconds}
    end
  end

  defp leg_outcome({:done, _source}), do: :ok
  defp leg_outcome(:not_found), do: {:terminal, "deleted"}
  defp leg_outcome({:rejected, reason}), do: {:terminal, "rejected: #{reason}"}
  defp leg_outcome({:failed, reason}), do: {:failed, reason}
  defp leg_outcome({:error, reason}), do: {:error, reason}
  defp leg_outcome(:running), do: :running

  # A CDS job is terminally unreachable — either 404 (deleted) or rejected
  # by the server (over the per-user cap). Polling will never succeed, but
  # the tile-month's data still needs to exist, so we drop the row, clean
  # up the sibling CDS job (best-effort), and ask Era5SubmitWorker to
  # re-submit. The submit worker's CDS cap guard will snooze the retry
  # until there's room on the server. Returning `:discard` keeps Oban from
  # wasting attempts on a retry that would hit the same terminal state.
  defp resubmit_vanished(row, leg, reason) do
    Logger.warning(
      "Era5Poll: CDS job terminal (#{leg} #{reason}) for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon} — re-submitting"
    )

    delete_cds_jobs(row)
    Repo.delete!(row)
    enqueue_resubmit(row)
    {:discard, "#{leg} CDS job terminal (#{reason}); re-submitted tile-month"}
  end

  defp enqueue_resubmit(row) do
    %{
      "year" => row.year,
      "month" => row.month,
      "tile_lat" => row.tile_lat,
      "tile_lon" => row.tile_lon
    }
    |> Era5SubmitWorker.new()
    |> Oban.insert()
  end

  defp handle_both_done(row, single_source, pressure_source) do
    tmp_dir = System.tmp_dir!()

    tag =
      "#{row.year}_#{pad(row.month)}_#{row.tile_lat}_#{row.tile_lon}_#{System.unique_integer([:positive])}"

    single_path = Path.join(tmp_dir, "era5_single_#{tag}.grib2")
    pressure_path = Path.join(tmp_dir, "era5_pressure_#{tag}.grib2")

    try do
      with :ok <-
             Era5Client.download_source_to_file(single_source, api_key(), single_path),
           :ok <-
             Era5Client.download_source_to_file(pressure_source, api_key(), pressure_path),
           {:ok, inserted} <-
             Era5BatchClient.decode_and_insert(
               row.year,
               row.month,
               row.tile_lat,
               row.tile_lon,
               single_path,
               pressure_path
             ) do
        Repo.delete!(row)
        delete_cds_jobs(row)

        Logger.info(
          "Era5Poll: stored #{inserted} profiles for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}"
        )

        {:ok, inserted}
      else
        {:error, reason} ->
          Logger.warning(
            "Era5Poll: download/decode failed for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}: #{inspect(reason)}"
          )

          {:error, reason}
      end
    after
      File.rm(single_path)
      File.rm(pressure_path)
    end
  end

  defp fail_row(row, reason) do
    Logger.warning(
      "Era5Poll: CDS failure for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}: #{reason}"
    )

    delete_cds_jobs(row)
    Repo.delete!(row)
    {:error, reason}
  end

  defp delete_cds_jobs(row) do
    # Best-effort — don't block progress on a CDS cleanup failure.
    for job_id <- [row.single_job_id, row.pressure_job_id] do
      case Era5Client.delete_job(job_id) do
        :ok ->
          :ok

        {:error, reason} ->
          Logger.warning("Era5Poll: failed to delete CDS job #{job_id}: #{reason}")
          :ok
      end
    end
  end

  defp bump_poll_count(row) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    row
    |> Era5CdsJob.changeset(%{poll_count: row.poll_count + 1, last_polled_at: now})
    |> Repo.update!()
  end

  defp api_key, do: System.get_env("ERA5_CDS_API_KEY")

  defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
end