defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do @moduledoc false use Oban.Worker, queue: :enqueue, max_attempts: 3 alias Microwaveprop.Radio alias Microwaveprop.Weather alias Microwaveprop.Weather.HrrrClient alias Microwaveprop.Workers.HrrrFetchWorker alias Microwaveprop.Workers.IemreFetchWorker alias Microwaveprop.Workers.TerrainProfileWorker alias Microwaveprop.Workers.WeatherFetchWorker @asos_radius_km 150 @sounding_radius_km 300 # Oban jobs have ~149 params each; PG limit is 65535 params per query @insert_batch_size 400 @doc """ Enqueue all enrichment jobs (weather, HRRR, terrain, IEMRE) for a single contact. Called directly from submission flow — no Oban indirection. Optionally pass a list of types to limit which enrichments run. """ def enqueue_for_contact(contact, types \\ [:hrrr, :weather, :terrain, :iemre]) do contact = Radio.ensure_positions!(contact) weather_jobs = if :weather in types, do: build_weather_jobs([contact]), else: [] hrrr_jobs = if :hrrr in types, do: build_hrrr_jobs([contact]), else: [] terrain_jobs = if :terrain in types, do: build_terrain_jobs([contact]), else: [] iemre_jobs = if :iemre in types, do: build_iemre_jobs([contact]), else: [] all_jobs = weather_jobs ++ hrrr_jobs ++ terrain_jobs ++ iemre_jobs if all_jobs != [] do insert_all_chunked(all_jobs) end ids = [contact.id] if contact.pos1 do if :weather in types, do: mark_status!(ids, :weather_status, weather_jobs) if :hrrr in types, do: mark_status!(ids, :hrrr_status, hrrr_jobs) if :iemre in types, do: mark_status!(ids, :iemre_status, iemre_jobs) end if contact.pos1 && contact.pos2 do if :terrain in types, do: mark_status!(ids, :terrain_status, terrain_jobs) end :ok end @impl Oban.Worker def perform(%Oban.Job{}) do enqueue_weather_jobs() enqueue_hrrr_jobs() enqueue_terrain_jobs() enqueue_iemre_jobs() :ok end defp enqueue_weather_jobs do case Radio.unprocessed_contacts() do [] -> :ok contacts -> Radio.backfill_distances(contacts) jobs = build_weather_jobs(contacts) if jobs != [] do insert_all_chunked(jobs) end ids = Enum.map(contacts, & &1.id) Radio.mark_weather_queued!(ids) enqueue_weather_jobs() end end defp enqueue_hrrr_jobs do case Radio.unprocessed_hrrr_contacts() do [] -> :ok contacts -> jobs = build_hrrr_jobs(contacts) if jobs != [] do insert_all_chunked(jobs) end ids = Enum.map(contacts, & &1.id) Radio.mark_hrrr_queued!(ids) enqueue_hrrr_jobs() end end defp enqueue_terrain_jobs do case Radio.unprocessed_terrain_contacts() do [] -> :ok contacts -> jobs = build_terrain_jobs(contacts) if jobs != [] do insert_all_chunked(jobs) end ids = Enum.map(contacts, & &1.id) Radio.mark_terrain_queued!(ids) enqueue_terrain_jobs() end end def build_terrain_jobs(contacts) do contacts |> Enum.filter(fn contact -> contact.pos1 && contact.pos2 end) |> Enum.map(fn contact -> TerrainProfileWorker.new(%{"contact_id" => contact.id}) end) |> Enum.uniq_by(fn changeset -> changeset.changes.args end) end def build_weather_jobs(contacts) do contacts |> Enum.flat_map(&jobs_for_contact/1) |> Enum.uniq_by(fn changeset -> changeset.changes.args end) end def build_hrrr_jobs(contacts) do contacts |> Enum.flat_map(&hrrr_points_for_contact/1) |> Enum.group_by(fn {_point, hour} -> hour end, fn {point, _hour} -> point end) |> Enum.map(fn {hour, points} -> unique_points = points |> Enum.uniq() |> Enum.map(fn {lat, lon} -> %{"lat" => lat, "lon" => lon} end) HrrrFetchWorker.new(%{ "points" => unique_points, "valid_time" => DateTime.to_iso8601(hour) }) end) end def build_iemre_jobs(contacts) do contacts |> Enum.flat_map(&iemre_job_for_contact/1) |> Enum.uniq_by(fn changeset -> changeset.changes.args end) end defp jobs_for_contact(contact) do contact |> Radio.contact_path_points() |> Enum.flat_map(fn {lat, lon} -> build_asos_jobs(lat, lon, contact.qso_timestamp) ++ build_raob_jobs(lat, lon, contact.qso_timestamp) end) end defp enqueue_iemre_jobs do case Radio.unprocessed_iemre_contacts() do [] -> :ok contacts -> jobs = build_iemre_jobs(contacts) if jobs != [] do insert_all_chunked(jobs) end ids = Enum.map(contacts, & &1.id) Radio.mark_iemre_queued!(ids) enqueue_iemre_jobs() end end defp hrrr_points_for_contact(%{pos1: nil}), do: [] defp hrrr_points_for_contact(contact) do rounded_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp) contact |> Radio.contact_path_points() |> Enum.flat_map(fn {lat, lon} -> {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon) if Weather.has_hrrr_profile?(rlat, rlon, rounded_time) do [] else [{{rlat, rlon}, rounded_time}] end end) end defp build_asos_jobs(lat, lon, timestamp) do start_dt = DateTime.add(timestamp, -2 * 3600, :second) 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) stations |> Enum.reject(fn station -> MapSet.member?(covered, station.id) end) |> Enum.map(fn station -> WeatherFetchWorker.new(%{ "fetch_type" => "asos", "station_id" => station.id, "station_code" => station.station_code, "start_dt" => DateTime.to_iso8601(start_dt), "end_dt" => DateTime.to_iso8601(end_dt) }) end) end defp iemre_job_for_contact(%{pos1: nil}), do: [] defp iemre_job_for_contact(contact) do date = DateTime.to_date(contact.qso_timestamp) contact |> Radio.contact_path_points() |> Enum.flat_map(fn {lat, lon} -> {rlat, rlon} = Weather.round_to_iemre_grid(lat, lon) if Weather.has_iemre_observation?(rlat, rlon, date) do [] else [ IemreFetchWorker.new(%{ "lat" => rlat, "lon" => rlon, "date" => Date.to_iso8601(date) }) ] end end) end defp build_raob_jobs(lat, lon, timestamp) do sounding_times = Weather.sounding_times_around(timestamp) stations = Weather.nearby_stations(lat, lon, "sounding", @sounding_radius_km) station_ids = Enum.map(stations, & &1.id) covered = Weather.station_ids_with_soundings(station_ids, sounding_times) for station <- stations, sounding_time <- sounding_times, not MapSet.member?(covered, {station.id, sounding_time}) do WeatherFetchWorker.new(%{ "fetch_type" => "raob", "station_id" => station.id, "station_code" => station.station_code, "sounding_time" => DateTime.to_iso8601(sounding_time) }) end end defp insert_all_chunked(jobs) do jobs |> Enum.chunk_every(@insert_batch_size) |> Enum.each(&Oban.insert_all/1) end defp mark_status!(ids, field, [_ | _]), do: Radio.set_enrichment_status!(ids, field, :queued) defp mark_status!(ids, field, []), do: Radio.set_enrichment_status!(ids, field, :complete) end