From a2e2776e7cfa77cc38aa90889ad96d61b435df95 Mon Sep 17 00:00:00 2001
From: Graham McInitre <graham.mcintire@island.io>
Date: Wed, 15 Jul 2026 14:18:43 -0500
Subject: [PATCH] fix: batch pskr calibration upsert into 1,000-row chunks to
 avoid Postgres 65,535 param limit

---
 lib/microwaveprop/pskr/calibration_sampler.ex | 57 +++++++++++--------
 1 file changed, 33 insertions(+), 24 deletions(-)

diff --git a/lib/microwaveprop/pskr/calibration_sampler.ex b/lib/microwaveprop/pskr/calibration_sampler.ex
index 331d4744..6e216e46 100644
--- a/lib/microwaveprop/pskr/calibration_sampler.ex
+++ b/lib/microwaveprop/pskr/calibration_sampler.ex
@@ -267,35 +267,44 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
 
   # ── Upsert ──────────────────────────────────────────────────────
 
+  # Each sample has ~20 fields. PostgreSQL's parameter limit is 65,535,
+  # so a single insert_all with > ~3,200 rows will fail. Batch at 1,000
+  # to stay well under the limit even at peak hours.
+  @upsert_batch_size 1_000
+
   defp upsert([]), do: 0
 
   defp upsert(samples) do
-    {count, _} =
-      Repo.insert_all(CalibrationSample, samples,
-        on_conflict:
-          from(c in CalibrationSample,
-            update: [
-              set: [
-                spot_count: fragment("EXCLUDED.spot_count"),
-                distinct_paths: fragment("EXCLUDED.distinct_paths"),
-                median_distance_km: fragment("EXCLUDED.median_distance_km"),
-                modes: fragment("EXCLUDED.modes"),
-                surface_temp_c: fragment("EXCLUDED.surface_temp_c"),
-                surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"),
-                pwat_mm: fragment("EXCLUDED.pwat_mm"),
-                surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"),
-                min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"),
-                hpbl_m: fragment("EXCLUDED.hpbl_m"),
-                hrrr_ducting_detected: fragment("EXCLUDED.hrrr_ducting_detected"),
-                kp_index: fragment("EXCLUDED.kp_index"),
-                updated_at: fragment("EXCLUDED.updated_at")
+    samples
+    |> Enum.chunk_every(@upsert_batch_size)
+    |> Enum.reduce(0, fn batch, total ->
+      {count, _} =
+        Repo.insert_all(CalibrationSample, batch,
+          on_conflict:
+            from(c in CalibrationSample,
+              update: [
+                set: [
+                  spot_count: fragment("EXCLUDED.spot_count"),
+                  distinct_paths: fragment("EXCLUDED.distinct_paths"),
+                  median_distance_km: fragment("EXCLUDED.median_distance_km"),
+                  modes: fragment("EXCLUDED.modes"),
+                  surface_temp_c: fragment("EXCLUDED.surface_temp_c"),
+                  surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"),
+                  pwat_mm: fragment("EXCLUDED.pwat_mm"),
+                  surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"),
+                  min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"),
+                  hpbl_m: fragment("EXCLUDED.hpbl_m"),
+                  hrrr_ducting_detected: fragment("EXCLUDED.hrrr_ducting_detected"),
+                  kp_index: fragment("EXCLUDED.kp_index"),
+                  updated_at: fragment("EXCLUDED.updated_at")
+                ]
               ]
-            ]
-          ),
-        conflict_target: [:hour_utc, :band, :midpoint_lat, :midpoint_lon]
-      )
+            ),
+          conflict_target: [:hour_utc, :band, :midpoint_lat, :midpoint_lon]
+        )
 
-    count
+      total + count
+    end)
   end
 
   defp align_to_hour(%DateTime{} = dt) do