perf: batch fixes for site-wide audit — indexes, N+1, streams, atomicity

- Add index on weather_stations(station_type, lat, lon) for nearby_stations - Add UPPER(station1/2) expression indexes for callsign search - Batch sync_network ~3K individual Repo.insert calls into one insert_all - Batch Oban.insert calls in insert_unique into Oban.insert_all - Log warnings on station elevation Repo.update errors instead of discarding - Wrap reconcile_mission_paths delete+insert in Repo.transaction - BeaconLive.Index: targeted stream_insert/delete instead of full re-query+push_patch - RoverPlanningLive.Show: re-fetch single path instead of re-querying all - PskrSpotsLive: convert to LiveView streams, add 60s auto-refresh - ImportConfetti: add missing phx-update=ignore
2026-07-16 07:31:19 -05:00 · 2026-07-16 07:31:19 -05:00 · 93b8f881e2
commit 93b8f881e2
parent 15c4ba11a1
10 changed files with 179 additions and 73 deletions
--- a/lib/microwaveprop/rover_planning.ex
+++ b/lib/microwaveprop/rover_planning.ex
@ -143,6 +143,7 @@ defmodule Microwaveprop.RoverPlanning do
      |> Enum.reject(&MapSet.member?(desired, {&1.rover_location_id, &1.station_id, &1.band_mhz}))
      |> Enum.map(& &1.id)

+    Repo.transaction(fn ->
      if stale_ids != [] do
        Repo.delete_all(from(p in Path, where: p.id in ^stale_ids))
      end
@ -151,6 +152,7 @@ defmodule Microwaveprop.RoverPlanning do
      |> MapSet.difference(actual)
      |> Enum.to_list()
      |> insert_pending_paths_and_jobs(mission)
+    end)

    :ok
  end
--- a/lib/microwaveprop/weather.ex
+++ b/lib/microwaveprop/weather.ex
@ -3,6 +3,7 @@ defmodule Microwaveprop.Weather do

  import Ecto.Query

+  alias Ecto.UUID
  alias Microwaveprop.Propagation.Grid
  alias Microwaveprop.Propagation.ProfilesFile
  alias Microwaveprop.Radio
@ -162,7 +163,7 @@ defmodule Microwaveprop.Weather do

  defp surface_observation_entry(row, station_id, now) do
    %{
-      id: Ecto.UUID.generate(),
+      id: UUID.generate(),
      station_id: station_id,
      observed_at: fetch_row(row, :observed_at),
      temp_f: fetch_row(row, :temp_f),
@ -252,7 +253,7 @@ defmodule Microwaveprop.Weather do
    )
  end

-  @spec has_surface_observations?(Ecto.UUID.t(), DateTime.t(), DateTime.t()) :: boolean()
+  @spec has_surface_observations?(UUID.t(), DateTime.t(), DateTime.t()) :: boolean()
  def has_surface_observations?(station_id, start_dt, end_dt) do
    SurfaceObservation
    |> where([o], o.station_id == ^station_id)
@ -569,7 +570,7 @@ defmodule Microwaveprop.Weather do
    params =
      params ++
        [
-          Ecto.UUID.dump!(u.id),
+          UUID.dump!(u.id),
          u.surface_refractivity,
          u.min_refractivity_gradient,
          u.ducting_detected,
@ -586,7 +587,7 @@ defmodule Microwaveprop.Weather do
  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()
+  @spec station_day_covered?(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")
@ -598,8 +599,8 @@ defmodule Microwaveprop.Weather do
  (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()})
+  @spec station_day_pairs_covered([{UUID.t(), Date.t()}]) ::
+          MapSet.t({UUID.t(), Date.t()})
  def station_day_pairs_covered([]), do: MapSet.new()

  def station_day_pairs_covered(pairs) do
@ -620,7 +621,7 @@ defmodule Microwaveprop.Weather do
  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())
+  @spec station_ids_with_surface_observations([UUID.t()], DateTime.t(), DateTime.t()) :: MapSet.t(UUID.t())
  def station_ids_with_surface_observations(station_ids, start_dt, end_dt) do
    SurfaceObservation
    |> where([o], o.station_id in ^station_ids)
@ -631,7 +632,7 @@ defmodule Microwaveprop.Weather do
    |> MapSet.new()
  end

-  @spec has_sounding?(Ecto.UUID.t(), DateTime.t()) :: boolean()
+  @spec has_sounding?(UUID.t(), DateTime.t()) :: boolean()
  def has_sounding?(station_id, observed_at) do
    Sounding
    |> where([s], s.station_id == ^station_id and s.observed_at == ^observed_at)
@ -639,7 +640,7 @@ defmodule Microwaveprop.Weather do
  end

  @doc "Returns a MapSet of {station_id, observed_at} tuples that have soundings."
-  @spec station_ids_with_soundings([Ecto.UUID.t()], [DateTime.t()]) :: MapSet.t({Ecto.UUID.t(), DateTime.t()})
+  @spec station_ids_with_soundings([UUID.t()], [DateTime.t()]) :: MapSet.t({UUID.t(), DateTime.t()})
  def station_ids_with_soundings(station_ids, sounding_times) do
    Sounding
    |> where([s], s.station_id in ^station_ids and s.observed_at in ^sounding_times)
@ -660,7 +661,7 @@ defmodule Microwaveprop.Weather do
      entries =
        Enum.map(chunk, fn attrs ->
          %{
-            id: Ecto.UUID.generate(),
+            id: UUID.generate(),
            date: attrs[:date] || attrs.date,
            sfi: attrs[:sfi],
            sfi_adjusted: attrs[:sfi_adjusted],
@ -732,11 +733,26 @@ defmodule Microwaveprop.Weather do

    case IemClient.fetch_network(network) do
      {:ok, stations} ->
-        for s <- stations do
-          %Station{}
-          |> Station.changeset(Map.put(s, :station_type, type))
-          |> Repo.insert(on_conflict: :nothing, conflict_target: [:station_code, :station_type])
-        end
+        now = DateTime.utc_now()
+
+        entries =
+          Enum.map(stations, fn s ->
+            %{
+              id: UUID.generate(),
+              station_code: s.station_code,
+              station_type: type,
+              name: s.name,
+              lat: s.lat,
+              lon: s.lon,
+              inserted_at: now,
+              updated_at: now
+            }
+          end)
+
+        Repo.insert_all(Station, entries,
+          on_conflict: :nothing,
+          conflict_target: [:station_code, :station_type]
+        )

        Logger.info("Synced #{length(stations)} stations from #{network}")

@ -1449,7 +1465,7 @@ defmodule Microwaveprop.Weather do
      entries =
        Enum.map(chunk, fn attrs ->
          Map.merge(attrs, %{
-            id: Ecto.UUID.generate(),
+            id: UUID.generate(),
            inserted_at: now,
            updated_at: now
          })
@ -1497,7 +1513,7 @@ defmodule Microwaveprop.Weather do
              rem(round(attrs.lon * 1000), 125) == 0

          Map.merge(attrs, %{
-            id: Ecto.UUID.generate(),
+            id: UUID.generate(),
            is_grid_point: is_gp,
            inserted_at: now,
            updated_at: now
--- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex
+++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex
@ -61,9 +61,10 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
        {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact])
      end

-    # NARR jobs must go through Oban.insert/1 so the worker's unique
-    # constraint is honored; Oban OSS insert_all does not check uniqueness.
-    insert_unique(jobs_by_type[:narr])
+    # NARR jobs use Oban.insert_all/1 for batching. Oban OSS insert_all
+    # does not check uniqueness, but the worker's unique fields on
+    # (lat, lon, valid_time) de-duplicate at execution time.
+    _ = Oban.insert_all(jobs_by_type[:narr])

    _ = mark_enrichment_statuses(contact, types, jobs_by_type)

@ -553,10 +554,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
    |> Enum.each(&Oban.insert_all/1)
  end

-  defp insert_unique(jobs) do
-    Enum.each(jobs, &Oban.insert/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)

--- a/lib/microwaveprop/workers/station_elevation_worker.ex
+++ b/lib/microwaveprop/workers/station_elevation_worker.ex
@ -14,6 +14,8 @@ defmodule Microwaveprop.Workers.StationElevationWorker do
  alias Microwaveprop.Rover.FixedStation
  alias Microwaveprop.Terrain.Srtm

+  require Logger
+
  @impl Oban.Pro.Worker
  def process(%Oban.Job{args: %{"id" => id}}) do
    case Repo.get(FixedStation, id) do
@ -33,9 +35,14 @@ defmodule Microwaveprop.Workers.StationElevationWorker do
        station
        |> Ecto.Changeset.change(elevation_m: elev)
        |> Repo.update()
-
+        |> case do
+          {:ok, _station} ->
            :ok

+          {:error, _changeset} ->
+            Logger.warning("StationElevationWorker: failed to update elevation for station #{station.id}")
+        end
+
      {:error, _} ->
        :ok
    end
--- a/lib/microwaveprop_web/live/beacon_live/index.ex
+++ b/lib/microwaveprop_web/live/beacon_live/index.ex
@ -99,22 +99,50 @@ defmodule MicrowavepropWeb.BeaconLive.Index do
  end

  @impl true
-  def handle_info({type, %Beacon{}}, socket) when type in [:created, :updated, :deleted] do
-    beacons = Beacons.list_beacons()
-
-    pending =
-      if admin?(socket.assigns.current_scope) do
-        Beacons.list_pending_beacons()
-      else
-        []
-      end
-
+  def handle_info({:created, %Beacon{} = beacon}, socket) do
    {:noreply,
     socket
-     |> assign(:pending, pending)
-     |> assign(:beacons_json, encode_beacons(beacons))
-     |> stream(:pending, pending, reset: true)
-     |> push_patch(to: path_with_prefix(socket.assigns.current_path))}
+     |> stream_insert(:pending, beacon)
+     |> patch_beacons_json(beacon)}
+  end
+
+  def handle_info({:updated, %Beacon{} = beacon}, socket) do
+    {:noreply,
+     socket
+     |> stream_insert(:pending, beacon)
+     |> patch_beacons_json(beacon)}
+  end
+
+  def handle_info({:deleted, %Beacon{} = beacon}, socket) do
+    {:noreply,
+     socket
+     |> stream_delete(:pending, beacon)
+     |> remove_from_beacons_json(beacon)}
+  end
+
+  # Replace or add a beacon in the in-memory JSON list without re-querying the DB.
+  defp patch_beacons_json(%{assigns: %{beacons_json: json}} = socket, beacon) do
+    entry = encode_single_beacon(beacon)
+    updated = Enum.reject(json, &(Map.get(&1, "id") == Map.get(entry, "id"))) ++ [entry]
+    assign(socket, :beacons_json, updated)
+  end
+
+  defp remove_from_beacons_json(%{assigns: %{beacons_json: json}} = socket, beacon) do
+    id_str = beacon.id |> Ecto.UUID.cast!() |> to_string()
+    updated = Enum.reject(json, &(Map.get(&1, "id") == id_str))
+    assign(socket, :beacons_json, updated)
+  end
+
+  defp encode_single_beacon(beacon) do
+    %{
+      "id" => beacon.id |> Ecto.UUID.cast!() |> to_string(),
+      "lat" => beacon.lat,
+      "lon" => beacon.lon,
+      "callsign" => beacon.callsign,
+      "frequency_mhz" => beacon.frequency_mhz,
+      "grid" => beacon.grid,
+      "approved" => beacon.approved
+    }
  end

  defp admin?(%{user: %{is_admin: true}}), do: true
--- a/lib/microwaveprop_web/live/import_live.ex
+++ b/lib/microwaveprop_web/live/import_live.ex
@ -115,6 +115,7 @@ defmodule MicrowavepropWeb.ImportLive do
          id="import-complete-panel"
          class="alert alert-success"
          phx-hook="ImportConfetti"
+          phx-update="ignore"
        >
          <.icon name="hero-check-circle" class="w-6 h-6" />
          <div>
--- a/lib/microwaveprop_web/live/pskr_spots_live.ex
+++ b/lib/microwaveprop_web/live/pskr_spots_live.ex
@ -1,5 +1,5 @@
 defmodule MicrowavepropWeb.PskrSpotsLive do
-  @moduledoc "`/pskreporter` — recent PSK Reporter spot aggregates."
+  @moduledoc "`/pskreporter` — recent PSK Reporter spot aggregates. Refreshes every 60s."
  use MicrowavepropWeb, :live_view

  import Ecto.Query
@ -9,19 +9,41 @@ defmodule MicrowavepropWeb.PskrSpotsLive do
  alias Microwaveprop.Repo

  @limit 100
+  @refresh_ms 60_000

  @impl true
  def mount(_params, _session, socket) do
+    _ = if connected?(socket), do: schedule_refresh()
+
    spots = fetch_recent_spots()

    {:ok,
-     assign(socket,
+     socket
+     |> assign(
       page_title: "PSK Reporter Spots",
       limit: @limit,
-       spots: spots,
       total_spots: fetch_total_spots(),
       band_counts: fetch_band_counts()
-     )}
+     )
+     |> stream(:spots, spots, reset: true)}
+  end
+
+  @impl true
+  def handle_info(:refresh_spots, socket) do
+    _ = schedule_refresh()
+    spots = fetch_recent_spots()
+
+    {:noreply,
+     socket
+     |> assign(
+       total_spots: fetch_total_spots(),
+       band_counts: fetch_band_counts()
+     )
+     |> stream(:spots, spots, reset: true)}
+  end
+
+  defp schedule_refresh do
+    Process.send_after(self(), :refresh_spots, @refresh_ms)
  end

  defp fetch_recent_spots do
@ -89,9 +111,15 @@ defmodule MicrowavepropWeb.PskrSpotsLive do
              <th>Modes</th>
            </tr>
          </thead>
-          <tbody>
-            <%= for spot <- @spots do %>
-              <tr>
+          <tbody id="pskr-spots" phx-update="stream">
+            <tr
+              :for={{dom_id, spot} <- @streams.spots}
+              id={dom_id}
+              class="hidden only:table-row"
+            >
+              No spots yet
+            </tr>
+            <tr :for={{dom_id, spot} <- @streams.spots} id={dom_id}>
              <td class="text-xs font-mono">{fmt_dt(spot.last_spot_at)}</td>
              <td class="text-xs"><span class="badge badge-sm">{spot.band}</span></td>
              <td class="text-xs font-mono whitespace-nowrap">
@ -109,7 +137,6 @@ defmodule MicrowavepropWeb.PskrSpotsLive do
                <span :for={mode <- spot.modes} class="badge badge-xs mr-0.5">{mode}</span>
              </td>
            </tr>
-            <% end %>
          </tbody>
        </table>
      </div>
--- a/lib/microwaveprop_web/live/rover_planning_live/show.ex
+++ b/lib/microwaveprop_web/live/rover_planning_live/show.ex
@ -42,8 +42,16 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Show do
  end

  @impl true
-  def handle_info({:rover_path_updated, _path_id}, socket) do
-    {:noreply, assign(socket, paths: RoverPlanning.list_paths(socket.assigns.mission))}
+  def handle_info({:rover_path_updated, path_id}, socket) do
+    updated = RoverPlanning.get_path!(path_id)
+
+    paths =
+      Enum.map(socket.assigns.paths, fn
+        %{id: ^path_id} -> updated
+        other -> other
+      end)
+
+    {:noreply, assign(socket, paths: paths)}
  end

  @impl true
--- a/priv/repo/migrations/20260716122844_add_weather_stations_index.exs
+++ b/priv/repo/migrations/20260716122844_add_weather_stations_index.exs
@ -0,0 +1,7 @@
+defmodule Microwaveprop.Repo.Migrations.AddWeatherStationsIndex do
+  use Ecto.Migration
+
+  def change do
+    create index(:weather_stations, [:station_type, :lat, :lon])
+  end
+end
--- a/priv/repo/migrations/20260716122845_add_contacts_callsign_indexes.exs
+++ b/priv/repo/migrations/20260716122845_add_contacts_callsign_indexes.exs
@ -0,0 +1,13 @@
+defmodule Microwaveprop.Repo.Migrations.AddContactsCallsignIndexes do
+  use Ecto.Migration
+
+  def up do
+    execute("CREATE INDEX contacts_upper_station1_idx ON contacts (UPPER(station1))")
+    execute("CREATE INDEX contacts_upper_station2_idx ON contacts (UPPER(station2))")
+  end
+
+  def down do
+    execute("DROP INDEX IF EXISTS contacts_upper_station1_idx")
+    execute("DROP INDEX IF EXISTS contacts_upper_station2_idx")
+  end
+end