diff --git a/lib/microwaveprop/accounts.ex b/lib/microwaveprop/accounts.ex index 7d5c25b0..2acfafa4 100644 --- a/lib/microwaveprop/accounts.ex +++ b/lib/microwaveprop/accounts.ex @@ -9,6 +9,7 @@ defmodule Microwaveprop.Accounts do alias Microwaveprop.Accounts.UserApiToken alias Microwaveprop.Accounts.UserNotifier alias Microwaveprop.Accounts.UserToken + alias Microwaveprop.ContextHelpers alias Microwaveprop.Repo ## Database getters alias Microwaveprop.Workers.UserHomeQthLookupWorker @@ -152,11 +153,7 @@ defmodule Microwaveprop.Accounts do defp maybe_enqueue_home_qth_lookup({:ok, %User{id: id, callsign: call} = user}) when is_binary(call) do _ = if home_qth_lookup_enabled?() do - worker = UserHomeQthLookupWorker - - if Code.ensure_loaded?(worker) and function_exported?(worker, :new, 1) do - _ = %{user_id: id} |> worker.new() |> Oban.insert() - end + ContextHelpers.safe_enqueue(UserHomeQthLookupWorker, %{user_id: id}) end {:ok, user} @@ -516,22 +513,24 @@ defmodule Microwaveprop.Accounts do @spec revoke_api_token(User.t(), Ecto.UUID.t()) :: {:ok, UserApiToken.t()} | {:error, :not_found} def revoke_api_token(%User{id: user_id}, token_id) do - case Repo.get_by(UserApiToken, id: token_id, user_id: user_id) do - nil -> + case Ecto.UUID.cast(token_id) do + {:ok, uuid} -> + case Repo.get_by(UserApiToken, id: uuid, user_id: user_id) do + nil -> + {:error, :not_found} + + %UserApiToken{revoked_at: nil} = token -> + token + |> Ecto.Changeset.change(revoked_at: DateTime.utc_now(:second)) + |> Repo.update() + + %UserApiToken{} = token -> + {:ok, token} + end + + :error -> {:error, :not_found} - - %UserApiToken{revoked_at: nil} = token -> - token - |> Ecto.Changeset.change(revoked_at: DateTime.utc_now(:second)) - |> Repo.update() - - %UserApiToken{} = token -> - {:ok, token} end - rescue - # Malformed UUID in the URL — treat as a clean 404 instead of - # bubbling a 500 out of `Repo.get_by`'s UUID cast. - Ecto.Query.CastError -> {:error, :not_found} end ## Token helper diff --git a/lib/microwaveprop/accounts/user.ex b/lib/microwaveprop/accounts/user.ex index 31017878..96c291df 100644 --- a/lib/microwaveprop/accounts/user.ex +++ b/lib/microwaveprop/accounts/user.ex @@ -4,7 +4,7 @@ defmodule Microwaveprop.Accounts.User do import Ecto.Changeset - alias Microwaveprop.Radio.Maidenhead + alias Microwaveprop.Radio.MaidenheadChangesetHelpers, as: MCH @admin_email "graham@mcintire.me" @@ -227,10 +227,10 @@ defmodule Microwaveprop.Accounts.User do user |> cast(attrs, [:home_grid, :home_lat, :home_lon, :home_elevation_m]) |> normalize_home_grid() - |> validate_home_grid_format() - |> derive_home_latlon_from_grid() - |> derive_home_grid_from_latlon() - |> validate_home_latlon_range() + |> MCH.validate_grid(:home_grid, blank_error: false) + |> MCH.derive_latlon(:home_grid, :home_lat, :home_lon) + |> MCH.derive_grid(:home_grid, :home_lat, :home_lon, 10) + |> MCH.validate_latlon(:home_lat, :home_lon) |> require_home_qth_pair() end @@ -243,86 +243,10 @@ defmodule Microwaveprop.Accounts.User do put_change(changeset, :home_grid, nil) grid when is_binary(grid) -> - put_change(changeset, :home_grid, normalize_grid(grid)) + put_change(changeset, :home_grid, MCH.normalize_grid(grid)) end end - defp normalize_grid(grid) do - trimmed = String.trim(grid) - - case String.length(trimmed) do - 4 -> String.upcase(trimmed) - 6 -> String.upcase(String.slice(trimmed, 0, 4)) <> String.downcase(String.slice(trimmed, 4, 2)) - _ -> trimmed - end - end - - defp validate_home_grid_format(changeset) do - case get_field(changeset, :home_grid) do - nil -> - changeset - - grid -> - if Maidenhead.valid?(grid), - do: changeset, - else: add_error(changeset, :home_grid, "must be a 4 or 6 character Maidenhead locator") - end - end - - defp derive_home_latlon_from_grid(changeset) do - cond do - not changeset.valid? -> - changeset - - get_change(changeset, :home_lat) || get_change(changeset, :home_lon) -> - changeset - - grid = get_change(changeset, :home_grid) -> - case Maidenhead.to_latlon(grid) do - {:ok, {lat, lon}} -> - changeset - |> put_change(:home_lat, lat) - |> put_change(:home_lon, lon) - - :error -> - changeset - end - - true -> - changeset - end - end - - # If lat/lon were provided without a grid, encode a 10-char Maidenhead - # grid from the coordinates so the UI always has a human-readable label - # to render and downstream callers can rely on `home_grid` being set - # whenever the QTH is set. Skipped on invalid changesets (lat/lon range - # check has not run yet) and when both fields aren't numeric. - defp derive_home_grid_from_latlon(changeset) do - lat = get_field(changeset, :home_lat) - lon = get_field(changeset, :home_lon) - - cond do - not changeset.valid? -> - changeset - - not (is_number(lat) and is_number(lon)) -> - changeset - - get_field(changeset, :home_grid) not in [nil, ""] -> - changeset - - true -> - put_change(changeset, :home_grid, Maidenhead.from_latlon(lat * 1.0, lon * 1.0, 10)) - end - end - - defp validate_home_latlon_range(changeset) do - changeset - |> validate_number(:home_lat, greater_than_or_equal_to: -90.0, less_than_or_equal_to: 90.0) - |> validate_number(:home_lon, greater_than_or_equal_to: -180.0, less_than_or_equal_to: 180.0) - end - defp require_home_qth_pair(changeset) do if home_qth_satisfied?(changeset), do: changeset, diff --git a/lib/microwaveprop/beacon_measurements.ex b/lib/microwaveprop/beacon_measurements.ex index ecfb4bf6..c28a17d7 100644 --- a/lib/microwaveprop/beacon_measurements.ex +++ b/lib/microwaveprop/beacon_measurements.ex @@ -48,32 +48,25 @@ defmodule Microwaveprop.BeaconMeasurements do defp lookup_beacon(attrs) do attrs = Map.new(attrs, fn {k, v} -> {to_string(k), v} end) - case Map.get(attrs, "beacon_id") do - id when is_binary(id) and byte_size(id) > 0 -> - # Only let monitors report for approved on-the-air beacons. The - # client drops 404s without retry, so an admin un-approving a - # beacon cleanly stops the monitor pointing at it. - query = - from b in Microwaveprop.Beacons.Beacon, - where: b.id == ^id and b.approved == true and b.on_the_air == true, - select: b.id + with id when is_binary(id) and byte_size(id) > 0 <- Map.get(attrs, "beacon_id"), + {:ok, uuid} <- Ecto.UUID.cast(id) do + # Only let monitors report for approved on-the-air beacons. The + # client drops 404s without retry, so an admin un-approving a + # beacon cleanly stops the monitor pointing at it. + query = + from b in Microwaveprop.Beacons.Beacon, + where: b.id == ^uuid and b.approved == true and b.on_the_air == true, + select: b.id - case safe_one(query) do - nil -> :not_found - uuid -> {:ok, uuid} - end - - _ -> - :not_found + case Repo.one(query) do + nil -> :not_found + uuid -> {:ok, uuid} + end + else + _ -> :not_found end end - defp safe_one(query) do - Repo.one(query) - rescue - Ecto.Query.CastError -> nil - end - # The wire payload uses string keys (Phoenix JSON params). Make sure # we don't drop a value because some intermediate handed us atom keys. defp normalize_attrs(attrs) do diff --git a/lib/microwaveprop/beacon_monitors.ex b/lib/microwaveprop/beacon_monitors.ex index 9bbf9ece..afdb13e9 100644 --- a/lib/microwaveprop/beacon_monitors.ex +++ b/lib/microwaveprop/beacon_monitors.ex @@ -8,7 +8,6 @@ defmodule Microwaveprop.BeaconMonitors do import Ecto.Query - alias Ecto.Query.CastError alias Microwaveprop.Accounts.User alias Microwaveprop.BeaconMonitors.BeaconMonitor alias Microwaveprop.Repo @@ -143,16 +142,20 @@ defmodule Microwaveprop.BeaconMonitors do @spec delete_monitor(User.t(), Ecto.UUID.t()) :: {:ok, BeaconMonitor.t()} | {:error, :not_found} def delete_monitor(%User{id: user_id}, monitor_id) do - query = - from m in BeaconMonitor, - where: m.id == ^monitor_id and m.user_id == ^user_id + case Ecto.UUID.cast(monitor_id) do + {:ok, uuid} -> + query = + from m in BeaconMonitor, + where: m.id == ^uuid and m.user_id == ^user_id - case Repo.one(query) do - nil -> {:error, :not_found} - monitor -> Repo.delete(monitor) + case Repo.one(query) do + nil -> {:error, :not_found} + monitor -> Repo.delete(monitor) + end + + :error -> + {:error, :not_found} end - rescue - CastError -> {:error, :not_found} end @doc """ @@ -160,12 +163,16 @@ defmodule Microwaveprop.BeaconMonitors do """ @spec delete_monitor!(Ecto.UUID.t()) :: {:ok, BeaconMonitor.t()} | {:error, :not_found} def delete_monitor!(monitor_id) do - case Repo.get(BeaconMonitor, monitor_id) do - nil -> {:error, :not_found} - monitor -> Repo.delete(monitor) + case Ecto.UUID.cast(monitor_id) do + {:ok, uuid} -> + case Repo.get(BeaconMonitor, uuid) do + nil -> {:error, :not_found} + monitor -> Repo.delete(monitor) + end + + :error -> + {:error, :not_found} end - rescue - CastError -> {:error, :not_found} end # ── Auth / heartbeat ───────────────────────────────────────────── diff --git a/lib/microwaveprop/beacon_monitors/beacon_monitor.ex b/lib/microwaveprop/beacon_monitors/beacon_monitor.ex index a9492c9a..befadbc4 100644 --- a/lib/microwaveprop/beacon_monitors/beacon_monitor.ex +++ b/lib/microwaveprop/beacon_monitors/beacon_monitor.ex @@ -14,6 +14,7 @@ defmodule Microwaveprop.BeaconMonitors.BeaconMonitor do alias Microwaveprop.Accounts.User alias Microwaveprop.Beacons.Beacon + alias Microwaveprop.Radio.MaidenheadChangesetHelpers, as: MCH @primary_key {:id, :binary_id, autogenerate: true} @foreign_key_type :binary_id @@ -90,8 +91,7 @@ defmodule Microwaveprop.BeaconMonitors.BeaconMonitor do |> validate_length(:antenna_type, max: 100) |> validate_length(:callsign, max: 20) |> validate_number(:antenna_gain_dbi, greater_than_or_equal_to: -20, less_than_or_equal_to: 60) - |> validate_number(:lat, greater_than_or_equal_to: -90, less_than_or_equal_to: 90) - |> validate_number(:lon, greater_than_or_equal_to: -180, less_than_or_equal_to: 180) + |> MCH.validate_latlon(:lat, :lon) |> foreign_key_constraint(:user_id) |> foreign_key_constraint(:assigned_by_id) end diff --git a/lib/microwaveprop/beacons.ex b/lib/microwaveprop/beacons.ex index 2caa036d..8c9e1adc 100644 --- a/lib/microwaveprop/beacons.ex +++ b/lib/microwaveprop/beacons.ex @@ -104,9 +104,12 @@ defmodule Microwaveprop.Beacons do end defp fetch_beacon(id) do - Beacon |> Repo.get(id) |> Repo.preload(:user) - rescue - Ecto.Query.CastError -> nil + with {:ok, uuid} <- Ecto.UUID.cast(id), + %Beacon{} = beacon <- Repo.get(Beacon, uuid) do + Repo.preload(beacon, :user) + else + _ -> nil + end end defp can_view?(%Beacon{approved: true}, _viewer), do: true diff --git a/lib/microwaveprop/beacons/beacon.ex b/lib/microwaveprop/beacons/beacon.ex index ba5eb364..05c13593 100644 --- a/lib/microwaveprop/beacons/beacon.ex +++ b/lib/microwaveprop/beacons/beacon.ex @@ -9,7 +9,7 @@ defmodule Microwaveprop.Beacons.Beacon do import Ecto.Changeset - alias Microwaveprop.Radio.Maidenhead + alias Microwaveprop.Radio.MaidenheadChangesetHelpers, as: MCH @keying_entries [ {"on_off", "On/Off"}, @@ -180,12 +180,12 @@ defmodule Microwaveprop.Beacons.Beacon do :beamwidth_deg, :notes ]) - |> update_change(:callsign, fn cs -> cs && String.upcase(String.trim(cs)) end) + |> MCH.normalize_callsign(:callsign) |> update_change(:lat, &round_coord/1) |> update_change(:lon, &round_coord/1) |> update_change(:height_ft, &round_int/1) - |> maybe_fill_latlon() - |> maybe_fill_grid() + |> MCH.derive_latlon(:grid, :lat, :lon) + |> MCH.derive_grid(:grid, :lat, :lon) |> normalize_bearing_change() |> validate_required(@required_fields) |> validate_inclusion(:keying, @keyings) @@ -194,10 +194,9 @@ defmodule Microwaveprop.Beacons.Beacon do |> validate_number(:frequency_mhz, greater_than: 0) |> validate_number(:power_mw, greater_than_or_equal_to: 0) |> validate_number(:height_ft, greater_than_or_equal_to: 0) - |> validate_number(:lat, greater_than_or_equal_to: -90, less_than_or_equal_to: 90) - |> validate_number(:lon, greater_than_or_equal_to: -180, less_than_or_equal_to: 180) + |> MCH.validate_latlon(:lat, :lon) |> validate_length(:callsign, min: 3, max: 10) - |> validate_grid_format() + |> MCH.validate_grid(:grid) |> foreign_key_constraint(:user_id) end @@ -210,40 +209,6 @@ defmodule Microwaveprop.Beacons.Beacon do defp round_int(v) when is_integer(v), do: v defp round_int(v), do: v - # If grid is blank but lat/lon are valid, derive it. - defp maybe_fill_grid(changeset) do - grid = get_field(changeset, :grid) - lat = get_field(changeset, :lat) - lon = get_field(changeset, :lon) - - if grid in [nil, ""] and is_number(lat) and is_number(lon) do - put_change(changeset, :grid, Maidenhead.from_latlon(lat, lon, 6)) - else - changeset - end - end - - # If lat/lon are blank but grid is a valid Maidenhead, derive them. - defp maybe_fill_latlon(changeset) do - grid = get_field(changeset, :grid) - lat = get_field(changeset, :lat) - lon = get_field(changeset, :lon) - - if is_binary(grid) and grid != "" and (lat in [nil, ""] or lon in [nil, ""]) do - case Maidenhead.to_latlon(grid) do - {:ok, {new_lat, new_lon}} -> - changeset - |> put_change(:lat, Float.round(new_lat * 1.0, 6)) - |> put_change(:lon, Float.round(new_lon * 1.0, 6)) - - :error -> - changeset - end - else - changeset - end - end - # Normalize bearing: trim, treat nil/blank/"omni" (any case) as "omni". # Anything else is left for validate_bearing to check as a number. defp normalize_bearing_change(changeset) do @@ -289,32 +254,4 @@ defmodule Microwaveprop.Beacons.Beacon do changeset end end - - defp validate_grid_format(changeset) do - case get_field(changeset, :grid) do - nil -> - add_error(changeset, :grid, "can't be blank") - - grid -> - if Maidenhead.valid?(grid) do - update_change(changeset, :grid, &normalize_grid/1) - else - add_error(changeset, :grid, "is not a valid Maidenhead grid") - end - end - end - - # Field uppercase, square digits, subsquare lowercase (standard form) - defp normalize_grid(nil), do: nil - - defp normalize_grid(grid) do - grid = String.trim(grid) - len = String.length(grid) - - if len >= 6 do - grid |> String.slice(0, 4) |> String.upcase() |> Kernel.<>(String.downcase(String.slice(grid, 4, len - 4))) - else - String.upcase(grid) - end - end end diff --git a/lib/microwaveprop/context_helpers.ex b/lib/microwaveprop/context_helpers.ex new file mode 100644 index 00000000..d424d501 --- /dev/null +++ b/lib/microwaveprop/context_helpers.ex @@ -0,0 +1,67 @@ +defmodule Microwaveprop.ContextHelpers do + @moduledoc """ + Shared helpers for context modules: owner-scoped fetches, safe Oban + enqueuing, and UUID casting. + """ + + alias Microwaveprop.Accounts.User + alias Microwaveprop.Repo + + @doc """ + Fetches a record by id, scoped to the given user. Admins bypass + ownership — they can fetch any record. Regular users can only fetch + records where `user_id` matches. + + Validates that `id` is a UUID before hitting the database; non-UUID + strings return `{:error, :not_found}` instead of raising CastError. + + Returns `{:ok, record}` or `{:error, :not_found}`. + + ## Examples + + fetch_owned(MySchema, "some-uuid", admin_user) + fetch_owned(MySchema, "some-uuid", regular_user) + """ + @spec fetch_owned(module(), Ecto.UUID.t(), User.t()) :: {:ok, Ecto.Schema.t()} | {:error, :not_found} + def fetch_owned(schema, id, user) do + with {:ok, uuid} <- cast_uuid(id) do + do_fetch_owned(schema, uuid, user) + end + end + + defp do_fetch_owned(schema, uuid, %User{is_admin: true}) do + case Repo.get(schema, uuid) do + nil -> {:error, :not_found} + record -> {:ok, record} + end + end + + defp do_fetch_owned(schema, uuid, %User{id: user_id}) do + case Repo.get(schema, uuid) do + %{user_id: ^user_id} = record -> {:ok, record} + _ -> {:error, :not_found} + end + end + + @doc """ + Safe Oban enqueue — only enqueues if the worker module is loaded and + exports `new/1`. Returns whatever `Oban.insert/1` returns on success, + or `{:error, :worker_not_available}` if the worker is not available. + """ + @spec safe_enqueue(module(), map()) :: {:ok, Oban.Job.t()} | {:error, :worker_not_available | Ecto.Changeset.t()} + def safe_enqueue(worker, args \\ %{}) do + if Code.ensure_loaded?(worker) and function_exported?(worker, :new, 1) do + args |> worker.new() |> Oban.insert() + else + {:error, :worker_not_available} + end + end + + @doc """ + Safely casts a string to a UUID. Returns `{:ok, uuid}` or `:error`. + Use this at API/controller boundaries so contexts never receive bad UUIDs. + """ + @spec cast_uuid(binary()) :: {:ok, Ecto.UUID.t()} | :error + def cast_uuid(id) when is_binary(id), do: Ecto.UUID.cast(id) + def cast_uuid(_), do: :error +end diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 60525872..a934957b 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -4,14 +4,13 @@ defmodule Microwaveprop.Propagation do import Ecto.Query alias Microwaveprop.Propagation.BandConfig - alias Microwaveprop.Propagation.Grid - alias Microwaveprop.Propagation.ProfilesFile alias Microwaveprop.Propagation.RunTiming - alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.Scorer - alias Microwaveprop.Propagation.ScoresFile + alias Microwaveprop.Propagation.ScoreStore alias Microwaveprop.Repo - alias Microwaveprop.Weather.ScalarFile + + # ── ML Model Lifecycle ────────────────────────────────────────────── + alias Microwaveprop.Weather.SoundingParams require Logger @@ -41,6 +40,8 @@ defmodule Microwaveprop.Propagation do end else Logger.info("PropagationML: ML dependencies not available") + + # ── Scoring ───────────────────────────────────────────────────────── :ok end end @@ -174,494 +175,30 @@ defmodule Microwaveprop.Propagation do max(hrrr_rate, nexrad_rate) end - @doc """ - Replace every propagation score for `valid_time` with `scores`. + defdelegate replace_scores(scores, valid_time), to: ScoreStore + defdelegate prune_old_scores(), to: ScoreStore + defdelegate retain_scores_window(run_time), to: ScoreStore + defdelegate available_valid_times(band_mhz), to: ScoreStore + # ── Delegates to ScoreStore (file I/O + cache) ────────────────────── + defdelegate hot_cache_window(), to: ScoreStore + defdelegate scores_at(band_mhz, valid_time, bounds \\ nil), to: ScoreStore + defdelegate scores_at_fresh(band_mhz, valid_time, bounds \\ nil), to: ScoreStore + defdelegate warm_cache_and_broadcast(band_mhz, valid_time), to: ScoreStore + defdelegate latest_scores(band_mhz, bounds \\ nil), to: ScoreStore + defdelegate point_forecast(band_mhz, lat, lon), to: ScoreStore + defdelegate point_detail(band_mhz, lat, lon, valid_time \\ nil), to: ScoreStore + defdelegate latest_valid_time(), to: ScoreStore - Used by `PropagationGridWorker` on the hot path. Scores are written - as binary files on disk via `ScoresFile.write!/3`, one file per - band. - - Consumes `scores` in a single streaming pass that folds each score - straight into a per-band accumulator. Previously this function ran - `Enum.to_list/1` followed by `Enum.group_by/2`, which held two full - copies of the ~460k-entry grid (list + grouped list) in memory at - once — the hot path's largest transient spike after native-duct - merge. The single-pass reduce keeps only one copy and buys back - ~100 MB of headroom per forecast-hour step. - """ - @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} - def replace_scores(scores, %DateTime{} = valid_time) do - do_replace_scores(scores, valid_time) - end - - defp do_replace_scores(scores, valid_time) do - # Pure grouping phase runs outside the telemetry span — typically - # <10ms on small result sets, and the span's two dispatches - # (~100µs each) would otherwise dominate. The span now wraps only - # the per-band writes, which is where the actual DB cost lives. - {per_band, total} = - Enum.reduce(scores, {%{}, 0}, fn score, {acc, count} -> - {Map.update(acc, score.band_mhz, [score], &[score | &1]), count + 1} - end) - - Microwaveprop.Instrument.span( - [:db, :replace_scores], - %{valid_time: valid_time}, - fn -> - Enum.each(per_band, fn {band_mhz, band_scores} -> - try do - ScoresFile.write!(band_mhz, valid_time, band_scores) - rescue - e -> - Logger.warning("Propagation: ScoresFile write failed for band=#{band_mhz} vt=#{valid_time}: #{inspect(e)}") - end - end) - - {:ok, total} - end - ) - end + # ── Scoring helpers called from ScoreStore ────────────────────────── + defdelegate latest_valid_time(band_mhz), to: ScoreStore @doc """ - Remove score files with valid_times older than 3 hours. Called on - a cron by `Microwaveprop.Workers.PropagationPruneWorker`. - - The cutoff sits one hour beyond HRRR's ~2h publish lag: the hourly - seeder picks `run_time = now - 2h`, so the f00 analysis file is - written at valid_time = now - 2h. A 2h cutoff deletes it within - minutes; a 3h cutoff keeps it alive until the next hourly run - supersedes it. + Rebuild the factor breakdown for a clicked grid cell by rescoring + the persisted HRRR profile. Made public so ScoreStore can call it + after reading the profile from disk. """ - @spec prune_old_scores() :: :ok - def prune_old_scores do - cutoff = DateTime.shift(DateTime.utc_now(), hour: -3) - file_deleted = ScoresFile.prune_older_than(cutoff) - profiles_deleted = ProfilesFile.prune_older_than(cutoff) - scalar_deleted = ScalarFile.prune_older_than(cutoff) - - total = file_deleted + profiles_deleted + scalar_deleted - - if total > 0 do - Logger.info( - "PropagationScores: pruned #{file_deleted} old score files + " <> - "#{profiles_deleted} profile files + #{scalar_deleted} scalar dirs " <> - "(before #{cutoff})" - ) - end - - # Sweep orphaned .tmp.* files left by crashed atomic-write processes - tmp_deleted = - Enum.reduce([ScoresFile.base_dir(), ProfilesFile.base_dir(), ScalarFile.base_dir()], 0, &sweep_tmp_dir/2) - - if tmp_deleted > 0 do - Logger.info("PropagationScores: swept #{tmp_deleted} orphaned .tmp files") - end - - :ok - end - - defp sweep_tmp_dir(dir, acc) do - case File.ls(dir) do - {:ok, entries} -> - Enum.reduce(entries, acc, &sweep_tmp_entry(dir, &1, &2)) - - _ -> - acc - end - end - - defp sweep_tmp_entry(parent, entry, acc) do - full = Path.join(parent, entry) - - case File.ls(full) do - {:ok, _} -> - sweep_tmp_dir(full, acc) - - {:error, _} -> - if String.contains?(entry, ".tmp.") do - _ = File.rm_rf(full) - acc + 1 - else - acc - end - end - end - - @doc """ - Retains score files through GEFS's 168-hour horizon and profile/scalar - files through HRRR's 48-hour horizon, deleting files older than - `run_time`. Called - by `NotifyListener` after chain completion to keep `/data/scores` - within bounds. - - Mirrors `ScoreCache.prune_outside_window` for the on-disk tier. - """ - @spec retain_scores_window(DateTime.t()) :: :ok - def retain_scores_window(%DateTime{} = run_time) do - scores_deleted = ScoresFile.retain_window(run_time, 168) - profiles_deleted = ProfilesFile.retain_window(run_time, 48) - scalars_deleted = ScalarFile.retain_window(run_time, 48) - total = scores_deleted + profiles_deleted + scalars_deleted - - if total > 0 do - Logger.info( - "Propagation.retain_scores_window: deleted #{total} stale files (#{scores_deleted} scores, #{profiles_deleted} profiles, #{scalars_deleted} scalars)" - ) - end - - :ok - end - - @doc """ - Returns distinct valid_times for a band, ordered ascending. Always - reads from the on-disk `ScoresFile` store — the `ScoreCache` only - holds whatever hours have been fetched or broadcast, which can be a - partial view of what's actually on disk, so using it as the source - of truth for the timeline makes new forecast hours invisible until - the cache happens to catch up. Filters out times more than 1 hour in - the past, but always includes the most recent valid_time so there's - always data to display. - """ - # HRRR forecast horizon: f00..f48 covers the next 48 hours from cycle - # time (f01-f18 hourly, f21-f48 3-hourly). Anything beyond that in the - # score store is a leftover from a stale cycle and clutters the - # timeline without adding information. - @hrrr_forecast_horizon_hours 48 - - @spec available_valid_times(non_neg_integer()) :: [DateTime.t()] - def available_valid_times(band_mhz) do - {past_cutoff, future_cutoff} = hot_cache_window() - - case ScoresFile.list_valid_times(band_mhz) do - [] -> [] - times -> filter_or_latest(times, past_cutoff, future_cutoff) - end - end - - @doc """ - The active forecast window for the `/map` UI: one hour in the past - through HRRR's 48-hour forecast horizon. Used by `NotifyListener` to - bound ETS growth — long-horizon GEFS `.prop` files on disk must not - balloon `propagation_score_cache` past the memory the UI actually - reads. - """ - @spec hot_cache_window() :: {DateTime.t(), DateTime.t()} - def hot_cache_window do - now = DateTime.utc_now() - past = DateTime.shift(now, hour: -1) - future = DateTime.shift(now, hour: @hrrr_forecast_horizon_hours) - {past, future} - end - - defp filter_or_latest(times, past_cutoff, future_cutoff) do - fresh = - Enum.filter(times, fn t -> - DateTime.compare(t, past_cutoff) != :lt and DateTime.compare(t, future_cutoff) != :gt - end) - - if fresh == [] do - [Enum.max(times, DateTime)] - else - fresh - end - end - - @doc """ - Get scores for a band at a specific valid_time, optionally within a bounding box. - If valid_time is nil, uses the earliest available (current analysis hour). - Excludes factors for performance. - """ - @spec scores_at(non_neg_integer(), DateTime.t() | nil, %{optional(String.t()) => float()} | nil) :: - [%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}] - def scores_at(band_mhz, valid_time, bounds \\ nil) do - time = valid_time || earliest_valid_time(band_mhz) - - case time do - nil -> [] - _ -> scores_at_fetch(band_mhz, time, bounds) - end - end - - # Cache-hit path is the map's most frequent LiveView call (~every - # pan + click). A wrapping Instrument.span fires 2 telemetry handler - # dispatches that dominate the ~10µs ETS lookup — skip the span on - # hits and rely on the cheap hit/miss counter for the cache-ratio - # panel. The miss path still wraps the disk read where duration is - # the meaningful signal. - defp scores_at_fetch(band_mhz, time, bounds) do - case ScoreCache.fetch_bounds(band_mhz, time, bounds) do - {:ok, scores} -> - :telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: true}) - Enum.map(scores, &Map.put(&1, :valid_time, time)) - - :miss -> - :telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: false}) - - Microwaveprop.Instrument.span([:propagation, :scores_at], %{band_mhz: band_mhz}, fn -> - read_from_disk_and_cache(band_mhz, time, bounds) - end) - end - end - - @doc """ - Variant of `scores_at/3` that always reads from the `.prop` file on - disk and overwrites the cache entry, rather than returning whatever - the cache happens to hold. Use from update paths (the map's - `propagation_updated` handler) where the underlying file has just - been rewritten but the cache may still contain the previous chain's - scores because of the race between `propagation:cache` fan-out and - `propagation:updated` delivery. - """ - @spec scores_at_fresh(non_neg_integer(), DateTime.t(), %{optional(String.t()) => float()} | nil) :: - [%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}] - def scores_at_fresh(band_mhz, %DateTime{} = valid_time, bounds \\ nil) do - read_from_disk_and_cache(band_mhz, valid_time, bounds) - end - - defp read_from_disk_and_cache(band_mhz, time, bounds) do - full = ScoresFile.read_bounds(band_mhz, time) - ScoreCache.put(band_mhz, time, full) - - full - |> filter_bounds(bounds) - |> Enum.map(&Map.put(&1, :valid_time, time)) - end - - @doc """ - Load the full North America score set for `{band_mhz, valid_time}` from - the on-disk binary files (HRRR `.prop` + HRDPS `.hrdps.prop`, merged) and - broadcast it to every `ScoreCache` in the cluster. Called from - `PropagationGridWorker` after each forecast hour so all pods have a warm - cache by the time clients begin requesting the new hour. - - Returns `{:error, :enoent}` only when neither file exists. A single - missing file (HRDPS pre-cycle, HRRR briefly absent) is OK — the cache - warms with whichever side is available. - """ - @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) :: - :ok | {:error, :enoent | :invalid_format} - def warm_cache_and_broadcast(band_mhz, valid_time) do - hrrr = read_score_points(&ScoresFile.read/2, band_mhz, valid_time) - hrdps = read_score_points(&ScoresFile.read_hrdps/2, band_mhz, valid_time) - - # Merge priority: HRRR > HRDPS > GEFS. GEFS provides extended-horizon - # coverage beyond HRRR's 48h window. Within the f24-f48 overlap, cells - # already present in HRRR/HRDPS are skipped so the coarser GEFS scores - # don't override the higher-resolution ones. - merged = (hrrr || []) ++ (hrdps || []) - gefs = read_score_points(&ScoresFile.read_gefs/2, band_mhz, valid_time) || [] - scores = ScoresFile.merge_preferred(merged, gefs) - - case {hrrr, hrdps, scores} do - {nil, nil, []} -> - {:error, :enoent} - - _ -> - ScoreCache.broadcast_put(band_mhz, valid_time, scores) - :ok - end - end - - defp read_score_points(reader, band_mhz, valid_time) do - case reader.(band_mhz, valid_time) do - {:ok, payload} -> ScoresFile.extract_points(payload, nil) - _ -> nil - end - end - - defp filter_bounds(scores, nil), do: scores - - defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do - Enum.filter(scores, fn %{lat: lat, lon: lon} -> - lat >= s and lat <= n and lon >= w and lon <= e - end) - end - - @doc "Get the latest scores for a band (alias for scores_at with earliest valid_time)." - @spec latest_scores(non_neg_integer(), %{optional(String.t()) => float()} | nil) :: - [%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}] - def latest_scores(band_mhz, bounds \\ nil) do - scores_at(band_mhz, nil, bounds) - end - - defp earliest_valid_time(band_mhz) do - case ScoresFile.list_valid_times(band_mhz) do - [earliest | _] -> earliest - [] -> nil - end - end - - @doc "Get scores across all forecast hours for a single grid point (for sparkline)." - @spec point_forecast(non_neg_integer(), float(), float()) :: - [%{valid_time: DateTime.t(), score: non_neg_integer()}] - def point_forecast(band_mhz, lat, lon) do - Microwaveprop.Instrument.span([:propagation, :point_forecast], %{band_mhz: band_mhz}, fn -> - {snapped_lat, snapped_lon} = snap_to_grid(lat, lon) - now = DateTime.utc_now() - - # Use the on-disk .prop file list as the authoritative timeline so - # the chart never falls behind the main-map timeline (which also - # reads the disk). The cache is still consulted per-hour for a - # fast score lookup; a miss falls through to the file. - # Fan the per-hour disk lookups across 4 tasks. Each ScoresFile - # read_point is an NFS stat + pread of ~100 bytes (keyed byte at - # row*cols+col), so the ceiling is NFS RTT × number of hours — - # sequential ran ~4–5× the wall time of the slowest read. - band_mhz - |> ScoresFile.list_valid_times() - |> forecast_window(now) - |> Task.async_stream( - &point_forecast_entry(band_mhz, &1, snapped_lat, snapped_lon), - max_concurrency: 4, - ordered: true, - timeout: 5_000 - ) - |> Enum.flat_map(fn - {:ok, nil} -> - [] - - {:ok, entry} -> - [entry] - - {:exit, reason} -> - Logger.error( - "Propagation.point_forecast async lookup failed: band_mhz=#{band_mhz} lat=#{snapped_lat} lon=#{snapped_lon} reason=#{inspect(reason)}" - ) - - [] - end) - end) - end - - defp point_forecast_entry(band_mhz, valid_time, lat, lon) do - case ScoreCache.fetch_point(band_mhz, valid_time, lat, lon) do - {:ok, score} -> - %{valid_time: valid_time, score: score} - - :miss -> - case ScoresFile.read_point(band_mhz, valid_time, lat, lon) do - nil -> nil - score -> %{valid_time: valid_time, score: score} - end - end - end - - # Select the set of valid_times the forecast chart should render. - # Mirrors `available_valid_times`: keep everything from one hour - # before now onward so the most recent analysis hour (typically - # ~30–60 min behind wall clock due to HRRR publishing lag) sits at - # the left edge of the chart as "now". When every hour on disk is - # older than that cutoff, fall back to just the newest entry so the - # chart can still render a single data point. - defp forecast_window([], _now), do: [] - - defp forecast_window(times, now) do - past_cutoff = DateTime.shift(now, hour: -1) - future_cutoff = DateTime.shift(now, hour: @hrrr_forecast_horizon_hours) - filter_or_latest(times, past_cutoff, future_cutoff) - end - - defp snap_to_grid(lat, lon) do - step = Grid.step() - {Float.round(Float.round(lat / step) * step, 3), Float.round(Float.round(lon / step) * step, 3)} - end - - @doc """ - Get the full score and factors for a specific grid point, snapped to - the nearest grid cell. - - `:profile_source` describes where the factor breakdown came from: - - * `:exact` — rescored from this `valid_time`'s own profile file. - * `{:fallback, fallback_valid_time}` — the requested hour's profile - was missing, so we rescored from the most recent analysis profile - within the lookback window. Treat as approximate. - * `:unavailable` — no profile available; `factors` is `%{}`. - """ - @spec point_detail(non_neg_integer(), float(), float(), DateTime.t() | nil) :: - %{ - lat: float(), - lon: float(), - score: non_neg_integer(), - factors: map(), - profile_source: :exact | {:fallback, DateTime.t()} | :unavailable, - valid_time: DateTime.t() - } - | nil - def point_detail(band_mhz, lat, lon, valid_time \\ nil) do - Microwaveprop.Instrument.span([:propagation, :point_detail], %{band_mhz: band_mhz}, fn -> - do_point_detail(band_mhz, lat, lon, valid_time) - end) - end - - defp do_point_detail(band_mhz, lat, lon, valid_time) do - {snapped_lat, snapped_lon} = snap_to_grid(lat, lon) - time = valid_time || latest_valid_time(band_mhz) - - case time do - nil -> - nil - - _ -> - case ScoresFile.read_point(band_mhz, time, snapped_lat, snapped_lon) do - nil -> - nil - - score -> - {factors, source} = factors_for(band_mhz, time, snapped_lat, snapped_lon) - - %{ - lat: snapped_lat, - lon: snapped_lon, - score: score, - factors: factors, - profile_source: source, - valid_time: time - } - end - end - end - - # Rebuild the factor breakdown for a clicked grid cell by rescoring - # the persisted HRRR profile. - # - # The Rust pipeline (`prop_grid_rs`) writes a per-cell profile file - # for every chain step (f00..f18 since Phase 2 cutover), so the - # `:exact` branch covers a healthy production state. The fallback to - # a recent analysis profile remains as a safety net for missed - # cycles or partial chain runs — when it kicks in we tag the result - # `{:fallback, profile_valid_time}` so the UI can label the - # breakdown as approximated rather than silently misrepresent it. - @fallback_profile_lookback_hours 24 - @spec factors_for(non_neg_integer(), DateTime.t(), float(), float()) :: - {map(), :exact | {:fallback, DateTime.t()} | :unavailable} - defp factors_for(band_mhz, valid_time, lat, lon) do - case ProfilesFile.read_point(valid_time, lat, lon) do - nil -> - factors_from_fallback_profile(band_mhz, valid_time, lat, lon) - - profile -> - {factors_from_profile(band_mhz, valid_time, profile, lat, lon), :exact} - end - end - - defp factors_from_fallback_profile(band_mhz, valid_time, lat, lon) do - case latest_profile_time_within_lookback(valid_time) do - nil -> - {%{}, :unavailable} - - fallback_time -> - case ProfilesFile.read_point(fallback_time, lat, lon) do - nil -> - {%{}, :unavailable} - - profile -> - {factors_from_profile(band_mhz, fallback_time, profile, lat, lon), {:fallback, fallback_time}} - end - end - end - - defp factors_from_profile(band_mhz, valid_time, profile, lat, lon) do + @spec factors_from_profile(non_neg_integer(), DateTime.t(), map(), float(), float()) :: map() + def factors_from_profile(band_mhz, valid_time, profile, lat, lon) do profile |> Map.put(:kp_index, current_kp_index()) |> score_grid_point(valid_time, lat, lon) @@ -684,38 +221,10 @@ defmodule Microwaveprop.Propagation do %{estimated_kp: kp} when is_number(kp) -> trunc(kp) _ -> nil end + + # ── Run timings ───────────────────────────────────────────────────── end - defp latest_profile_time_within_lookback(%DateTime{} = valid_time) do - lookback_cutoff = DateTime.shift(valid_time, hour: -@fallback_profile_lookback_hours) - - ProfilesFile.list_valid_times() - |> Enum.filter(fn t -> - DateTime.compare(t, valid_time) != :gt and DateTime.compare(t, lookback_cutoff) != :lt - end) - |> case do - [] -> nil - past -> Enum.max(past, DateTime) - end - end - - @doc "Get the latest valid_time across all bands." - @spec latest_valid_time() :: DateTime.t() | nil - def latest_valid_time do - ScoresFile.latest_valid_time() - end - - @doc "Get the latest valid_time for a specific band." - @spec latest_valid_time(non_neg_integer()) :: DateTime.t() | nil - def latest_valid_time(band_mhz) do - case ScoresFile.list_valid_times(band_mhz) do - [] -> nil - times -> Enum.max(times, DateTime) - end - end - - ## Run timings - @doc """ Record wall-clock duration for a single forecast-hour chain step. @@ -746,6 +255,8 @@ defmodule Microwaveprop.Propagation do |> Repo.all() end + # ── Derived factors ───────────────────────────────────────────────── + # Prefer the persisted scalar — `hrrr_profiles` already stored this at # ingestion time and AsosAdjustmentWorker loads 92k rows per tick without # the JSONB `profile` column to avoid a Jason.decode! storm on the DB pool. diff --git a/lib/microwaveprop/propagation/score_store.ex b/lib/microwaveprop/propagation/score_store.ex new file mode 100644 index 00000000..d07f8d23 --- /dev/null +++ b/lib/microwaveprop/propagation/score_store.ex @@ -0,0 +1,528 @@ +defmodule Microwaveprop.Propagation.ScoreStore do + @moduledoc false + + alias Microwaveprop.Instrument + alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Propagation.ProfilesFile + alias Microwaveprop.Propagation.ScoreCache + alias Microwaveprop.Propagation.ScoresFile + alias Microwaveprop.Weather.ScalarFile + + require Logger + + @hrrr_forecast_horizon_hours 48 + @fallback_profile_lookback_hours 24 + + @doc """ + Replace every propagation score for `valid_time` with `scores`. + + Used by `PropagationGridWorker` on the hot path. Scores are written + as binary files on disk via `ScoresFile.write!/3`, one file per + band. + + Consumes `scores` in a single streaming pass that folds each score + straight into a per-band accumulator. Previously this function ran + `Enum.to_list/1` followed by `Enum.group_by/2`, which held two full + copies of the ~460k-entry grid (list + grouped list) in memory at + once — the hot path's largest transient spike after native-duct + merge. The single-pass reduce keeps only one copy and buys back + ~100 MB of headroom per forecast-hour step. + """ + @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} + def replace_scores(scores, %DateTime{} = valid_time) do + do_replace_scores(scores, valid_time) + end + + defp do_replace_scores(scores, valid_time) do + # Pure grouping phase runs outside the telemetry span — typically + # <10ms on small result sets, and the span's two dispatches + # (~100µs each) would otherwise dominate. The span now wraps only + # the per-band writes, which is where the actual DB cost lives. + {per_band, total} = + Enum.reduce(scores, {%{}, 0}, fn score, {acc, count} -> + {Map.update(acc, score.band_mhz, [score], &[score | &1]), count + 1} + end) + + Instrument.span( + [:db, :replace_scores], + %{valid_time: valid_time}, + fn -> + Enum.each(per_band, fn {band_mhz, band_scores} -> + try do + ScoresFile.write!(band_mhz, valid_time, band_scores) + rescue + e -> + Logger.warning("Propagation: ScoresFile write failed for band=#{band_mhz} vt=#{valid_time}: #{inspect(e)}") + end + end) + + {:ok, total} + end + ) + end + + @doc """ + Remove score files with valid_times older than 3 hours. Called on + a cron by `Microwaveprop.Workers.PropagationPruneWorker`. + + The cutoff sits one hour beyond HRRR's ~2h publish lag: the hourly + seeder picks `run_time = now - 2h`, so the f00 analysis file is + written at valid_time = now - 2h. A 2h cutoff deletes it within + minutes; a 3h cutoff keeps it alive until the next hourly run + supersedes it. + """ + @spec prune_old_scores() :: :ok + def prune_old_scores do + cutoff = DateTime.shift(DateTime.utc_now(), hour: -3) + file_deleted = ScoresFile.prune_older_than(cutoff) + profiles_deleted = ProfilesFile.prune_older_than(cutoff) + scalar_deleted = ScalarFile.prune_older_than(cutoff) + + total = file_deleted + profiles_deleted + scalar_deleted + + if total > 0 do + Logger.info( + "PropagationScores: pruned #{file_deleted} old score files + " <> + "#{profiles_deleted} profile files + #{scalar_deleted} scalar dirs " <> + "(before #{cutoff})" + ) + end + + # Sweep orphaned .tmp.* files left by crashed atomic-write processes + tmp_deleted = + Enum.reduce([ScoresFile.base_dir(), ProfilesFile.base_dir(), ScalarFile.base_dir()], 0, &sweep_tmp_dir/2) + + if tmp_deleted > 0 do + Logger.info("PropagationScores: swept #{tmp_deleted} orphaned .tmp files") + end + + :ok + end + + defp sweep_tmp_dir(dir, acc) do + case File.ls(dir) do + {:ok, entries} -> + Enum.reduce(entries, acc, &sweep_tmp_entry(dir, &1, &2)) + + _ -> + acc + end + end + + defp sweep_tmp_entry(parent, entry, acc) do + full = Path.join(parent, entry) + + case File.ls(full) do + {:ok, _} -> + sweep_tmp_dir(full, acc) + + {:error, _} -> + if String.contains?(entry, ".tmp.") do + _ = File.rm_rf(full) + acc + 1 + else + acc + end + end + end + + @doc """ + Retains score files through GEFS's 168-hour horizon and profile/scalar + files through HRRR's 48-hour horizon, deleting files older than + `run_time`. Called + by `NotifyListener` after chain completion to keep `/data/scores` + within bounds. + + Mirrors `ScoreCache.prune_outside_window` for the on-disk tier. + """ + @spec retain_scores_window(DateTime.t()) :: :ok + def retain_scores_window(%DateTime{} = run_time) do + scores_deleted = ScoresFile.retain_window(run_time, 168) + profiles_deleted = ProfilesFile.retain_window(run_time, 48) + scalars_deleted = ScalarFile.retain_window(run_time, 48) + total = scores_deleted + profiles_deleted + scalars_deleted + + if total > 0 do + Logger.info( + "Propagation.retain_scores_window: deleted #{total} stale files (#{scores_deleted} scores, #{profiles_deleted} profiles, #{scalars_deleted} scalars)" + ) + end + + :ok + end + + @doc """ + Returns distinct valid_times for a band, ordered ascending. Always + reads from the on-disk `ScoresFile` store — the `ScoreCache` only + holds whatever hours have been fetched or broadcast, which can be a + partial view of what's actually on disk, so using it as the source + of truth for the timeline makes new forecast hours invisible until + the cache happens to catch up. Filters out times more than 1 hour in + the past, but always includes the most recent valid_time so there's + always data to display. + """ + # HRRR forecast horizon: f00..f48 covers the next 48 hours from cycle + # time (f01-f18 hourly, f21-f48 3-hourly). Anything beyond that in the + # score store is a leftover from a stale cycle and clutters the + # timeline without adding information. + @spec available_valid_times(non_neg_integer()) :: [DateTime.t()] + def available_valid_times(band_mhz) do + {past_cutoff, future_cutoff} = hot_cache_window() + + case ScoresFile.list_valid_times(band_mhz) do + [] -> [] + times -> filter_or_latest(times, past_cutoff, future_cutoff) + end + end + + @doc """ + The active forecast window for the `/map` UI: one hour in the past + through HRRR's 48-hour forecast horizon. Used by `NotifyListener` to + bound ETS growth — long-horizon GEFS `.prop` files on disk must not + balloon `propagation_score_cache` past the memory the UI actually + reads. + """ + @spec hot_cache_window() :: {DateTime.t(), DateTime.t()} + def hot_cache_window do + now = DateTime.utc_now() + past = DateTime.shift(now, hour: -1) + future = DateTime.shift(now, hour: @hrrr_forecast_horizon_hours) + {past, future} + end + + defp filter_or_latest(times, past_cutoff, future_cutoff) do + fresh = + Enum.filter(times, fn t -> + DateTime.compare(t, past_cutoff) != :lt and DateTime.compare(t, future_cutoff) != :gt + end) + + if fresh == [] do + [Enum.max(times, DateTime)] + else + fresh + end + end + + @doc """ + Get scores for a band at a specific valid_time, optionally within a bounding box. + If valid_time is nil, uses the earliest available (current analysis hour). + Excludes factors for performance. + """ + @spec scores_at(non_neg_integer(), DateTime.t() | nil, %{optional(String.t()) => float()} | nil) :: + [%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}] + def scores_at(band_mhz, valid_time, bounds \\ nil) do + time = valid_time || earliest_valid_time(band_mhz) + + case time do + nil -> [] + _ -> scores_at_fetch(band_mhz, time, bounds) + end + end + + # Cache-hit path is the map's most frequent LiveView call (~every + # pan + click). A wrapping Instrument.span fires 2 telemetry handler + # dispatches that dominate the ~10µs ETS lookup — skip the span on + # hits and rely on the cheap hit/miss counter for the cache-ratio + # panel. The miss path still wraps the disk read where duration is + # the meaningful signal. + defp scores_at_fetch(band_mhz, time, bounds) do + case ScoreCache.fetch_bounds(band_mhz, time, bounds) do + {:ok, scores} -> + :telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: true}) + Enum.map(scores, &Map.put(&1, :valid_time, time)) + + :miss -> + :telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: false}) + + Instrument.span([:propagation, :scores_at], %{band_mhz: band_mhz}, fn -> + read_from_disk_and_cache(band_mhz, time, bounds) + end) + end + end + + @doc """ + Variant of `scores_at/3` that always reads from the `.prop` file on + disk and overwrites the cache entry, rather than returning whatever + the cache happens to hold. Use from update paths (the map's + `propagation_updated` handler) where the underlying file has just + been rewritten but the cache may still contain the previous chain's + scores because of the race between `propagation:cache` fan-out and + `propagation:updated` delivery. + """ + @spec scores_at_fresh(non_neg_integer(), DateTime.t(), %{optional(String.t()) => float()} | nil) :: + [%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}] + def scores_at_fresh(band_mhz, %DateTime{} = valid_time, bounds \\ nil) do + read_from_disk_and_cache(band_mhz, valid_time, bounds) + end + + defp read_from_disk_and_cache(band_mhz, time, bounds) do + full = ScoresFile.read_bounds(band_mhz, time) + ScoreCache.put(band_mhz, time, full) + + full + |> filter_bounds(bounds) + |> Enum.map(&Map.put(&1, :valid_time, time)) + end + + @doc """ + Load the full North America score set for `{band_mhz, valid_time}` from + the on-disk binary files (HRRR `.prop` + HRDPS `.hrdps.prop`, merged) and + broadcast it to every `ScoreCache` in the cluster. Called from + `PropagationGridWorker` after each forecast hour so all pods have a warm + cache by the time clients begin requesting the new hour. + + Returns `{:error, :enoent}` only when neither file exists. A single + missing file (HRDPS pre-cycle, HRRR briefly absent) is OK — the cache + warms with whichever side is available. + """ + @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) :: + :ok | {:error, :enoent | :invalid_format} + def warm_cache_and_broadcast(band_mhz, valid_time) do + hrrr = read_score_points(&ScoresFile.read/2, band_mhz, valid_time) + hrdps = read_score_points(&ScoresFile.read_hrdps/2, band_mhz, valid_time) + + # Merge priority: HRRR > HRDPS > GEFS. GEFS provides extended-horizon + # coverage beyond HRRR's 48h window. Within the f24-f48 overlap, cells + # already present in HRRR/HRDPS are skipped so the coarser GEFS scores + # don't override the higher-resolution ones. + merged = (hrrr || []) ++ (hrdps || []) + gefs = read_score_points(&ScoresFile.read_gefs/2, band_mhz, valid_time) || [] + scores = ScoresFile.merge_preferred(merged, gefs) + + case {hrrr, hrdps, scores} do + {nil, nil, []} -> + {:error, :enoent} + + _ -> + ScoreCache.broadcast_put(band_mhz, valid_time, scores) + :ok + end + end + + defp read_score_points(reader, band_mhz, valid_time) do + case reader.(band_mhz, valid_time) do + {:ok, payload} -> ScoresFile.extract_points(payload, nil) + _ -> nil + end + end + + defp filter_bounds(scores, nil), do: scores + + defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do + Enum.filter(scores, fn %{lat: lat, lon: lon} -> + lat >= s and lat <= n and lon >= w and lon <= e + end) + end + + @doc "Get the latest scores for a band (alias for scores_at with earliest valid_time)." + @spec latest_scores(non_neg_integer(), %{optional(String.t()) => float()} | nil) :: + [%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}] + def latest_scores(band_mhz, bounds \\ nil) do + scores_at(band_mhz, nil, bounds) + end + + defp earliest_valid_time(band_mhz) do + case ScoresFile.list_valid_times(band_mhz) do + [earliest | _] -> earliest + [] -> nil + end + end + + @doc "Get scores across all forecast hours for a single grid point (for sparkline)." + @spec point_forecast(non_neg_integer(), float(), float()) :: + [%{valid_time: DateTime.t(), score: non_neg_integer()}] + def point_forecast(band_mhz, lat, lon) do + Instrument.span([:propagation, :point_forecast], %{band_mhz: band_mhz}, fn -> + {snapped_lat, snapped_lon} = snap_to_grid(lat, lon) + now = DateTime.utc_now() + + # Use the on-disk .prop file list as the authoritative timeline so + # the chart never falls behind the main-map timeline (which also + # reads the disk). The cache is still consulted per-hour for a + # fast score lookup; a miss falls through to the file. + # Fan the per-hour disk lookups across 4 tasks. Each ScoresFile + # read_point is an NFS stat + pread of ~100 bytes (keyed byte at + # row*cols+col), so the ceiling is NFS RTT × number of hours — + # sequential ran ~4–5× the wall time of the slowest read. + band_mhz + |> ScoresFile.list_valid_times() + |> forecast_window(now) + |> Task.async_stream( + &point_forecast_entry(band_mhz, &1, snapped_lat, snapped_lon), + max_concurrency: 4, + ordered: true, + timeout: 5_000 + ) + |> Enum.flat_map(fn + {:ok, nil} -> + [] + + {:ok, entry} -> + [entry] + + {:exit, reason} -> + Logger.error( + "Propagation.point_forecast async lookup failed: band_mhz=#{band_mhz} lat=#{snapped_lat} lon=#{snapped_lon} reason=#{inspect(reason)}" + ) + + [] + end) + end) + end + + defp point_forecast_entry(band_mhz, valid_time, lat, lon) do + case ScoreCache.fetch_point(band_mhz, valid_time, lat, lon) do + {:ok, score} -> + %{valid_time: valid_time, score: score} + + :miss -> + case ScoresFile.read_point(band_mhz, valid_time, lat, lon) do + nil -> nil + score -> %{valid_time: valid_time, score: score} + end + end + end + + # Select the set of valid_times the forecast chart should render. + # Mirrors `available_valid_times`: keep everything from one hour + # before now onward so the most recent analysis hour (typically + # ~30–60 min behind wall clock due to HRRR publishing lag) sits at + # the left edge of the chart as "now". When every hour on disk is + # older than that cutoff, fall back to just the newest entry so the + # chart can still render a single data point. + defp forecast_window([], _now), do: [] + + defp forecast_window(times, now) do + past_cutoff = DateTime.shift(now, hour: -1) + future_cutoff = DateTime.shift(now, hour: @hrrr_forecast_horizon_hours) + filter_or_latest(times, past_cutoff, future_cutoff) + end + + defp snap_to_grid(lat, lon) do + step = Grid.step() + {Float.round(Float.round(lat / step) * step, 3), Float.round(Float.round(lon / step) * step, 3)} + end + + @doc """ + Get the full score and factors for a specific grid point, snapped to + the nearest grid cell. + + `:profile_source` describes where the factor breakdown came from: + + * `:exact` — rescored from this `valid_time`'s own profile file. + * `{:fallback, fallback_valid_time}` — the requested hour's profile + was missing, so we rescored from the most recent analysis profile + within the lookback window. Treat as approximate. + * `:unavailable` — no profile available; `factors` is `%{}`. + """ + @spec point_detail(non_neg_integer(), float(), float(), DateTime.t() | nil) :: + %{ + lat: float(), + lon: float(), + score: non_neg_integer(), + factors: map(), + profile_source: :exact | {:fallback, DateTime.t()} | :unavailable, + valid_time: DateTime.t() + } + | nil + def point_detail(band_mhz, lat, lon, valid_time \\ nil) do + Instrument.span([:propagation, :point_detail], %{band_mhz: band_mhz}, fn -> + do_point_detail(band_mhz, lat, lon, valid_time) + end) + end + + defp do_point_detail(band_mhz, lat, lon, valid_time) do + {snapped_lat, snapped_lon} = snap_to_grid(lat, lon) + time = valid_time || latest_valid_time(band_mhz) + + case time do + nil -> + nil + + _ -> + case ScoresFile.read_point(band_mhz, time, snapped_lat, snapped_lon) do + nil -> + nil + + score -> + {factors, source} = factors_for(band_mhz, time, snapped_lat, snapped_lon) + + %{ + lat: snapped_lat, + lon: snapped_lon, + score: score, + factors: factors, + profile_source: source, + valid_time: time + } + end + end + end + + # Rebuild the factor breakdown for a clicked grid cell by rescoring + # the persisted HRRR profile. + # + # The Rust pipeline (`prop_grid_rs`) writes a per-cell profile file + # for every chain step (f00..f18 since Phase 2 cutover), so the + # `:exact` branch covers a healthy production state. The fallback to + # a recent analysis profile remains as a safety net for missed + # cycles or partial chain runs — when it kicks in we tag the result + # `{:fallback, profile_valid_time}` so the UI can label the + # breakdown as approximated rather than silently misrepresent it. + @spec factors_for(non_neg_integer(), DateTime.t(), float(), float()) :: + {map(), :exact | {:fallback, DateTime.t()} | :unavailable} + defp factors_for(band_mhz, valid_time, lat, lon) do + case ProfilesFile.read_point(valid_time, lat, lon) do + nil -> + factors_from_fallback_profile(band_mhz, valid_time, lat, lon) + + profile -> + {Microwaveprop.Propagation.factors_from_profile(band_mhz, valid_time, profile, lat, lon), :exact} + end + end + + defp factors_from_fallback_profile(band_mhz, valid_time, lat, lon) do + case latest_profile_time_within_lookback(valid_time) do + nil -> + {%{}, :unavailable} + + fallback_time -> + case ProfilesFile.read_point(fallback_time, lat, lon) do + nil -> + {%{}, :unavailable} + + profile -> + {Microwaveprop.Propagation.factors_from_profile(band_mhz, fallback_time, profile, lat, lon), + {:fallback, fallback_time}} + end + end + end + + defp latest_profile_time_within_lookback(%DateTime{} = valid_time) do + lookback_cutoff = DateTime.shift(valid_time, hour: -@fallback_profile_lookback_hours) + + ProfilesFile.list_valid_times() + |> Enum.filter(fn t -> + DateTime.compare(t, valid_time) != :gt and DateTime.compare(t, lookback_cutoff) != :lt + end) + |> case do + [] -> nil + past -> Enum.max(past, DateTime) + end + end + + @doc "Get the latest valid_time across all bands." + @spec latest_valid_time() :: DateTime.t() | nil + def latest_valid_time do + ScoresFile.latest_valid_time() + end + + @doc "Get the latest valid_time for a specific band." + @spec latest_valid_time(non_neg_integer()) :: DateTime.t() | nil + def latest_valid_time(band_mhz) do + case ScoresFile.list_valid_times(band_mhz) do + [] -> nil + times -> Enum.max(times, DateTime) + end + end +end diff --git a/lib/microwaveprop/radio/contact.ex b/lib/microwaveprop/radio/contact.ex index ecb1bf7c..1304aaa5 100644 --- a/lib/microwaveprop/radio/contact.ex +++ b/lib/microwaveprop/radio/contact.ex @@ -5,7 +5,7 @@ defmodule Microwaveprop.Radio.Contact do import Ecto.Changeset alias Microwaveprop.Accounts.User - alias Microwaveprop.Radio.Maidenhead + alias Microwaveprop.Radio.MaidenheadChangesetHelpers, as: MCH @primary_key {:id, :binary_id, autogenerate: true} @foreign_key_type :binary_id @@ -110,15 +110,15 @@ defmodule Microwaveprop.Radio.Contact do |> cast(attrs, @submission_fields) |> validate_required(@submission_required) |> validate_user_or_email() - |> sanitize_callsign(:station1) - |> sanitize_callsign(:station2) - |> sanitize_grid(:grid1) - |> sanitize_grid(:grid2) + |> MCH.normalize_callsign(:station1) + |> MCH.normalize_callsign(:station2) + |> update_change(:grid1, &MCH.normalize_grid/1) + |> update_change(:grid2, &MCH.normalize_grid/1) |> normalize_blank_mode() |> validate_callsign(:station1) |> validate_callsign(:station2) - |> validate_grid_format(:grid1) - |> validate_grid_format(:grid2) + |> MCH.validate_grid(:grid1, blank_error: false) + |> MCH.validate_grid(:grid2, blank_error: false) |> validate_mode_inclusion() |> validate_inclusion(:band, @allowed_bands) |> validate_email_format() @@ -204,34 +204,8 @@ defmodule Microwaveprop.Radio.Contact do end end - # Strip whitespace and upcase callsigns - defp sanitize_callsign(changeset, field) do - case get_change(changeset, field) do - nil -> changeset - val -> put_change(changeset, field, val |> String.trim() |> String.upcase()) - end - end - - # Upcase grid squares (Maidenhead uses uppercase letters + digits) - defp sanitize_grid(changeset, field) do - case get_change(changeset, field) do - nil -> changeset - val -> put_change(changeset, field, val |> String.trim() |> String.upcase()) - end - end - # Callsigns: letters, digits, and / only (e.g. W5XD, KG5CCI/P, VE3/W5XD) defp validate_callsign(changeset, field) do validate_format(changeset, field, ~r/^[A-Z0-9\/]+$/, message: "must contain only letters, digits, and /") end - - defp validate_grid_format(changeset, field) do - validate_change(changeset, field, fn _, value -> - if Maidenhead.valid?(value) do - [] - else - [{field, "is not a valid Maidenhead grid square"}] - end - end) - end end diff --git a/lib/microwaveprop/radio/maidenhead_changeset_helpers.ex b/lib/microwaveprop/radio/maidenhead_changeset_helpers.ex new file mode 100644 index 00000000..7995b9a6 --- /dev/null +++ b/lib/microwaveprop/radio/maidenhead_changeset_helpers.ex @@ -0,0 +1,156 @@ +defmodule Microwaveprop.Radio.MaidenheadChangesetHelpers do + @moduledoc """ + Shared changeset helpers for Maidenhead grid squares, callsigns, and + lat/lon coordinate validation. Used across multiple domain schemas + (Beacon, User, FixedStation, Contact, Location, BeaconMonitor). + """ + + import Ecto.Changeset + + alias Microwaveprop.Radio.Maidenhead + + @doc """ + Trims whitespace and uppercases a callsign field if a change is present. + Returns the changeset unchanged when the field has no change or the + change is `nil`. + """ + @spec normalize_callsign(Ecto.Changeset.t(), atom()) :: Ecto.Changeset.t() + def normalize_callsign(changeset, field) do + update_change(changeset, field, fn + nil -> nil + val when is_binary(val) -> val |> String.trim() |> String.upcase() + val -> val + end) + end + + @doc """ + Validates a Maidenhead grid field on the changeset. + + ## Options + + * `:blank_error` — when `true` (default), a `nil` or empty-string value + adds a "can't be blank" error. When `false`, blank values pass through + silently. + * `:normalizer` — a function of one argument called to normalize the + grid when it passes validation. Defaults to `normalize_grid/1`. + """ + @spec validate_grid(Ecto.Changeset.t(), atom(), keyword()) :: Ecto.Changeset.t() + def validate_grid(changeset, field, opts \\ []) do + blank_error = Keyword.get(opts, :blank_error, true) + normalizer = Keyword.get(opts, :normalizer, &normalize_grid/1) + + case get_field(changeset, field) do + nil -> + if blank_error do + add_error(changeset, field, "can't be blank") + else + changeset + end + + "" -> + if blank_error do + add_error(changeset, field, "can't be blank") + else + changeset + end + + value -> + if Maidenhead.valid?(value) do + update_change(changeset, field, normalizer) + else + add_error(changeset, field, "is not a valid Maidenhead grid") + end + end + end + + @doc """ + Normalizes a Maidenhead grid string to standard form. + + * `nil` → `nil` + * Trims leading and trailing whitespace. + * For grids of 6 or more characters: uppercases the first 4 characters + (field + square) and downcases the rest (subsquares and beyond). + * For grids of fewer than 6 characters: uppercases the entire string. + """ + @spec normalize_grid(nil | String.t()) :: nil | String.t() + def normalize_grid(nil), do: nil + + def normalize_grid(grid) when is_binary(grid) do + trimmed = String.trim(grid) + + if String.length(trimmed) >= 6 do + String.upcase(String.slice(trimmed, 0, 4)) <> String.downcase(String.slice(trimmed, 4, String.length(trimmed) - 4)) + else + String.upcase(trimmed) + end + end + + @doc """ + Derives lat/lon coordinates from a Maidenhead grid field when lat/lon + are not already set. Only runs when the changeset is valid. + + Uses the centre point of the grid square, rounded to 6 decimal places. + """ + @spec derive_latlon( + Ecto.Changeset.t(), + atom(), + atom(), + atom() + ) :: Ecto.Changeset.t() + def derive_latlon(changeset, grid_field \\ :grid, lat_field \\ :lat, lon_field \\ :lon) do + grid = get_field(changeset, grid_field) + lat = get_field(changeset, lat_field) + lon = get_field(changeset, lon_field) + + if changeset.valid? and is_binary(grid) and grid != "" and + (not is_number(lat) or not is_number(lon)) do + case Maidenhead.to_latlon(grid) do + {:ok, {derived_lat, derived_lon}} -> + changeset + |> put_change(lat_field, Float.round(derived_lat * 1.0, 6)) + |> put_change(lon_field, Float.round(derived_lon * 1.0, 6)) + + :error -> + changeset + end + else + changeset + end + end + + @doc """ + Derives a Maidenhead grid from lat/lon coordinates when the grid field + is blank and lat/lon are numbers. + + The `precision` must be an even number between 4 and 10 (default 6). + """ + @spec derive_grid( + Ecto.Changeset.t(), + atom(), + atom(), + atom(), + pos_integer() + ) :: Ecto.Changeset.t() + def derive_grid(changeset, grid_field \\ :grid, lat_field \\ :lat, lon_field \\ :lon, precision \\ 6) do + grid = get_field(changeset, grid_field) + lat = get_field(changeset, lat_field) + lon = get_field(changeset, lon_field) + + if grid in [nil, ""] and is_number(lat) and is_number(lon) do + put_change(changeset, grid_field, Maidenhead.from_latlon(lat * 1.0, lon * 1.0, precision)) + else + changeset + end + end + + @doc """ + Validates that latitude is between -90 and 90 and longitude is between + -180 and 180. + """ + @spec validate_latlon(Ecto.Changeset.t(), atom(), atom()) :: Ecto.Changeset.t() + def validate_latlon(changeset, lat_field \\ :lat, lon_field \\ :lon) do + changeset + |> validate_number(lat_field, greater_than_or_equal_to: -90.0, less_than_or_equal_to: 90.0) + |> validate_number(lon_field, greater_than_or_equal_to: -180.0, less_than_or_equal_to: 180.0) + end +end diff --git a/lib/microwaveprop/rover.ex b/lib/microwaveprop/rover.ex index 7762bd23..754cc3d7 100644 --- a/lib/microwaveprop/rover.ex +++ b/lib/microwaveprop/rover.ex @@ -11,6 +11,7 @@ defmodule Microwaveprop.Rover do import Ecto.Query alias Microwaveprop.Accounts.User + alias Microwaveprop.ContextHelpers alias Microwaveprop.Radio.Maidenhead alias Microwaveprop.Repo alias Microwaveprop.Rover.FixedStation @@ -56,7 +57,7 @@ defmodule Microwaveprop.Rover do @spec update_station(User.t(), Ecto.UUID.t(), map()) :: {:ok, FixedStation.t()} | {:error, :not_found | Ecto.Changeset.t()} def update_station(%User{} = user, id, attrs) do - case fetch_owned(user, id) do + case ContextHelpers.fetch_owned(FixedStation, id, user) do {:ok, station} -> station |> FixedStation.changeset(attrs) @@ -70,7 +71,7 @@ defmodule Microwaveprop.Rover do @spec delete_station(User.t(), Ecto.UUID.t()) :: {:ok, FixedStation.t()} | {:error, :not_found} def delete_station(%User{} = user, id) do - case fetch_owned(user, id) do + case ContextHelpers.fetch_owned(FixedStation, id, user) do {:ok, station} -> Repo.delete(station) @@ -82,7 +83,7 @@ defmodule Microwaveprop.Rover do @spec toggle_selected(User.t(), Ecto.UUID.t()) :: {:ok, FixedStation.t()} | {:error, :not_found | Ecto.Changeset.t()} def toggle_selected(%User{} = user, id) do - case fetch_owned(user, id) do + case ContextHelpers.fetch_owned(FixedStation, id, user) do {:ok, station} -> station |> Ecto.Changeset.change(selected: not station.selected) @@ -117,13 +118,6 @@ defmodule Microwaveprop.Rover do end) end - defp fetch_owned(%User{id: user_id}, id) do - case Repo.get(FixedStation, id) do - %FixedStation{user_id: ^user_id} = station -> {:ok, station} - _ -> {:error, :not_found} - end - end - # ── Rover locations (globally visible, logged-in to mutate) ────────── @doc """ @@ -148,7 +142,7 @@ defmodule Microwaveprop.Rover do @spec update_location(User.t(), Ecto.UUID.t(), map()) :: {:ok, Location.t()} | {:error, :not_found | Ecto.Changeset.t()} def update_location(%User{} = user, id, attrs) do - case fetch_owned_location(user, id) do + case ContextHelpers.fetch_owned(Location, id, user) do {:ok, location} -> location |> Location.changeset(attrs) @@ -162,26 +156,12 @@ defmodule Microwaveprop.Rover do @spec delete_location(User.t(), Ecto.UUID.t()) :: {:ok, Location.t()} | {:error, :not_found} def delete_location(%User{} = user, id) do - case fetch_owned_location(user, id) do + case ContextHelpers.fetch_owned(Location, id, user) do {:ok, location} -> Repo.delete(location) {:error, :not_found} -> {:error, :not_found} end end - defp fetch_owned_location(%User{is_admin: true}, id) do - case Repo.get(Location, id) do - %Location{} = loc -> {:ok, loc} - _ -> {:error, :not_found} - end - end - - defp fetch_owned_location(%User{id: user_id}, id) do - case Repo.get(Location, id) do - %Location{user_id: ^user_id} = loc -> {:ok, loc} - _ -> {:error, :not_found} - end - end - defp next_position(%User{id: user_id}) do query = from s in FixedStation, @@ -198,14 +178,7 @@ defmodule Microwaveprop.Rover do defp maybe_enqueue_elevation(other), do: other - # Wrapped behind a function-exported guard so this module compiles - # cleanly while Task 8's worker is still being authored — the worker - # is referenced by atom and resolved at runtime once it's loaded. defp enqueue_station_elevation(id) do - worker = Microwaveprop.Workers.StationElevationWorker - - if Code.ensure_loaded?(worker) and function_exported?(worker, :new, 1) do - %{id: id} |> worker.new() |> Oban.insert() - end + ContextHelpers.safe_enqueue(Microwaveprop.Workers.StationElevationWorker, %{id: id}) end end diff --git a/lib/microwaveprop/rover/fixed_station.ex b/lib/microwaveprop/rover/fixed_station.ex index 158a0aec..a347e695 100644 --- a/lib/microwaveprop/rover/fixed_station.ex +++ b/lib/microwaveprop/rover/fixed_station.ex @@ -14,7 +14,7 @@ defmodule Microwaveprop.Rover.FixedStation do import Ecto.Changeset - alias Microwaveprop.Radio.Maidenhead + alias Microwaveprop.Radio.MaidenheadChangesetHelpers, as: MCH @callsign_regex ~r/^[A-Z0-9\/]{3,12}$/ @@ -40,29 +40,18 @@ defmodule Microwaveprop.Rover.FixedStation do def changeset(station, attrs) do station |> cast(attrs, [:callsign, :grid, :lat, :lon, :elevation_m, :selected, :position]) - |> normalize_callsign() + |> MCH.normalize_callsign(:callsign) |> normalize_grid() |> validate_required([:callsign]) |> validate_format(:callsign, @callsign_regex) - |> validate_grid_format() - |> derive_latlon_from_grid() + |> MCH.validate_grid(:grid, blank_error: false, normalizer: &normalize_grid_string/1) + |> MCH.derive_latlon(:grid, :lat, :lon) |> 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) + |> MCH.validate_latlon(:lat, :lon) |> foreign_key_constraint(:user_id) |> unique_constraint(:callsign, name: :fixed_stations_user_id_callsign_index) end - defp normalize_callsign(changeset) do - case get_change(changeset, :callsign) do - nil -> - changeset - - call when is_binary(call) -> - put_change(changeset, :callsign, call |> String.trim() |> String.upcase()) - end - end - defp normalize_grid(changeset) do case get_change(changeset, :grid) do nil -> changeset @@ -93,46 +82,4 @@ defmodule Microwaveprop.Rover.FixedStation do end end) end - - defp validate_grid_format(changeset) do - case get_field(changeset, :grid) do - nil -> - changeset - - grid -> - if Maidenhead.valid?(grid) do - changeset - else - add_error(changeset, :grid, "is not a valid Maidenhead grid") - end - end - end - - defp derive_latlon_from_grid(changeset) do - lat = get_field(changeset, :lat) - lon = get_field(changeset, :lon) - grid = get_field(changeset, :grid) - - cond do - not changeset.valid? -> - changeset - - is_number(lat) and is_number(lon) -> - changeset - - is_binary(grid) -> - case Maidenhead.to_latlon(grid) do - {:ok, {derived_lat, derived_lon}} -> - changeset - |> put_change(:lat, derived_lat) - |> put_change(:lon, derived_lon) - - :error -> - changeset - end - - true -> - changeset - end - end end diff --git a/lib/microwaveprop/rover/location.ex b/lib/microwaveprop/rover/location.ex index 3859f70e..77cec53b 100644 --- a/lib/microwaveprop/rover/location.ex +++ b/lib/microwaveprop/rover/location.ex @@ -11,6 +11,8 @@ defmodule Microwaveprop.Rover.Location do import Ecto.Changeset + alias Microwaveprop.Radio.MaidenheadChangesetHelpers, as: MCH + @statuses [:good, :bad] @primary_key {:id, :binary_id, autogenerate: true} @@ -36,8 +38,7 @@ defmodule Microwaveprop.Rover.Location do location |> cast(attrs, [:lat, :lon, :status, :notes]) |> validate_required([:lat, :lon, :status]) - |> 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) + |> MCH.validate_latlon(:lat, :lon) |> validate_inclusion(:status, @statuses) |> validate_length(:notes, max: 4000) |> foreign_key_constraint(:user_id) diff --git a/lib/microwaveprop/rover_planning.ex b/lib/microwaveprop/rover_planning.ex index 6d971dd8..466a81bb 100644 --- a/lib/microwaveprop/rover_planning.ex +++ b/lib/microwaveprop/rover_planning.ex @@ -14,6 +14,7 @@ defmodule Microwaveprop.RoverPlanning do import Ecto.Query alias Microwaveprop.Accounts.User + alias Microwaveprop.ContextHelpers alias Microwaveprop.Repo alias Microwaveprop.Rover.Location alias Microwaveprop.RoverPlanning.Mission @@ -81,7 +82,7 @@ defmodule Microwaveprop.RoverPlanning do @spec update_mission(User.t(), Ecto.UUID.t(), map()) :: {:ok, Mission.t()} | {:error, :not_found | Ecto.Changeset.t()} def update_mission(%User{} = user, id, attrs) do - case fetch_owned(user, id) do + case ContextHelpers.fetch_owned(Mission, id, user) do {:ok, mission} -> mission |> Repo.preload(stations: from(s in Station, order_by: s.position)) @@ -105,7 +106,7 @@ defmodule Microwaveprop.RoverPlanning do @spec delete_mission(User.t(), Ecto.UUID.t()) :: {:ok, Mission.t()} | {:error, :not_found} def delete_mission(%User{} = user, id) do - case fetch_owned(user, id) do + case ContextHelpers.fetch_owned(Mission, id, user) do {:ok, mission} -> Repo.delete(mission) {:error, :not_found} -> {:error, :not_found} end @@ -305,18 +306,4 @@ defmodule Microwaveprop.RoverPlanning do _ = Oban.insert_all(changesets) :ok end - - defp fetch_owned(%User{is_admin: true}, id) do - case get_mission(id) do - nil -> {:error, :not_found} - mission -> {:ok, mission} - end - end - - defp fetch_owned(%User{id: user_id}, id) do - case get_mission(id) do - %Mission{user_id: ^user_id} = mission -> {:ok, mission} - _ -> {:error, :not_found} - end - end end diff --git a/lib/microwaveprop_web/live/beacon_live/form.ex b/lib/microwaveprop_web/live/beacon_live/form.ex index e8467128..75db9add 100644 --- a/lib/microwaveprop_web/live/beacon_live/form.ex +++ b/lib/microwaveprop_web/live/beacon_live/form.ex @@ -2,6 +2,8 @@ defmodule MicrowavepropWeb.BeaconLive.Form do @moduledoc "Shared new/edit form for beacon records, used by `BeaconLive.Index`." use MicrowavepropWeb, :live_view + import MicrowavepropWeb.LiveHelpers, only: [current_user: 1] + alias Microwaveprop.Beacons alias Microwaveprop.Beacons.Beacon @@ -158,9 +160,6 @@ defmodule MicrowavepropWeb.BeaconLive.Form do end end - defp current_user(%{user: user}), do: user - defp current_user(_), do: nil - defp return_path("index", _beacon), do: ~p"/beacons" defp return_path("show", beacon), do: ~p"/beacons/#{beacon}" end diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex index f6941408..9886e2ec 100644 --- a/lib/microwaveprop_web/live/contact_live/show.ex +++ b/lib/microwaveprop_web/live/contact_live/show.ex @@ -4,6 +4,7 @@ defmodule MicrowavepropWeb.ContactLive.Show do import Ecto.Query import MicrowavepropWeb.Components.SkewTChart + import MicrowavepropWeb.LiveHelpers, only: [current_user: 1, subscribe: 2] alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.PathAnalysis @@ -83,7 +84,7 @@ defmodule MicrowavepropWeb.ContactLive.Show do socket = if connected?(socket) do - :ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:#{contact.id}") + subscribe(socket, "contact_enrichment:#{contact.id}") kickoff_hydration(socket, contact) else @@ -215,9 +216,6 @@ defmodule MicrowavepropWeb.ContactLive.Show do end end - defp current_user(%{current_scope: %{user: %{} = user}}), do: user - defp current_user(_), do: nil - defp admin?(%{current_scope: %{user: %{is_admin: true}}}), do: true defp admin?(_), do: false diff --git a/lib/microwaveprop_web/live/eme_live.ex b/lib/microwaveprop_web/live/eme_live.ex index b05a9307..ca19f236 100644 --- a/lib/microwaveprop_web/live/eme_live.ex +++ b/lib/microwaveprop_web/live/eme_live.ex @@ -18,7 +18,7 @@ defmodule MicrowavepropWeb.EmeLive do """ use MicrowavepropWeb, :live_view - import MicrowavepropWeb.LiveHelpers, only: [parse_float: 2, parse_int: 2] + import MicrowavepropWeb.LiveHelpers, only: [parse_float: 2, parse_int: 2, assign_url_params: 2] alias Microwaveprop.Moon alias Microwaveprop.Propagation.BandConfig @@ -64,7 +64,7 @@ defmodule MicrowavepropWeb.EmeLive do @impl true def handle_params(params, _uri, socket) do - p = Map.merge(@defaults, Map.take(params, Map.keys(@defaults))) + p = assign_url_params(params, @defaults) socket = assign(socket, diff --git a/lib/microwaveprop_web/live/import_live.ex b/lib/microwaveprop_web/live/import_live.ex index bd6dcbe1..4c47f7de 100644 --- a/lib/microwaveprop_web/live/import_live.ex +++ b/lib/microwaveprop_web/live/import_live.ex @@ -8,11 +8,11 @@ defmodule MicrowavepropWeb.ImportLive do """ use MicrowavepropWeb, :live_view + import MicrowavepropWeb.LiveHelpers, only: [subscribe: 2] + alias Microwaveprop.Radio.ImportRun alias Microwaveprop.Repo - @pubsub Microwaveprop.PubSub - @impl true def mount(%{"id" => id}, _session, socket) do case load_run(id) do @@ -24,9 +24,7 @@ defmodule MicrowavepropWeb.ImportLive do 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 + subscribe(socket, "csv_import:#{run.id}") {:ok, socket diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 848723d6..1483ed24 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -4,6 +4,7 @@ defmodule MicrowavepropWeb.MapLive do use LiveStash, stored_keys: [:selected_band, :selected_time] import MicrowavepropWeb.Layouts, only: [admin_links: 1] + import MicrowavepropWeb.LiveHelpers, only: [subscribe: 2] alias Microwaveprop.Propagation alias Microwaveprop.Propagation.BandConfig @@ -97,8 +98,8 @@ defmodule MicrowavepropWeb.MapLive do defp subscribe_if_connected(socket) do if connected?(socket) do - :ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated") - :ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:pipeline") + subscribe(socket, "propagation:updated") + subscribe(socket, "propagation:pipeline") socket = schedule_pipeline_status_timer(socket) schedule_advance_timer(socket) else diff --git a/lib/microwaveprop_web/live/path_live.ex b/lib/microwaveprop_web/live/path_live.ex index 59e4770d..cd257976 100644 --- a/lib/microwaveprop_web/live/path_live.ex +++ b/lib/microwaveprop_web/live/path_live.ex @@ -3,7 +3,7 @@ defmodule MicrowavepropWeb.PathLive do use MicrowavepropWeb, :live_view import MicrowavepropWeb.Components.SkewTChart - import MicrowavepropWeb.LiveHelpers, only: [parse_float: 2, parse_int: 2] + import MicrowavepropWeb.LiveHelpers, only: [parse_float: 2, parse_int: 2, subscribe: 2, assign_url_params: 2] alias Microwaveprop.Buildings.Index, as: BuildingsIndex alias Microwaveprop.Buildings.Loader, as: BuildingsLoader @@ -31,10 +31,7 @@ defmodule MicrowavepropWeb.PathLive do @impl true def mount(_params, _session, socket) do - _ = - if connected?(socket) do - Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated") - end + subscribe(socket, "propagation:updated") {:ok, assign(socket, @@ -101,7 +98,7 @@ defmodule MicrowavepropWeb.PathLive do end def handle_params(params, _uri, socket) do - p = Map.merge(@defaults, Map.take(params, @url_params)) + p = assign_url_params(params, @defaults) is_gps = p["source"] == "gps" diff --git a/lib/microwaveprop_web/live/rover_live.ex b/lib/microwaveprop_web/live/rover_live.ex index a6005d3f..79614892 100644 --- a/lib/microwaveprop_web/live/rover_live.ex +++ b/lib/microwaveprop_web/live/rover_live.ex @@ -8,7 +8,7 @@ defmodule MicrowavepropWeb.RoverLive do use MicrowavepropWeb, :live_view - import MicrowavepropWeb.LiveHelpers, only: [parse_int: 2] + import MicrowavepropWeb.LiveHelpers, only: [parse_int: 2, current_user: 1] alias Microwaveprop.Accounts.User alias Microwaveprop.Buildings.Index, as: BuildingsIndex @@ -160,13 +160,6 @@ defmodule MicrowavepropWeb.RoverLive do end end - defp current_user(socket) do - case socket.assigns[:current_scope] do - %{user: %User{} = user} -> user - _ -> nil - end - end - defp home_label(lat, lon), do: Maidenhead.from_latlon(lat, lon, 10) defp drive_radius_km(max_distance_mi), do: max_distance_mi * @km_per_mi diff --git a/lib/microwaveprop_web/live/rover_locations_live.ex b/lib/microwaveprop_web/live/rover_locations_live.ex index 7f914583..bab9a928 100644 --- a/lib/microwaveprop_web/live/rover_locations_live.ex +++ b/lib/microwaveprop_web/live/rover_locations_live.ex @@ -8,6 +8,7 @@ defmodule MicrowavepropWeb.RoverLocationsLive do use MicrowavepropWeb.LiveTableResource, schema: Microwaveprop.Rover.Location import Ecto.Query + import MicrowavepropWeb.LiveHelpers, only: [current_user: 1] alias Microwaveprop.Accounts.User alias Microwaveprop.Radio.Maidenhead @@ -258,9 +259,6 @@ defmodule MicrowavepropWeb.RoverLocationsLive do value |> Float.round(6) |> Float.to_string() end - defp current_user(%Phoenix.LiveView.Socket{assigns: assigns}), do: scope_user(assigns) - defp current_user(assigns) when is_map(assigns), do: scope_user(assigns) - defp scope_user(assigns) do case assigns[:current_scope] do %{user: %User{} = user} -> user diff --git a/lib/microwaveprop_web/live/rover_planning_live.ex b/lib/microwaveprop_web/live/rover_planning_live.ex index 564b1b3a..f216f14f 100644 --- a/lib/microwaveprop_web/live/rover_planning_live.ex +++ b/lib/microwaveprop_web/live/rover_planning_live.ex @@ -6,6 +6,8 @@ defmodule MicrowavepropWeb.RoverPlanningLive do use MicrowavepropWeb, :live_view use MicrowavepropWeb.LiveTableResource, schema: Microwaveprop.RoverPlanning.Mission + import MicrowavepropWeb.LiveHelpers, only: [current_user: 1] + alias Microwaveprop.Accounts.User alias Microwaveprop.RoverPlanning alias Microwaveprop.RoverPlanning.Mission @@ -79,16 +81,6 @@ defmodule MicrowavepropWeb.RoverPlanningLive do |> assign(:options, updated_options) end - defp current_user(%Phoenix.LiveView.Socket{assigns: assigns}), do: scope_user(assigns) - defp current_user(assigns) when is_map(assigns), do: scope_user(assigns) - - defp scope_user(assigns) do - case assigns[:current_scope] do - %{user: %User{} = user} -> user - _ -> nil - end - end - defp can_modify?(scope, %{user_id: user_id}) do case scope do %{user: %User{is_admin: true}} -> true diff --git a/lib/microwaveprop_web/live/rover_planning_live/form.ex b/lib/microwaveprop_web/live/rover_planning_live/form.ex index 5fec8c2b..6283c318 100644 --- a/lib/microwaveprop_web/live/rover_planning_live/form.ex +++ b/lib/microwaveprop_web/live/rover_planning_live/form.ex @@ -2,6 +2,8 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Form do @moduledoc "New / edit form for `/rover-planning` missions." use MicrowavepropWeb, :live_view + import MicrowavepropWeb.LiveHelpers, only: [current_user: 1] + alias Microwaveprop.Accounts.User alias Microwaveprop.RoverPlanning alias Microwaveprop.RoverPlanning.Mission @@ -198,13 +200,6 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Form do |> Kernel.+(1) end - defp current_user(%Phoenix.LiveView.Socket{assigns: assigns}) do - case assigns[:current_scope] do - %{user: %User{} = user} -> user - _ -> nil - end - end - defp can_modify?(%User{is_admin: true}, _), do: true defp can_modify?(%User{id: id}, %Mission{user_id: id}) when not is_nil(id), do: true defp can_modify?(_, _), do: false diff --git a/lib/microwaveprop_web/live/rover_planning_live/show.ex b/lib/microwaveprop_web/live/rover_planning_live/show.ex index 9fb716a9..4ce1fcff 100644 --- a/lib/microwaveprop_web/live/rover_planning_live/show.ex +++ b/lib/microwaveprop_web/live/rover_planning_live/show.ex @@ -6,6 +6,8 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Show do """ use MicrowavepropWeb, :live_view + import MicrowavepropWeb.LiveHelpers, only: [subscribe: 2] + alias Microwaveprop.Accounts.User alias Microwaveprop.Radio.Maidenhead alias Microwaveprop.Rover @@ -24,10 +26,7 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Show do |> push_navigate(to: ~p"/rover-planning")} %Mission{} = mission -> - _ = - if connected?(socket) do - Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "rover_planning:#{mission.id}") - end + subscribe(socket, "rover_planning:#{mission.id}") {:ok, assign(socket, diff --git a/lib/microwaveprop_web/live/status_live.ex b/lib/microwaveprop_web/live/status_live.ex index f4f964f6..7a34bfa1 100644 --- a/lib/microwaveprop_web/live/status_live.ex +++ b/lib/microwaveprop_web/live/status_live.ex @@ -3,6 +3,7 @@ defmodule MicrowavepropWeb.StatusLive do use MicrowavepropWeb, :live_view import Ecto.Query + import MicrowavepropWeb.LiveHelpers, only: [subscribe: 2] alias Microwaveprop.Cache alias Microwaveprop.Format @@ -15,16 +16,13 @@ defmodule MicrowavepropWeb.StatusLive do @impl true def mount(_params, _session, socket) do - _ = - if connected?(socket) do - :ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:contact_status") - :ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:oban_jobs") - # Rust prop-grid-rs emits NOTIFY propagation_ready on completion — - # PropagationNotifyListener fans it out as `propagation:updated` so - # the grid_tasks panel transitions from "running" → "done" without - # a single refresh. - Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated") - end + # Rust prop-grid-rs emits NOTIFY propagation_ready on completion — + # PropagationNotifyListener fans it out as `propagation:updated` so + # the grid_tasks panel transitions from "running" → "done" without + # a single refresh. + subscribe(socket, "db:contact_status") + subscribe(socket, "db:oban_jobs") + subscribe(socket, "propagation:updated") all = fetch_all_stats() diff --git a/lib/microwaveprop_web/live/submit_live.ex b/lib/microwaveprop_web/live/submit_live.ex index 61bad998..4491a958 100644 --- a/lib/microwaveprop_web/live/submit_live.ex +++ b/lib/microwaveprop_web/live/submit_live.ex @@ -3,6 +3,11 @@ defmodule MicrowavepropWeb.SubmitLive do use MicrowavepropWeb, :live_view use LiveStash, stored_keys: [:active_tab] + import MicrowavepropWeb.LiveHelpers, only: [current_user: 1] + import MicrowavepropWeb.SubmitLive.AdifUploadComponent + import MicrowavepropWeb.SubmitLive.CsvUploadComponent + import MicrowavepropWeb.SubmitLive.PreviewComponent + alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Radio alias Microwaveprop.Radio.AdifImport @@ -270,11 +275,6 @@ defmodule MicrowavepropWeb.SubmitLive do end end - defp error_to_string(:too_large), do: "File is too large" - defp error_to_string(:too_many_files), do: "Only one file allowed" - defp error_to_string(:not_accepted), do: "File type not accepted" - defp error_to_string(other), do: to_string(other) - @impl true def render(assigns) do ~H""" @@ -359,10 +359,6 @@ defmodule MicrowavepropWeb.SubmitLive do """ end - # In-template variant — takes an assigns map directly (not a socket). - defp current_user(%{current_scope: %{user: %{} = user}}), do: user - defp current_user(_), do: nil - defp single_contact_form(assigns) do ~H""" <.form for={@form} id="contact-form" phx-change="validate" phx-submit="save" class="space-y-4"> @@ -475,470 +471,4 @@ defmodule MicrowavepropWeb.SubmitLive do """ end - - defp csv_upload_form(assigns) do - ~H""" -
-
-
-

Upload a CSV file with multiple contacts. Columns:

- - station1, station2, grid1, grid2, band, mode, qso_timestamp, notes - -
    -
  • - Timestamps in most formats accepted (e.g. 2024-06-15T14:30:00Z, 6/15/2024 2:30 PM, 2024-06-15 14:30). All times assumed UTC. -
    • -
  • - Grid squares should be as detailed as possible (8 characters preferred, e.g. EM12kp37) -
    • -
  • Band in MHz (e.g. 10000, 24000)
    • -
  • - Mode is optional — omit the mode column entirely - (6 columns total) or leave its value blank. -
    • -
  • - Notes is optional — free-form operator commentary up to - 2000 characters. Leave the cell blank to store NULL. -
    • -
-

- - <.icon name="hero-arrow-down-tray" class="w-4 h-4" /> Download sample CSV - -

-
-
- -
-
- -
- -
-
- {entry.client_name} - - {entry.progress}% - - -
- -
- {error_to_string(err)} -
-
- -
- {error_to_string(err)} -
- - <%= if @current_user do %> - - <% end %> - - - -
- -
-
-
- """ - end - - defp adif_upload_form(assigns) do - assigns = assign(assigns, :mode_mapping, AdifImport.mode_mapping_reference()) - - ~H""" -
-
-
-

- Upload an ADIF (.adi / .adif) file exported from your logging program. -

-
    -
  • - Required fields: CALL, STATION_CALLSIGN - (or OPERATOR), GRIDSQUARE, MY_GRIDSQUARE, QSO_DATE, TIME_ON, - and FREQ - or BAND -
    • -
  • Contacts on amateur bands from 50 MHz and up will be imported
    • -
  • HF contacts (below 50 MHz) are silently skipped
    • -
  • Frequencies are fuzzy-matched to the nearest amateur band
    • -
  • - Operator commentary from NOTES - is carried over (falling back to COMMENT) -
    • -
-
-
- -
-
-

Mode handling

-

- ADIF modes are normalized to the six modes we track. SUBMODE takes - precedence when present (e.g. MODE=MFSK, SUBMODE=FT8 imports as FT8). -

- - - - - - - - - - - - - -
ADIF valueStored as
{adif}{mapped}
-
-
- -
-
- -
- -
-
- {entry.client_name} - - {entry.progress}% - - -
- -
- {error_to_string(err)} -
-
- -
- {error_to_string(err)} -
- - <%= if @current_user do %> - - <% end %> - - - -
- -
-
-
- """ - end - - defp csv_preview(assigns) do - assigns = - assign(assigns, - valid_count: length(assigns.preview.valid), - invalid_count: length(assigns.preview.invalid), - duplicate_count: length(assigns.preview.duplicates), - refinement_count: length(assigns.preview.refinements || []), - valid_sample: Enum.take(assigns.preview.valid, 20), - refinement_sample: Enum.take(assigns.preview.refinements || [], 50) - ) - - ~H""" -
-
-

Review before submitting

-

- Processed {@preview.total_rows} {if @preview.total_rows == 1, do: "row", else: "rows"}. - Nothing has been inserted yet — confirm below to import. -

-
- -
- <.summary_card - tone="success" - label="Valid" - value={@valid_count} - hint="Will be inserted" - /> - <.summary_card - tone="info" - label="Refinements" - value={@refinement_count} - hint="Will update existing contacts" - /> - <.summary_card - tone="warning" - label="Duplicates" - value={@duplicate_count} - hint="Match existing or earlier rows" - /> - <.summary_card - tone="error" - label="Invalid" - value={@invalid_count} - hint="Skipped — see errors" - /> -
- -
0}> -

Invalid rows ({@invalid_count})

-
- - - - - - - - - - - - - -
RowErrors
Row {row.row_num} -
{msg}
-
-
-

100} class="text-xs opacity-60 mt-1"> - Showing first 100 of {@invalid_count} invalid rows. -

-
- -
0}> -

Refinements ({@refinement_count})

-

- These rows match an existing contact but add more precise data. - Confirming will update the existing contact in place. -

-
- - - - - - - - - - - - - - - - - - - -
RowStation 1Station 2BandChanges
Row {row.row_num}{row.attrs["station1"]} {row.attrs["grid1"]}{row.attrs["station2"]} {row.attrs["grid2"]}{row.attrs["band"]} -
- {field}: - {value} -
-
-
-

length(@refinement_sample)} class="text-xs opacity-60 mt-1"> - Showing first {length(@refinement_sample)} of {@refinement_count} refinements. -

-
- -
0}> -

Duplicates ({@duplicate_count})

-

- Same two callsigns at the same grids on the same band within an hour. -

-
- - - - - - - - - - - - - - - - - - - - - -
RowStation 1Station 2BandTimestampSource
Row {row.row_num}{row.attrs["station1"]} {row.attrs["grid1"]}{row.attrs["station2"]} {row.attrs["grid2"]}{row.attrs["band"]} - {Calendar.strftime(row.timestamp, "%Y-%m-%d %H:%M UTC")} - {duplicate_source(row.reason)}
-
-

100} class="text-xs opacity-60 mt-1"> - Showing first 100 of {@duplicate_count} duplicates. -

-
- -
0}> -

Valid rows ready to import ({@valid_count})

-
- - - - - - - - - - - - - - - - - - - - - -
RowStation 1Station 2BandModeTimestamp
Row {row.row_num}{row.attrs["station1"]} {row.attrs["grid1"]}{row.attrs["station2"]} {row.attrs["grid2"]}{row.attrs["band"]}{row.attrs["mode"]} - {Calendar.strftime(row.timestamp, "%Y-%m-%d %H:%M UTC")} -
-
-

length(@valid_sample)} class="text-xs opacity-60 mt-1"> - Showing first {length(@valid_sample)} of {@valid_count} valid rows. -

-
- -
- - -
-
- """ - end - - attr :tone, :string, required: true - attr :label, :string, required: true - attr :value, :integer, required: true - attr :hint, :string, required: true - - defp summary_card(assigns) do - ~H""" -
"border-success/30 bg-success/10" - "warning" -> "border-warning/30 bg-warning/10" - "error" -> "border-error/30 bg-error/10" - "info" -> "border-info/30 bg-info/10" - _ -> "border-base-300 bg-base-200" - end - ]}> -
{@label}
-
{@value}
-
{@hint}
-
- """ - end - - defp duplicate_source(:existing_contact), do: "Already in database" - defp duplicate_source(:earlier_in_upload), do: "Earlier row in this upload" - defp duplicate_source(_), do: "Duplicate" - - defp pluralize(1, word), do: word - defp pluralize(_, word), do: word <> "s" - - # An upload entry is actively streaming when its progress is partway - # between 0 and 100. With `auto_upload: false` entries sit at - # `progress: 0` after the user picks a file and only start streaming - # when the form is submitted, so gating the progress bar and percent - # text on the 0 < progress < 100 window keeps them hidden until the - # upload is actually in flight. - defp uploading?(%{progress: progress}), do: progress > 0 and progress < 100 - defp uploading?(_), do: false - - defp confirm_button_label(valid, 0) do - "Looks good — submit #{valid} #{pluralize(valid, "contact")}" - end - - defp confirm_button_label(0, refined) do - "Looks good — refine #{refined} existing #{pluralize(refined, "contact")}" - end - - defp confirm_button_label(valid, refined) do - "Looks good — submit #{valid} and refine #{refined}" - end - - defp confirm_prompt(valid, 0), do: "Import #{valid} contacts?" - defp confirm_prompt(0, refined), do: "Refine #{refined} existing contacts?" - - defp confirm_prompt(valid, refined) do - "Import #{valid} contacts and refine #{refined} existing?" - end end diff --git a/lib/microwaveprop_web/live/submit_live/adif_upload_component.ex b/lib/microwaveprop_web/live/submit_live/adif_upload_component.ex new file mode 100644 index 00000000..ee64f7d8 --- /dev/null +++ b/lib/microwaveprop_web/live/submit_live/adif_upload_component.ex @@ -0,0 +1,131 @@ +defmodule MicrowavepropWeb.SubmitLive.AdifUploadComponent do + @moduledoc false + use MicrowavepropWeb, :html + + alias Microwaveprop.Radio.AdifImport + + @doc false + @spec adif_upload_form(map()) :: Phoenix.LiveView.Rendered.t() + def adif_upload_form(assigns) do + assigns = assign(assigns, mode_mapping: AdifImport.mode_mapping_reference()) + + ~H""" +
+
+
+

+ Upload an ADIF (.adi / .adif) file exported from your logging program. +

+
    +
  • + Required fields: CALL, STATION_CALLSIGN + (or OPERATOR), GRIDSQUARE, MY_GRIDSQUARE, QSO_DATE, TIME_ON, + and FREQ + or BAND +
    • +
  • Contacts on amateur bands from 50 MHz and up will be imported
    • +
  • HF contacts (below 50 MHz) are silently skipped
    • +
  • Frequencies are fuzzy-matched to the nearest amateur band
    • +
  • + Operator commentary from NOTES + is carried over (falling back to COMMENT) +
    • +
+
+
+ +
+
+

Mode handling

+

+ ADIF modes are normalized to the six modes we track. SUBMODE takes + precedence when present (e.g. MODE=MFSK, SUBMODE=FT8 imports as FT8). +

+ + + + + + + + + + + + + +
ADIF valueStored as
{adif}{mapped}
+
+
+ +
+
+ +
+ +
+
+ {entry.client_name} + + {entry.progress}% + + +
+ +
+ {error_to_string(err)} +
+
+ +
+ {error_to_string(err)} +
+ + <%= if @current_user do %> + + <% end %> + + + +
+ +
+
+
+ """ + end + + defp error_to_string(:too_large), do: "File is too large" + defp error_to_string(:too_many_files), do: "Only one file allowed" + defp error_to_string(:not_accepted), do: "File type not accepted" + defp error_to_string(other), do: to_string(other) + + defp uploading?(%{progress: progress}), do: progress > 0 and progress < 100 + defp uploading?(_), do: false +end diff --git a/lib/microwaveprop_web/live/submit_live/csv_upload_component.ex b/lib/microwaveprop_web/live/submit_live/csv_upload_component.ex new file mode 100644 index 00000000..50211a7f --- /dev/null +++ b/lib/microwaveprop_web/live/submit_live/csv_upload_component.ex @@ -0,0 +1,112 @@ +defmodule MicrowavepropWeb.SubmitLive.CsvUploadComponent do + @moduledoc false + use MicrowavepropWeb, :html + + @doc false + @spec csv_upload_form(map()) :: Phoenix.LiveView.Rendered.t() + def csv_upload_form(assigns) do + ~H""" +
+
+
+

Upload a CSV file with multiple contacts. Columns:

+ + station1, station2, grid1, grid2, band, mode, qso_timestamp, notes + +
    +
  • + Timestamps in most formats accepted (e.g. 2024-06-15T14:30:00Z, 6/15/2024 2:30 PM, 2024-06-15 14:30). All times assumed UTC. +
    • +
  • + Grid squares should be as detailed as possible (8 characters preferred, e.g. EM12kp37) +
    • +
  • Band in MHz (e.g. 10000, 24000)
    • +
  • + Mode is optional — omit the mode column entirely + (6 columns total) or leave its value blank. +
    • +
  • + Notes is optional — free-form operator commentary up to + 2000 characters. Leave the cell blank to store NULL. +
    • +
+

+ + <.icon name="hero-arrow-down-tray" class="w-4 h-4" /> Download sample CSV + +

+
+
+ +
+
+ +
+ +
+
+ {entry.client_name} + + {entry.progress}% + + +
+ +
+ {error_to_string(err)} +
+
+ +
+ {error_to_string(err)} +
+ + <%= if @current_user do %> + + <% end %> + + + +
+ +
+
+
+ """ + end + + defp error_to_string(:too_large), do: "File is too large" + defp error_to_string(:too_many_files), do: "Only one file allowed" + defp error_to_string(:not_accepted), do: "File type not accepted" + defp error_to_string(other), do: to_string(other) + + # An upload entry is actively streaming when its progress is partway + # between 0 and 100. With `auto_upload: false` entries sit at + # `progress: 0` after the user picks a file and only start streaming + # when the form is submitted, so gating the progress bar and percent + # text on the 0 < progress < 100 window keeps them hidden until the + # upload is actually in flight. + defp uploading?(%{progress: progress}), do: progress > 0 and progress < 100 + defp uploading?(_), do: false +end diff --git a/lib/microwaveprop_web/live/submit_live/preview_component.ex b/lib/microwaveprop_web/live/submit_live/preview_component.ex new file mode 100644 index 00000000..82b50ea5 --- /dev/null +++ b/lib/microwaveprop_web/live/submit_live/preview_component.ex @@ -0,0 +1,260 @@ +defmodule MicrowavepropWeb.SubmitLive.PreviewComponent do + @moduledoc false + use MicrowavepropWeb, :html + + alias Phoenix.LiveView.Rendered + + @doc false + @spec csv_preview(map()) :: Rendered.t() + def csv_preview(assigns) do + assigns = + assign(assigns, + valid_count: length(assigns.preview.valid), + invalid_count: length(assigns.preview.invalid), + duplicate_count: length(assigns.preview.duplicates), + refinement_count: length(assigns.preview.refinements || []), + valid_sample: Enum.take(assigns.preview.valid, 20), + refinement_sample: Enum.take(assigns.preview.refinements || [], 50) + ) + + ~H""" +
+
+

Review before submitting

+

+ Processed {@preview.total_rows} {if @preview.total_rows == 1, do: "row", else: "rows"}. + Nothing has been inserted yet — confirm below to import. +

+
+ +
+ <.summary_card + tone="success" + label="Valid" + value={@valid_count} + hint="Will be inserted" + /> + <.summary_card + tone="info" + label="Refinements" + value={@refinement_count} + hint="Will update existing contacts" + /> + <.summary_card + tone="warning" + label="Duplicates" + value={@duplicate_count} + hint="Match existing or earlier rows" + /> + <.summary_card + tone="error" + label="Invalid" + value={@invalid_count} + hint="Skipped — see errors" + /> +
+ +
0}> +

Invalid rows ({@invalid_count})

+
+ + + + + + + + + + + + + +
RowErrors
Row {row.row_num} +
{msg}
+
+
+

100} class="text-xs opacity-60 mt-1"> + Showing first 100 of {@invalid_count} invalid rows. +

+
+ +
0}> +

Refinements ({@refinement_count})

+

+ These rows match an existing contact but add more precise data. + Confirming will update the existing contact in place. +

+
+ + + + + + + + + + + + + + + + + + + +
RowStation 1Station 2BandChanges
Row {row.row_num}{row.attrs["station1"]} {row.attrs["grid1"]}{row.attrs["station2"]} {row.attrs["grid2"]}{row.attrs["band"]} +
+ {field}: + {value} +
+
+
+

length(@refinement_sample)} class="text-xs opacity-60 mt-1"> + Showing first {length(@refinement_sample)} of {@refinement_count} refinements. +

+
+ +
0}> +

Duplicates ({@duplicate_count})

+

+ Same two callsigns at the same grids on the same band within an hour. +

+
+ + + + + + + + + + + + + + + + + + + + + +
RowStation 1Station 2BandTimestampSource
Row {row.row_num}{row.attrs["station1"]} {row.attrs["grid1"]}{row.attrs["station2"]} {row.attrs["grid2"]}{row.attrs["band"]} + {Calendar.strftime(row.timestamp, "%Y-%m-%d %H:%M UTC")} + {duplicate_source(row.reason)}
+
+

100} class="text-xs opacity-60 mt-1"> + Showing first 100 of {@duplicate_count} duplicates. +

+
+ +
0}> +

Valid rows ready to import ({@valid_count})

+
+ + + + + + + + + + + + + + + + + + + + + +
RowStation 1Station 2BandModeTimestamp
Row {row.row_num}{row.attrs["station1"]} {row.attrs["grid1"]}{row.attrs["station2"]} {row.attrs["grid2"]}{row.attrs["band"]}{row.attrs["mode"]} + {Calendar.strftime(row.timestamp, "%Y-%m-%d %H:%M UTC")} +
+
+

length(@valid_sample)} class="text-xs opacity-60 mt-1"> + Showing first {length(@valid_sample)} of {@valid_count} valid rows. +

+
+ +
+ + +
+
+ """ + end + + attr :tone, :string, required: true + attr :label, :string, required: true + attr :value, :integer, required: true + attr :hint, :string, required: true + + @doc false + @spec summary_card(map()) :: Rendered.t() + def summary_card(assigns) do + ~H""" +
"border-success/30 bg-success/10" + "warning" -> "border-warning/30 bg-warning/10" + "error" -> "border-error/30 bg-error/10" + "info" -> "border-info/30 bg-info/10" + _ -> "border-base-300 bg-base-200" + end + ]}> +
{@label}
+
{@value}
+
{@hint}
+
+ """ + end + + defp duplicate_source(:existing_contact), do: "Already in database" + defp duplicate_source(:earlier_in_upload), do: "Earlier row in this upload" + defp duplicate_source(_), do: "Duplicate" + + defp pluralize(1, word), do: word + defp pluralize(_, word), do: word <> "s" + + defp confirm_button_label(valid, 0) do + "Looks good — submit #{valid} #{pluralize(valid, "contact")}" + end + + defp confirm_button_label(0, refined) do + "Looks good — refine #{refined} existing #{pluralize(refined, "contact")}" + end + + defp confirm_button_label(valid, refined) do + "Looks good — submit #{valid} and refine #{refined}" + end + + defp confirm_prompt(valid, 0), do: "Import #{valid} contacts?" + defp confirm_prompt(0, refined), do: "Refine #{refined} existing contacts?" + + defp confirm_prompt(valid, refined) do + "Import #{valid} contacts and refine #{refined} existing?" + end +end diff --git a/lib/microwaveprop_web/live/weather_ca_map_live.ex b/lib/microwaveprop_web/live/weather_ca_map_live.ex index 0b1f9662..5849e864 100644 --- a/lib/microwaveprop_web/live/weather_ca_map_live.ex +++ b/lib/microwaveprop_web/live/weather_ca_map_live.ex @@ -7,6 +7,8 @@ defmodule MicrowavepropWeb.WeatherCaMapLive do """ use MicrowavepropWeb, :live_view + import MicrowavepropWeb.LiveHelpers, only: [subscribe: 2] + alias Microwaveprop.Weather alias Microwaveprop.Weather.MapLayers alias MicrowavepropWeb.WeatherMapComponent @@ -24,11 +26,8 @@ defmodule MicrowavepropWeb.WeatherCaMapLive do @impl true def mount(params, _session, socket) do - _ = - if connected?(socket) do - :ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "weather:updated") - Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:pipeline") - end + subscribe(socket, "weather:updated") + subscribe(socket, "propagation:pipeline") valid_times = recent_valid_times(Weather.available_hrdps_valid_times()) initial_vt = pick_initial_valid_time(valid_times) diff --git a/lib/microwaveprop_web/live/weather_map_live.ex b/lib/microwaveprop_web/live/weather_map_live.ex index c382520e..876d332c 100644 --- a/lib/microwaveprop_web/live/weather_map_live.ex +++ b/lib/microwaveprop_web/live/weather_map_live.ex @@ -2,6 +2,8 @@ defmodule MicrowavepropWeb.WeatherMapLive do @moduledoc "`/weather` — HRRR-derived forecast fields (temp, Td, wind, refractivity) on a map." use MicrowavepropWeb, :live_view + import MicrowavepropWeb.LiveHelpers, only: [subscribe: 2] + alias Microwaveprop.Weather alias Microwaveprop.Weather.MapLayers alias MicrowavepropWeb.WeatherMapComponent @@ -26,11 +28,8 @@ defmodule MicrowavepropWeb.WeatherMapLive do @impl true def mount(params, _session, socket) do - _ = - if connected?(socket) do - :ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "weather:updated") - Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:pipeline") - end + subscribe(socket, "weather:updated") + subscribe(socket, "propagation:pipeline") # Default the timeline cursor to the valid_time closest to "now" so # users land on current conditions, not a +11h forecast hour (which diff --git a/lib/microwaveprop_web/live_helpers.ex b/lib/microwaveprop_web/live_helpers.ex index 72e00c8b..20482789 100644 --- a/lib/microwaveprop_web/live_helpers.ex +++ b/lib/microwaveprop_web/live_helpers.ex @@ -4,6 +4,10 @@ defmodule MicrowavepropWeb.LiveHelpers do view-specific logic belongs in its own LiveView. """ + # ── Parse helpers ──────────────────────────────────────────────────── + + alias Phoenix.LiveView.Socket + @doc """ Parses a value into a float, returning `default` for nil, empty, or unparseable input. Accepts strings, numbers, or any value with a @@ -34,4 +38,46 @@ defmodule MicrowavepropWeb.LiveHelpers do :error -> default end end + + # ── User helpers ───────────────────────────────────────────────────── + + @doc """ + Extracts the current user from a LiveView socket or assigns map. + Returns the `%User{}` struct or `nil` if not authenticated. + """ + @spec current_user(Socket.t() | map()) :: Ecto.Schema.t() | nil + def current_user(%Socket{assigns: assigns}) do + current_user(assigns) + end + + @spec current_user(map()) :: Ecto.Schema.t() | nil + def current_user(%{current_scope: %{user: %_{} = user}}), do: user + def current_user(%{user: %_{} = user}), do: user + def current_user(_), do: nil + + # ── PubSub helpers ─────────────────────────────────────────────────── + + @doc """ + Subscribes the socket to a Phoenix.PubSub topic, but only when the + socket is connected (WebSocket). Safe to call unconditionally in mount/3. + """ + @spec subscribe(Socket.t(), String.t()) :: :ok + def subscribe(%Socket{} = socket, topic) when is_binary(topic) do + if Phoenix.LiveView.connected?(socket) do + Phoenix.PubSub.subscribe(Microwaveprop.PubSub, topic) + end + + :ok + end + + # ── URL helpers ────────────────────────────────────────────────────── + + @doc """ + Merges URL params with defaults for handle_params/3. Returns a map of + merged values suitable for assigning to the socket. + """ + @spec assign_url_params(map(), map()) :: map() + def assign_url_params(params, defaults) when is_map(params) and is_map(defaults) do + Map.merge(defaults, Map.take(params, Map.keys(defaults))) + end end diff --git a/test/microwaveprop/radio/contact_submission_test.exs b/test/microwaveprop/radio/contact_submission_test.exs index 171fbf98..b541b6ec 100644 --- a/test/microwaveprop/radio/contact_submission_test.exs +++ b/test/microwaveprop/radio/contact_submission_test.exs @@ -50,13 +50,13 @@ defmodule Microwaveprop.Radio.ContactSubmissionTest do test "rejects invalid grid1" do attrs = Map.put(@valid_attrs, :grid1, "ZZ99") changeset = Contact.submission_changeset(%Contact{}, attrs) - assert "is not a valid Maidenhead grid square" in errors_on(changeset).grid1 + assert "is not a valid Maidenhead grid" in errors_on(changeset).grid1 end test "rejects invalid grid2" do attrs = Map.put(@valid_attrs, :grid2, "ZZ99") changeset = Contact.submission_changeset(%Contact{}, attrs) - assert "is not a valid Maidenhead grid square" in errors_on(changeset).grid2 + assert "is not a valid Maidenhead grid" in errors_on(changeset).grid2 end test "accepts valid 6-char grids" do diff --git a/test/microwaveprop_web/controllers/user_settings_controller_test.exs b/test/microwaveprop_web/controllers/user_settings_controller_test.exs index 0ee53ae9..cdb40e33 100644 --- a/test/microwaveprop_web/controllers/user_settings_controller_test.exs +++ b/test/microwaveprop_web/controllers/user_settings_controller_test.exs @@ -94,7 +94,7 @@ defmodule MicrowavepropWeb.UserSettingsControllerTest do response = html_response(conn, 200) assert response =~ "Home QTH" - assert response =~ "must be a 4 or 6 character" + assert response =~ "is not a valid Maidenhead" end end