From 3874fc173de95679ef38fa9e74d9afa0a31b573f Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Tue, 31 Mar 2026 10:35:59 -0500
Subject: [PATCH] Atomic score upserts and auto-prune old propagation data

- Wrap upsert_scores in Repo.transaction for all-or-nothing visibility
- Prune scores older than the 2 most recent valid_times after each upsert
- Add band-specific latest_valid_time/1 to eliminate N+1 query
- Add require Logger to Propagation module
---
 lib/microwaveprop/propagation.ex | 56 +++++++++++++++++++++++++-------
 1 file changed, 44 insertions(+), 12 deletions(-)

diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex
index d7d4daed..a7af74d0 100644
--- a/lib/microwaveprop/propagation.ex
+++ b/lib/microwaveprop/propagation.ex
@@ -10,6 +10,8 @@ defmodule Microwaveprop.Propagation do
   alias Microwaveprop.Repo
   alias Microwaveprop.Weather.SoundingParams
 
+  require Logger
+
   @doc """
   Score a single grid point across all bands using HRRR profile data.
   Returns a list of %{band_mhz, score, factors} maps.
@@ -54,7 +56,7 @@ defmodule Microwaveprop.Propagation do
     end)
   end
 
-  @doc "Upsert propagation scores in batches."
+  @doc "Upsert propagation scores in batches within a transaction so readers see all-or-nothing."
   def upsert_scores(scores) do
     now = DateTime.truncate(DateTime.utc_now(), :second)
 
@@ -73,20 +75,50 @@ defmodule Microwaveprop.Propagation do
         }
       end)
 
-    total =
-      entries
-      |> Enum.chunk_every(500)
-      |> Enum.reduce(0, fn chunk, acc ->
-        {count, _} =
-          Repo.insert_all(GridScore, chunk,
-            on_conflict: {:replace, [:score, :factors, :updated_at]},
-            conflict_target: [:lat, :lon, :valid_time, :band_mhz]
-          )
+    result =
+      Repo.transaction(fn ->
+        entries
+        |> Enum.chunk_every(500)
+        |> Enum.reduce(0, fn chunk, acc ->
+          {count, _} =
+            Repo.insert_all(GridScore, chunk,
+              on_conflict: {:replace, [:score, :factors, :updated_at]},
+              conflict_target: [:lat, :lon, :valid_time, :band_mhz]
+            )
 
-        acc + count
+          acc + count
+        end)
       end)
 
-    {:ok, total}
+    case result do
+      {:ok, _count} -> prune_old_scores()
+      _ -> :ok
+    end
+
+    result
+  end
+
+  defp prune_old_scores do
+    keep_times =
+      from(gs in GridScore,
+        select: gs.valid_time,
+        distinct: gs.valid_time,
+        order_by: [desc: gs.valid_time],
+        limit: 2
+      )
+      |> Repo.all()
+
+    if length(keep_times) == 2 do
+      oldest_kept = List.last(keep_times)
+
+      {deleted, _} =
+        from(gs in GridScore, where: gs.valid_time < ^oldest_kept)
+        |> Repo.delete_all()
+
+      if deleted > 0 do
+        Logger.info("PropagationScores: pruned #{deleted} old scores, keeping 2 most recent valid_times")
+      end
+    end
   end
 
   @doc "Get the latest scores for a band, optionally within a bounding box. Excludes factors for performance."