diff --git a/config/runtime.exs b/config/runtime.exs index bbf8b4d7..ebf04c15 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -46,6 +46,20 @@ if config_env() == :prod do host = System.get_env("PHX_HOST") || "example.com" + config :libcluster, + topologies: [ + k8s: [ + strategy: Cluster.Strategy.Kubernetes, + config: [ + mode: :ip, + kubernetes_ip_lookup_mode: :pods, + kubernetes_node_basename: "microwaveprop", + kubernetes_selector: "app=prop", + kubernetes_namespace: "prop" + ] + ] + ] + # ## SSL Support # # To get SSL working, you will need to add the `https` key @@ -130,20 +144,6 @@ if config_env() == :prod do ]} ] - config :libcluster, - topologies: [ - k8s: [ - strategy: Cluster.Strategy.Kubernetes, - config: [ - mode: :ip, - kubernetes_ip_lookup_mode: :pods, - kubernetes_node_basename: "microwaveprop", - kubernetes_selector: "app=prop", - kubernetes_namespace: "prop" - ] - ] - ] - config :microwaveprop, :email_from, {"NTMS Propagation", "graham@w5isp.com"} config :microwaveprop, hrrr_base_url: System.get_env("HRRR_BASE_URL", "http://skippy.w5isp.com:8080") config :microwaveprop, srtm_tiles_dir: "/srtm" diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex index c0ac4086..c88266de 100644 --- a/lib/microwaveprop/workers/backfill_enqueue_worker.ex +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -14,25 +14,24 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do require Logger @enrichable [:pending, :queued, :failed] + @valid_types ~w(hrrr weather terrain iemre) @impl Oban.Worker - def perform(%Oban.Job{args: %{"limit" => limit}}) do + def perform(%Oban.Job{args: %{"limit" => limit} = args}) do + types = parse_types(args) + contacts = - Repo.all( - from(c in Contact, - where: - (not is_nil(c.pos1) or not is_nil(c.grid1)) and - (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 - ) - ) + Contact + |> where([c], not is_nil(c.pos1) or not is_nil(c.grid1)) + |> where(^type_filter(types)) + |> order_by([c], desc: c.qso_timestamp) + |> limit(^limit) + |> Repo.all() count = length(contacts) - Logger.info("BackfillEnqueue: processing #{count} contacts") + Logger.info("BackfillEnqueue: processing #{count} contacts (types: #{inspect(types)})") - Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1) + Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact(&1, types)) Phoenix.PubSub.broadcast( Microwaveprop.PubSub, @@ -42,4 +41,17 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do :ok end + + defp parse_types(%{"types" => types}) when is_list(types) do + types |> Enum.filter(&(&1 in @valid_types)) |> Enum.map(&String.to_existing_atom/1) + end + + defp parse_types(_), do: [:hrrr, :weather, :terrain, :iemre] + + defp type_filter(types) do + Enum.reduce(types, dynamic(false), fn type, acc -> + field = :"#{type}_status" + dynamic([c], ^acc or field(c, ^field) in ^@enrichable) + end) + end end diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index c1989347..727e40f8 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -18,13 +18,14 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do @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) do + def enqueue_for_contact(contact, types \\ [:hrrr, :weather, :terrain, :iemre]) do contact = Radio.ensure_positions!(contact) - weather_jobs = build_weather_jobs([contact]) - hrrr_jobs = build_hrrr_jobs([contact]) - terrain_jobs = build_terrain_jobs([contact]) - iemre_jobs = build_iemre_jobs([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 @@ -35,13 +36,13 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do ids = [contact.id] if contact.pos1 do - mark_status!(ids, :weather_status, weather_jobs) - mark_status!(ids, :hrrr_status, hrrr_jobs) - mark_status!(ids, :iemre_status, iemre_jobs) + 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 - mark_status!(ids, :terrain_status, terrain_jobs) + if :terrain in types, do: mark_status!(ids, :terrain_status, terrain_jobs) end :ok diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex index 65e3c94b..433ffba7 100644 --- a/lib/microwaveprop/workers/weather_fetch_worker.ex +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -5,14 +5,14 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do max_attempts: 20, unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]] - require Logger - alias Microwaveprop.Repo alias Microwaveprop.Weather alias Microwaveprop.Weather.IemClient alias Microwaveprop.Weather.SoundingParams alias Microwaveprop.Weather.Station + require Logger + @impl Oban.Worker def backoff(%Oban.Job{attempt: attempt}) do min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600) diff --git a/lib/microwaveprop_web/live/backfill_live.ex b/lib/microwaveprop_web/live/backfill_live.ex index 5b9ec8da..4905e952 100644 --- a/lib/microwaveprop_web/live/backfill_live.ex +++ b/lib/microwaveprop_web/live/backfill_live.ex @@ -41,10 +41,18 @@ defmodule MicrowavepropWeb.BackfillLive do end @impl true - def handle_event("enqueue", %{"limit" => limit_str}, socket) do + def handle_event("enqueue", %{"limit" => limit_str} = params, socket) do limit = String.to_integer(limit_str) - %{"limit" => limit} + types = + ~w(hrrr weather terrain iemre) + |> Enum.filter(&(params[&1] == "true")) + |> then(fn + [] -> ~w(hrrr weather terrain iemre) + types -> types + end) + + %{"limit" => limit, "types" => types} |> BackfillEnqueueWorker.new() |> Oban.insert() @@ -295,7 +303,8 @@ defmodule MicrowavepropWeb.BackfillLive do
<%= for {label, complete} <- [{"HRRR", @unprocessed.total - @unprocessed.hrrr}, {"Weather", @unprocessed.total - @unprocessed.weather}, {"Terrain", @unprocessed.total - @unprocessed.terrain}, {"IEMRE", @unprocessed.total - @unprocessed.iemre}] do %> - <% pct = if @unprocessed.total > 0, do: round(complete * 100 / @unprocessed.total), else: 0 %> + <% pct = + if @unprocessed.total > 0, do: round(complete * 100 / @unprocessed.total), else: 0 %>
{label}
-
+
+
+ <%= for type <- ~w(hrrr weather terrain iemre) do %> + + <% end %> +