defmodule MicrowavepropWeb.BackfillLive do @moduledoc false use MicrowavepropWeb, :live_view use LiveStash import Ecto.Query alias Microwaveprop.Radio.Contact alias Microwaveprop.Repo alias Microwaveprop.Workers.ContactWeatherEnqueueWorker @refresh_interval 2_000 @impl true def mount(_params, _session, socket) do if connected?(socket) do Process.send_after(self(), :refresh_stats, @refresh_interval) end stats = fetch_stats() unprocessed = count_unprocessed() db_stats = fetch_db_stats() case LiveStash.recover_state(socket) do {:recovered, socket} -> {:ok, assign(socket, stats: stats, unprocessed: unprocessed, db_stats: db_stats)} _ -> {:ok, assign(socket, page_title: "Backfill", limit: 500, stats: stats, unprocessed: unprocessed, db_stats: db_stats, last_enqueued: nil, enqueuing: false )} end end @impl true def handle_event("enqueue", %{"limit" => limit_str}, socket) do limit = String.to_integer(limit_str) # Run in a task to not block the LiveView pid = self() Task.start(fn -> count = enqueue_batch(limit) send(pid, {:enqueued, count}) end) {:noreply, socket |> assign(enqueuing: true, limit: limit) |> LiveStash.stash_assigns([:limit])} end @impl true def handle_info({:enqueued, count}, socket) do {:noreply, assign(socket, last_enqueued: count, enqueuing: false, stats: fetch_stats(), unprocessed: count_unprocessed(), db_stats: fetch_db_stats() )} end def handle_info(:refresh_stats, socket) do Process.send_after(self(), :refresh_stats, @refresh_interval) {:noreply, assign(socket, stats: fetch_stats(), unprocessed: count_unprocessed(), db_stats: fetch_db_stats() )} end @enrichable [:pending, :failed] defp enqueue_batch(limit) do contacts = 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 ) |> Repo.all() # ensure_positions! is called inside enqueue_for_contact Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1) length(contacts) end defp count_unprocessed do incomplete = [:pending, :queued, :processing, :failed] terrain = Repo.one(from(c in Contact, where: c.terrain_status in ^incomplete and not is_nil(c.pos1), select: count())) 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: terrain, hrrr: hrrr, weather: weather, total: max(terrain, max(hrrr, weather)) } end defp fetch_stats do jobs = Repo.all( from(j in "oban_jobs", where: j.state in ["available", "executing", "scheduled", "retryable"], group_by: [j.worker, j.state], select: {j.worker, j.state, count(j.id)} ) ) by_worker = Enum.group_by(jobs, &elem(&1, 0), fn {_, state, count} -> {state, count} end) |> Enum.map(fn {worker, states} -> short_name = worker |> String.split(".") |> List.last() total = Enum.reduce(states, 0, fn {_, c}, acc -> acc + c end) executing = Enum.find_value(states, 0, fn {s, c} -> if s == "executing", do: c end) available = Enum.find_value(states, 0, fn {s, c} -> if s == "available", do: c end) retryable = Enum.find_value(states, 0, fn {s, c} -> if s == "retryable", do: c end) %{ worker: short_name, total: total, executing: executing, available: available, retryable: retryable } end) |> Enum.sort_by(& &1.total, :desc) completed_1h = Repo.one( from(j in "oban_jobs", where: j.state == "completed" and j.completed_at > ago(1, "hour"), select: count(j.id) ) ) %{by_worker: by_worker, completed_1h: completed_1h} end defp fetch_db_stats do %{rows: rows} = Repo.query!(""" SELECT relname as table, n_live_tup as rows, pg_total_relation_size(c.oid) as bytes FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace JOIN pg_stat_user_tables s ON s.relid = c.oid WHERE n.nspname = 'public' AND c.relkind = 'r' AND relname NOT LIKE 'oban_%' AND relname NOT LIKE 'schema_%' AND n_live_tup > 0 ORDER BY pg_total_relation_size(c.oid) DESC LIMIT 20 """) tables = Enum.map(rows, fn [table, row_count, bytes] -> %{table: table, rows: row_count, size: format_bytes(bytes)} end) %{rows: total_rows} = Repo.query!("SELECT pg_database_size(current_database())") [[db_bytes]] = total_rows # Enrichment status breakdown %{rows: status_rows} = Repo.query!(""" SELECT 'hrrr' as type, hrrr_status as status, count(*) as cnt FROM qsos GROUP BY hrrr_status UNION ALL SELECT 'weather', weather_status, count(*) FROM qsos GROUP BY weather_status UNION ALL SELECT 'terrain', terrain_status, count(*) FROM qsos GROUP BY terrain_status UNION ALL SELECT 'iemre', iemre_status, count(*) FROM qsos GROUP BY iemre_status ORDER BY 1, 2 """) statuses = Enum.group_by(status_rows, fn [type, _, _] -> type end, fn [_, status, count] -> {status, count} end) # Per-contact averages %{rows: avg_rows} = Repo.query!(""" SELECT (SELECT count(*) FROM hrrr_profiles) as hrrr_profiles, (SELECT count(*) FROM terrain_profiles) as terrain_profiles, (SELECT count(*) FROM surface_observations) as surface_obs, (SELECT count(*) FROM soundings) as soundings, (SELECT count(*) FROM iemre_observations) as iemre_obs, (SELECT count(*) FROM qsos WHERE pos1 IS NOT NULL) as contacts """) [[hrrr_count, terrain_count, obs_count, sounding_count, iemre_count, contact_count]] = avg_rows contact_count = max(contact_count, 1) %{ db_size: format_bytes(db_bytes), tables: tables, statuses: statuses, totals: %{ hrrr_profiles: hrrr_count, terrain_profiles: terrain_count, surface_observations: obs_count, soundings: sounding_count, iemre_observations: iemre_count, contacts: contact_count }, averages: %{ hrrr_per_contact: Float.round(hrrr_count / contact_count, 1), obs_per_contact: Float.round(obs_count / contact_count, 1), soundings_per_contact: Float.round(sounding_count / contact_count, 1) } } end defp format_number(n) when is_integer(n) do n |> Integer.to_string() |> String.graphemes() |> Enum.reverse() |> Enum.chunk_every(3) |> Enum.map_join(",", &Enum.reverse/1) |> String.reverse() |> then(fn s -> String.replace(s, ~r/^,/, "") end) end defp format_number(n) when is_float(n), do: format_number(round(n)) defp format_number(n), do: to_string(n) defp status_class("complete"), do: "text-success" defp status_class("pending"), do: "text-warning" defp status_class("queued"), do: "text-info" defp status_class("processing"), do: "text-info" defp status_class("failed"), do: "text-error" defp status_class("unavailable"), do: "opacity-40" defp status_class(_), do: "" defp format_bytes(bytes) when bytes >= 1_073_741_824, do: "#{Float.round(bytes / 1_073_741_824, 1)} GB" defp format_bytes(bytes) when bytes >= 1_048_576, do: "#{Float.round(bytes / 1_048_576, 1)} MB" defp format_bytes(bytes) when bytes >= 1024, do: "#{Float.round(bytes / 1024, 1)} KB" defp format_bytes(bytes), do: "#{bytes} B" @impl true def render(assigns) do ~H""" <.header> Backfill Dashboard <:subtitle>Enqueue and monitor contact enrichment jobs
Enrichment Progress
{@unprocessed.total} remaining
{@db_stats.totals.contacts - @unprocessed.total} / {@db_stats.totals.contacts} complete
Terrain: {@unprocessed.terrain} HRRR: {@unprocessed.hrrr} Weather: {@unprocessed.weather}
Active/Queued Jobs
{Enum.reduce(@stats.by_worker, 0, fn w, acc -> acc + w.total end)}
Completed (last hour)
{@stats.completed_1h}
<%= if @last_enqueued do %> Enqueued {@last_enqueued} contacts <% end %>

Job Queue Status

<%= if @stats.by_worker == [] do %>

No active jobs.

<% else %>
<%= for w <- @stats.by_worker do %> <% end %>
Worker Executing Available Retryable Total
{w.worker} <%= if w.executing > 0 do %> {w.executing} <% else %> 0 <% end %> {w.available} <%= if w.retryable > 0 do %> {w.retryable} <% else %> 0 <% end %> {w.total}
<% end %>

Enrichment Status by Type

<%= for {type, label} <- [{"hrrr", "HRRR"}, {"weather", "Weather"}, {"terrain", "Terrain"}, {"iemre", "IEMRE"}] do %>
{label}
<%= for {status, count} <- Map.get(@db_stats.statuses, type, []) do %>
{status} {format_number(count)}
<% end %>
<% end %>

Database

Total DB Size
{@db_stats.db_size}
HRRR Profiles
{format_number(@db_stats.totals.hrrr_profiles)}
{@db_stats.averages.hrrr_per_contact} per contact
Surface Observations
{format_number(@db_stats.totals.surface_observations)}
{@db_stats.averages.obs_per_contact} per contact
Soundings
{format_number(@db_stats.totals.soundings)}
{@db_stats.averages.soundings_per_contact} per contact

Table Sizes

<%= for t <- @db_stats.tables do %> <% end %>
Table Rows Size
{t.table} {format_number(t.rows)} {t.size}

Auto-refreshes every 2 seconds.

""" end end