diff --git a/lib/microwaveprop/rover_planning.ex b/lib/microwaveprop/rover_planning.ex index 54c741bb..bbf7478f 100644 --- a/lib/microwaveprop/rover_planning.ex +++ b/lib/microwaveprop/rover_planning.ex @@ -143,14 +143,16 @@ defmodule Microwaveprop.RoverPlanning do |> Enum.reject(&MapSet.member?(desired, {&1.rover_location_id, &1.station_id, &1.band_mhz})) |> Enum.map(& &1.id) - if stale_ids != [] do - Repo.delete_all(from(p in Path, where: p.id in ^stale_ids)) - end + Repo.transaction(fn -> + if stale_ids != [] do + Repo.delete_all(from(p in Path, where: p.id in ^stale_ids)) + end - desired - |> MapSet.difference(actual) - |> Enum.to_list() - |> insert_pending_paths_and_jobs(mission) + desired + |> MapSet.difference(actual) + |> Enum.to_list() + |> insert_pending_paths_and_jobs(mission) + end) :ok end diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 824fa1ee..492ccc1c 100644 --- 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 diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index 021a0c07..ad12a584 100644 --- 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) diff --git a/lib/microwaveprop/workers/station_elevation_worker.ex b/lib/microwaveprop/workers/station_elevation_worker.ex index 3b8caf74..d85de881 100644 --- 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,8 +35,13 @@ defmodule Microwaveprop.Workers.StationElevationWorker do station |> Ecto.Changeset.change(elevation_m: elev) |> Repo.update() + |> case do + {:ok, _station} -> + :ok - :ok + {:error, _changeset} -> + Logger.warning("StationElevationWorker: failed to update elevation for station #{station.id}") + end {:error, _} -> :ok diff --git a/lib/microwaveprop_web/live/beacon_live/index.ex b/lib/microwaveprop_web/live/beacon_live/index.ex index 7d5b85f8..24b0b638 100644 --- 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 diff --git a/lib/microwaveprop_web/live/import_live.ex b/lib/microwaveprop_web/live/import_live.ex index a2cb8e2c..466c4046 100644 --- 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" />
diff --git a/lib/microwaveprop_web/live/pskr_spots_live.ex b/lib/microwaveprop_web/live/pskr_spots_live.ex index 3b5bb9e9..5845db61 100644 --- 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,27 +111,32 @@ defmodule MicrowavepropWeb.PskrSpotsLive do Modes - - <%= for spot <- @spots do %> - - {fmt_dt(spot.last_spot_at)} - {spot.band} - - {fmt_callsigns(spot.sender_callsigns)} - - {spot.sender_grid} - - {fmt_callsigns(spot.receiver_callsigns)} - - {spot.receiver_grid} - {Format.distance_km(spot.distance_km)} - {Format.number(spot.spot_count)} - {fmt_snr(spot.min_snr_db, spot.max_snr_db)} - - {mode} - - - <% end %> + + + No spots yet + + + {fmt_dt(spot.last_spot_at)} + {spot.band} + + {fmt_callsigns(spot.sender_callsigns)} + + {spot.sender_grid} + + {fmt_callsigns(spot.receiver_callsigns)} + + {spot.receiver_grid} + {Format.distance_km(spot.distance_km)} + {Format.number(spot.spot_count)} + {fmt_snr(spot.min_snr_db, spot.max_snr_db)} + + {mode} + +
diff --git a/lib/microwaveprop_web/live/rover_planning_live/show.ex b/lib/microwaveprop_web/live/rover_planning_live/show.ex index 1016e925..9fb716a9 100644 --- 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 diff --git a/priv/repo/migrations/20260716122844_add_weather_stations_index.exs b/priv/repo/migrations/20260716122844_add_weather_stations_index.exs new file mode 100644 index 00000000..bbdf457a --- /dev/null +++ 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 diff --git a/priv/repo/migrations/20260716122845_add_contacts_callsign_indexes.exs b/priv/repo/migrations/20260716122845_add_contacts_callsign_indexes.exs new file mode 100644 index 00000000..6b6603e4 --- /dev/null +++ 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