fix: resolve 18 bugs across LiveViews, schemas, tests, and logic
Some checks failed
Build and Push / Build and Push Docker Image (push) Failing after 4m35s

- MonitorLive.Show: safe nil-guard on current_scope for anonymous access
- Admin.MonitorLive.Index: add phx-update=stream to enable stream ops
- ImportLive: require owner/admin authorization, not_found redirect
- MapLive: store timer refs in assigns, cancel before reschedule
- 10 schemas: add missing foreign_key_constraint on belongs_to
- Soundings: preload :station to eliminate N+1 in path analysis
- PathAnalysis: defensive preload of :station on soundings
- GridTaskEnqueuer: wrap reclaim_stale_running in Repo.transaction()
- HrdpsClient: replace String.to_atom with compile-time atom literals
- Contacts: fix extract_latlon false return for lon=0.0
- Tests: remove duplicate Mox.defmock, unblock swallowed task exits,
  bump refute_receive timeouts from 50ms to 200ms
This commit is contained in:
Graham McIntire 2026-07-29 07:46:54 -05:00
parent c4e71e41b3
commit d31e783776
24 changed files with 158 additions and 94 deletions

View file

@ -36,6 +36,13 @@ defmodule Microwaveprop.Accounts.UserApiToken do
@type t :: %__MODULE__{}
@spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
def changeset(token, attrs) do
token
|> cast(attrs, [:name, :user_id])
|> foreign_key_constraint(:user_id)
end
@doc "Returns the bearer-token prefix (e.g. `mwp_`)."
@spec token_prefix() :: String.t()
def token_prefix, do: @prefix

View file

@ -198,6 +198,7 @@ defmodule Microwaveprop.Beacons.Beacon do
|> validate_number(:lon, greater_than_or_equal_to: -180, less_than_or_equal_to: 180)
|> validate_length(:callsign, min: 3, max: 10)
|> validate_grid_format()
|> foreign_key_constraint(:user_id)
end
defp round_coord(nil), do: nil

View file

@ -169,25 +169,35 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
cutoff = DateTime.utc_now() |> DateTime.add(-cutoff_seconds, :second) |> DateTime.truncate(:second)
now = DateTime.truncate(DateTime.utc_now(), :second)
{failed_n, _} =
"grid_tasks"
|> where(
[t],
t.status == "running" and t.claimed_at < ^cutoff and t.attempt >= ^@max_reclaim_attempts
)
|> Repo.update_all(
set: [
status: "failed",
error: "reclaim-orphan: exceeded max_reclaim_attempts",
claimed_at: nil,
updated_at: now
]
)
{failed_n, requeued_n} =
fn ->
{failed_count, _} =
"grid_tasks"
|> where(
[t],
t.status == "running" and t.claimed_at < ^cutoff and t.attempt >= ^@max_reclaim_attempts
)
|> Repo.update_all(
set: [
status: "failed",
error: "reclaim-orphan: exceeded max_reclaim_attempts",
claimed_at: nil,
updated_at: now
]
)
{requeued_n, _} =
"grid_tasks"
|> where([t], t.status == "running" and t.claimed_at < ^cutoff)
|> Repo.update_all(set: [status: "queued", claimed_at: nil, updated_at: now])
{rq_count, _} =
"grid_tasks"
|> where([t], t.status == "running" and t.claimed_at < ^cutoff)
|> Repo.update_all(set: [status: "queued", claimed_at: nil, updated_at: now])
{failed_count, rq_count}
end
|> Repo.transaction()
|> case do
{:ok, {fc, rqc}} -> {fc, rqc}
{:error, reason} -> raise "reclaim_stale_running transaction failed: #{inspect(reason)}"
end
if failed_n > 0 or requeued_n > 0 do
Logger.info("GridTaskEnqueuer: reclaimed stale-running grid_tasks (requeued=#{requeued_n}, failed=#{failed_n})")

View file

@ -11,6 +11,7 @@ defmodule Microwaveprop.Propagation.PathAnalysis do
duplicated duct-detection and mechanism-classification logic.
"""
alias Microwaveprop.Repo
alias Microwaveprop.Terrain.ElevationClient
alias Microwaveprop.Terrain.TerrainAnalysis
@ -564,6 +565,7 @@ defmodule Microwaveprop.Propagation.PathAnalysis do
defp boundary_layer_detail(_), do: nil
defp sounding_ducting_detail(soundings) when is_list(soundings) do
soundings = Repo.preload(soundings, :station)
ducting = Enum.filter(soundings, & &1.ducting_detected)
if ducting == [] do

View file

@ -53,6 +53,7 @@ defmodule Microwaveprop.Pskr.FeatureBin do
record
|> cast(attrs, @cast_fields)
|> validate_required(@required_fields)
|> foreign_key_constraint(:run_id)
|> unique_constraint([:run_id, :band, :feature, :bin_label])
end
end

View file

@ -90,6 +90,8 @@ defmodule Microwaveprop.Radio.Contact do
contact
|> cast(attrs, @required_fields ++ @optional_fields)
|> validate_required(@required_fields)
|> foreign_key_constraint(:user_id)
|> foreign_key_constraint(:flagged_by_user_id)
end
@submission_fields ~w(station1 station2 qso_timestamp mode band grid1 grid2 submitter_email user_declared_prop_mode height1_ft height2_ft private notes)a
@ -127,6 +129,8 @@ defmodule Microwaveprop.Radio.Contact do
|> validate_height(:height2_ft)
|> normalize_blank_notes()
|> validate_length(:notes, max: 2000)
|> foreign_key_constraint(:user_id)
|> foreign_key_constraint(:flagged_by_user_id)
end
# Collapse whitespace-only notes to `nil` so the DB column reflects

View file

@ -42,6 +42,7 @@ defmodule Microwaveprop.Radio.ContactCommonVolumeRadar do
row
|> cast(attrs, @required ++ @optional)
|> validate_required(@required)
|> foreign_key_constraint(:contact_id)
|> unique_constraint(:contact_id)
end
end

View file

@ -308,7 +308,7 @@ defmodule Microwaveprop.Radio.Contacts do
def contact_path_points(%{pos1: pos1, pos2: pos2}), do: build_path_points(extract_latlon(pos1), extract_latlon(pos2))
defp extract_latlon(%{"lat" => lat} = pos) when is_number(lat), do: if(pos["lon"], do: {lat, pos["lon"]})
defp extract_latlon(%{"lat" => lat} = pos) when is_number(lat), do: if(is_number(pos["lon"]), do: {lat, pos["lon"]})
defp extract_latlon(_), do: nil
defp build_path_points({lat1, lon1}, {lat2, lon2}) do

View file

@ -49,6 +49,7 @@ defmodule Microwaveprop.Rover.FixedStation do
|> validate_required([:lat, :lon])
|> validate_number(:lat, greater_than_or_equal_to: -90.0, less_than_or_equal_to: 90.0)
|> validate_number(:lon, greater_than_or_equal_to: -180.0, less_than_or_equal_to: 180.0)
|> foreign_key_constraint(:user_id)
|> unique_constraint(:callsign, name: :fixed_stations_user_id_callsign_index)
end

View file

@ -40,5 +40,6 @@ defmodule Microwaveprop.Rover.Location do
|> validate_number(:lon, greater_than_or_equal_to: -180.0, less_than_or_equal_to: 180.0)
|> validate_inclusion(:status, @statuses)
|> validate_length(:notes, max: 4000)
|> foreign_key_constraint(:user_id)
end
end

View file

@ -72,6 +72,7 @@ defmodule Microwaveprop.RoverPlanning.Mission do
|> validate_number(:station_height_ft, greater_than_or_equal_to: 0.0)
|> validate_length(:notes, max: 4000)
|> validate_at_least_one_station()
|> foreign_key_constraint(:user_id)
end
# The multi-checkbox form sends a hidden empty-string sentinel so

View file

@ -52,6 +52,9 @@ defmodule Microwaveprop.RoverPlanning.Path do
|> validate_required([:mission_id, :rover_location_id, :station_id, :band_mhz, :status])
|> validate_inclusion(:status, @statuses)
|> validate_number(:band_mhz, greater_than: 0)
|> foreign_key_constraint(:mission_id)
|> foreign_key_constraint(:rover_location_id)
|> foreign_key_constraint(:station_id)
|> unique_constraint([:mission_id, :rover_location_id, :station_id, :band_mhz],
name: :rover_mission_paths_unique
)

View file

@ -38,6 +38,7 @@ defmodule Microwaveprop.RoverPlanning.Station do
|> validate_required([:lat, :lon])
|> validate_number(:lat, greater_than_or_equal_to: -90.0, less_than_or_equal_to: 90.0)
|> validate_number(:lon, greater_than_or_equal_to: -180.0, less_than_or_equal_to: 180.0)
|> foreign_key_constraint(:mission_id)
end
# When `input` looks like a callsign or grid, geocode it and populate

View file

@ -53,12 +53,12 @@ defmodule Microwaveprop.Weather.HrdpsClient do
@pressure_var_kinds [{:tmp, "TMP"}, {:depr, "DEPR"}, {:hgt, "HGT"}]
# All variable atoms are generated from compile-time constants
# (@grid_pressure_levels + @pressure_var_kinds = 21 atoms max) — safe
# to use String.to_atom/1 here; not reachable from user input.
# All pressure-level variable atoms are created as atom literals at compile
# time. The bounded cross-product of @grid_pressure_levels × @pressure_var_kinds
# (7 levels × 3 vars = 21 atoms) ensures no runtime atom creation.
@pressure_vars (for level <- @grid_pressure_levels,
{kind, msc} <- @pressure_var_kinds do
atom = String.to_atom("#{kind}_#{level}mb")
atom = :"#{kind}_#{level}mb"
msc_level = "ISBL_#{level |> Integer.to_string() |> String.pad_leading(4, "0")}"
{atom, msc, msc_level}
end)

View file

@ -195,7 +195,8 @@ defmodule Microwaveprop.Weather.Soundings do
s.observed_at,
^timestamp
),
limit: 1
limit: 1,
preload: [:station]
)
|> Repo.one()
|> case do

View file

@ -108,7 +108,7 @@ defmodule MicrowavepropWeb.Admin.MonitorLive.Index do
<th class="w-1"></th>
</tr>
</thead>
<tbody>
<tbody id="monitors" phx-update="stream">
<tr :for={{id, monitor} <- @streams.monitors}>
<td class="font-semibold">{monitor.name}</td>
<td class="text-sm">

View file

@ -16,22 +16,42 @@ defmodule MicrowavepropWeb.ImportLive do
@impl true
def mount(%{"id" => id}, _session, socket) do
case load_run(id) do
nil ->
{:ok,
socket
|> put_flash(:error, "Import not found.")
|> push_navigate(to: ~p"/submit")}
nil -> not_found_redirect(socket)
%ImportRun{} = run -> mount_authorized(socket, run)
end
end
%ImportRun{} = run ->
_ =
if connected?(socket) do
Phoenix.PubSub.subscribe(@pubsub, "csv_import:#{run.id}")
end
defp mount_authorized(socket, run) do
case authorized_to_view?(socket, run) do
:ok ->
if connected?(socket) do
Phoenix.PubSub.subscribe(@pubsub, "csv_import:#{run.id}")
end
{:ok,
socket
|> assign(:page_title, "Import ##{String.slice(run.id, 0, 8)}")
|> assign(:run, run)}
:not_found ->
not_found_redirect(socket)
end
end
defp not_found_redirect(socket) do
{:ok,
socket
|> put_flash(:error, "Import not found.")
|> push_navigate(to: ~p"/")}
end
defp authorized_to_view?(socket, run) do
viewer = socket.assigns[:current_scope] && socket.assigns.current_scope.user
cond do
viewer && viewer.is_admin -> :ok
viewer && viewer.email == run.submitter_email -> :ok
true -> :not_found
end
end

View file

@ -87,7 +87,9 @@ defmodule MicrowavepropWeb.MapLive do
grid_visible: false,
radar_visible: false,
deploy_iso: Calendar.strftime(build_ts, "%Y-%m-%d %H:%M UTC"),
deploy_ago: format_deploy_ago(build_ts)
deploy_ago: format_deploy_ago(build_ts),
pipeline_timer: nil,
advance_timer: nil
)}
end
@ -95,13 +97,29 @@ defmodule MicrowavepropWeb.MapLive do
if connected?(socket) do
:ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated")
:ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:pipeline")
_ = Process.send_after(self(), :refresh_pipeline_status, @pipeline_status_refresh_ms)
_ = Process.send_after(self(), :advance_now_cursor, @advance_now_cursor_ms)
socket = schedule_pipeline_status_timer(socket)
schedule_advance_timer(socket)
else
socket
end
socket
end
defp schedule_pipeline_status_timer(socket) do
cancel_timer_if_exists(socket.assigns[:pipeline_timer])
ref = Process.send_after(self(), :refresh_pipeline_status, @pipeline_status_refresh_ms)
assign(socket, :pipeline_timer, ref)
end
defp schedule_advance_timer(socket) do
cancel_timer_if_exists(socket.assigns[:advance_timer])
ref = Process.send_after(self(), :advance_now_cursor, @advance_now_cursor_ms)
assign(socket, :advance_timer, ref)
end
defp cancel_timer_if_exists(nil), do: :ok
defp cancel_timer_if_exists(ref) when is_reference(ref), do: Process.cancel_timer(ref)
defp cancel_timer_if_exists(_), do: :ok
# Compact relative-time rendering of the build timestamp. Matches the
# format used by Layouts.deploy_stamp/1 so the nav and the map
# sidebar stay consistent.
@ -325,7 +343,7 @@ defmodule MicrowavepropWeb.MapLive do
end
def handle_info(:advance_now_cursor, socket) do
Process.send_after(self(), :advance_now_cursor, @advance_now_cursor_ms)
socket = schedule_advance_timer(socket)
if socket.assigns.tracking_now? do
new_time = closest_to_now(socket.assigns.valid_times)
@ -350,7 +368,7 @@ defmodule MicrowavepropWeb.MapLive do
end
def handle_info(:refresh_pipeline_status, socket) do
Process.send_after(self(), :refresh_pipeline_status, @pipeline_status_refresh_ms)
socket = schedule_pipeline_status_timer(socket)
new_status = PipelineStatus.current()

View file

@ -12,25 +12,33 @@ defmodule MicrowavepropWeb.MonitorLive.Show do
@impl true
def mount(%{"id" => id}, _session, socket) do
user = socket.assigns.current_scope.user
viewer = socket.assigns[:current_scope] && socket.assigns.current_scope.user
monitor = BeaconMonitors.get_monitor!(id)
if monitor.user_id == user.id do
beacons = Repo.all(from b in Beacon, where: b.approved == true, order_by: b.callsign)
measurements = BeaconMeasurements.list_recent_for_monitor(monitor.id, 20)
cond do
viewer && monitor.user_id == viewer.id ->
beacons = Repo.all(from b in Beacon, where: b.approved == true, order_by: b.callsign)
measurements = BeaconMeasurements.list_recent_for_monitor(monitor.id, 20)
{:ok,
socket
|> assign(:page_title, "My Monitor: #{monitor.name}")
|> assign(:monitor, monitor)
|> assign(:beacons, beacons)
|> assign(:measurements, measurements)
|> assign_form(monitor)}
else
{:ok,
socket
|> put_flash(:error, "You don't have access to that monitor.")
|> push_navigate(to: ~p"/users/settings")}
{:ok,
socket
|> assign(:page_title, "My Monitor: #{monitor.name}")
|> assign(:monitor, monitor)
|> assign(:beacons, beacons)
|> assign(:measurements, measurements)
|> assign_form(monitor)}
viewer ->
{:ok,
socket
|> put_flash(:error, "You don't have access to that monitor.")
|> push_navigate(to: ~p"/users/settings")}
true ->
{:ok,
socket
|> put_flash(:info, "Please log in to view this monitor.")
|> push_navigate(to: ~p"/users/log-in")}
end
end

View file

@ -198,7 +198,7 @@ defmodule Microwaveprop.ObanErrorReporterTest do
capture_log(fn -> execute_exception(meta_transient) end)
assert_receive {:transient, 303}
refute_receive {:permanent, 303}, 50
refute_receive {:permanent, 303}, 200
meta_permanent =
build_metadata(%{
@ -213,7 +213,7 @@ defmodule Microwaveprop.ObanErrorReporterTest do
capture_log(fn -> execute_exception(meta_permanent) end)
assert_receive {:permanent, 404}
refute_receive {:transient, 404}, 50
refute_receive {:transient, 404}, 200
end
test "a raising callback does not propagate, and the log still lands" do

View file

@ -657,7 +657,7 @@ defmodule Microwaveprop.PropagationTest do
_ = Propagation.scores_at(10_000, valid_time)
assert_receive {:telemetry, [:microwaveprop, :propagation, :scores_at, :cache], %{hit: true}}
refute_receive {:telemetry, [:microwaveprop, :propagation, :scores_at, :stop], _}, 50
refute_receive {:telemetry, [:microwaveprop, :propagation, :scores_at, :stop], _}, 200
after
:telemetry.detach(handler_id)
end

View file

@ -52,7 +52,7 @@ defmodule Microwaveprop.RepoListenerTest do
msg = notification("contact_status_changed", "not-json-at-all{")
assert {:noreply, %{pid: nil}} = RepoListener.handle_info(msg, %{pid: nil})
refute_receive {:contact_status_changed, _}, 50
refute_receive {:contact_status_changed, _}, 200
end
end
@ -76,7 +76,7 @@ defmodule Microwaveprop.RepoListenerTest do
msg = notification("oban_job_changed", "{\"broken\":")
assert {:noreply, %{pid: nil}} = RepoListener.handle_info(msg, %{pid: nil})
refute_receive {:oban_job_changed, _}, 50
refute_receive {:oban_job_changed, _}, 200
end
end
@ -88,8 +88,8 @@ defmodule Microwaveprop.RepoListenerTest do
msg = notification("contact_status_changed", Jason.encode!(%{"contact_id" => "abc"}))
assert {:noreply, _} = RepoListener.handle_info(msg, %{pid: nil})
refute_receive {:oban_job_changed, _}, 50
refute_receive {:contact_status_changed, _}, 50
refute_receive {:oban_job_changed, _}, 200
refute_receive {:contact_status_changed, _}, 200
end
test "oban_job_changed does not broadcast on db:contact_status" do
@ -99,8 +99,8 @@ defmodule Microwaveprop.RepoListenerTest do
msg = notification("oban_job_changed", Jason.encode!(%{"id" => 1}))
assert {:noreply, _} = RepoListener.handle_info(msg, %{pid: nil})
refute_receive {:contact_status_changed, _}, 50
refute_receive {:oban_job_changed, _}, 50
refute_receive {:contact_status_changed, _}, 200
refute_receive {:oban_job_changed, _}, 200
end
end
@ -117,8 +117,8 @@ defmodule Microwaveprop.RepoListenerTest do
msg = notification("some_other_channel", Jason.encode!(%{"x" => 1}))
assert {:noreply, %{pid: nil}} = RepoListener.handle_info(msg, %{pid: nil})
refute_receive {:contact_status_changed, _}, 50
refute_receive {:oban_job_changed, _}, 50
refute_receive {:contact_status_changed, _}, 200
refute_receive {:oban_job_changed, _}, 200
end
test "ignores arbitrary info messages without crashing" do

View file

@ -8,14 +8,10 @@ defmodule Microwaveprop.ValkeyTest do
import Mox
# Defined here (not in test_helper.exs) so test_helper can load even
# when the precommit chain has purged Mox from the code path.
alias Microwaveprop.Valkey
alias Microwaveprop.Valkey.Conn
alias Microwaveprop.Valkey.MockAdapter
Mox.defmock(MockAdapter, for: Microwaveprop.Valkey.Adapter)
# Register Mox mock as the adapter and fake a Redix connection so
# configured?() returns true, letting command/pipeline run through the mock.
setup do

View file

@ -99,13 +99,9 @@ defmodule Mix.Tasks.Prop.CompareTest do
# read_history is exercised.
File.write!(history_path, "not-valid-json\n", [:append])
try do
ExUnit.CaptureIO.capture_io(fn ->
Compare.run(["--days", "1", "--samples", "10", "--output", output_dir])
end)
catch
:exit, _ -> :ok
end
ExUnit.CaptureIO.capture_io(fn ->
Compare.run(["--days", "1", "--samples", "10", "--output", output_dir])
end)
# The history file remains; new entries appended.
assert File.exists?(history_path)
@ -119,13 +115,9 @@ defmodule Mix.Tasks.Prop.CompareTest do
end
end
try do
ExUnit.CaptureIO.capture_io(fn ->
Compare.run(["--days", "1", "--samples", "100", "--output", output_dir])
end)
catch
:exit, _ -> :ok
end
ExUnit.CaptureIO.capture_io(fn ->
Compare.run(["--days", "1", "--samples", "100", "--output", output_dir])
end)
latest_path = Path.join(output_dir, "latest.json")
@ -140,13 +132,9 @@ defmodule Mix.Tasks.Prop.CompareTest do
test "seeded contact + matching HRRR walks the scoring + write path", %{output_dir: output_dir} do
_ = seed_contact_and_hrrr()
try do
ExUnit.CaptureIO.capture_io(fn ->
Compare.run(["--days", "1", "--samples", "10", "--output", output_dir])
end)
catch
:exit, _ -> :ok
end
ExUnit.CaptureIO.capture_io(fn ->
Compare.run(["--days", "1", "--samples", "10", "--output", output_dir])
end)
# Each successful run writes latest.json + history.jsonl. If the
# task short-circuited (no rows / no model), at least one of these