From 1f2a729d401750ca1acde6068a6fc11a7e9ac6d6 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Thu, 16 Apr 2026 08:26:17 -0500
Subject: [PATCH] Drop dead Era5*/CDS code paths

After the NARR backfill landed, nothing reachable from the app calls
any of these modules. Net removal: ~2700 lines.

Deleted:
- Microwaveprop.Workers.{Era5Fetch,Era5Submit,Era5Poll,Era5MonthBatch}Worker
- Microwaveprop.Weather.{Era5BatchClient,Era5Client}
- Mix.Tasks.Era5Backfill (superseded by BackfillEnqueueWorker cron)
- Matching test files (~1100 LOC of test)
- era5/era5_submit/era5_poll/era5_batch queue blocks from runtime.exs
- era5_req_options Req.Test stub config from test.exs

Kept (actively used or intentionally preserved):
- Era5Profile schema (the era5_profiles table is the NARR target)
- Era5CdsJob schema + its test (inspection handle on the era5_cds_jobs
  table, which retains rows from the failed CDS runs)
- :era5 type key in ContactWeatherEnqueueWorker / BackfillEnqueueWorker
  (the symbol still means "historical backfill", just routed to NARR now)

Swapped the one remaining live Era5FetchWorker call site in
contact_live/show.ex's maybe_enqueue_era5 path to NarrFetchWorker
(with NarrClient.snap_to_analysis_hour for the 3-hourly slot).

Test suite: 1513 tests, 0 failures (-50 from the deleted era5 test
files). No compile warnings. Credo clean.
---
 config/runtime.exs                            |  23 +-
 config/test.exs                               |   1 -
 .../weather/era5_batch_client.ex              | 376 ------------
 lib/microwaveprop/weather/era5_client.ex      | 541 ------------------
 lib/microwaveprop/weather/narr_client.ex      |  13 +-
 .../workers/era5_fetch_worker.ex              |  80 ---
 .../workers/era5_month_batch_worker.ex        |  42 --
 lib/microwaveprop/workers/era5_poll_worker.ex | 243 --------
 .../workers/era5_submit_worker.ex             | 241 --------
 .../live/contact_live/show.ex                 |   7 +-
 lib/mix/tasks/era5_backfill.ex                |  82 ---
 .../weather/era5_batch_client_test.exs        |  93 ---
 .../weather/era5_client_test.exs              | 225 --------
 .../workers/era5_fetch_worker_test.exs        | 100 ----
 .../workers/era5_month_batch_worker_test.exs  |  49 --
 .../workers/era5_poll_worker_test.exs         | 337 -----------
 .../workers/era5_submit_worker_test.exs       | 258 ---------
 17 files changed, 18 insertions(+), 2693 deletions(-)
 delete mode 100644 lib/microwaveprop/weather/era5_batch_client.ex
 delete mode 100644 lib/microwaveprop/weather/era5_client.ex
 delete mode 100644 lib/microwaveprop/workers/era5_fetch_worker.ex
 delete mode 100644 lib/microwaveprop/workers/era5_month_batch_worker.ex
 delete mode 100644 lib/microwaveprop/workers/era5_poll_worker.ex
 delete mode 100644 lib/microwaveprop/workers/era5_submit_worker.ex
 delete mode 100644 lib/mix/tasks/era5_backfill.ex
 delete mode 100644 test/microwaveprop/weather/era5_batch_client_test.exs
 delete mode 100644 test/microwaveprop/weather/era5_client_test.exs
 delete mode 100644 test/microwaveprop/workers/era5_fetch_worker_test.exs
 delete mode 100644 test/microwaveprop/workers/era5_month_batch_worker_test.exs
 delete mode 100644 test/microwaveprop/workers/era5_poll_worker_test.exs
 delete mode 100644 test/microwaveprop/workers/era5_submit_worker_test.exs

diff --git a/config/runtime.exs b/config/runtime.exs
index 2dd013d3..48624822 100644
--- a/config/runtime.exs
+++ b/config/runtime.exs
@@ -168,24 +168,13 @@ if config_env() == :prod do
       hrrr: 1,
       terrain: 3,
       iemre: 3,
+      # Historical backfill for pre-2014 contacts (pre-HRRR archive).
+      # NARR is fetched anonymously from NCEI with no quota, no job
+      # queue. Replaced the ERA5/CDS pipeline which never produced a
+      # single row in prod. era5_cds_jobs table is kept for inspection
+      # but its workers/client are gone. See
+      # docs/plans/2026-04-15-merra2-historical-backfill.md.
       narr: 6,
-      # ERA5 queues are DEPRECATED in favor of NARR via NarrFetchWorker
-      # (see docs/plans/2026-04-15-merra2-historical-backfill.md). The
-      # CDS pipeline never produced a single era5_profiles row in prod
-      # — every poll hit either CDS-rejected or long-running stuck
-      # states. NARR fetches anonymously from NCEI with no quota, no
-      # job queue, and the same era5_profiles table on the receiving
-      # side. These queues stay paused while the dead Era5* workers
-      # and the era5_cds_jobs table are slated for deletion in a
-      # follow-up PR.
-      era5: [local_limit: 2, paused: true],
-      era5_submit: [
-        local_limit: 4,
-        rate_limit: [allowed: 30, period: {1, :hour}],
-        paused: true
-      ],
-      era5_poll: [local_limit: 20, paused: true],
-      era5_batch: [local_limit: 1, paused: true],
       rtma: 2,
       backfill_enqueue: 1,
       admin: 1,
diff --git a/config/test.exs b/config/test.exs
index 90aee311..7bcd4612 100644
--- a/config/test.exs
+++ b/config/test.exs
@@ -58,7 +58,6 @@ config :microwaveprop,
 config :microwaveprop, cache_contact_count: false
 config :microwaveprop, cache_contact_map: false
 config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Terrain.ElevationClient}, retry: false]
-config :microwaveprop, era5_req_options: [plug: {Req.Test, Microwaveprop.Weather.Era5Client}, retry: false]
 config :microwaveprop, giro_req_options: [plug: {Req.Test, Microwaveprop.Ionosphere.GiroClient}, retry: false]
 config :microwaveprop, hrrr_req_options: [plug: {Req.Test, Microwaveprop.Weather.HrrrClient}, retry: false]
 config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false]
diff --git a/lib/microwaveprop/weather/era5_batch_client.ex b/lib/microwaveprop/weather/era5_batch_client.ex
deleted file mode 100644
index 2bd81be4..00000000
--- a/lib/microwaveprop/weather/era5_batch_client.ex
+++ /dev/null
@@ -1,376 +0,0 @@
-defmodule Microwaveprop.Weather.Era5BatchClient do
-  @moduledoc """
-  Fetches ERA5 reanalysis data from Copernicus Climate Data Store (CDS) one
-  month-tile at a time and bulk-inserts the resulting profiles into
-  `era5_profiles`.
-
-  The per-point `Era5Client.fetch_profile/3` path is tragically slow because
-  every point-hour triggers its own asynchronous CDS job (submit → poll →
-  assemble → download). For backfill this balloons into thousands of
-  independent CDS jobs. This module instead groups requests by month and a
-  small lat/lon tile (2° × 2° by default) so one submit/poll/download cycle
-  produces many thousands of profile rows.
-
-  Idempotency: `fetch_month_into_db/1` is a no-op when the requested
-  (year, month, tile) already has profile rows in `era5_profiles`. Combined
-  with Oban uniqueness on `Era5MonthBatchWorker`, redundant fetches are
-  collapsed.
-  """
-
-  import Ecto.Query
-
-  alias Microwaveprop.Repo
-  alias Microwaveprop.Weather.Era5Client
-  alias Microwaveprop.Weather.Era5Profile
-  alias Microwaveprop.Weather.Grib2.Wgrib2
-  alias Microwaveprop.Weather.SoundingParams
-  alias Microwaveprop.Weather.ThetaE
-
-  require Logger
-
-  # ERA5 hourly reanalysis runs on a 0.25° global grid. Tiles are aligned
-  # to an even-degree lattice (2° × 2°), producing 9 × 9 = 81 grid points
-  # per tile per hour. A month-tile expands to ~60k profiles.
-  @tile_degrees 2
-  @era5_step 0.25
-
-  @doc """
-  Returns the integer tile coordinate `{tile_lat, tile_lon}` for a point,
-  where `tile_lat` is the south edge and `tile_lon` is the west edge.
-  """
-  @spec tile_for_point(float(), float()) :: {integer(), integer()}
-  def tile_for_point(lat, lon) do
-    {floor_to_tile(lat), floor_to_tile(lon)}
-  end
-
-  defp floor_to_tile(value) do
-    value
-    |> Kernel./(@tile_degrees)
-    |> Float.floor()
-    |> trunc()
-    |> Kernel.*(@tile_degrees)
-  end
-
-  @doc false
-  @spec tile_degrees() :: pos_integer()
-  def tile_degrees, do: @tile_degrees
-
-  @doc """
-  Fetches one month of ERA5 data covering the given tile and stores every
-  resulting hourly profile in `era5_profiles`. Returns
-  `{:ok, inserted_count}` or `{:error, reason}`.
-  """
-  @spec fetch_month_into_db(%{
-          year: pos_integer(),
-          month: pos_integer(),
-          tile_lat: integer(),
-          tile_lon: integer()
-        }) :: {:ok, non_neg_integer()} | {:error, term()}
-  def fetch_month_into_db(%{year: year, month: month, tile_lat: tile_lat, tile_lon: tile_lon}) do
-    if month_tile_cached?(year, month, tile_lat, tile_lon) do
-      Logger.info("ERA5Batch: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already cached")
-      {:ok, 0}
-    else
-      do_fetch_month(year, month, tile_lat, tile_lon)
-    end
-  end
-
-  defp do_fetch_month(year, month, tile_lat, tile_lon) do
-    area = tile_area(tile_lat, tile_lon)
-    days = days_in_month(year, month)
-
-    Logger.info("ERA5Batch: fetching #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} (#{days} days)")
-
-    tmp_dir = System.tmp_dir!()
-    job_tag = "#{year}_#{pad(month)}_#{tile_lat}_#{tile_lon}_#{System.unique_integer([:positive])}"
-    single_path = Path.join(tmp_dir, "era5_single_#{job_tag}.grib2")
-    pressure_path = Path.join(tmp_dir, "era5_pressure_#{job_tag}.grib2")
-
-    # Fire both CDS submits in parallel — they have no ordering dependency and
-    # each spends 30+ min blocking on CDS queue/download. Running them serially
-    # doubles wall time for zero benefit. Task.async links to the Oban worker,
-    # so if the worker exits (deploy, crash) both tasks die with it. We still
-    # shut down any unfinished task on early error so we don't waste CDS quota.
-    # Both downloads stream directly to disk (Req `into: File.stream!`), so
-    # peak heap stays O(chunk) rather than O(month-tile GRIB size ~50-200 MB).
-    single_task =
-      Task.async(fn -> fetch_single_level_month(year, month, days, area, single_path) end)
-
-    pressure_task =
-      Task.async(fn -> fetch_pressure_level_month(year, month, days, area, pressure_path) end)
-
-    try do
-      with :ok <- Task.await(single_task, :infinity),
-           :ok <- Task.await(pressure_task, :infinity) do
-        decode_and_insert(year, month, tile_lat, tile_lon, single_path, pressure_path)
-      end
-    after
-      Task.shutdown(single_task, :brutal_kill)
-      Task.shutdown(pressure_task, :brutal_kill)
-      File.rm(single_path)
-      File.rm(pressure_path)
-    end
-  end
-
-  @doc """
-  Decodes a downloaded single-level + pressure-level GRIB2 pair from disk
-  and bulk-inserts the resulting month-tile profiles into `era5_profiles`.
-
-  Shared between the legacy inline `do_fetch_month/4` path and the split
-  `Era5PollWorker` path.
-  """
-  @spec decode_and_insert(
-          pos_integer(),
-          pos_integer(),
-          integer(),
-          integer(),
-          Path.t(),
-          Path.t()
-        ) :: {:ok, non_neg_integer()} | {:error, term()}
-  def decode_and_insert(year, month, tile_lat, tile_lon, single_path, pressure_path) do
-    with {:ok, profiles} <- build_month_profiles(tile_lat, tile_lon, single_path, pressure_path) do
-      inserted = bulk_insert_profiles(profiles)
-
-      Logger.info("ERA5Batch: stored #{inserted} profiles for #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}")
-
-      {:ok, inserted}
-    end
-  end
-
-  defp month_tile_cached?(year, month, tile_lat, tile_lon) do
-    {first, last} = month_bounds(year, month)
-
-    Era5Profile
-    |> where(
-      [p],
-      p.valid_time >= ^first and p.valid_time < ^last and
-        p.lat >= ^(tile_lat * 1.0) and p.lat <= ^(tile_lat * 1.0 + @tile_degrees) and
-        p.lon >= ^(tile_lon * 1.0) and p.lon <= ^(tile_lon * 1.0 + @tile_degrees)
-    )
-    |> Repo.exists?()
-  end
-
-  defp month_bounds(year, month) do
-    {:ok, first} = Date.new(year, month, 1)
-    {:ok, first_dt} = DateTime.new(first, ~T[00:00:00], "Etc/UTC")
-    last_dt = DateTime.add(first_dt, days_in_month(year, month) * 86_400, :second)
-    {first_dt, last_dt}
-  end
-
-  defp days_in_month(year, month), do: Calendar.ISO.days_in_month(year, month)
-
-  defp tile_area(tile_lat, tile_lon) do
-    # CDS area is [north, west, south, east]
-    Enum.map([tile_lat + @tile_degrees, tile_lon, tile_lat, tile_lon + @tile_degrees], &(&1 * 1.0))
-  end
-
-  defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
-
-  # -- CDS requests ----------------------------------------------------------
-
-  defp fetch_single_level_month(year, month, days, area, dest_path) do
-    request = %{
-      "product_type" => ["reanalysis"],
-      "variable" => Era5Client.single_level_vars(),
-      "year" => [Integer.to_string(year)],
-      "month" => [pad(month)],
-      "day" => Enum.map(1..days, &pad/1),
-      "time" => Enum.map(0..23, &"#{pad(&1)}:00"),
-      "area" => area,
-      "data_format" => "grib"
-    }
-
-    Era5Client.submit_and_download_to_file("reanalysis-era5-single-levels", request, dest_path)
-  end
-
-  defp fetch_pressure_level_month(year, month, days, area, dest_path) do
-    request = %{
-      "product_type" => ["reanalysis"],
-      "variable" => Era5Client.pressure_level_vars(),
-      "pressure_level" => Era5Client.pressure_levels(),
-      "year" => [Integer.to_string(year)],
-      "month" => [pad(month)],
-      "day" => Enum.map(1..days, &pad/1),
-      "time" => Enum.map(0..23, &"#{pad(&1)}:00"),
-      "area" => area,
-      "data_format" => "grib"
-    }
-
-    Era5Client.submit_and_download_to_file("reanalysis-era5-pressure-levels", request, dest_path)
-  end
-
-  # -- GRIB extraction -------------------------------------------------------
-
-  defp grid_spec_for_tile(tile_lat, tile_lon) do
-    count = round(@tile_degrees / @era5_step) + 1
-
-    %{
-      lon_start: tile_lon * 1.0,
-      lon_count: count,
-      lon_step: @era5_step,
-      lat_start: tile_lat * 1.0,
-      lat_count: count,
-      lat_step: @era5_step
-    }
-  end
-
-  defp build_month_profiles(tile_lat, tile_lon, single_path, pressure_path) do
-    grid_spec = grid_spec_for_tile(tile_lat, tile_lon)
-
-    with {:ok, single_msgs} <-
-           Wgrib2.extract_grid_messages_from_file(
-             single_path,
-             ":(TMP|DPT|PRES|UGRD|VGRD|PWAT|HPBL):",
-             grid_spec
-           ),
-         {:ok, pressure_msgs} <-
-           Wgrib2.extract_grid_messages_from_file(
-             pressure_path,
-             ":(TMP|SPFH|HGT):",
-             grid_spec
-           ) do
-      # Index both message sets by (datetime, lat, lon) → var:level map.
-      points_single = index_messages(single_msgs)
-      points_pressure = index_messages(pressure_msgs)
-
-      all_keys =
-        MapSet.union(
-          MapSet.new(Map.keys(points_single)),
-          MapSet.new(Map.keys(points_pressure))
-        )
-
-      profiles =
-        all_keys
-        |> Enum.map(fn {dt, lat, lon} ->
-          single_fields = Map.get(points_single, {dt, lat, lon}, %{})
-          pressure_fields = Map.get(points_pressure, {dt, lat, lon}, %{})
-          build_profile_attrs(lat, lon, dt, single_fields, pressure_fields)
-        end)
-        |> Enum.reject(&is_nil/1)
-
-      {:ok, profiles}
-    end
-  end
-
-  defp index_messages(messages) do
-    Enum.reduce(messages, %{}, fn msg, acc ->
-      key_name = "#{msg.var}:#{msg.level}"
-
-      Enum.reduce(msg.values, acc, fn {{lat, lon}, value}, inner ->
-        Map.update(
-          inner,
-          {msg.datetime, lat, lon},
-          %{key_name => value},
-          &Map.put(&1, key_name, value)
-        )
-      end)
-    end)
-  end
-
-  defp build_profile_attrs(_lat, _lon, nil, _single, _pressure), do: nil
-
-  defp build_profile_attrs(lat, lon, datetime, single_fields, pressure_fields) do
-    temp_c = kelvin_to_celsius(single_fields["TMP:2 m above ground"])
-    dewpoint_c = kelvin_to_celsius(single_fields["DPT:2 m above ground"])
-    pressure_mb = pa_to_mb(single_fields["PRES:surface"])
-    hpbl_m = single_fields["HPBL:surface"]
-    pwat_mm = single_fields["PWAT:entire atmosphere"]
-
-    profile =
-      Era5Client.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_fields["TMP:#{pres_key}"]),
-          "dwpc" =>
-            dewpoint_from_spfh(
-              pressure_fields["SPFH:#{pres_key}"],
-              level * 100.0
-            ),
-          "hght" => geopotential_to_height(pressure_fields["HGT:#{pres_key}"])
-        }
-      end)
-      |> Enum.reject(&is_nil(&1["tmpc"]))
-
-    if profile == [] and is_nil(temp_c) do
-      nil
-    else
-      params = SoundingParams.derive(profile)
-
-      base = %{
-        valid_time: datetime,
-        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
-      }
-
-      if params do
-        Map.merge(base, %{
-          surface_refractivity: params.surface_refractivity,
-          min_refractivity_gradient: params.min_refractivity_gradient,
-          ducting_detected: params.ducting_detected || false,
-          duct_characteristics: params.duct_characteristics
-        })
-      else
-        base
-      end
-    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
-
-  # ERA5 pressure-level moisture is specific humidity (kg/kg). Convert to
-  # dewpoint in °C via the Magnus-Tetens inverse already used for HRRR native
-  # profiles. `level_pa` is the pressure level itself in Pa.
-  # ThetaE.dewpoint_from_spfh/2 returns Kelvin. Profile `dwpc` entries
-  # are Celsius (the downstream SoundingParams.compute_refractivity_profile
-  # passes p["dwpc"] straight to the Buck saturation-vapor-pressure
-  # equation which expects °C). Convert here.
-  defp dewpoint_from_spfh(nil, _level_pa), do: nil
-  defp dewpoint_from_spfh(spfh, _level_pa) when spfh <= 0, do: nil
-  defp dewpoint_from_spfh(spfh, level_pa), do: ThetaE.dewpoint_from_spfh(spfh, level_pa) - 273.15
-
-  # -- Bulk insert -----------------------------------------------------------
-
-  # Insert in chunks to keep each query under Postgres' parameter limit
-  # (65,535 params / ~15 fields ≈ 4,000 rows per statement).
-  @chunk_size 4_000
-
-  defp bulk_insert_profiles(profiles) do
-    now = DateTime.truncate(DateTime.utc_now(), :second)
-
-    profiles
-    |> Enum.chunk_every(@chunk_size)
-    |> Enum.reduce(0, fn chunk, acc ->
-      rows =
-        Enum.map(chunk, fn attrs ->
-          attrs
-          |> Map.put(:inserted_at, now)
-          |> Map.put(:updated_at, now)
-        end)
-
-      {count, _} =
-        Repo.insert_all(Era5Profile, rows,
-          on_conflict: :nothing,
-          conflict_target: [:lat, :lon, :valid_time]
-        )
-
-      acc + count
-    end)
-  end
-end
diff --git a/lib/microwaveprop/weather/era5_client.ex b/lib/microwaveprop/weather/era5_client.ex
deleted file mode 100644
index 9431c25e..00000000
--- a/lib/microwaveprop/weather/era5_client.ex
+++ /dev/null
@@ -1,541 +0,0 @@
-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
diff --git a/lib/microwaveprop/weather/narr_client.ex b/lib/microwaveprop/weather/narr_client.ex
index 33822458..9cab24f7 100644
--- a/lib/microwaveprop/weather/narr_client.ex
+++ b/lib/microwaveprop/weather/narr_client.ex
@@ -147,9 +147,11 @@ defmodule Microwaveprop.Weather.NarrClient do
   Pure-ish: shells out to `cdo -outputtab,name,lev,value -remapnn,lon=…_lat=…`
   on the local file. No network.
 
-  Returns `{:ok, attrs}` where `attrs` has the same shape
-  `Era5BatchClient.build_profile_attrs/5` produces, or `{:error, reason}`
-  on missing file, cdo failure, or empty parser output.
+  Returns `{:ok, attrs}` — a map with the `era5_profiles` fields
+  (`:surface_temp_c`, `:surface_dewpoint_c`, `:surface_pressure_mb`,
+  `:hpbl_m`, `:pwat_mm`, `:profile` list plus the derived fields from
+  `SoundingParams.derive/1`) — or `{:error, reason}` on missing file,
+  cdo failure, or empty parser output.
   """
   @spec extract_profile_from_file(Path.t(), float(), float()) ::
           {:ok, map()} | {:error, term()}
@@ -167,8 +169,9 @@ defmodule Microwaveprop.Weather.NarrClient do
   GRIB1 with `cdo -merge`, then extracts the profile at `{lat, lon}`
   using `extract_profile_from_file/3`.
 
-  Returns `{:ok, profile_attrs}` shaped like `Era5BatchClient.build_profile_attrs/5`
-  or `{:error, reason}`. Temp files are always cleaned up via `try/after`.
+  Returns `{:ok, profile_attrs}` shaped for direct `Repo.insert_all/3`
+  into the `era5_profiles` table, or `{:error, reason}`. Temp files are
+  always cleaned up via `try/after`.
   """
   @spec fetch_profile_at(DateTime.t(), {float(), float()}) ::
           {:ok, map()} | {:error, term()}
diff --git a/lib/microwaveprop/workers/era5_fetch_worker.ex b/lib/microwaveprop/workers/era5_fetch_worker.ex
deleted file mode 100644
index 10846a35..00000000
--- a/lib/microwaveprop/workers/era5_fetch_worker.ex
+++ /dev/null
@@ -1,80 +0,0 @@
-defmodule Microwaveprop.Workers.Era5FetchWorker do
-  @moduledoc """
-  Thin router that ensures an ERA5 profile is available for a single
-  `{lat, lon, valid_time}`. On a cache miss it enqueues the much more
-  efficient `Era5MonthBatchWorker` for the month-tile that contains the
-  requested point, rather than hitting the CDS API point-by-point.
-
-  Jobs are unique on `{lat, lon, valid_time}` so duplicate enrichment
-  requests collapse. The enqueued month-batch worker is itself unique on
-  `(year, month, tile_lat, tile_lon)` so the first enqueue wins and later
-  ones are no-ops.
-  """
-  use Oban.Worker,
-    queue: :era5,
-    max_attempts: 5,
-    unique: [
-      period: :infinity,
-      states: [:available, :scheduled, :executing, :retryable]
-    ]
-
-  alias Microwaveprop.Repo
-  alias Microwaveprop.Weather.Era5BatchClient
-  alias Microwaveprop.Weather.Era5Profile
-  alias Microwaveprop.Workers.Era5MonthBatchWorker
-
-  require Logger
-
-  @impl Oban.Worker
-  def backoff(%Oban.Job{attempt: attempt}) do
-    # Enqueueing the batch is cheap; retry fast so we stop blocking downstream.
-    min(30 * Integer.pow(2, attempt - 1), _one_hour = 3_600)
-  end
-
-  @impl Oban.Worker
-  def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do
-    {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
-
-    rlat = Float.round(lat * 4) / 4
-    rlon = Float.round(lon * 4) / 4
-
-    if has_era5_profile?(rlat, rlon, valid_time) do
-      :ok
-    else
-      {tile_lat, tile_lon} = Era5BatchClient.tile_for_point(rlat, rlon)
-
-      Logger.info(
-        "ERA5: enqueueing month batch for #{valid_time.year}-#{valid_time.month} tile #{tile_lat},#{tile_lon} (triggered by #{rlat},#{rlon} @ #{valid_time_str})"
-      )
-
-      %{
-        year: valid_time.year,
-        month: valid_time.month,
-        tile_lat: tile_lat,
-        tile_lon: tile_lon
-      }
-      |> Era5MonthBatchWorker.new()
-      |> Oban.insert()
-
-      :ok
-    end
-  end
-
-  defp has_era5_profile?(lat, lon, valid_time) do
-    import Ecto.Query
-
-    dlat = 0.13
-    dlon = 0.13
-    time_start = DateTime.add(valid_time, -1800, :second)
-    time_end = DateTime.add(valid_time, 1800, :second)
-
-    Era5Profile
-    |> where(
-      [p],
-      p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and
-        p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and
-        p.valid_time >= ^time_start and p.valid_time <= ^time_end
-    )
-    |> Repo.exists?()
-  end
-end
diff --git a/lib/microwaveprop/workers/era5_month_batch_worker.ex b/lib/microwaveprop/workers/era5_month_batch_worker.ex
deleted file mode 100644
index c9a89ecb..00000000
--- a/lib/microwaveprop/workers/era5_month_batch_worker.ex
+++ /dev/null
@@ -1,42 +0,0 @@
-defmodule Microwaveprop.Workers.Era5MonthBatchWorker do
-  @moduledoc """
-  Legacy entry point for the ERA5 month-tile backfill. **Forwards** to
-  `Era5SubmitWorker` in the new split-worker pipeline — kept in place so
-  already-enqueued callers (`Era5FetchWorker`, `mix era5.backfill`, any
-  in-flight Oban rows) don't need to be changed on the same deploy.
-
-  On the next deploy cycle, callers will be updated to enqueue
-  `Era5SubmitWorker` directly and this module can be removed.
-  """
-  use Oban.Worker,
-    queue: :era5_submit,
-    max_attempts: 5,
-    unique: [
-      period: :infinity,
-      states: [:available, :scheduled, :executing, :retryable],
-      keys: [:year, :month, :tile_lat, :tile_lon]
-    ]
-
-  alias Microwaveprop.Workers.Era5SubmitWorker
-
-  require Logger
-
-  @impl Oban.Worker
-  def backoff(%Oban.Job{attempt: attempt}) do
-    # Forwarding is cheap; retry fast on transient failure.
-    min(30 * Integer.pow(2, attempt - 1), 3_600)
-  end
-
-  @impl Oban.Worker
-  def perform(%Oban.Job{args: args}) do
-    Logger.debug("Era5MonthBatchWorker forwarding to Era5SubmitWorker: #{inspect(args)}")
-
-    args
-    |> Era5SubmitWorker.new()
-    |> Oban.insert()
-    |> case do
-      {:ok, _job} -> :ok
-      {:error, reason} -> {:error, reason}
-    end
-  end
-end
diff --git a/lib/microwaveprop/workers/era5_poll_worker.ex b/lib/microwaveprop/workers/era5_poll_worker.ex
deleted file mode 100644
index d6eccc6c..00000000
--- a/lib/microwaveprop/workers/era5_poll_worker.ex
+++ /dev/null
@@ -1,243 +0,0 @@
-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
-
-  # Snooze delay after a transient CDS HTTP error (5xx, network blip).
-  # Shorter than the normal poll interval so recovery is fast, but a
-  # snooze instead of an attempt so a sustained outage doesn't burn
-  # through max_attempts and orphan the row.
-  @error_snooze_seconds 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. Tightened from 18h to 8h while diagnosing the
-  # zero-ingestion problem: 18h was letting rows poll for ~13h without
-  # ever producing a profile, 8h recycles stuck rows within a work
-  # shift while still staying above the ~90 min normal pressure-level
-  # runtime.
-  @stuck_after_seconds 8 * 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}, _} ->
-        Logger.warning("Era5Poll: transient single-level check_status error (snoozing): #{reason}")
-        {:snooze, @error_snooze_seconds}
-
-      {_, {:error, reason}} ->
-        Logger.warning("Era5Poll: transient pressure-level check_status error (snoozing): #{reason}")
-        {:snooze, @error_snooze_seconds}
-
-      _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
diff --git a/lib/microwaveprop/workers/era5_submit_worker.ex b/lib/microwaveprop/workers/era5_submit_worker.ex
deleted file mode 100644
index 0e75e720..00000000
--- a/lib/microwaveprop/workers/era5_submit_worker.ex
+++ /dev/null
@@ -1,241 +0,0 @@
-defmodule Microwaveprop.Workers.Era5SubmitWorker do
-  @moduledoc """
-  First half of the split ERA5 backfill pipeline.
-
-  Submits the two CDS jobs for a month-tile (single-level + pressure-level)
-  in parallel, persists the returned CDS job IDs to `era5_cds_jobs`, and
-  enqueues an `Era5PollWorker` job scheduled a few minutes out to check for
-  completion.
-
-  The fast submit → enqueue-poll split means an Oban slot is held for ~1
-  second (the time to POST and write a row), not the 30-45 minutes CDS
-  takes to queue, assemble, and return a month-tile. A pod restart between
-  submit and poll does not lose the CDS work: the row in `era5_cds_jobs`
-  persists the job IDs and the poll worker picks up where we left off.
-
-  Idempotent: if the month-tile already has profiles (`era5_profiles`) or
-  an in-flight `era5_cds_jobs` row, no submit is made.
-  """
-  use Oban.Worker,
-    queue: :era5_submit,
-    max_attempts: 5,
-    unique: [
-      period: :infinity,
-      states: [:available, :scheduled, :executing, :retryable],
-      keys: [:year, :month, :tile_lat, :tile_lon]
-    ]
-
-  import Ecto.Query
-
-  alias Microwaveprop.Repo
-  alias Microwaveprop.Weather.Era5BatchClient
-  alias Microwaveprop.Weather.Era5CdsJob
-  alias Microwaveprop.Weather.Era5Client
-  alias Microwaveprop.Weather.Era5Profile
-  alias Microwaveprop.Workers.Era5PollWorker
-
-  require Logger
-
-  @poll_delay_seconds 5 * 60
-
-  # CDS enforces a server-side per-user cap of 150 in-flight jobs. Each
-  # tile-month submit lives in ONE era5_cds_jobs row but carries TWO CDS
-  # job IDs (single-level + pressure-level), so the real in-flight count
-  # CDS sees is 2 × row_count. The cap guard multiplies accordingly.
-  # Snoozing (instead of failing or sleeping) releases the Oban slot
-  # immediately and lets the poll workers drain in-flight jobs before we
-  # push more in.
-  #
-  # Headroom is deliberately generous (30 job slots, effective ceiling
-  # 120 jobs = 60 rows) because the cap check races between concurrent
-  # workers: two workers that both observe row_count=59 can both submit,
-  # landing us at row_count=61 (122 jobs) after both finish. With 30
-  # slots of headroom the cluster would have to race 15 workers wide to
-  # clip CDS's hard 150 ceiling. Observed in prod: the old 1× math was
-  # allowing 74 rows (148 jobs) which triggered CDS reject storms.
-  @cds_ceiling 150
-  @cap_headroom 30
-  @snooze_seconds 5 * 60
-
-  @impl Oban.Worker
-  def backoff(%Oban.Job{attempt: attempt}) do
-    # The submit itself is cheap; retry fast so transient CDS/HTTP blips
-    # don't stall the tile for an hour waiting on exponential backoff.
-    min(30 * Integer.pow(2, attempt - 1), _one_hour = 3_600)
-  end
-
-  @impl Oban.Worker
-  def perform(%Oban.Job{args: args}) do
-    year = fetch_int!(args, "year")
-    month = fetch_int!(args, "month")
-    tile_lat = fetch_int!(args, "tile_lat")
-    tile_lon = fetch_int!(args, "tile_lon")
-
-    cond do
-      month_tile_cached?(year, month, tile_lat, tile_lon) ->
-        Logger.info("Era5Submit: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already cached — skip")
-
-        {:ok, :cached}
-
-      existing = find_in_flight(year, month, tile_lat, tile_lon) ->
-        Logger.info(
-          "Era5Submit: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already submitted (cds_job_id=#{existing.id}) — skip"
-        )
-
-        {:ok, :already_submitted}
-
-      cds_queue_full?() ->
-        row_count = Repo.aggregate(Era5CdsJob, :count, :id)
-        in_flight_jobs = row_count * 2
-
-        Logger.info(
-          "Era5Submit: CDS in-flight cap reached (#{in_flight_jobs}/#{@cds_ceiling} jobs across #{row_count} rows) — snoozing #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}"
-        )
-
-        {:snooze, @snooze_seconds}
-
-      true ->
-        submit_and_persist(year, month, tile_lat, tile_lon)
-    end
-  end
-
-  defp cds_queue_full? do
-    in_flight_jobs_after_submit = Repo.aggregate(Era5CdsJob, :count, :id) * 2 + 2
-    in_flight_jobs_after_submit > @cds_ceiling - @cap_headroom
-  end
-
-  defp submit_and_persist(year, month, tile_lat, tile_lon) do
-    days = Calendar.ISO.days_in_month(year, month)
-    area = tile_area(tile_lat, tile_lon)
-
-    # Run both submits in parallel — they're independent HTTP POSTs.
-    single_task =
-      Task.async(fn ->
-        Era5Client.submit_job("reanalysis-era5-single-levels", single_request(year, month, days, area))
-      end)
-
-    pressure_task =
-      Task.async(fn ->
-        Era5Client.submit_job("reanalysis-era5-pressure-levels", pressure_request(year, month, days, area))
-      end)
-
-    try do
-      with {:ok, single_id} <- Task.await(single_task, :infinity),
-           {:ok, pressure_id} <- Task.await(pressure_task, :infinity),
-           {:ok, row} <-
-             insert_cds_job_row(year, month, tile_lat, tile_lon, single_id, pressure_id) do
-        enqueue_poll(row)
-
-        Logger.info(
-          "Era5Submit: submitted #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} (single=#{single_id}, pressure=#{pressure_id})"
-        )
-
-        {:ok, row}
-      end
-    after
-      Task.shutdown(single_task, :brutal_kill)
-      Task.shutdown(pressure_task, :brutal_kill)
-    end
-  end
-
-  defp insert_cds_job_row(year, month, tile_lat, tile_lon, single_id, pressure_id) do
-    %Era5CdsJob{}
-    |> Era5CdsJob.changeset(%{
-      year: year,
-      month: month,
-      tile_lat: tile_lat,
-      tile_lon: tile_lon,
-      single_job_id: single_id,
-      pressure_job_id: pressure_id,
-      submitted_at: DateTime.truncate(DateTime.utc_now(), :second)
-    })
-    |> Repo.insert()
-    |> case do
-      {:ok, row} -> {:ok, row}
-      {:error, changeset} -> {:error, "era5_cds_jobs insert failed: #{inspect(changeset.errors)}"}
-    end
-  end
-
-  defp enqueue_poll(%Era5CdsJob{id: id}) do
-    %{"era5_cds_job_id" => id}
-    |> Era5PollWorker.new(schedule_in: @poll_delay_seconds)
-    |> Oban.insert()
-  end
-
-  defp find_in_flight(year, month, tile_lat, tile_lon) do
-    Repo.one(
-      from j in Era5CdsJob,
-        where:
-          j.year == ^year and j.month == ^month and j.tile_lat == ^tile_lat and
-            j.tile_lon == ^tile_lon,
-        limit: 1
-    )
-  end
-
-  defp month_tile_cached?(year, month, tile_lat, tile_lon) do
-    {first_dt, last_dt} = month_bounds(year, month)
-    tile_degrees = Era5BatchClient.tile_degrees()
-
-    Era5Profile
-    |> where(
-      [p],
-      p.valid_time >= ^first_dt and p.valid_time < ^last_dt and
-        p.lat >= ^(tile_lat * 1.0) and p.lat <= ^(tile_lat * 1.0 + tile_degrees) and
-        p.lon >= ^(tile_lon * 1.0) and p.lon <= ^(tile_lon * 1.0 + tile_degrees)
-    )
-    |> Repo.exists?()
-  end
-
-  defp month_bounds(year, month) do
-    {:ok, first} = Date.new(year, month, 1)
-    {:ok, first_dt} = DateTime.new(first, ~T[00:00:00], "Etc/UTC")
-    days = Calendar.ISO.days_in_month(year, month)
-    last_dt = DateTime.add(first_dt, days * 86_400, :second)
-    {first_dt, last_dt}
-  end
-
-  defp tile_area(tile_lat, tile_lon) do
-    tile_degrees = Era5BatchClient.tile_degrees()
-
-    Enum.map(
-      [tile_lat + tile_degrees, tile_lon, tile_lat, tile_lon + tile_degrees],
-      &(&1 * 1.0)
-    )
-  end
-
-  defp single_request(year, month, days, area) do
-    %{
-      "product_type" => ["reanalysis"],
-      "variable" => Era5Client.single_level_vars(),
-      "year" => [Integer.to_string(year)],
-      "month" => [pad(month)],
-      "day" => Enum.map(1..days, &pad/1),
-      "time" => Enum.map(0..23, &"#{pad(&1)}:00"),
-      "area" => area,
-      "data_format" => "grib"
-    }
-  end
-
-  defp pressure_request(year, month, days, area) do
-    %{
-      "product_type" => ["reanalysis"],
-      "variable" => Era5Client.pressure_level_vars(),
-      "pressure_level" => Era5Client.pressure_levels(),
-      "year" => [Integer.to_string(year)],
-      "month" => [pad(month)],
-      "day" => Enum.map(1..days, &pad/1),
-      "time" => Enum.map(0..23, &"#{pad(&1)}:00"),
-      "area" => area,
-      "data_format" => "grib"
-    }
-  end
-
-  defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
-
-  defp fetch_int!(args, key) do
-    case Map.fetch!(args, key) do
-      v when is_integer(v) -> v
-      v when is_binary(v) -> String.to_integer(v)
-    end
-  end
-end
diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex
index 43848277..24e84e51 100644
--- a/lib/microwaveprop_web/live/contact_live/show.ex
+++ b/lib/microwaveprop_web/live/contact_live/show.ex
@@ -13,9 +13,10 @@ defmodule MicrowavepropWeb.ContactLive.Show do
   alias Microwaveprop.Weather
   alias Microwaveprop.Weather.HrrrClient
   alias Microwaveprop.Weather.HrrrNativeProfile
+  alias Microwaveprop.Weather.NarrClient
   alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
-  alias Microwaveprop.Workers.Era5FetchWorker
   alias Microwaveprop.Workers.HrrrFetchWorker
+  alias Microwaveprop.Workers.NarrFetchWorker
   alias Microwaveprop.Workers.SolarIndexWorker
   alias Microwaveprop.Workers.TerrainProfileWorker
 
@@ -671,14 +672,14 @@ defmodule MicrowavepropWeb.ContactLive.Show do
     lon = contact.pos1 && (contact.pos1["lon"] || contact.pos1["lng"])
 
     if lat && lon do
-      valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0}
+      valid_time = NarrClient.snap_to_analysis_hour(contact.qso_timestamp)
 
       %{
         "lat" => lat,
         "lon" => lon,
         "valid_time" => DateTime.to_iso8601(valid_time)
       }
-      |> Era5FetchWorker.new()
+      |> NarrFetchWorker.new()
       |> Oban.insert()
     end
 
diff --git a/lib/mix/tasks/era5_backfill.ex b/lib/mix/tasks/era5_backfill.ex
deleted file mode 100644
index 0b220e7a..00000000
--- a/lib/mix/tasks/era5_backfill.ex
+++ /dev/null
@@ -1,82 +0,0 @@
-defmodule Mix.Tasks.Era5Backfill do
-  @shortdoc "Enqueue ERA5 month-batch jobs for contacts without HRRR data"
-  @moduledoc """
-  Finds contacts where HRRR status is `:unavailable` (pre-2014 or missing
-  from the archive) and enqueues the month-tile batch worker for every
-  tile-month their path points need.
-
-  Usage:
-      mix era5_backfill              # scan up to 1000 contacts
-      mix era5_backfill 5000         # scan up to 5000 contacts
-
-  Enqueues `Era5MonthBatchWorker` jobs deduplicated by
-  `(year, month, tile_lat, tile_lon)`. The per-point `Era5FetchWorker`
-  path still works for one-off enrichment, but this task goes straight
-  to the batch worker so backfills don't pay the per-point enqueue cost.
-  """
-  use Mix.Task
-
-  import Ecto.Query
-
-  alias Microwaveprop.Radio
-  alias Microwaveprop.Radio.Contact
-  alias Microwaveprop.Repo
-  alias Microwaveprop.Weather.Era5BatchClient
-  alias Microwaveprop.Workers.Era5MonthBatchWorker
-
-  @impl Mix.Task
-  def run(args) do
-    Mix.Task.run("app.start")
-    Oban.pause_all_queues(Oban)
-
-    limit =
-      case args do
-        [n | _] -> String.to_integer(n)
-        _ -> 1000
-      end
-
-    contacts =
-      Contact
-      |> where([c], c.hrrr_status == :unavailable and not is_nil(c.pos1))
-      |> order_by([c], asc: c.qso_timestamp)
-      |> limit(^limit)
-      |> Repo.all()
-
-    Mix.shell().info("Found #{length(contacts)} contacts needing ERA5 data")
-
-    batches =
-      contacts
-      |> Enum.flat_map(fn contact ->
-        valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0}
-
-        contact
-        |> Radio.contact_path_points()
-        |> Enum.map(fn {lat, lon} ->
-          {tile_lat, tile_lon} = Era5BatchClient.tile_for_point(lat, lon)
-
-          %{
-            year: valid_time.year,
-            month: valid_time.month,
-            tile_lat: tile_lat,
-            tile_lon: tile_lon
-          }
-        end)
-      end)
-      |> Enum.uniq()
-
-    if batches == [] do
-      Mix.shell().info("No ERA5 batches to enqueue")
-    else
-      # Use Oban.insert/1 (not insert_all) so the worker's unique
-      # constraint collapses duplicate tile-month jobs against anything
-      # already queued.
-      Enum.each(batches, fn args ->
-        args
-        |> Era5MonthBatchWorker.new()
-        |> Oban.insert()
-      end)
-
-      Mix.shell().info("Enqueued #{length(batches)} ERA5 month-tile batches")
-    end
-  end
-end
diff --git a/test/microwaveprop/weather/era5_batch_client_test.exs b/test/microwaveprop/weather/era5_batch_client_test.exs
deleted file mode 100644
index 47f0a75c..00000000
--- a/test/microwaveprop/weather/era5_batch_client_test.exs
+++ /dev/null
@@ -1,93 +0,0 @@
-defmodule Microwaveprop.Weather.Era5BatchClientTest do
-  use Microwaveprop.DataCase, async: true
-
-  alias Microwaveprop.Weather.Era5BatchClient
-  alias Microwaveprop.Weather.Era5Profile
-
-  describe "tile_for_point/2" do
-    test "floors positive coordinates to a 2° lattice" do
-      assert {32, 96} = Era5BatchClient.tile_for_point(32.75, 97.5)
-      assert {32, 96} = Era5BatchClient.tile_for_point(33.99, 97.99)
-      assert {34, 98} = Era5BatchClient.tile_for_point(34.0, 98.0)
-    end
-
-    test "floors negative longitudes toward -infinity" do
-      assert {32, -98} = Era5BatchClient.tile_for_point(32.75, -97.25)
-      assert {32, -98} = Era5BatchClient.tile_for_point(33.0, -97.01)
-      assert {32, -100} = Era5BatchClient.tile_for_point(33.0, -99.0)
-    end
-
-    test "floors negative latitudes toward -infinity" do
-      assert {-2, 0} = Era5BatchClient.tile_for_point(-1.25, 0.5)
-      assert {-34, -60} = Era5BatchClient.tile_for_point(-33.5, -59.5)
-    end
-
-    test "edges land on the tile to the right/north" do
-      # 30.0 exactly → tile 30 (south/west edge)
-      assert {30, 96} = Era5BatchClient.tile_for_point(30.0, 97.5)
-    end
-  end
-
-  describe "fetch_month_into_db/1 idempotency" do
-    # This describe block exercises the uncached path which would otherwise
-    # hit the real CDS API if the developer has ERA5_CDS_API_KEY set via
-    # direnv. Clear the env var for these tests so the expected "API key not
-    # configured" short-circuit fires and the tests stay hermetic.
-    setup do
-      previous = System.get_env("ERA5_CDS_API_KEY")
-      System.delete_env("ERA5_CDS_API_KEY")
-
-      on_exit(fn ->
-        if previous, do: System.put_env("ERA5_CDS_API_KEY", previous)
-      end)
-
-      :ok
-    end
-
-    test "returns {:ok, 0} when the month-tile already has any profile" do
-      # Seed one profile inside the 2014-03, tile (32, -98) window.
-      %Era5Profile{}
-      |> Era5Profile.changeset(%{
-        lat: 32.5,
-        lon: -97.25,
-        valid_time: ~U[2014-03-15 12:00:00Z],
-        profile: []
-      })
-      |> Repo.insert!()
-
-      # No CDS API key configured in tests → if the cache check fails, the
-      # function would return an ERA5_CDS_API_KEY error. {:ok, 0} means the
-      # short-circuit path was taken.
-      assert {:ok, 0} =
-               Era5BatchClient.fetch_month_into_db(%{
-                 year: 2014,
-                 month: 3,
-                 tile_lat: 32,
-                 tile_lon: -98
-               })
-    end
-
-    test "a profile outside the tile window does not count as cached" do
-      # This profile is in tile (30, -100), not (32, -98).
-      %Era5Profile{}
-      |> Era5Profile.changeset(%{
-        lat: 30.5,
-        lon: -99.25,
-        valid_time: ~U[2014-03-15 12:00:00Z],
-        profile: []
-      })
-      |> Repo.insert!()
-
-      # The (32, -98) tile is still uncached, so fetch_month_into_db tries
-      # to reach CDS and fails with missing API key, which bubbles up
-      # (rather than :ok, 0).
-      assert {:error, _} =
-               Era5BatchClient.fetch_month_into_db(%{
-                 year: 2014,
-                 month: 3,
-                 tile_lat: 32,
-                 tile_lon: -98
-               })
-    end
-  end
-end
diff --git a/test/microwaveprop/weather/era5_client_test.exs b/test/microwaveprop/weather/era5_client_test.exs
deleted file mode 100644
index c1a61a24..00000000
--- a/test/microwaveprop/weather/era5_client_test.exs
+++ /dev/null
@@ -1,225 +0,0 @@
-defmodule Microwaveprop.Weather.Era5ClientTest do
-  # async: false — tests mutate the global ERA5_CDS_API_KEY env var to match
-  # production's lookup path. Running concurrently with Era5BatchClientTest
-  # (which *deletes* the env var to exercise the missing-key error path)
-  # leaks state in both directions.
-  use ExUnit.Case, async: false
-
-  alias Microwaveprop.Weather.Era5Client
-
-  setup do
-    # A real key — stubs don't care about value, only presence.
-    System.put_env("ERA5_CDS_API_KEY", "test-key")
-    on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
-    :ok
-  end
-
-  describe "submit_job/2" do
-    test "returns {:ok, job_id} when CDS accepts the request" do
-      Req.Test.stub(Era5Client, fn conn ->
-        assert conn.method == "POST"
-        assert String.ends_with?(conn.request_path, "/reanalysis-era5-single-levels/execution")
-        Req.Test.json(conn, %{"jobID" => "cds-job-123", "status" => "accepted"})
-      end)
-
-      assert {:ok, "cds-job-123"} =
-               Era5Client.submit_job("reanalysis-era5-single-levels", %{"foo" => "bar"})
-    end
-
-    test "returns {:error, _} when CDS returns 400" do
-      Req.Test.stub(Era5Client, fn conn ->
-        conn
-        |> Plug.Conn.put_resp_content_type("application/json")
-        |> Plug.Conn.resp(400, ~s({"detail":"bad request"}))
-      end)
-
-      assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
-      assert reason =~ "HTTP 400"
-    end
-
-    test "returns {:error, _} when API key is missing" do
-      System.delete_env("ERA5_CDS_API_KEY")
-      assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
-      assert reason =~ "ERA5_CDS_API_KEY"
-    end
-  end
-
-  describe "check_status/1" do
-    test "returns :running while CDS job is accepted" do
-      Req.Test.stub(Era5Client, fn conn ->
-        assert conn.method == "GET"
-        assert String.ends_with?(conn.request_path, "/jobs/cds-job-123")
-        Req.Test.json(conn, %{"status" => "accepted"})
-      end)
-
-      assert :running = Era5Client.check_status("cds-job-123")
-    end
-
-    test "returns :running while CDS job is running" do
-      Req.Test.stub(Era5Client, fn conn ->
-        Req.Test.json(conn, %{"status" => "running"})
-      end)
-
-      assert :running = Era5Client.check_status("cds-job-123")
-    end
-
-    test "returns {:done, {:url, href}} when CDS job completes with an asset href" do
-      Req.Test.stub(Era5Client, fn conn ->
-        case conn.request_path do
-          "/api/retrieve/v1/jobs/cds-job-123" ->
-            Req.Test.json(conn, %{"status" => "successful"})
-
-          "/api/retrieve/v1/jobs/cds-job-123/results" ->
-            Req.Test.json(conn, %{
-              "asset" => %{"value" => %{"href" => "https://example.com/file.grib"}}
-            })
-        end
-      end)
-
-      assert {:done, {:url, "https://example.com/file.grib"}} =
-               Era5Client.check_status("cds-job-123")
-    end
-
-    test "returns {:failed, _} when CDS reports the job failed" do
-      Req.Test.stub(Era5Client, fn conn ->
-        assert conn.request_path == "/api/retrieve/v1/jobs/cds-job-123"
-        Req.Test.json(conn, %{"status" => "failed", "message" => "input out of range"})
-      end)
-
-      assert {:failed, reason} = Era5Client.check_status("cds-job-123")
-      assert reason =~ "input out of range"
-    end
-
-    test "returns :not_found when CDS returns 404 for the job" do
-      # 404 means the CDS job was deleted or never existed. This is terminal —
-      # retrying will always 404 — so callers should treat it as a distinct
-      # state from the generic {:error, _} retryable bucket.
-      Req.Test.stub(Era5Client, fn conn ->
-        conn
-        |> Plug.Conn.put_resp_content_type("application/json")
-        |> Plug.Conn.resp(
-          404,
-          ~s({"detail":"job cds-job-123 deleted","status":404,"title":"job not found"})
-        )
-      end)
-
-      assert :not_found = Era5Client.check_status("cds-job-123")
-    end
-
-    test "returns {:rejected, reason} when CDS marks the job as rejected" do
-      # CDS rejects submits when the user's in-flight queue is over the cap
-      # (150 jobs). The rejected job still gets a jobID and a GET /jobs/:id
-      # returns a body with status=rejected. This is terminal — re-polling
-      # will never transition to successful — so callers should treat it
-      # like {:failed, _} and re-submit the tile-month.
-      Req.Test.stub(Era5Client, fn conn ->
-        Req.Test.json(conn, %{
-          "jobID" => "cds-job-123",
-          "status" => "rejected",
-          "metadata" => %{"origin" => "api"}
-        })
-      end)
-
-      assert {:rejected, reason} = Era5Client.check_status("cds-job-123")
-      assert is_binary(reason)
-    end
-  end
-
-  describe "delete_job/1" do
-    test "sends DELETE to the CDS jobs endpoint and returns :ok on 204" do
-      test_pid = self()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        send(test_pid, {:delete_request, conn.method, conn.request_path})
-        Plug.Conn.resp(conn, 204, "")
-      end)
-
-      assert :ok = Era5Client.delete_job("cds-job-123")
-      assert_received {:delete_request, "DELETE", "/api/retrieve/v1/jobs/cds-job-123"}
-    end
-
-    test "also treats 200/202/404 as success (CDS sometimes returns 200 or the job may already be gone)" do
-      for status <- [200, 202, 404] do
-        Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, status, "") end)
-        assert :ok = Era5Client.delete_job("cds-job-#{status}")
-      end
-    end
-
-    test "returns {:error, _} for unexpected statuses" do
-      Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, 500, "server error") end)
-      assert {:error, reason} = Era5Client.delete_job("cds-job-500")
-      assert reason =~ "HTTP 500"
-    end
-  end
-
-  describe "download_source_to_file/3" do
-    test "writes a {:body, binary} source directly to the target path" do
-      path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
-
-      try do
-        assert :ok = Era5Client.download_source_to_file({:body, "grib-payload"}, nil, path)
-        assert File.read!(path) == "grib-payload"
-      after
-        File.rm(path)
-      end
-    end
-
-    test "streams a {:url, href} source through Req into the target path" do
-      Req.Test.stub(Era5Client, fn conn ->
-        assert conn.method == "GET"
-
-        conn
-        |> Plug.Conn.put_resp_content_type("application/octet-stream")
-        |> Plug.Conn.resp(200, "streamed-grib-data")
-      end)
-
-      path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
-
-      try do
-        assert :ok =
-                 Era5Client.download_source_to_file(
-                   {:url, "https://example.com/file.grib"},
-                   "test-key",
-                   path
-                 )
-
-        assert File.read!(path) == "streamed-grib-data"
-      after
-        File.rm(path)
-      end
-    end
-
-    test "removes partial file and returns {:error, _} on HTTP error" do
-      Req.Test.stub(Era5Client, fn conn ->
-        Plug.Conn.resp(conn, 503, "")
-      end)
-
-      path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
-
-      assert {:error, reason} =
-               Era5Client.download_source_to_file(
-                 {:url, "https://example.com/file.grib"},
-                 "test-key",
-                 path
-               )
-
-      assert reason =~ "HTTP 503"
-      refute File.exists?(path)
-    end
-  end
-
-  describe "pressure_levels/0" do
-    test "covers upper troposphere and lower stratosphere for the skew-T plot" do
-      levels = Enum.map(Era5Client.pressure_levels(), &String.to_integer/1)
-
-      # Boundary-layer resolution near the surface stays intact.
-      assert 1000 in levels
-      assert 700 in levels
-
-      # Upper-air levels needed to fill out the skew-T above 700 mb.
-      for level <- [500, 300, 200, 100] do
-        assert level in levels, "expected #{level} mb in Era5Client.pressure_levels/0"
-      end
-    end
-  end
-end
diff --git a/test/microwaveprop/workers/era5_fetch_worker_test.exs b/test/microwaveprop/workers/era5_fetch_worker_test.exs
deleted file mode 100644
index a6a3dfcf..00000000
--- a/test/microwaveprop/workers/era5_fetch_worker_test.exs
+++ /dev/null
@@ -1,100 +0,0 @@
-defmodule Microwaveprop.Workers.Era5FetchWorkerTest do
-  use Microwaveprop.DataCase
-  use Oban.Testing, repo: Microwaveprop.Repo
-
-  alias Microwaveprop.Weather.Era5Profile
-  alias Microwaveprop.Workers.Era5FetchWorker
-  alias Microwaveprop.Workers.Era5MonthBatchWorker
-
-  describe "perform/1" do
-    test "enqueues a month-tile batch worker when no profile is cached" do
-      args = %{"lat" => 32.75, "lon" => -97.25, "valid_time" => "2014-01-15T12:00:00Z"}
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert :ok = Era5FetchWorker.perform(%Oban.Job{args: args})
-
-        jobs = all_enqueued(worker: Era5MonthBatchWorker)
-        assert [job] = jobs
-        assert job.args["year"] == 2014
-        assert job.args["month"] == 1
-        # 32.75 → tile 32 (floor to even degree), -97.25 → tile -98
-        assert job.args["tile_lat"] == 32
-        assert job.args["tile_lon"] == -98
-      end)
-    end
-
-    test "is a no-op when the profile is already cached" do
-      valid_time = ~U[2014-01-15 12:00:00Z]
-
-      %Era5Profile{}
-      |> Era5Profile.changeset(%{
-        lat: 32.75,
-        lon: -97.25,
-        valid_time: valid_time,
-        profile: []
-      })
-      |> Repo.insert!()
-
-      args = %{"lat" => 32.75, "lon" => -97.25, "valid_time" => DateTime.to_iso8601(valid_time)}
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert :ok = Era5FetchWorker.perform(%Oban.Job{args: args})
-        assert [] = all_enqueued(worker: Era5MonthBatchWorker)
-      end)
-    end
-  end
-
-  describe "unique constraint" do
-    test "deduplicates identical (lat, lon, valid_time) jobs across inserts" do
-      args = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        {:ok, job1} = Oban.insert(Era5FetchWorker.new(args))
-        {:ok, job2} = Oban.insert(Era5FetchWorker.new(args))
-
-        # Second insert should hit the unique constraint and return the
-        # existing job (same id) instead of creating a new one.
-        assert job1.id == job2.id
-
-        count = Repo.aggregate(Oban.Job, :count, :id)
-        assert count == 1
-      end)
-    end
-
-    test "allows distinct hours on the same date" do
-      args_00 = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
-      args_06 = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T06:00:00Z"}
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        {:ok, job1} = Oban.insert(Era5FetchWorker.new(args_00))
-        {:ok, job2} = Oban.insert(Era5FetchWorker.new(args_06))
-
-        assert job1.id != job2.id
-        assert Repo.aggregate(Oban.Job, :count, :id) == 2
-      end)
-    end
-
-    test "allows distinct grid points for the same valid_time" do
-      args_a = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
-      args_b = %{"lat" => 33.00, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        {:ok, job1} = Oban.insert(Era5FetchWorker.new(args_a))
-        {:ok, job2} = Oban.insert(Era5FetchWorker.new(args_b))
-
-        assert job1.id != job2.id
-        assert Repo.aggregate(Oban.Job, :count, :id) == 2
-      end)
-    end
-
-    test "deduplicates across repeated Oban.insert calls" do
-      args = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        Enum.each(1..5, fn _ -> Oban.insert(Era5FetchWorker.new(args)) end)
-
-        assert Repo.aggregate(Oban.Job, :count, :id) == 1
-      end)
-    end
-  end
-end
diff --git a/test/microwaveprop/workers/era5_month_batch_worker_test.exs b/test/microwaveprop/workers/era5_month_batch_worker_test.exs
deleted file mode 100644
index 699a86ea..00000000
--- a/test/microwaveprop/workers/era5_month_batch_worker_test.exs
+++ /dev/null
@@ -1,49 +0,0 @@
-defmodule Microwaveprop.Workers.Era5MonthBatchWorkerTest do
-  use Microwaveprop.DataCase
-
-  alias Microwaveprop.Workers.Era5MonthBatchWorker
-
-  describe "worker config" do
-    test "max_attempts bumped to 5 so CDS slowness doesn't discard tiles after 3 tries" do
-      assert Era5MonthBatchWorker.__opts__()[:max_attempts] == 5
-    end
-  end
-
-  describe "unique constraint" do
-    test "collapses identical (year, month, tile_lat, tile_lon) jobs" do
-      args = %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        {:ok, job1} = Oban.insert(Era5MonthBatchWorker.new(args))
-        {:ok, job2} = Oban.insert(Era5MonthBatchWorker.new(args))
-
-        assert job1.id == job2.id
-        assert Repo.aggregate(Oban.Job, :count, :id) == 1
-      end)
-    end
-
-    test "different months for the same tile are distinct jobs" do
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        {:ok, j1} =
-          Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}))
-
-        {:ok, j2} =
-          Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 4, tile_lat: 32, tile_lon: -98}))
-
-        assert j1.id != j2.id
-      end)
-    end
-
-    test "different tiles in the same month are distinct jobs" do
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        {:ok, j1} =
-          Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}))
-
-        {:ok, j2} =
-          Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 30, tile_lon: -98}))
-
-        assert j1.id != j2.id
-      end)
-    end
-  end
-end
diff --git a/test/microwaveprop/workers/era5_poll_worker_test.exs b/test/microwaveprop/workers/era5_poll_worker_test.exs
deleted file mode 100644
index 1f2988bc..00000000
--- a/test/microwaveprop/workers/era5_poll_worker_test.exs
+++ /dev/null
@@ -1,337 +0,0 @@
-defmodule Microwaveprop.Workers.Era5PollWorkerTest do
-  use Microwaveprop.DataCase, async: false
-  use Oban.Testing, repo: Microwaveprop.Repo
-
-  alias Microwaveprop.Weather.Era5CdsJob
-  alias Microwaveprop.Weather.Era5Client
-  alias Microwaveprop.Workers.Era5PollWorker
-  alias Microwaveprop.Workers.Era5SubmitWorker
-
-  setup do
-    System.put_env("ERA5_CDS_API_KEY", "test-key")
-    on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
-    :ok
-  end
-
-  defp insert_cds_job(overrides \\ %{}) do
-    # Default to "just now" so the stuck-after guard doesn't misfire
-    # and re-submit in tests that want to exercise status-specific
-    # branches. Tests that care about the stuck path pass a stale
-    # submitted_at explicitly.
-    fresh_submit = DateTime.truncate(DateTime.utc_now(), :second)
-
-    attrs =
-      Map.merge(
-        %{
-          year: 2014,
-          month: 3,
-          tile_lat: 32,
-          tile_lon: -98,
-          single_job_id: "cds-single-123",
-          pressure_job_id: "cds-pressure-456",
-          submitted_at: fresh_submit
-        },
-        overrides
-      )
-
-    {:ok, row} = %Era5CdsJob{} |> Era5CdsJob.changeset(attrs) |> Repo.insert()
-    row
-  end
-
-  describe "perform/1 — row lifecycle" do
-    test "returns :ok and does nothing when the era5_cds_jobs row no longer exists" do
-      # Row was already processed/deleted by a previous poll run; snooze is
-      # scheduled behind us in time; just no-op.
-      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted" end)
-
-      missing_id = Ecto.UUID.generate()
-
-      assert :ok =
-               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => missing_id}})
-    end
-  end
-
-  describe "perform/1 — still running" do
-    test "returns {:snooze, seconds} and bumps poll_count when either CDS job is still running" do
-      row = insert_cds_job()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        assert conn.method == "GET"
-        Req.Test.json(conn, %{"status" => "running"})
-      end)
-
-      assert {:snooze, snooze_s} =
-               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-
-      assert snooze_s > 0
-
-      reloaded = Repo.get!(Era5CdsJob, row.id)
-      assert reloaded.poll_count == 1
-      assert reloaded.last_polled_at
-    end
-  end
-
-  describe "perform/1 — CDS job vanished (404)" do
-    test "deletes the row, deletes the other CDS job, re-enqueues Era5SubmitWorker, and discards" do
-      row = insert_cds_job()
-      test_pid = self()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        cond do
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-single-123") ->
-            conn
-            |> Plug.Conn.put_resp_content_type("application/json")
-            |> Plug.Conn.resp(404, ~s({"detail":"job deleted","title":"job not found"}))
-
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
-            Req.Test.json(conn, %{"status" => "running"})
-
-          conn.method == "DELETE" ->
-            send(test_pid, {:delete, conn.request_path})
-            Plug.Conn.resp(conn, 204, "")
-        end
-      end)
-
-      # Run with manual Oban testing so the re-enqueued Era5SubmitWorker is
-      # inserted into the queue but not executed inline — we only want to
-      # assert that the row was deleted and the re-submit was scheduled.
-      result =
-        Oban.Testing.with_testing_mode(:manual, fn ->
-          Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-        end)
-
-      assert {:discard, reason} = result
-      assert reason =~ "deleted"
-      refute Repo.get(Era5CdsJob, row.id)
-
-      # The sibling CDS job gets cleaned up too so it doesn't linger on CDS.
-      assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"}
-
-      # A fresh Era5SubmitWorker job is enqueued so the month gets re-fetched.
-      assert_enqueued(
-        worker: Era5SubmitWorker,
-        args: %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
-      )
-    end
-  end
-
-  describe "perform/1 — CDS job rejected (over-cap)" do
-    test "deletes the row, cleans up the sibling CDS job, re-enqueues submit, and discards" do
-      row = insert_cds_job()
-      test_pid = self()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        cond do
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-single-123") ->
-            Req.Test.json(conn, %{
-              "jobID" => "cds-single-123",
-              "status" => "rejected",
-              "metadata" => %{"origin" => "api"}
-            })
-
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
-            Req.Test.json(conn, %{"status" => "running"})
-
-          conn.method == "DELETE" ->
-            send(test_pid, {:delete, conn.request_path})
-            Plug.Conn.resp(conn, 204, "")
-        end
-      end)
-
-      result =
-        Oban.Testing.with_testing_mode(:manual, fn ->
-          Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-        end)
-
-      assert {:discard, reason} = result
-      assert reason =~ "rejected"
-      refute Repo.get(Era5CdsJob, row.id)
-
-      # Sibling CDS job is cleaned up so we don't burn queue slots waiting on
-      # a request we've already abandoned.
-      assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"}
-
-      # A fresh Era5SubmitWorker job is enqueued so the tile-month gets
-      # re-submitted once the CDS cap has room.
-      assert_enqueued(
-        worker: Era5SubmitWorker,
-        args: %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
-      )
-    end
-  end
-
-  describe "perform/1 — stuck in CDS queue" do
-    # A CDS job that's been in 'accepted' state for many hours is stuck in
-    # the server's queue and unlikely to ever run. We track submitted_at
-    # on era5_cds_jobs, so once the row ages past @stuck_after, poll
-    # treats it as terminal and re-submits — same path as :not_found /
-    # :rejected.
-    #
-    # Threshold is 8h (see @stuck_after_seconds in Era5PollWorker). We
-    # observed the previous 18h threshold held a row polling for ~13h
-    # without catching a stuck state while 0 profiles were being
-    # ingested. 8h is a diagnostic compromise: short enough to recycle
-    # stuck rows within a shift, still above the ~90 min pressure-level
-    # normal completion time so healthy runs aren't interrupted.
-    test "re-submits when submitted_at is older than 8h and at least one leg is still running" do
-      stale_submit =
-        DateTime.utc_now()
-        |> DateTime.add(-9 * 3600, :second)
-        |> DateTime.truncate(:second)
-
-      row = insert_cds_job(%{submitted_at: stale_submit})
-      test_pid = self()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        cond do
-          conn.method == "GET" ->
-            Req.Test.json(conn, %{"status" => "accepted"})
-
-          conn.method == "DELETE" ->
-            send(test_pid, {:delete, conn.request_path})
-            Plug.Conn.resp(conn, 204, "")
-        end
-      end)
-
-      result =
-        Oban.Testing.with_testing_mode(:manual, fn ->
-          Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-        end)
-
-      assert {:discard, reason} = result
-      assert reason =~ "stuck"
-      refute Repo.get(Era5CdsJob, row.id)
-
-      assert_enqueued(
-        worker: Era5SubmitWorker,
-        args: %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
-      )
-    end
-
-    test "does NOT re-submit a 6h-old row that's still running (under the 8h threshold)" do
-      # Observed in prod: CDS pressure-level completes in ~90 min, so
-      # anything under the 8h diagnostic threshold should keep snoozing,
-      # not trigger a re-submit storm. The single-level leg is allowed
-      # to keep running — only the combined age is stale, and 6h is
-      # still well within the envelope.
-      slowish_submit =
-        DateTime.utc_now()
-        |> DateTime.add(-6 * 3600, :second)
-        |> DateTime.truncate(:second)
-
-      row = insert_cds_job(%{submitted_at: slowish_submit})
-
-      Req.Test.stub(Era5Client, fn conn ->
-        Req.Test.json(conn, %{"status" => "running"})
-      end)
-
-      assert {:snooze, _} =
-               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-
-      assert Repo.get(Era5CdsJob, row.id)
-    end
-
-    test "does NOT re-submit when submitted_at is recent even if running" do
-      # A 30-minute-old row with both legs running is normal — snooze.
-      fresh_submit =
-        DateTime.utc_now()
-        |> DateTime.add(-30 * 60, :second)
-        |> DateTime.truncate(:second)
-
-      row = insert_cds_job(%{submitted_at: fresh_submit})
-
-      Req.Test.stub(Era5Client, fn conn ->
-        Req.Test.json(conn, %{"status" => "running"})
-      end)
-
-      assert {:snooze, _} =
-               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-
-      assert Repo.get(Era5CdsJob, row.id)
-    end
-  end
-
-  describe "perform/1 — transient CDS HTTP error" do
-    # Observed in prod 2026-04-15: a brief CDS network blip returned HTTP
-    # 5xx on check_status. The old {:error, _} branches consumed attempts
-    # and max_attempts=10 burned out in seconds, so 13 rows became
-    # orphaned (discarded with no cleanup, no active poll worker). The
-    # fix: treat HTTP errors as transient — snooze without counting an
-    # attempt. Terminal states (:failed, :rejected, :not_found) still
-    # drive the fail/resubmit paths.
-    test "snoozes and keeps the row when single-level check_status returns a transient HTTP error" do
-      row = insert_cds_job()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        cond do
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-single-123") ->
-            conn
-            |> Plug.Conn.put_resp_content_type("application/json")
-            |> Plug.Conn.resp(503, ~s({"error":"service unavailable"}))
-
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
-            Req.Test.json(conn, %{"status" => "running"})
-        end
-      end)
-
-      assert {:snooze, snooze_s} =
-               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-
-      assert snooze_s > 0
-      assert Repo.get(Era5CdsJob, row.id)
-    end
-
-    test "snoozes and keeps the row when pressure-level check_status returns a transient HTTP error" do
-      row = insert_cds_job()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        cond do
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-single-123") ->
-            Req.Test.json(conn, %{"status" => "running"})
-
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
-            conn
-            |> Plug.Conn.put_resp_content_type("application/json")
-            |> Plug.Conn.resp(502, ~s({"error":"bad gateway"}))
-        end
-      end)
-
-      assert {:snooze, snooze_s} =
-               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-
-      assert snooze_s > 0
-      assert Repo.get(Era5CdsJob, row.id)
-    end
-  end
-
-  describe "perform/1 — failure" do
-    test "returns {:error, _}, deletes the row, and deletes both CDS jobs when a leg fails" do
-      row = insert_cds_job()
-      test_pid = self()
-
-      Req.Test.stub(Era5Client, fn conn ->
-        cond do
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-single-123") ->
-            Req.Test.json(conn, %{"status" => "failed", "message" => "bad input"})
-
-          conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
-            Req.Test.json(conn, %{"status" => "running"})
-
-          conn.method == "DELETE" ->
-            send(test_pid, {:delete, conn.request_path})
-            Plug.Conn.resp(conn, 204, "")
-        end
-      end)
-
-      assert {:error, reason} =
-               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
-
-      assert reason =~ "bad input"
-      refute Repo.get(Era5CdsJob, row.id)
-
-      # Both CDS jobs are deleted to free server-side quota.
-      assert_received {:delete, "/api/retrieve/v1/jobs/cds-single-123"}
-      assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"}
-    end
-  end
-end
diff --git a/test/microwaveprop/workers/era5_submit_worker_test.exs b/test/microwaveprop/workers/era5_submit_worker_test.exs
deleted file mode 100644
index d3cdc214..00000000
--- a/test/microwaveprop/workers/era5_submit_worker_test.exs
+++ /dev/null
@@ -1,258 +0,0 @@
-defmodule Microwaveprop.Workers.Era5SubmitWorkerTest do
-  use Microwaveprop.DataCase, async: false
-  use Oban.Testing, repo: Microwaveprop.Repo
-
-  import Ecto.Query
-
-  alias Microwaveprop.Weather.Era5CdsJob
-  alias Microwaveprop.Weather.Era5Client
-  alias Microwaveprop.Weather.Era5Profile
-  alias Microwaveprop.Workers.Era5PollWorker
-  alias Microwaveprop.Workers.Era5SubmitWorker
-
-  @default_args %{"year" => 2014, "month" => 3, "tile_lat" => 32, "tile_lon" => -98}
-
-  setup do
-    System.put_env("ERA5_CDS_API_KEY", "test-key")
-    on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
-    :ok
-  end
-
-  describe "perform/1 — short-circuit paths" do
-    test "returns {:ok, :cached} when the month-tile already has era5_profiles" do
-      # Seed a profile inside the (2014-03, tile 32,-98) window.
-      %Era5Profile{}
-      |> Era5Profile.changeset(%{
-        lat: 32.5,
-        lon: -97.25,
-        valid_time: ~U[2014-03-15 12:00:00Z],
-        profile: []
-      })
-      |> Repo.insert!()
-
-      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted on cache hit" end)
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert {:ok, :cached} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
-        assert [] = all_enqueued(worker: Era5PollWorker)
-      end)
-    end
-
-    test "returns {:ok, :already_submitted} when an era5_cds_jobs row already exists" do
-      {:ok, _} =
-        %Era5CdsJob{}
-        |> Era5CdsJob.changeset(%{
-          year: 2014,
-          month: 3,
-          tile_lat: 32,
-          tile_lon: -98,
-          single_job_id: "pre-existing-single",
-          pressure_job_id: "pre-existing-pressure",
-          submitted_at: DateTime.truncate(DateTime.utc_now(), :second)
-        })
-        |> Repo.insert()
-
-      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted on duplicate submit" end)
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert {:ok, :already_submitted} =
-                 Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
-      end)
-    end
-  end
-
-  describe "perform/1 — happy path" do
-    test "submits both CDS jobs, persists the row, and enqueues a poll worker" do
-      Req.Test.stub(Era5Client, fn conn ->
-        assert conn.method == "POST"
-
-        dataset_id =
-          cond do
-            String.contains?(conn.request_path, "single-levels") -> "single-job-123"
-            String.contains?(conn.request_path, "pressure-levels") -> "pressure-job-456"
-          end
-
-        Req.Test.json(conn, %{"jobID" => dataset_id})
-      end)
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert {:ok, %Era5CdsJob{} = row} =
-                 Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
-
-        assert row.single_job_id == "single-job-123"
-        assert row.pressure_job_id == "pressure-job-456"
-        assert row.year == 2014
-        assert row.month == 3
-        assert row.tile_lat == 32
-        assert row.tile_lon == -98
-
-        # Row is persisted in the DB.
-        assert Repo.aggregate(
-                 from(j in Era5CdsJob,
-                   where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98
-                 ),
-                 :count,
-                 :id
-               ) == 1
-
-        # A poll worker was enqueued for this row, scheduled ~5 min from now.
-        [poll_job] = all_enqueued(worker: Era5PollWorker)
-        assert poll_job.args["era5_cds_job_id"] == row.id
-        assert DateTime.after?(poll_job.scheduled_at, DateTime.utc_now())
-      end)
-    end
-
-    test "returns {:error, _} and persists no row if either CDS submit fails" do
-      Req.Test.stub(Era5Client, fn conn ->
-        if String.contains?(conn.request_path, "pressure-levels") do
-          Plug.Conn.resp(conn, 400, ~s({"detail":"bad pressure request"}))
-        else
-          Req.Test.json(conn, %{"jobID" => "single-ok"})
-        end
-      end)
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert {:error, reason} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
-        assert is_binary(reason)
-
-        # No orphaned row.
-        assert Repo.aggregate(Era5CdsJob, :count, :id) == 0
-
-        # No poll worker enqueued.
-        assert [] = all_enqueued(worker: Era5PollWorker)
-      end)
-    end
-  end
-
-  describe "CDS in-flight cap" do
-    # CDS rejects with "Number of queued requests is limited to 150" when
-    # a user has too many pending jobs. Each tile-month lives in ONE
-    # era5_cds_jobs row but carries TWO CDS job IDs (single-level +
-    # pressure-level), so the real in-flight count is 2 × row_count. The
-    # cap guard multiplies accordingly. The exact threshold (headroom) is
-    # tuned in the worker; the tests below exercise both the obvious
-    # "way above the cap" path and a row count that should only trip the
-    # snooze if the 2× math is in place.
-    @cds_ceiling 150
-
-    test "snoozes when era5_cds_jobs count is at or above the snooze threshold" do
-      # 140 in-flight is above any reasonable snooze threshold.
-      seed_count = @cds_ceiling - 10
-
-      rows =
-        for i <- 1..seed_count do
-          %{
-            id: Ecto.UUID.generate(),
-            year: 2014,
-            month: 3,
-            tile_lat: 32,
-            tile_lon: -98 - i,
-            single_job_id: "seed-single-#{i}",
-            pressure_job_id: "seed-pressure-#{i}",
-            submitted_at: DateTime.truncate(DateTime.utc_now(), :second),
-            inserted_at: DateTime.truncate(DateTime.utc_now(), :second),
-            updated_at: DateTime.truncate(DateTime.utc_now(), :second),
-            poll_count: 0
-          }
-        end
-
-      Repo.insert_all(Era5CdsJob, rows)
-
-      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted when capped" end)
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert {:snooze, seconds} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
-        assert is_integer(seconds) and seconds > 0
-      end)
-
-      # No new era5_cds_jobs row inserted for the requested tile.
-      refute Repo.one(
-               from j in Era5CdsJob,
-                 where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98
-             )
-    end
-
-    test "snoozes when row count × 2 crosses the cap (each row = two CDS jobs)" do
-      # 65 rows = 130 in-flight CDS jobs — above the effective snooze
-      # threshold even though row count alone is nowhere near 150. This
-      # is the scenario that was escaping the old guard and driving the
-      # per-user-cap reject storms in prod.
-      rows =
-        for i <- 1..65 do
-          %{
-            id: Ecto.UUID.generate(),
-            year: 2014,
-            month: rem(i - 1, 12) + 1,
-            tile_lat: 32,
-            tile_lon: -98 - i,
-            single_job_id: "x2-single-#{i}",
-            pressure_job_id: "x2-pressure-#{i}",
-            submitted_at: DateTime.truncate(DateTime.utc_now(), :second),
-            inserted_at: DateTime.truncate(DateTime.utc_now(), :second),
-            updated_at: DateTime.truncate(DateTime.utc_now(), :second),
-            poll_count: 0
-          }
-        end
-
-      Repo.insert_all(Era5CdsJob, rows)
-
-      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted when capped" end)
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert {:snooze, seconds} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
-        assert is_integer(seconds) and seconds > 0
-      end)
-
-      refute Repo.one(
-               from j in Era5CdsJob,
-                 where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98
-             )
-    end
-
-    test "submits normally when in-flight count is well below the ceiling" do
-      # 10 in-flight — comfortably under the cap.
-      rows =
-        for i <- 1..10 do
-          %{
-            id: Ecto.UUID.generate(),
-            year: 2013,
-            month: i,
-            tile_lat: 40,
-            tile_lon: -90,
-            single_job_id: "warm-single-#{i}",
-            pressure_job_id: "warm-pressure-#{i}",
-            submitted_at: DateTime.truncate(DateTime.utc_now(), :second),
-            inserted_at: DateTime.truncate(DateTime.utc_now(), :second),
-            updated_at: DateTime.truncate(DateTime.utc_now(), :second),
-            poll_count: 0
-          }
-        end
-
-      Repo.insert_all(Era5CdsJob, rows)
-
-      Req.Test.stub(Era5Client, fn conn ->
-        dataset_id =
-          cond do
-            String.contains?(conn.request_path, "single-levels") -> "fresh-single"
-            String.contains?(conn.request_path, "pressure-levels") -> "fresh-pressure"
-          end
-
-        Req.Test.json(conn, %{"jobID" => dataset_id})
-      end)
-
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        assert {:ok, %Era5CdsJob{}} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
-      end)
-    end
-  end
-
-  describe "unique constraint" do
-    test "collapses duplicate (year, month, tile_lat, tile_lon) Oban enqueues" do
-      Oban.Testing.with_testing_mode(:manual, fn ->
-        {:ok, j1} = Oban.insert(Era5SubmitWorker.new(@default_args))
-        {:ok, j2} = Oban.insert(Era5SubmitWorker.new(@default_args))
-        assert j1.id == j2.id
-      end)
-    end
-  end
-end