perf(weather): asos_day — one job per unique (station, UTC date)

Replaces per-QSO-endpoint asos_batch granularity with per-station-day
granularity. Each contact in the same UTC day near the same station
now produces identical unique-args and collapses to ONE Oban job, so
the backlog shrinks as cross-contact duplicates dedup.

Each fetch also covers a full 24h window (vs the previous 4h) so one
IEM request returns ~24 hourly observations instead of ~1-2. The
effective bytes-per-request is 10x higher, amortizing the
IemRateLimiter gap + 429 retry tail across far more data.

Changes:
- Weather.station_day_covered?/2 + station_day_pairs_covered/1 — cheap
  UTC-day coverage checks for the enqueuer + worker skip paths.
- WeatherFetchWorker gains an "asos_day" clause that fetches the full
  UTC day for one station and upserts. Skips if the day is already
  covered in the DB.
- ContactWeatherEnqueueWorker.build_asos_jobs/3 now emits one asos_day
  job per (station_id, UTC date) pair. ±2h windows crossing midnight
  enumerate both dates.
- Microwaveprop.Weather.RebatchAsos.to_day_jobs/1 migrates any
  already-queued asos/asos_batch jobs into the new shape. Idempotent;
  supports dry_run.
- asos_batch worker clause retained for jobs already in flight at
  deploy time.

2835 tests + credo green.

lib/microwaveprop/weather.ex

@@ -260,6 +260,45 @@ defmodule Microwaveprop.Weather do
     |> Repo.exists?()
   end
 
+  @doc """
+  True if the station already has at least one surface observation
+  anywhere within the given UTC date. Used by the `asos_day` worker
+  to short-circuit the IEM fetch when the day has already been
+  ingested by a prior run.
+  """
+  @spec station_day_covered?(Ecto.UUID.t(), Date.t()) :: boolean()
+  def station_day_covered?(station_id, date) do
+    {:ok, start_dt} = DateTime.new(date, ~T[00:00:00], "Etc/UTC")
+    {:ok, end_dt} = DateTime.new(date, ~T[23:59:59], "Etc/UTC")
+    has_surface_observations?(station_id, start_dt, end_dt)
+  end
+
+  @doc """
+  Returns the MapSet of `{station_id, date}` tuples already covered
+  (at least one obs in that UTC day). Used by the enqueuer to avoid
+  emitting jobs for combinations we've already fetched.
+  """
+  @spec station_day_pairs_covered([{Ecto.UUID.t(), Date.t()}]) ::
+          MapSet.t({Ecto.UUID.t(), Date.t()})
+  def station_day_pairs_covered([]), do: MapSet.new()
+
+  def station_day_pairs_covered(pairs) do
+    station_ids = pairs |> Enum.map(&elem(&1, 0)) |> Enum.uniq()
+    dates = pairs |> Enum.map(&elem(&1, 1)) |> Enum.uniq()
+    {:ok, start_dt} = DateTime.new(Enum.min(dates, Date), ~T[00:00:00], "Etc/UTC")
+    {:ok, end_dt} = DateTime.new(Enum.max(dates, Date), ~T[23:59:59], "Etc/UTC")
+
+    rows =
+      SurfaceObservation
+      |> where([o], o.station_id in ^station_ids)
+      |> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
+      |> select([o], {o.station_id, fragment("(?::date)", o.observed_at)})
+      |> distinct(true)
+      |> Repo.all()
+
+    MapSet.new(rows)
+  end
+
   @doc "Returns a MapSet of station_ids that have surface observations in the time window."
   @spec station_ids_with_surface_observations([Ecto.UUID.t()], DateTime.t(), DateTime.t()) :: MapSet.t(Ecto.UUID.t())
   def station_ids_with_surface_observations(station_ids, start_dt, end_dt) do

lib/microwaveprop/weather/rebatch_asos.ex

@@ -42,6 +42,162 @@ defmodule Microwaveprop.Weather.RebatchAsos do
     end
   end
 
+  @doc """
+  Converts pending `asos_batch` and `asos` jobs into the newer
+  `asos_day` shape: one job per unique `(station_id, station_code, date)`
+  tuple, where `date` enumerates every UTC date the source window
+  touched. Cross-contact dedup is automatic via Oban's unique-args.
+
+  Expected to shrink the pending queue substantially because many
+  QSOs reference the same (station, date) and currently have their
+  own 4h-window job. Idempotent; dry-run supported.
+  """
+  @spec to_day_jobs(keyword()) :: %{
+          source_jobs: non_neg_integer(),
+          unique_day_keys: non_neg_integer(),
+          inserted: non_neg_integer(),
+          cancelled: non_neg_integer()
+        }
+  def to_day_jobs(opts \\ []) do
+    dry_run? = Keyword.get(opts, :dry_run, false)
+    jobs = load_pending_window_jobs()
+
+    if jobs == [] do
+      IO.puts("No pending `asos`/`asos_batch` jobs. Nothing to do.")
+      %{source_jobs: 0, unique_day_keys: 0, inserted: 0, cancelled: 0}
+    else
+      do_to_day(jobs, dry_run?)
+    end
+  end
+
+  defp load_pending_window_jobs do
+    query =
+      from(j in Oban.Job,
+        where:
+          j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and
+            j.state in ["available", "scheduled", "retryable"] and
+            fragment("?->>'fetch_type' IN ('asos', 'asos_batch')", j.args),
+        select: %{id: j.id, args: j.args}
+      )
+
+    Repo.all(query)
+  end
+
+  defp do_to_day(jobs, dry_run?) do
+    # Expand each source job into a list of (station_id, station_code, date)
+    # tuples. asos_batch → one tuple per (station_code, date-in-window);
+    # asos → one tuple per (single station_code, date-in-window).
+    expanded =
+      Enum.flat_map(jobs, fn %{id: id, args: a} ->
+        a
+        |> expand_tuples()
+        |> Enum.map(fn {sid, code, date} -> {sid, code, date, id} end)
+      end)
+
+    # Group by the unique dedup key (station_id, date) and keep one
+    # station_code per group (always the same since station_id is the
+    # key). Track source_ids to cancel afterwards.
+    by_key =
+      Enum.group_by(expanded, fn {sid, _code, date, _id} -> {sid, date} end)
+
+    IO.puts("Expanded #{length(jobs)} source jobs into #{map_size(by_key)} unique (station, date) keys.")
+
+    inserts =
+      Enum.map(by_key, fn {{sid, date}, members} ->
+        {_, code, _, _} = hd(members)
+        source_ids = members |> Enum.map(fn {_, _, _, id} -> id end) |> Enum.uniq()
+
+        %{
+          "fetch_type" => "asos_day",
+          "station_id" => sid,
+          "station_code" => code,
+          "date" => Date.to_iso8601(date),
+          "_source_ids" => source_ids
+        }
+      end)
+
+    if dry_run? do
+      Enum.each(Enum.take(inserts, 20), fn a ->
+        IO.puts("[dry-run] #{a["station_code"]} #{a["date"]} (collapses #{length(a["_source_ids"])} sources)")
+      end)
+
+      %{source_jobs: length(jobs), unique_day_keys: length(inserts), inserted: 0, cancelled: 0}
+    else
+      apply_day_conversion(inserts, length(jobs))
+    end
+  end
+
+  defp expand_tuples(%{"fetch_type" => "asos_batch"} = a) do
+    case window_dates(a["start_dt"], a["end_dt"]) do
+      [] ->
+        []
+
+      dates ->
+        ids = a["station_ids"] || []
+        codes = a["station_codes"] || []
+
+        for {sid, code} <- Enum.zip(ids, codes), date <- dates do
+          {sid, code, date}
+        end
+    end
+  end
+
+  defp expand_tuples(%{"fetch_type" => "asos"} = a) do
+    case window_dates(a["start_dt"], a["end_dt"]) do
+      [] -> []
+      dates -> Enum.map(dates, fn date -> {a["station_id"], a["station_code"], date} end)
+    end
+  end
+
+  defp window_dates(nil, _), do: []
+  defp window_dates(_, nil), do: []
+
+  defp window_dates(start_str, end_str) do
+    with {:ok, start_dt, _} <- DateTime.from_iso8601(start_str),
+         {:ok, end_dt, _} <- DateTime.from_iso8601(end_str) do
+      start_date = DateTime.to_date(start_dt)
+      end_date = DateTime.to_date(end_dt)
+
+      if Date.compare(start_date, end_date) == :eq do
+        [start_date]
+      else
+        [start_date, end_date]
+      end
+    else
+      _ -> []
+    end
+  end
+
+  defp apply_day_conversion(inserts, source_jobs) do
+    {inserted, cancelled} =
+      Enum.reduce(inserts, {0, 0}, fn a, {ins, can} ->
+        source_ids = a["_source_ids"]
+        job_args = Map.delete(a, "_source_ids")
+
+        Repo.transaction(fn ->
+          # Unique-args dedup means duplicate inserts are silently no-ops.
+          _ = job_args |> WeatherFetchWorker.new() |> Oban.insert!()
+        end)
+
+        {ins + 1, can + length(source_ids)}
+      end)
+
+    # Cancel the source jobs in bulk after inserts succeed.
+    {:ok, cancelled_n} =
+      Oban.cancel_all_jobs(
+        from(j in Oban.Job,
+          where:
+            j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and
+              j.state in ["available", "scheduled", "retryable"] and
+              fragment("?->>'fetch_type' IN ('asos', 'asos_batch')", j.args)
+        )
+      )
+
+    IO.puts("Converted: inserted #{inserted} asos_day jobs, cancelled #{cancelled_n} source jobs.")
+
+    %{source_jobs: source_jobs, unique_day_keys: length(inserts), inserted: inserted, cancelled: cancelled}
+  end
+
   defp load_pending_asos_jobs do
     query =
       from(j in Oban.Job,

lib/microwaveprop/workers/contact_weather_enqueue_worker.ex

@@ -389,34 +389,45 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
     end_dt = DateTime.add(timestamp, 2 * 3600, :second)
 
     stations = Weather.nearby_stations(lat, lon, "asos", @asos_radius_km)
-    station_ids = Enum.map(stations, & &1.id)
-    covered = Weather.station_ids_with_surface_observations(station_ids, start_dt, end_dt)
 
-    # Group every uncovered station for this (lat, lon, window) into one
+    # A ±2h window around the QSO can straddle UTC midnight, so enumerate
-    # batch job. IEM's ASOS CSV endpoint accepts multiple station= params
+    # every date the window touches. Each (station, date) becomes one
-    # in a single request and we group the result rows by station_code in
+    # asos_day job — Oban's unique-args dedup collapses cross-QSO
-    # the worker. Collapsing ~N→1 requests per endpoint was the largest
+    # repeats (contact A's KDFW/2021-08-21 and contact B's KDFW/2021-08-21
-    # lever against the IemRateLimiter gap + 429 retry tail.
+    # share args → one job). A full-day fetch returns ~24 hourly rows,
-    uncovered =
+    # amortizing the IemRateLimiter gap over 24× the previous payload.
-      stations
+    dates = window_dates(start_dt, end_dt)
-      |> Enum.reject(fn s -> MapSet.member?(covered, s.id) end)
-      # Sort for deterministic args → Oban unique dedup works across enqueues.
-      |> Enum.sort_by(& &1.station_code)
 
-    case uncovered do
+    pairs =
-      [] ->
+      for station <- stations, date <- dates do
-        []
+        {station, date}
+      end
 
-      stations ->
+    covered =
-        [
+      pairs
-          WeatherFetchWorker.new(%{
+      |> Enum.map(fn {s, d} -> {s.id, d} end)
-            "fetch_type" => "asos_batch",
+      |> Weather.station_day_pairs_covered()
-            "station_ids" => Enum.map(stations, & &1.id),
+
-            "station_codes" => Enum.map(stations, & &1.station_code),
+    pairs
-            "start_dt" => DateTime.to_iso8601(start_dt),
+    |> Enum.reject(fn {s, d} -> MapSet.member?(covered, {s.id, d}) end)
-            "end_dt" => DateTime.to_iso8601(end_dt)
+    |> Enum.map(fn {station, date} ->
-          })
+      WeatherFetchWorker.new(%{
-        ]
+        "fetch_type" => "asos_day",
+        "station_id" => station.id,
+        "station_code" => station.station_code,
+        "date" => Date.to_iso8601(date)
+      })
+    end)
+  end
+
+  defp window_dates(start_dt, end_dt) do
+    start_date = DateTime.to_date(start_dt)
+    end_date = DateTime.to_date(end_dt)
+
+    if Date.compare(start_date, end_date) == :eq do
+      [start_date]
+    else
+      [start_date, end_date]
     end
   end
 

lib/microwaveprop/workers/weather_fetch_worker.ex

@@ -41,6 +41,30 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do
     end
   end
 
+  def perform(%Oban.Job{args: %{"fetch_type" => "asos_day"} = args}) do
+    %{
+      "station_id" => station_id,
+      "station_code" => station_code,
+      "date" => date_str
+    } = args
+
+    {:ok, date} = Date.from_iso8601(date_str)
+
+    cond do
+      is_nil(Repo.get(Station, station_id)) ->
+        Logger.warning("WeatherFetch ASOS day: station #{station_id} (#{station_code}) missing, skipping")
+        :ok
+
+      Weather.station_day_covered?(station_id, date) ->
+        Logger.debug("WeatherFetch ASOS day: #{station_code} #{date_str} already covered, skipping")
+        :ok
+
+      true ->
+        station = Repo.get!(Station, station_id)
+        fetch_asos_day(station, station_id, station_code, date)
+    end
+  end
+
   def perform(%Oban.Job{args: %{"fetch_type" => "asos_batch"} = args}) do
     %{
       "station_ids" => station_ids,
@@ -155,6 +179,26 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do
     :ok
   end
 
+  # Fetch a full UTC day of ASOS obs for one station and upsert.
+  # Amortizes the per-request IemRateLimiter gap + 429 retry tail
+  # across ~24 hourly observations, vs ~1-2 for the previous 4h
+  # windows. Oban unique-args dedup means cross-QSO collisions on
+  # the same (station, date) never re-enqueue.
+  defp fetch_asos_day(station, station_id, station_code, date) do
+    {:ok, start_dt} = DateTime.new(date, ~T[00:00:00], "Etc/UTC")
+    {:ok, end_dt} = DateTime.new(date, ~T[23:59:59], "Etc/UTC")
+
+    Logger.info("WeatherFetch ASOS day: fetching #{station_code} for #{date}")
+
+    case IemClient.fetch_asos(station_code, start_dt, end_dt) do
+      {:ok, rows} ->
+        ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt)
+
+      {:error, reason} ->
+        handle_iem_error("ASOS day", station_code, reason)
+    end
+  end
+
   defp store_asos_stub(station, start_dt, end_dt) do
     midpoint = DateTime.add(start_dt, div(DateTime.diff(end_dt, start_dt), 2))
 

test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs

@@ -61,51 +61,67 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do
     station
   end
 
-  describe "build_weather_jobs/1" do
+  describe "build_weather_jobs/1 — asos_day" do
-    test "builds ASOS batch jobs grouping nearby stations per time window" do
+    test "emits one asos_day job per (station_id, date) pair" do
-      station = create_asos_station("KDFW", 32.90, -97.04)
+      s = create_asos_station("KDFW", 32.90, -97.04)
       contact = create_contact()
 
       jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact])
 
-      asos_jobs =
+      day_jobs =
         Enum.filter(jobs, fn j ->
-          j.changes.args["fetch_type"] == "asos_batch"
+          j.changes.args["fetch_type"] == "asos_day"
         end)
 
-      assert asos_jobs != []
+      assert day_jobs != []
-
+      job = Enum.find(day_jobs, &(&1.changes.args["station_id"] == s.id))
-      # Every uncovered station near this endpoint rides in one batch job,
+      assert job.changes.args["station_code"] == "KDFW"
-      # not one job per station.
+      assert job.changes.args["date"] == "2026-03-28"
-      job = Enum.find(asos_jobs, &(station.id in &1.changes.args["station_ids"]))
-      assert job
-      assert "KDFW" in job.changes.args["station_codes"]
-      assert job.changes.args["start_dt"]
-      assert job.changes.args["end_dt"]
     end
 
-    test "builds several batch jobs for nearby stations — one per endpoint window" do
+    test "dedups identical (station, date) across multiple contacts" do
       create_asos_station("KDFW", 32.90, -97.04)
-      create_asos_station("KFTW", 32.82, -97.36)
+
-      create_asos_station("KAFW", 32.98, -97.32)
+      # Different seconds, same UTC day, same nearby station → one job.
-      contact = create_contact()
+      q1 = create_contact(%{station1: "A1", qso_timestamp: ~U[2026-03-28 14:00:00Z]})
+      q2 = create_contact(%{station1: "A2", qso_timestamp: ~U[2026-03-28 18:00:00Z]})
+
+      jobs = ContactWeatherEnqueueWorker.build_weather_jobs([q1, q2])
+
+      day_jobs =
+        Enum.filter(jobs, fn j -> j.changes.args["fetch_type"] == "asos_day" end)
+
+      # pos1 near KDFW → 1 day job; pos2 and midpoint have no KDFW-nearby station,
+      # so each endpoint contributes at most 1 day job; shared (KDFW, 2026-03-28)
+      # across both QSOs should collapse.
+      kdfw_jobs =
+        Enum.filter(day_jobs, fn j -> j.changes.args["station_code"] == "KDFW" end)
+
+      assert length(kdfw_jobs) == 1
+    end
+
+    test "emits both UTC days when the window crosses midnight" do
+      create_asos_station("KDFW", 32.90, -97.04)
+
+      # QSO at 23:30 UTC → ±2h window spans 21:30..01:30 next day.
+      contact = create_contact(%{qso_timestamp: ~U[2026-03-28 23:30:00Z]})
 
       jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact])
 
-      asos_jobs =
+      kdfw_dates =
-        Enum.filter(jobs, fn j ->
+        jobs
-          j.changes.args["fetch_type"] == "asos_batch"
+        |> Enum.filter(fn j ->
+          j.changes.args["fetch_type"] == "asos_day" and
+            j.changes.args["station_code"] == "KDFW"
         end)
+        |> Enum.map(& &1.changes.args["date"])
 
-      # At least one batch covers pos1's cluster; all three nearby stations
+      assert "2026-03-28" in kdfw_dates
-      # should land in the same batch (shared 4-hour window, same endpoint).
+      assert "2026-03-29" in kdfw_dates
-      pos1_batch = Enum.find(asos_jobs, &(length(&1.changes.args["station_codes"]) == 3))
-      assert pos1_batch
-
-      assert MapSet.new(pos1_batch.changes.args["station_codes"]) ==
-               MapSet.new(["KDFW", "KFTW", "KAFW"])
     end
+  end
 
+  describe "build_weather_jobs/1" do
     test "builds RAOB jobs for nearby sounding stations" do
       station = create_sounding_station("FWD", 32.83, -97.30)
       contact = create_contact()
@@ -125,23 +141,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do
       assert job.changes.args["sounding_time"]
     end
 
-    test "deduplicates jobs with identical args" do
-      _station = create_asos_station("KDFW", 32.90, -97.04)
-
-      # Two QSOs at the exact same time produce identical batch args → deduped
-      q1 = create_contact(%{station1: "A1", qso_timestamp: ~U[2026-03-28 18:00:00Z]})
-      q2 = create_contact(%{station1: "A2", qso_timestamp: ~U[2026-03-28 18:00:00Z]})
-
-      jobs = ContactWeatherEnqueueWorker.build_weather_jobs([q1, q2])
-
-      asos_jobs =
-        Enum.filter(jobs, fn j ->
-          j.changes.args["fetch_type"] == "asos_batch"
-        end)
-
-      assert length(asos_jobs) == 1
-    end
-
     test "builds ASOS jobs for stations near pos2 and midpoint" do
       # Station near pos2 (30.3, -97.7) but NOT near pos1 (32.9, -97.0)
       station_near_pos2 = create_asos_station("KAUS", 30.20, -97.68)
@@ -149,10 +148,12 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do
 
       jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact])
 
-      all_station_ids =
+      station_ids =
-        Enum.flat_map(jobs, fn j -> j.changes.args["station_ids"] || [] end)
+        jobs
+        |> Enum.filter(&(&1.changes.args["fetch_type"] == "asos_day"))
+        |> Enum.map(& &1.changes.args["station_id"])
 
-      assert station_near_pos2.id in all_station_ids
+      assert station_near_pos2.id in station_ids
     end
 
     test "returns empty list when no stations nearby" do
@@ -175,7 +176,7 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do
 
       asos_jobs =
         Enum.filter(jobs, fn j ->
-          j.changes.args["fetch_type"] == "asos_batch"
+          j.changes.args["fetch_type"] == "asos_day"
         end)
 
       assert asos_jobs == []

test/microwaveprop/workers/weather_fetch_worker_test.exs

@@ -59,6 +59,80 @@ defmodule Microwaveprop.Workers.WeatherFetchWorkerTest do
     }
   end
 
+  describe "ASOS day perform/1" do
+    test "fetches full UTC day for one station and upserts returned rows" do
+      {:ok, station} =
+        Weather.find_or_create_station(%{
+          station_code: "KDFW",
+          station_type: "asos",
+          name: "DFW",
+          lat: 32.9,
+          lon: -97.0
+        })
+
+      Req.Test.stub(IemClient, fn conn ->
+        # Assert the worker requested the whole UTC day 2021-08-21.
+        qs = URI.decode_query(conn.query_string)
+        assert qs["year1"] == "2021"
+        assert qs["month1"] == "8"
+        assert qs["day1"] == "21"
+        assert qs["year2"] == "2021"
+        assert qs["day2"] == "21"
+
+        body = """
+        station,valid,tmpf,dwpf,relh,sknt,drct,mslp,alti,skyc1,p01i,wxcodes
+        KDFW,2021-08-21 00:53,80.0,70.0,71.0,8,180,1012.0,29.90,CLR,null,null
+        KDFW,2021-08-21 12:53,85.0,68.0,58.0,10,180,1013.0,29.92,FEW,null,null
+        KDFW,2021-08-21 23:53,78.0,69.0,73.0,6,160,1011.0,29.89,SCT,null,null
+        """
+
+        Req.Test.text(conn, body)
+      end)
+
+      args = %{
+        "fetch_type" => "asos_day",
+        "station_id" => station.id,
+        "station_code" => "KDFW",
+        "date" => "2021-08-21"
+      }
+
+      assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
+      assert Repo.aggregate(SurfaceObservation, :count) == 3
+    end
+
+    test "skips fetch when day is already covered" do
+      {:ok, station} =
+        Weather.find_or_create_station(%{
+          station_code: "KDFW",
+          station_type: "asos",
+          name: "DFW",
+          lat: 32.9,
+          lon: -97.0
+        })
+
+      # Seed a prior observation inside the target day.
+      Weather.upsert_surface_observation(station, %{
+        observed_at: ~U[2021-08-21 12:00:00Z],
+        temp_f: 85.0
+      })
+
+      Req.Test.stub(IemClient, fn conn ->
+        Plug.Conn.send_resp(conn, 500, "should not be called")
+      end)
+
+      args = %{
+        "fetch_type" => "asos_day",
+        "station_id" => station.id,
+        "station_code" => "KDFW",
+        "date" => "2021-08-21"
+      }
+
+      assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
+      # Still only the seeded row — no fetch happened.
+      assert Repo.aggregate(SurfaceObservation, :count) == 1
+    end
+  end
+
   describe "ASOS batch perform/1" do
     test "fetches one URL covering all stations and ingests per-station rows" do
       {:ok, s1} =