From 084e6b62245e467d299221dc6f3f5f05bc546ef6 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 14:22:15 -0500 Subject: [PATCH] refactor(db): use explicit on_conflict set clauses over :replace_all_except Convert every context/worker upsert that used the catch-all `:replace_all_except` form to the explicit `{:replace, [:col1, ...]}` form. Reviewers can now see exactly which columns an upsert overwrites, and adding a new column to a schema no longer silently opts it into the update path. Behaviour is preserved bit-for-bit: each new explicit list contains every column the old `:replace_all_except` would have overwritten. Touched: - Microwaveprop.SpaceWeather (Kp / F10.7 / X-ray shared helper) - Microwaveprop.Ionosphere (observation upsert) - NexradWorker, HrrrNativeGridWorker (insert_all grids) - CommonVolumeRadarWorker, RadarFrameWorker (per-contact stats) Also pins the conflict behaviour for the ContactCommonVolumeRadar upsert path in RadarFrameWorkerTest so a future column addition that isn't reflected in the explicit list fails loudly. --- lib/microwaveprop/ionosphere.ex | 9 +++-- lib/microwaveprop/space_weather.ex | 23 ++++++++++-- .../workers/common_volume_radar_worker.ex | 3 +- .../workers/hrrr_native_grid_worker.ex | 6 ++- lib/microwaveprop/workers/nexrad_worker.ex | 3 +- .../workers/radar_frame_worker.ex | 3 +- .../workers/radar_frame_worker_test.exs | 37 +++++++++++++++++++ 7 files changed, 73 insertions(+), 11 deletions(-) diff --git a/lib/microwaveprop/ionosphere.ex b/lib/microwaveprop/ionosphere.ex index fb51bdb9..6bc8dedd 100644 --- a/lib/microwaveprop/ionosphere.ex +++ b/lib/microwaveprop/ionosphere.ex @@ -9,10 +9,13 @@ defmodule Microwaveprop.Ionosphere do alias Microwaveprop.Ionosphere.Observation alias Microwaveprop.Repo + @observation_update_fields ~w(confidence_score fo_f2_mhz fo_e_mhz fo_es_mhz hm_f2_km mufd_mhz updated_at)a + @doc """ Upsert a list of parsed `GiroClient` observations for a station. - Conflicts on `(station_code, valid_time)` replace every column except - the primary key and inserted_at. Returns `{:ok, count}`. + Conflicts on `(station_code, valid_time)` replace the scaled + characteristic columns plus `updated_at`; `id`, `inserted_at`, + `station_code`, and `valid_time` are preserved. Returns `{:ok, count}`. """ @spec upsert_observations(String.t(), [map()]) :: {:ok, non_neg_integer()} def upsert_observations(_station_code, []), do: {:ok, 0} @@ -30,7 +33,7 @@ defmodule Microwaveprop.Ionosphere do {count, _} = Repo.insert_all(Observation, rows, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: {:replace, @observation_update_fields}, conflict_target: [:station_code, :valid_time] ) diff --git a/lib/microwaveprop/space_weather.ex b/lib/microwaveprop/space_weather.ex index 14c38781..8c72e4cf 100644 --- a/lib/microwaveprop/space_weather.ex +++ b/lib/microwaveprop/space_weather.ex @@ -15,11 +15,16 @@ defmodule Microwaveprop.SpaceWeather do # ---------- Kp ---------- + @kp_update_fields ~w(kp_index estimated_kp updated_at)a + @spec upsert_kp([map()]) :: {:ok, non_neg_integer()} def upsert_kp([]), do: {:ok, 0} def upsert_kp(rows) when is_list(rows) do - stamp_upsert(GeomagneticObservation, rows, conflict_target: :valid_time) + stamp_upsert(GeomagneticObservation, rows, + conflict_target: :valid_time, + update_fields: @kp_update_fields + ) end @spec latest_kp() :: GeomagneticObservation.t() | nil @@ -29,11 +34,16 @@ defmodule Microwaveprop.SpaceWeather do # ---------- F10.7 ---------- + @solar_flux_update_fields ~w(frequency_mhz flux_sfu reporting_schedule ninety_day_mean updated_at)a + @spec upsert_solar_flux([map()]) :: {:ok, non_neg_integer()} def upsert_solar_flux([]), do: {:ok, 0} def upsert_solar_flux(rows) when is_list(rows) do - stamp_upsert(SolarFluxObservation, rows, conflict_target: :valid_time) + stamp_upsert(SolarFluxObservation, rows, + conflict_target: :valid_time, + update_fields: @solar_flux_update_fields + ) end @spec latest_f107() :: SolarFluxObservation.t() | nil @@ -43,11 +53,16 @@ defmodule Microwaveprop.SpaceWeather do # ---------- GOES X-ray ---------- + @xray_update_fields ~w(satellite flux_wm2 electron_contaminated updated_at)a + @spec upsert_xray([map()]) :: {:ok, non_neg_integer()} def upsert_xray([]), do: {:ok, 0} def upsert_xray(rows) when is_list(rows) do - stamp_upsert(SolarXrayObservation, rows, conflict_target: [:valid_time, :energy_band]) + stamp_upsert(SolarXrayObservation, rows, + conflict_target: [:valid_time, :energy_band], + update_fields: @xray_update_fields + ) end @spec latest_xray() :: SolarXrayObservation.t() | nil @@ -69,7 +84,7 @@ defmodule Microwaveprop.SpaceWeather do {count, _} = Repo.insert_all(schema, stamped, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: {:replace, Keyword.fetch!(opts, :update_fields)}, conflict_target: Keyword.fetch!(opts, :conflict_target) ) diff --git a/lib/microwaveprop/workers/common_volume_radar_worker.ex b/lib/microwaveprop/workers/common_volume_radar_worker.ex index 1dcf862c..b3c7d6bf 100644 --- a/lib/microwaveprop/workers/common_volume_radar_worker.ex +++ b/lib/microwaveprop/workers/common_volume_radar_worker.ex @@ -265,7 +265,8 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do %ContactCommonVolumeRadar{} |> ContactCommonVolumeRadar.changeset(Map.merge(stats, %{contact_id: contact_id, observed_at: observed_at})) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :contact_id, :inserted_at]}, + on_conflict: {:replace, ~w(observed_at common_volume_km2 pixel_count rain_pixel_count heavy_rain_pixel_count + core_pixel_count max_dbz mean_dbz coverage_pct updated_at)a}, conflict_target: :contact_id ) end diff --git a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex index b57afc0b..4dd35386 100644 --- a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex +++ b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex @@ -157,7 +157,11 @@ defmodule Microwaveprop.Workers.HrrrNativeGridWorker do Repo.insert_all( HrrrNativeProfile, rows, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: {:replace, ~w(run_time level_count heights_m temp_k spfh pressure_pa + u_wind_ms v_wind_ms tke_m2s2 + surface_temp_k surface_spfh surface_pressure_pa + inversion_top_m bulk_richardson theta_e_jump_k shear_at_top_ms + ducts best_duct_band_ghz updated_at)a}, conflict_target: [:lat, :lon, :valid_time] ) diff --git a/lib/microwaveprop/workers/nexrad_worker.ex b/lib/microwaveprop/workers/nexrad_worker.ex index 63eed505..8ce87aab 100644 --- a/lib/microwaveprop/workers/nexrad_worker.ex +++ b/lib/microwaveprop/workers/nexrad_worker.ex @@ -83,7 +83,8 @@ defmodule Microwaveprop.Workers.NexradWorker do Repo.insert_all( NexradObservation, rows, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: + {:replace, ~w(mean_reflectivity_dbz max_reflectivity_dbz texture_variance pixel_count updated_at)a}, conflict_target: [:lat, :lon, :observed_at] ) diff --git a/lib/microwaveprop/workers/radar_frame_worker.ex b/lib/microwaveprop/workers/radar_frame_worker.ex index f8c23a39..2e1a9fbb 100644 --- a/lib/microwaveprop/workers/radar_frame_worker.ex +++ b/lib/microwaveprop/workers/radar_frame_worker.ex @@ -139,7 +139,8 @@ defmodule Microwaveprop.Workers.RadarFrameWorker do %ContactCommonVolumeRadar{} |> ContactCommonVolumeRadar.changeset(Map.merge(stats, %{contact_id: contact_id, observed_at: observed_at})) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :contact_id, :inserted_at]}, + on_conflict: {:replace, ~w(observed_at common_volume_km2 pixel_count rain_pixel_count heavy_rain_pixel_count + core_pixel_count max_dbz mean_dbz coverage_pct updated_at)a}, conflict_target: :contact_id ) end diff --git a/test/microwaveprop/workers/radar_frame_worker_test.exs b/test/microwaveprop/workers/radar_frame_worker_test.exs index 4f31e12a..776c966e 100644 --- a/test/microwaveprop/workers/radar_frame_worker_test.exs +++ b/test/microwaveprop/workers/radar_frame_worker_test.exs @@ -60,6 +60,43 @@ defmodule Microwaveprop.Workers.RadarFrameWorkerTest do assert %Contact{radar_status: :complete} = Repo.get!(Contact, c1.id) assert %Contact{radar_status: :complete} = Repo.get!(Contact, c2.id) end + + test "re-running process_frame/4 upserts in place (one row per contact, observed_at updated)" do + # Pins the on_conflict: {:replace, [...]} clause on contact_id. If a + # column gets added to ContactCommonVolumeRadar and the explicit list + # isn't updated, or if the list names a non-existent column, this + # test fails loudly. + contact = insert_contact(%{station1: "RR1"}) + + pixels = :binary.copy(<<0>>, 100) + width = 10 + + :ok = + RadarFrameWorker.process_frame( + [Repo.reload!(contact)], + ~U[2024-09-15 18:30:00Z], + pixels, + width + ) + + first = Repo.get_by!(ContactCommonVolumeRadar, contact_id: contact.id) + + :ok = + RadarFrameWorker.process_frame( + [Repo.reload!(contact)], + ~U[2024-09-15 18:35:00Z], + pixels, + width + ) + + rows = + Repo.all(from(r in ContactCommonVolumeRadar, where: r.contact_id == ^contact.id)) + + assert length(rows) == 1 + [second] = rows + assert second.id == first.id + assert second.observed_at == ~U[2024-09-15 18:35:00Z] + end end describe "perform/1" do