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,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]
       )
 

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)
       )
 

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

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]
       )
 

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]
           )
 

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

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