diff --git a/lib/microwaveprop/partition_manager.ex b/lib/microwaveprop/partition_manager.ex index d8235aa4..a92b81cf 100644 --- a/lib/microwaveprop/partition_manager.ex +++ b/lib/microwaveprop/partition_manager.ex @@ -40,6 +40,14 @@ defmodule Microwaveprop.PartitionManager do @default_parents ~w(hrrr_profiles hrdps_profiles) + @typedoc """ + Per-partition result. `parent` is the parent table name, `name` is + the resolved child partition name, and the status flags whether + this call performed the CREATE or found an existing partition that + already covered the range. + """ + @type result :: {parent :: String.t(), name :: String.t(), :created | :exists} + @doc """ For each configured parent table, ensure partitions covering the current quarter and the next `lookahead_quarters` quarters exist. @@ -47,8 +55,7 @@ defmodule Microwaveprop.PartitionManager do Returns a list of `{parent, partition_name, :created | :exists}` entries so the worker can log a single summary line. """ - @spec ensure_quarterly_partitions(non_neg_integer(), [String.t()]) :: - [{String.t(), String.t(), :created | :exists}] + @spec ensure_quarterly_partitions(non_neg_integer(), [String.t()]) :: [result()] def ensure_quarterly_partitions(lookahead_quarters \\ 4, parents \\ @default_parents) when is_integer(lookahead_quarters) and lookahead_quarters >= 0 and is_list(parents) do today = Date.utc_today() @@ -59,10 +66,8 @@ defmodule Microwaveprop.PartitionManager do end end - # Public for direct access from tests / mix tasks. Returns - # `{parent, partition_name, :created | :exists}`. - @spec ensure_partition(String.t(), Date.t(), Date.t()) :: - {String.t(), String.t(), :created | :exists} + # Public for direct access from tests / mix tasks. + @spec ensure_partition(String.t(), Date.t(), Date.t()) :: result() def ensure_partition(parent, %Date{} = q_start, %Date{} = q_end) do name = partition_name(parent, q_start) diff --git a/lib/microwaveprop/pskr/calibration_sampler.ex b/lib/microwaveprop/pskr/calibration_sampler.ex index 9e6deb42..a2a20cfc 100644 --- a/lib/microwaveprop/pskr/calibration_sampler.ex +++ b/lib/microwaveprop/pskr/calibration_sampler.ex @@ -150,6 +150,7 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do # a midpoint enqueue once, and let `HrrrPointEnqueuer.enqueue/1` # union the points into the existing `hrrr_fetch_tasks` row for # this hour rather than racing to insert duplicates. + @spec enqueue_missing_hrrr(map(), DateTime.t(), [map()]) :: :ok defp enqueue_missing_hrrr(cells, hour_utc, hrrr_index) do points = cells diff --git a/lib/microwaveprop/workers/partition_maintenance_worker.ex b/lib/microwaveprop/workers/partition_maintenance_worker.ex index b64cdc91..130aea6f 100644 --- a/lib/microwaveprop/workers/partition_maintenance_worker.ex +++ b/lib/microwaveprop/workers/partition_maintenance_worker.ex @@ -28,6 +28,7 @@ defmodule Microwaveprop.Workers.PartitionMaintenanceWorker do @lookahead_quarters 4 @impl Oban.Worker + @spec perform(Oban.Job.t()) :: :ok def perform(%Oban.Job{args: args}) do lookahead = lookahead(args) results = PartitionManager.ensure_quarterly_partitions(lookahead) @@ -46,6 +47,7 @@ defmodule Microwaveprop.Workers.PartitionMaintenanceWorker do :ok end + @spec lookahead(map()) :: non_neg_integer() defp lookahead(%{"lookahead_quarters" => n}) when is_integer(n) and n >= 0, do: n defp lookahead(_), do: @lookahead_quarters end diff --git a/test/microwaveprop/pskr/calibration_sampler_test.exs b/test/microwaveprop/pskr/calibration_sampler_test.exs index dd04ad1c..8872d818 100644 --- a/test/microwaveprop/pskr/calibration_sampler_test.exs +++ b/test/microwaveprop/pskr/calibration_sampler_test.exs @@ -205,6 +205,48 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do assert Enum.sort(sample.modes) == ["FT4", "FT8"] end + test "two-pass pipeline: first run NULL+enqueue, second run UPSERTs with HRRR" do + # End-to-end behaviour the producer was added for: a cell with no + # nearby HRRR profile gets a sample row written with NULL weather + # fields *and* a fetch task enqueued; once the Rust hrrr-point-worker + # delivers the profile (simulated here by directly inserting an + # hrrr_profiles row), the next sampler pass UPSERTs the same + # calibration_samples row, this time with non-NULL HRRR fields. + insert_spot!(%{midpoint_lat: 33.0, midpoint_lon: -97.0}) + + # ── Pass 1 ───────────────────────────────────────────────── + assert CalibrationSampler.build_for_hour(@hour) == 1 + [first_sample] = Repo.all(CalibrationSample) + assert first_sample.surface_temp_c == nil + assert first_sample.hpbl_m == nil + assert Repo.aggregate(from(t in "hrrr_fetch_tasks"), :count) == 1 + + # ── Rust worker drains the task (simulated) ──────────────── + insert_hrrr!(%{ + lat: 33.0, + lon: -97.0, + surface_temp_c: 24.0, + surface_dewpoint_c: 18.0, + pwat_mm: 31.0, + min_refractivity_gradient: -120.0, + hpbl_m: 480.0, + ducting_detected: true + }) + + # ── Pass 2 ───────────────────────────────────────────────── + assert CalibrationSampler.build_for_hour(@hour) == 1 + [second_sample] = Repo.all(CalibrationSample) + + # Same row, UPSERTed (id preserved, weather populated). + assert second_sample.id == first_sample.id + assert second_sample.surface_temp_c == 24.0 + assert second_sample.surface_dewpoint_c == 18.0 + assert second_sample.pwat_mm == 31.0 + assert second_sample.min_refractivity_gradient == -120.0 + assert second_sample.hpbl_m == 480.0 + assert second_sample.hrrr_ducting_detected == true + end + test "is idempotent — re-running the same hour upserts the row" do insert_spot!(%{}) insert_hrrr!(%{lat: 33.0, lon: -97.0})