diff --git a/config/config.exs b/config/config.exs index 429a9309..b8e89a2c 100644 --- a/config/config.exs +++ b/config/config.exs @@ -51,7 +51,6 @@ config :microwaveprop, Oban, {Oban.Plugins.Cron, crontab: [ {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, - {"*/30 * * * *", Microwaveprop.Workers.ContactWeatherEnqueueWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker} ]} diff --git a/config/dev.exs b/config/dev.exs index 552de9a6..9ec51295 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -78,8 +78,7 @@ config :microwaveprop, Oban, {Oban.Plugins.Cron, crontab: [ {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, - {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, - {"*/30 * * * *", Microwaveprop.Workers.ContactWeatherEnqueueWorker} + {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker} ]} ] diff --git a/lib/microwaveprop/radio.ex b/lib/microwaveprop/radio.ex index 98478bd4..2416840b 100644 --- a/lib/microwaveprop/radio.ex +++ b/lib/microwaveprop/radio.ex @@ -97,61 +97,43 @@ defmodule Microwaveprop.Radio do {field, dir} end - def unprocessed_contacts(limit \\ 500) do - Contact - |> where([q], q.weather_queued == false and not is_nil(q.pos1)) - |> order_by([q], asc: q.qso_timestamp) - |> limit(^limit) - |> Repo.all() + # ── Enrichment status queries ──────────────────────────────────── + + @enrichable_statuses [:pending, :failed] + + def contacts_needing_enrichment(field, limit \\ 500, extra_filter \\ nil) do + query = + Contact + |> where([q], field(q, ^field) in ^@enrichable_statuses and not is_nil(q.pos1)) + |> order_by([q], asc: q.qso_timestamp) + |> limit(^limit) + + query = if extra_filter, do: extra_filter.(query), else: query + Repo.all(query) end - def mark_weather_queued!(ids) do + def unprocessed_contacts(limit \\ 500), + do: contacts_needing_enrichment(:weather_status, limit) + + def unprocessed_hrrr_contacts(limit \\ 500), + do: contacts_needing_enrichment(:hrrr_status, limit) + + def unprocessed_terrain_contacts(limit \\ 500), + do: contacts_needing_enrichment(:terrain_status, limit, &where(&1, [q], not is_nil(q.pos2))) + + def unprocessed_iemre_contacts(limit \\ 500), + do: contacts_needing_enrichment(:iemre_status, limit) + + def set_enrichment_status!(ids, field, status) do Contact |> where([q], q.id in ^ids) - |> Repo.update_all(set: [weather_queued: true]) + |> Repo.update_all(set: [{field, status}]) end - def unprocessed_hrrr_contacts(limit \\ 500) do - Contact - |> where([q], q.hrrr_queued == false and not is_nil(q.pos1)) - |> order_by([q], asc: q.qso_timestamp) - |> limit(^limit) - |> Repo.all() - end - - def mark_hrrr_queued!(ids) do - Contact - |> where([q], q.id in ^ids) - |> Repo.update_all(set: [hrrr_queued: true]) - end - - def unprocessed_terrain_contacts(limit \\ 500) do - Contact - |> where([q], q.terrain_queued == false and not is_nil(q.pos1) and not is_nil(q.pos2)) - |> order_by([q], asc: q.qso_timestamp) - |> limit(^limit) - |> Repo.all() - end - - def mark_terrain_queued!(ids) do - Contact - |> where([q], q.id in ^ids) - |> Repo.update_all(set: [terrain_queued: true]) - end - - def unprocessed_iemre_contacts(limit \\ 500) do - Contact - |> where([q], q.iemre_queued == false and not is_nil(q.pos1)) - |> order_by([q], asc: q.qso_timestamp) - |> limit(^limit) - |> Repo.all() - end - - def mark_iemre_queued!(ids) do - Contact - |> where([q], q.id in ^ids) - |> Repo.update_all(set: [iemre_queued: true]) - end + def mark_weather_queued!(ids), do: set_enrichment_status!(ids, :weather_status, :queued) + def mark_hrrr_queued!(ids), do: set_enrichment_status!(ids, :hrrr_status, :queued) + def mark_terrain_queued!(ids), do: set_enrichment_status!(ids, :terrain_status, :queued) + def mark_iemre_queued!(ids), do: set_enrichment_status!(ids, :iemre_status, :queued) @doc """ Returns a list of {lat, lon} points along the contact path: pos1, midpoint, pos2. diff --git a/lib/microwaveprop/radio/contact.ex b/lib/microwaveprop/radio/contact.ex index 859ec15e..971b671b 100644 --- a/lib/microwaveprop/radio/contact.ex +++ b/lib/microwaveprop/radio/contact.ex @@ -20,10 +20,21 @@ defmodule Microwaveprop.Radio.Contact do field :mode, :string field :band, :decimal field :distance_km, :decimal - field :weather_queued, :boolean, default: false - field :hrrr_queued, :boolean, default: false - field :terrain_queued, :boolean, default: false - field :iemre_queued, :boolean, default: false + field :hrrr_status, Ecto.Enum, + values: [:pending, :queued, :processing, :complete, :failed, :unavailable], + default: :pending + + field :weather_status, Ecto.Enum, + values: [:pending, :queued, :processing, :complete, :failed, :unavailable], + default: :pending + + field :terrain_status, Ecto.Enum, + values: [:pending, :queued, :processing, :complete, :failed, :unavailable], + default: :pending + + field :iemre_status, Ecto.Enum, + values: [:pending, :queued, :processing, :complete, :failed, :unavailable], + default: :pending field :user_submitted, :boolean, default: false field :submitter_email, :string field :flagged_invalid, :boolean, default: false diff --git a/lib/microwaveprop/radio/enrichment_status.ex b/lib/microwaveprop/radio/enrichment_status.ex new file mode 100644 index 00000000..ee063c21 --- /dev/null +++ b/lib/microwaveprop/radio/enrichment_status.ex @@ -0,0 +1,57 @@ +defmodule Microwaveprop.Radio.EnrichmentStatus do + @moduledoc """ + State machine for contact enrichment lifecycle. + + States: + :pending → initial, no work started + :queued → Oban job inserted + :processing → worker actively running + :complete → data exists in DB + :failed → worker exhausted retries + :unavailable → data cannot be fetched (missing pos, archive gone, etc.) + + Valid transitions: + pending → queued, unavailable + queued → processing, failed + processing → complete, failed + failed → queued (retry) + unavailable → queued (retry with new data source) + complete → pending (reset for re-enrichment) + """ + + @states [:pending, :queued, :processing, :complete, :failed, :unavailable] + @fields [:hrrr_status, :weather_status, :terrain_status, :iemre_status] + + @transitions %{ + pending: [:queued, :unavailable], + queued: [:processing, :failed], + processing: [:complete, :failed], + failed: [:queued], + unavailable: [:queued], + complete: [:pending] + } + + def states, do: @states + def fields, do: @fields + + @doc "Returns true if transitioning from `from` to `to` is valid." + def valid_transition?(from, to) do + to in Map.get(@transitions, from, []) + end + + @doc "Validates and applies a status transition on a changeset." + def transition(changeset, field, new_status) when field in @fields and new_status in @states do + current = Ecto.Changeset.get_field(changeset, field) + + if valid_transition?(current, new_status) do + Ecto.Changeset.put_change(changeset, field, new_status) + else + Ecto.Changeset.add_error(changeset, field, "invalid transition from #{current} to #{new_status}") + end + end + + @doc "Force-sets status without transition validation (for migrations, resets)." + def force_status(changeset, field, status) when field in @fields and status in @states do + Ecto.Changeset.put_change(changeset, field, status) + end +end diff --git a/lib/microwaveprop/workers/terrain_profile_worker.ex b/lib/microwaveprop/workers/terrain_profile_worker.ex index 69660801..707ee2bf 100644 --- a/lib/microwaveprop/workers/terrain_profile_worker.ex +++ b/lib/microwaveprop/workers/terrain_profile_worker.ex @@ -56,6 +56,8 @@ defmodule Microwaveprop.Workers.TerrainProfileWorker do verdict: analysis.verdict }) + Radio.set_enrichment_status!([qso_id], :terrain_status, :complete) + Phoenix.PubSub.broadcast( Microwaveprop.PubSub, "contact_enrichment:#{qso_id}", diff --git a/lib/microwaveprop_web/live/backfill_live.ex b/lib/microwaveprop_web/live/backfill_live.ex index 67021018..0bcbaabe 100644 --- a/lib/microwaveprop_web/live/backfill_live.ex +++ b/lib/microwaveprop_web/live/backfill_live.ex @@ -68,13 +68,15 @@ defmodule MicrowavepropWeb.BackfillLive do )} end + @enrichable [:pending, :failed] + defp enqueue_batch(limit) do contacts = from(c in Contact, where: not is_nil(c.pos1) and - (c.hrrr_queued == false or c.weather_queued == false or - c.terrain_queued == false or c.iemre_queued == false), + (c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or + c.terrain_status in ^@enrichable or c.iemre_status in ^@enrichable), order_by: [desc: c.qso_timestamp], limit: ^limit ) @@ -85,42 +87,22 @@ defmodule MicrowavepropWeb.BackfillLive do end defp count_unprocessed do - missing_terrain = - Repo.one( - from(c in Contact, - left_join: t in Microwaveprop.Terrain.TerrainProfile, on: t.qso_id == c.id, - where: not is_nil(c.pos1) and not is_nil(c.pos2) and is_nil(t.id), - select: count(c.id) - ) - ) + incomplete = [:pending, :queued, :processing, :failed] - # Use Oban job counts as proxy for in-progress work since queued flags - # only indicate "was enqueued" not "data exists" - pending_hrrr = - Repo.one( - from(j in "oban_jobs", - where: - j.worker == "Microwaveprop.Workers.HrrrFetchWorker" and - j.state in ["available", "executing", "scheduled", "retryable"], - select: count(j.id) - ) - ) + terrain = + Repo.one(from(c in Contact, where: c.terrain_status in ^incomplete and not is_nil(c.pos1), select: count())) - pending_weather = - Repo.one( - from(j in "oban_jobs", - where: - j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and - j.state in ["available", "executing", "scheduled", "retryable"], - select: count(j.id) - ) - ) + hrrr = + Repo.one(from(c in Contact, where: c.hrrr_status in ^incomplete and not is_nil(c.pos1), select: count())) + + weather = + Repo.one(from(c in Contact, where: c.weather_status in ^incomplete and not is_nil(c.pos1), select: count())) %{ - terrain: missing_terrain, - hrrr: pending_hrrr, - weather: pending_weather, - total: missing_terrain + pending_hrrr + pending_weather + terrain: terrain, + hrrr: hrrr, + weather: weather, + total: max(terrain, max(hrrr, weather)) } end diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex index e574a589..5ffe0ba4 100644 --- a/lib/microwaveprop_web/live/contact_live/show.ex +++ b/lib/microwaveprop_web/live/contact_live/show.ex @@ -32,9 +32,10 @@ defmodule MicrowavepropWeb.ContactLive.Show do solar = load_solar(contact) hrrr_path = Weather.hrrr_profiles_for_path(contact) hrrr = List.first(hrrr_path) - {hrrr, hrrr_status} = maybe_enqueue_hrrr(hrrr, contact) + {hrrr, contact} = maybe_enqueue_hrrr(hrrr, contact) terrain = Terrain.get_terrain_profile(contact.id) - terrain_status = if terrain, do: :loaded, else: maybe_enqueue_terrain(terrain, contact) + contact = maybe_enqueue_terrain(terrain, contact) + contact = maybe_enqueue_weather(weather, contact) elevation_profile = compute_elevation_profile(contact, hrrr_path, weather.soundings) propagation_analysis = build_propagation_analysis(contact, hrrr, terrain, elevation_profile, weather.soundings) @@ -44,12 +45,9 @@ defmodule MicrowavepropWeb.ContactLive.Show do contact: contact, surface_observations: weather.surface_observations, soundings: weather.soundings, - weather_status: maybe_enqueue_weather(weather, contact), solar: solar, hrrr: hrrr, - hrrr_status: hrrr_status, terrain: terrain, - terrain_status: terrain_status, elevation_profile: elevation_profile, propagation_analysis: propagation_analysis, terrain_expanded: false, @@ -117,7 +115,7 @@ defmodule MicrowavepropWeb.ContactLive.Show do @impl true def handle_info({:terrain_ready, _qso_id}, socket) do - contact = socket.assigns.contact + contact = %{socket.assigns.contact | terrain_status: :complete} terrain = Terrain.get_terrain_profile(contact.id) soundings = socket.assigns.soundings hrrr_path = Weather.hrrr_profiles_for_path(contact) @@ -127,8 +125,8 @@ defmodule MicrowavepropWeb.ContactLive.Show do {:noreply, assign(socket, + contact: contact, terrain: terrain, - terrain_status: :loaded, elevation_profile: elevation_profile, propagation_analysis: propagation_analysis )} @@ -140,6 +138,7 @@ defmodule MicrowavepropWeb.ContactLive.Show do hrrr = List.first(hrrr_path) if hrrr do + contact = %{contact | hrrr_status: :complete} soundings = socket.assigns.soundings terrain = socket.assigns.terrain elevation_profile = compute_elevation_profile(contact, hrrr_path, soundings) @@ -147,8 +146,8 @@ defmodule MicrowavepropWeb.ContactLive.Show do {:noreply, assign(socket, + contact: contact, hrrr: hrrr, - hrrr_status: :loaded, elevation_profile: elevation_profile, propagation_analysis: propagation_analysis )} @@ -158,7 +157,7 @@ defmodule MicrowavepropWeb.ContactLive.Show do end def handle_info({:weather_ready, _info}, socket) do - contact = socket.assigns.contact + contact = %{socket.assigns.contact | weather_status: :complete} weather = load_weather(contact) soundings = weather.soundings @@ -166,34 +165,40 @@ defmodule MicrowavepropWeb.ContactLive.Show do hrrr = List.first(hrrr_path) terrain = socket.assigns.terrain - # Recompute elevation profile with new sounding duct data elevation_profile = compute_elevation_profile(contact, hrrr_path, soundings) propagation_analysis = build_propagation_analysis(contact, hrrr, terrain, elevation_profile, soundings) {:noreply, assign(socket, + contact: contact, surface_observations: weather.surface_observations, soundings: soundings, - weather_status: :loaded, elevation_profile: elevation_profile, propagation_analysis: propagation_analysis )} end defp maybe_enqueue_weather(weather, contact) do - has_data = weather.surface_observations != [] or weather.soundings != [] + if contact.weather_status == :complete do + contact + else + has_data = weather.surface_observations != [] or weather.soundings != [] - cond do - has_data -> - :loaded + cond do + has_data -> + Radio.set_enrichment_status!([contact.id], :weather_status, :complete) + %{contact | weather_status: :complete} - contact.pos1 -> - jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact]) - if jobs != [], do: Oban.insert_all(jobs) - :queued + contact.pos1 && contact.weather_status in [:pending, :failed] -> + jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact]) + if jobs != [], do: Oban.insert_all(jobs) + Radio.set_enrichment_status!([contact.id], :weather_status, :queued) + %{contact | weather_status: :queued} - true -> - :unavailable + true -> + Radio.set_enrichment_status!([contact.id], :weather_status, :unavailable) + %{contact | weather_status: :unavailable} + end end end @@ -268,24 +273,40 @@ defmodule MicrowavepropWeb.ContactLive.Show do end end - defp maybe_enqueue_terrain(terrain, _contact) when not is_nil(terrain), do: :loaded + defp maybe_enqueue_terrain(_terrain, %{terrain_status: :complete} = contact), do: contact + + defp maybe_enqueue_terrain(terrain, contact) when not is_nil(terrain) do + Radio.set_enrichment_status!([contact.id], :terrain_status, :complete) + %{contact | terrain_status: :complete} + end defp maybe_enqueue_terrain(nil, contact) do - if contact.pos1 && contact.pos2 do + if contact.pos1 && contact.pos2 && contact.terrain_status in [:pending, :failed] do Oban.insert(TerrainProfileWorker.new(%{"qso_id" => contact.id})) - :queued + Radio.set_enrichment_status!([contact.id], :terrain_status, :queued) + %{contact | terrain_status: :queued} else - :unavailable + if is_nil(contact.pos1) or is_nil(contact.pos2) do + Radio.set_enrichment_status!([contact.id], :terrain_status, :unavailable) + %{contact | terrain_status: :unavailable} + else + contact + end end end - defp maybe_enqueue_hrrr(hrrr, _contact) when not is_nil(hrrr), do: {hrrr, :loaded} + defp maybe_enqueue_hrrr(hrrr, %{hrrr_status: :complete} = contact), do: {hrrr, contact} + + defp maybe_enqueue_hrrr(hrrr, contact) when not is_nil(hrrr) do + Radio.set_enrichment_status!([contact.id], :hrrr_status, :complete) + {hrrr, %{contact | hrrr_status: :complete}} + end defp maybe_enqueue_hrrr(nil, contact) do lat = contact.pos1 && contact.pos1["lat"] lon = contact.pos1 && (contact.pos1["lon"] || contact.pos1["lng"]) - if lat && lon do + if lat && lon && contact.hrrr_status in [:pending, :failed] do valid_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp) {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon) @@ -297,9 +318,10 @@ defmodule MicrowavepropWeb.ContactLive.Show do }) ) - {nil, :queued} + Radio.set_enrichment_status!([contact.id], :hrrr_status, :queued) + {nil, %{contact | hrrr_status: :queued}} else - {nil, :unavailable} + {nil, contact} end end @@ -356,7 +378,7 @@ defmodule MicrowavepropWeb.ContactLive.Show do /> <% end %> - <%= if !@elevation_profile && @terrain_status == :queued do %> + <%= if !@elevation_profile && @contact.terrain_status == :queued do %>