From d880201713ede8532dc1f24fe3b9851df4020753 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 12 Apr 2026 12:11:25 -0500 Subject: [PATCH] Cache propagation scores in ETS, broadcast across cluster - Add ScoreCache GenServer with node-local ETS table keyed by {band, valid_time}, subscribed to "propagation:cache" PubSub topic so every pod stays in sync with a single hourly compute - scores_at/3 checks cache first, falls back to DB and populates on miss - PropagationGridWorker warms and broadcasts the cache for each band after every forecast hour upsert; prunes >2h old entries - Replace per-pixel string-keyed Map with flat Int8Array over the CONUS grid in propagation_map_hook.ts to eliminate allocations in the tile rasterization hot loop (interpolateScore / propagationReach) --- assets/js/propagation_map_hook.ts | 179 ++++++++++++------ lib/microwaveprop/application.ex | 1 + lib/microwaveprop/propagation.ex | 61 ++++-- lib/microwaveprop/propagation/score_cache.ex | 115 +++++++++++ .../workers/propagation_grid_worker.ex | 11 ++ .../propagation/score_cache_test.exs | 105 ++++++++++ test/microwaveprop/propagation_test.exs | 79 +++++++- 7 files changed, 474 insertions(+), 77 deletions(-) create mode 100644 lib/microwaveprop/propagation/score_cache.ex create mode 100644 test/microwaveprop/propagation/score_cache_test.exs diff --git a/assets/js/propagation_map_hook.ts b/assets/js/propagation_map_hook.ts index 5093cae0..0199c123 100644 --- a/assets/js/propagation_map_hook.ts +++ b/assets/js/propagation_map_hook.ts @@ -127,8 +127,7 @@ interface PropagationMapHook extends ViewHook { scoreOverlay: L.GridLayer | null scores: ScorePoint[] colorScale: ColorScaleEntry[] - gridLookup: Map - gridData: Map + scoreGrid: ScoreGrid bandInfo: BandInfo rangeCircles: L.LayerGroup gridLayer: L.LayerGroup @@ -155,6 +154,62 @@ interface PropagationMapHook extends ViewHook { // --- Constants --- +// CONUS propagation grid (must match Microwaveprop.Propagation.Grid in Elixir). +const GRID_LAT_MIN = 25.0 +const GRID_LON_MIN = -125.0 +const GRID_STEP = 0.125 +const GRID_LAT_COUNT = 201 // (50 - 25) / 0.125 + 1 +const GRID_LON_COUNT = 473 // (-66 - -125) / 0.125 + 1 +const GRID_MISSING = -1 // Int8Array sentinel for empty cell + +/** + * Flat typed-array grid storing propagation scores at 0.125° resolution across + * CONUS. Replaces a string-keyed Map — zero allocations on per-pixel reads, + * which matters because `interpolateScore` is called thousands of times per + * rendered canvas tile. + */ +class ScoreGrid { + private readonly data: Int8Array + + constructor() { + this.data = new Int8Array(GRID_LAT_COUNT * GRID_LON_COUNT) + this.data.fill(GRID_MISSING) + } + + reset(): void { + this.data.fill(GRID_MISSING) + } + + put(lat: number, lon: number, score: number): void { + const latIdx = Math.round((lat - GRID_LAT_MIN) / GRID_STEP) + const lonIdx = Math.round((lon - GRID_LON_MIN) / GRID_STEP) + if (latIdx < 0 || latIdx >= GRID_LAT_COUNT) return + if (lonIdx < 0 || lonIdx >= GRID_LON_COUNT) return + this.data[latIdx * GRID_LON_COUNT + lonIdx] = score + } + + get(lat: number, lon: number): number | null { + const latIdx = Math.round((lat - GRID_LAT_MIN) / GRID_STEP) + const lonIdx = Math.round((lon - GRID_LON_MIN) / GRID_STEP) + return this.getByIdx(latIdx, lonIdx) + } + + getByIdx(latIdx: number, lonIdx: number): number | null { + if (latIdx < 0 || latIdx >= GRID_LAT_COUNT) return null + if (lonIdx < 0 || lonIdx >= GRID_LON_COUNT) return null + const v = this.data[latIdx * GRID_LON_COUNT + lonIdx] + return v === GRID_MISSING ? null : v + } + + indexOf(lat: number, lon: number): number { + const latIdx = Math.round((lat - GRID_LAT_MIN) / GRID_STEP) + const lonIdx = Math.round((lon - GRID_LON_MIN) / GRID_STEP) + if (latIdx < 0 || latIdx >= GRID_LAT_COUNT) return -1 + if (lonIdx < 0 || lonIdx >= GRID_LON_COUNT) return -1 + return latIdx * GRID_LON_COUNT + lonIdx + } +} + // Weights must match BandConfig.weights() in band_config.ex // Recalibrated 2026-04-11 via gradient descent (loss 0.42 → 0.12) const FACTOR_META: Record = { @@ -280,30 +335,27 @@ function rangeEstimate(score: number, detail: PointDetail): string { * with score >= minScore. Returns array of {lat, lon} boundary points * as a convex hull for drawing a polygon. */ -function propagationReach(gridLookup: Map, startLat: number, startLon: number, minScore: number): LatLon[] { - const step = 0.125 +function propagationReach(grid: ScoreGrid, startLat: number, startLon: number, minScore: number): LatLon[] { const maxKm = 300 - const snap = (v: number): string => (Math.round(v / step) * step).toFixed(3) + const snap = (v: number): number => Math.round(v / GRID_STEP) * GRID_STEP - const startKey = `${snap(startLat)},${snap(startLon)}` - const startScore = gridLookup.get(startKey) + const snapLat = snap(startLat) + const snapLon = snap(startLon) + const startScore = grid.get(snapLat, snapLon) if (startScore == null || startScore < minScore) return [] const cosLat = Math.cos(startLat * Math.PI / 180) const kmPerDegLat = 111.0 const kmPerDegLon = 111.0 * cosLat - const visited = new Set() + const visited = new Set() const reachable: LatLon[] = [] - const queue: string[] = [startKey] - visited.add(startKey) + const queue: [number, number][] = [[snapLat, snapLon]] + visited.add(grid.indexOf(snapLat, snapLon)) // BFS flood fill through grid, capped at maxKm from origin while (queue.length > 0) { - const key = queue.shift()! - const [latStr, lonStr] = key.split(",") - const lat = parseFloat(latStr) - const lon = parseFloat(lonStr) + const [lat, lon] = queue.shift()! const dLat = (lat - startLat) * kmPerDegLat const dLon = (lon - startLon) * kmPerDegLon @@ -311,27 +363,25 @@ function propagationReach(gridLookup: Map, startLat: number, sta reachable.push({ lat, lon }) - // Check 4 neighbors const neighbors: [number, number][] = [ - [lat + step, lon], - [lat - step, lon], - [lat, lon + step], - [lat, lon - step] + [lat + GRID_STEP, lon], + [lat - GRID_STEP, lon], + [lat, lon + GRID_STEP], + [lat, lon - GRID_STEP] ] for (const [nlat, nlon] of neighbors) { - const nkey = `${nlat.toFixed(3)},${nlon.toFixed(3)}` - if (visited.has(nkey)) continue - visited.add(nkey) - const score = gridLookup.get(nkey) + const nidx = grid.indexOf(nlat, nlon) + if (nidx < 0 || visited.has(nidx)) continue + visited.add(nidx) + const score = grid.get(nlat, nlon) if (score != null && score >= minScore) { - queue.push(nkey) + queue.push([nlat, nlon]) } } } if (reachable.length < 3) return reachable - // Compute convex hull for polygon boundary return convexHull(reachable) } @@ -753,11 +803,11 @@ export const PropagationMap: Record & { } // Draw propagation reach polygon based on contiguous good cells - if (this.gridLookup && this.clickedLatLng) { + if (this.scoreGrid && this.clickedLatLng) { // Use MARGINAL threshold (50) as minimum for propagation reach const minScore = 50 const hull = propagationReach( - this.gridLookup, this.clickedLatLng[0], this.clickedLatLng[1], minScore + this.scoreGrid, this.clickedLatLng[0], this.clickedLatLng[1], minScore ) if (hull.length >= 3) { // Remove any previous reach polygon @@ -913,13 +963,12 @@ export const PropagationMap: Record & { } if (scores.length === 0) return - this.gridLookup = new Map() - this.gridData = new Map() - scores.forEach((s) => { - const key = `${s.lat.toFixed(3)},${s.lon.toFixed(3)}` - this.gridLookup.set(key, s.score) - this.gridData.set(key, s) - }) + if (!this.scoreGrid) { + this.scoreGrid = new ScoreGrid() + } else { + this.scoreGrid.reset() + } + scores.forEach((s) => this.scoreGrid.put(s.lat, s.lon, s.score)) // Pre-calculate RGBA strings for scores 0-100 const colorCache = new Array(101) @@ -971,33 +1020,40 @@ export const PropagationMap: Record & { this.scoreOverlay.addTo(this.map) }, - interpolateScore(this: PropagationMapHook, lat: number, lon: number, step: number): number | null { - const rlat = (Math.round(lat / step) * step).toFixed(3) - const rlon = (Math.round(lon / step) * step).toFixed(3) - if (this.gridLookup.has(`${rlat},${rlon}`)) return this.gridLookup.get(`${rlat},${rlon}`)! + interpolateScore(this: PropagationMapHook, lat: number, lon: number, _step: number): number | null { + // Snapped cell — hot path, zero allocations + const latIdxR = Math.round((lat - GRID_LAT_MIN) / GRID_STEP) + const lonIdxR = Math.round((lon - GRID_LON_MIN) / GRID_STEP) + const snapped = this.scoreGrid.getByIdx(latIdxR, lonIdxR) + if (snapped !== null) return snapped - const lat0 = Math.floor(lat / step) * step - const lat1 = lat0 + step - const lon0 = Math.floor(lon / step) * step - const lon1 = lon0 + step + // Bilinear interpolation across 4 neighbors + const latFloor = Math.floor((lat - GRID_LAT_MIN) / GRID_STEP) + const lonFloor = Math.floor((lon - GRID_LON_MIN) / GRID_STEP) + const tLat = ((lat - GRID_LAT_MIN) / GRID_STEP) - latFloor + const tLon = ((lon - GRID_LON_MIN) / GRID_STEP) - lonFloor - const s00 = this.gridLookup.get(`${lat0.toFixed(3)},${lon0.toFixed(3)}`) - const s10 = this.gridLookup.get(`${lat1.toFixed(3)},${lon0.toFixed(3)}`) - const s01 = this.gridLookup.get(`${lat0.toFixed(3)},${lon1.toFixed(3)}`) - const s11 = this.gridLookup.get(`${lat1.toFixed(3)},${lon1.toFixed(3)}`) + const s00 = this.scoreGrid.getByIdx(latFloor, lonFloor) + const s10 = this.scoreGrid.getByIdx(latFloor + 1, lonFloor) + const s01 = this.scoreGrid.getByIdx(latFloor, lonFloor + 1) + const s11 = this.scoreGrid.getByIdx(latFloor + 1, lonFloor + 1) - const corners = [s00, s10, s01, s11].filter((s): s is number => s !== undefined) - if (corners.length < 2) return null - - if (corners.length === 4) { - const tLat = (lat - lat0) / step - const tLon = (lon - lon0) / step + if (s00 !== null && s10 !== null && s01 !== null && s11 !== null) { return Math.round( - s00! * (1 - tLat) * (1 - tLon) + s10! * tLat * (1 - tLon) + - s01! * (1 - tLat) * tLon + s11! * tLat * tLon + s00 * (1 - tLat) * (1 - tLon) + s10 * tLat * (1 - tLon) + + s01 * (1 - tLat) * tLon + s11 * tLat * tLon ) } - return Math.round(corners.reduce((a, b) => a + b, 0) / corners.length) + + // Partial corners — average whatever we have (minimum 2 for a reasonable value) + let sum = 0 + let count = 0 + if (s00 !== null) { sum += s00; count++ } + if (s10 !== null) { sum += s10; count++ } + if (s01 !== null) { sum += s01; count++ } + if (s11 !== null) { sum += s11; count++ } + if (count < 2) return null + return Math.round(sum / count) }, scoreColorRGB(this: PropagationMapHook, score: number): RGB { @@ -1019,13 +1075,12 @@ export const PropagationMap: Record & { }, lookupPoint(this: PropagationMapHook, lat: number, lon: number): (ScorePoint & BandInfo) | null { - if (!this.gridData) return null - const step = 0.125 - const rlat = (Math.round(lat / step) * step).toFixed(3) - const rlon = (Math.round(lon / step) * step).toFixed(3) - const data = this.gridData.get(`${rlat},${rlon}`) - if (!data) return null - return { ...data, ...this.bandInfo } + if (!this.scoreGrid) return null + const rlat = Math.round(lat / GRID_STEP) * GRID_STEP + const rlon = Math.round(lon / GRID_STEP) * GRID_STEP + const score = this.scoreGrid.get(rlat, rlon) + if (score == null) return null + return { lat: rlat, lon: rlon, score, ...this.bandInfo } }, renderTimeline(this: PropagationMapHook) { diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index c30aa7f6..a4c3caea 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -14,6 +14,7 @@ defmodule Microwaveprop.Application do Microwaveprop.Repo, {Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]}, {Phoenix.PubSub, name: Microwaveprop.PubSub}, + Microwaveprop.Propagation.ScoreCache, {Oban, Application.fetch_env!(:microwaveprop, Oban)}, Microwaveprop.RepoListener, # Start to serve requests, typically the last entry diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 39311346..6d3129bf 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -6,6 +6,7 @@ defmodule Microwaveprop.Propagation do alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.Grid alias Microwaveprop.Propagation.GridScore + alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.Scorer alias Microwaveprop.Repo alias Microwaveprop.Weather.SoundingParams @@ -237,15 +238,55 @@ defmodule Microwaveprop.Propagation do [] _ -> - from(gs in GridScore, - where: gs.band_mhz == ^band_mhz and gs.valid_time == ^time, - select: %{lat: gs.lat, lon: gs.lon, score: gs.score, valid_time: gs.valid_time} - ) - |> maybe_filter_bounds(bounds) - |> Repo.all() + scores_at_fetch(band_mhz, time, bounds) end end + defp scores_at_fetch(band_mhz, time, bounds) do + case ScoreCache.fetch_bounds(band_mhz, time, bounds) do + {:ok, scores} -> + Enum.map(scores, &Map.put(&1, :valid_time, time)) + + :miss -> + full = load_scores_from_db(band_mhz, time) + ScoreCache.put(band_mhz, time, full) + + full + |> filter_bounds(bounds) + |> Enum.map(&Map.put(&1, :valid_time, time)) + end + end + + defp load_scores_from_db(band_mhz, time) do + Repo.all( + from(gs in GridScore, + where: gs.band_mhz == ^band_mhz and gs.valid_time == ^time, + select: %{lat: gs.lat, lon: gs.lon, score: gs.score} + ) + ) + end + + @doc """ + Load the full CONUS score set for `{band_mhz, valid_time}` from the DB and + broadcast it to every `ScoreCache` in the cluster. Intended to be called from + `PropagationGridWorker` after each upsert so all pods have a warm cache by + the time clients begin requesting the new hour. + """ + @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) :: :ok + def warm_cache_and_broadcast(band_mhz, valid_time) do + scores = load_scores_from_db(band_mhz, valid_time) + ScoreCache.broadcast_put(band_mhz, valid_time, scores) + :ok + end + + defp filter_bounds(scores, nil), do: scores + + defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do + Enum.filter(scores, fn %{lat: lat, lon: lon} -> + lat >= s and lat <= n and lon >= w and lon <= e + end) + end + @doc "Get the latest scores for a band (alias for scores_at with earliest valid_time)." @spec latest_scores(non_neg_integer(), map() | nil) :: [%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}] @@ -320,14 +361,6 @@ defmodule Microwaveprop.Propagation do end end - defp maybe_filter_bounds(query, nil), do: query - - defp maybe_filter_bounds(query, %{"south" => s, "north" => n, "west" => w, "east" => e}) do - from(gs in query, - where: gs.lat >= ^s and gs.lat <= ^n and gs.lon >= ^w and gs.lon <= ^e - ) - end - @doc "Get the latest valid_time across all scores." @spec latest_valid_time() :: DateTime.t() | nil def latest_valid_time do diff --git a/lib/microwaveprop/propagation/score_cache.ex b/lib/microwaveprop/propagation/score_cache.ex new file mode 100644 index 00000000..51d821ca --- /dev/null +++ b/lib/microwaveprop/propagation/score_cache.ex @@ -0,0 +1,115 @@ +defmodule Microwaveprop.Propagation.ScoreCache do + @moduledoc """ + Node-local ETS cache of propagation scores keyed by `{band_mhz, valid_time}`. + + Populated eagerly by `Microwaveprop.Workers.PropagationGridWorker` after each + hourly compute (via `broadcast_put/3`) and lazily by `Microwaveprop.Propagation.scores_at/3` + on a cache miss. `broadcast_put/3` fans the payload out across the cluster on + the `"propagation:cache"` PubSub topic so every BEAM node stays in sync with + a single compute. + """ + use GenServer + + alias Phoenix.PubSub + + @table :propagation_score_cache + @topic "propagation:cache" + @pubsub Microwaveprop.PubSub + + @type score :: %{lat: float(), lon: float(), score: non_neg_integer()} + @type bounds :: %{optional(String.t()) => float()} + + # ---------- Public API ---------- + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @spec fetch(non_neg_integer(), DateTime.t()) :: {:ok, [score()]} | :miss + def fetch(band_mhz, valid_time) do + case :ets.lookup(@table, {band_mhz, valid_time}) do + [{_, scores}] -> {:ok, scores} + [] -> :miss + end + end + + @spec fetch_bounds(non_neg_integer(), DateTime.t(), bounds() | nil) :: + {:ok, [score()]} | :miss + def fetch_bounds(band_mhz, valid_time, bounds) do + case fetch(band_mhz, valid_time) do + {:ok, scores} -> {:ok, filter_bounds(scores, bounds)} + :miss -> :miss + end + end + + @spec put(non_neg_integer(), DateTime.t(), [score()]) :: :ok + def put(band_mhz, valid_time, scores) do + :ets.insert(@table, {{band_mhz, valid_time}, scores}) + :ok + end + + @doc """ + Insert locally AND broadcast to peer nodes via PubSub. + + Call this from the single pod that computed the scores; every cluster member + (including the caller) will receive the PubSub message and populate its local + ETS. Use `put/3` directly if you already have a cluster-wide fan-out. + """ + @spec broadcast_put(non_neg_integer(), DateTime.t(), [score()]) :: :ok + def broadcast_put(band_mhz, valid_time, scores) do + PubSub.broadcast(@pubsub, @topic, {:cache_refresh, band_mhz, valid_time, scores}) + :ok + end + + @spec prune_older_than(DateTime.t()) :: non_neg_integer() + def prune_older_than(cutoff) do + match_spec = [ + {{{:_, :"$1"}, :_}, [{:<, :"$1", {:const, cutoff}}], [true]} + ] + + :ets.select_delete(@table, match_spec) + end + + @spec clear() :: :ok + def clear do + :ets.delete_all_objects(@table) + :ok + end + + @doc "Flush the GenServer mailbox so any in-flight cache_refresh messages are applied." + @spec sync() :: :ok + def sync do + GenServer.call(__MODULE__, :sync) + end + + # ---------- GenServer callbacks ---------- + + @impl true + def init(_opts) do + :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + PubSub.subscribe(@pubsub, @topic) + {:ok, %{}} + end + + @impl true + def handle_call(:sync, _from, state), do: {:reply, :ok, state} + + @impl true + def handle_info({:cache_refresh, band_mhz, valid_time, scores}, state) do + put(band_mhz, valid_time, scores) + {:noreply, state} + end + + def handle_info(_msg, state), do: {:noreply, state} + + # ---------- Internal ---------- + + defp filter_bounds(scores, nil), do: scores + + defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do + Enum.filter(scores, fn %{lat: lat, lon: lon} -> + lat >= s and lat <= n and lon >= w and lon <= e + end) + end +end diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index a7c9d283..b105fd6e 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -11,7 +11,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do unique: [period: 3600, states: [:available, :scheduled, :executing, :retryable]] alias Microwaveprop.Propagation + alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Weather alias Microwaveprop.Weather.HrrrClient alias Microwaveprop.Weather.HrrrNativeClient @@ -94,6 +96,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do case Propagation.upsert_scores(scores, prune: false) do {:ok, count} -> Logger.info("PropagationGrid: #{label} → #{count} scores for #{valid_time}") + warm_cache(valid_time) :ok error -> @@ -107,6 +110,14 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do end end + defp warm_cache(valid_time) do + Enum.each(BandConfig.all_bands(), fn band -> + Propagation.warm_cache_and_broadcast(band.freq_mhz, valid_time) + end) + + ScoreCache.prune_older_than(DateTime.add(DateTime.utc_now(), -2, :hour)) + end + defp timed(label, fun) do t0 = System.monotonic_time(:millisecond) result = fun.() diff --git a/test/microwaveprop/propagation/score_cache_test.exs b/test/microwaveprop/propagation/score_cache_test.exs new file mode 100644 index 00000000..d593f905 --- /dev/null +++ b/test/microwaveprop/propagation/score_cache_test.exs @@ -0,0 +1,105 @@ +defmodule Microwaveprop.Propagation.ScoreCacheTest do + use ExUnit.Case, async: false + + alias Microwaveprop.Propagation.ScoreCache + + setup do + ScoreCache.clear() + :ok + end + + describe "fetch/2" do + test "returns :miss when nothing is cached" do + assert ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z]) == :miss + end + + test "returns cached scores after put" do + scores = [%{lat: 32.0, lon: -97.0, score: 75}] + ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], scores) + assert {:ok, ^scores} = ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z]) + end + + test "isolates entries by band" do + ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], [%{lat: 1.0, lon: 1.0, score: 10}]) + ScoreCache.put(24_000, ~U[2026-04-12 12:00:00Z], [%{lat: 2.0, lon: 2.0, score: 20}]) + + assert {:ok, [%{score: 10}]} = ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z]) + assert {:ok, [%{score: 20}]} = ScoreCache.fetch(24_000, ~U[2026-04-12 12:00:00Z]) + end + + test "isolates entries by valid_time" do + ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], [%{lat: 1.0, lon: 1.0, score: 10}]) + ScoreCache.put(10_000, ~U[2026-04-12 13:00:00Z], [%{lat: 2.0, lon: 2.0, score: 20}]) + + assert {:ok, [%{score: 10}]} = ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z]) + assert {:ok, [%{score: 20}]} = ScoreCache.fetch(10_000, ~U[2026-04-12 13:00:00Z]) + end + end + + describe "fetch_bounds/3" do + setup do + scores = [ + %{lat: 32.0, lon: -97.0, score: 75}, + %{lat: 40.0, lon: -74.0, score: 50}, + %{lat: 34.0, lon: -98.0, score: 60} + ] + + ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], scores) + :ok + end + + test "returns :miss when no entry for band/time" do + bounds = %{"south" => 30.0, "north" => 50.0, "west" => -100.0, "east" => -70.0} + assert ScoreCache.fetch_bounds(24_000, ~U[2026-04-12 12:00:00Z], bounds) == :miss + end + + test "returns all scores when bounds is nil" do + assert {:ok, all} = ScoreCache.fetch_bounds(10_000, ~U[2026-04-12 12:00:00Z], nil) + assert length(all) == 3 + end + + test "returns only scores inside the bounds" do + bounds = %{"south" => 31.0, "north" => 35.0, "west" => -100.0, "east" => -95.0} + + assert {:ok, filtered} = + ScoreCache.fetch_bounds(10_000, ~U[2026-04-12 12:00:00Z], bounds) + + assert length(filtered) == 2 + assert Enum.all?(filtered, fn %{lat: lat} -> lat >= 31.0 and lat <= 35.0 end) + end + + test "includes scores exactly on the boundary" do + bounds = %{"south" => 32.0, "north" => 32.0, "west" => -97.0, "east" => -97.0} + + assert {:ok, [%{lat: 32.0, lon: -97.0, score: 75}]} = + ScoreCache.fetch_bounds(10_000, ~U[2026-04-12 12:00:00Z], bounds) + end + end + + describe "prune_older_than/1" do + test "removes entries whose valid_time is strictly before the cutoff" do + ScoreCache.put(10_000, ~U[2026-04-12 10:00:00Z], [%{lat: 1.0, lon: 1.0, score: 1}]) + ScoreCache.put(10_000, ~U[2026-04-12 14:00:00Z], [%{lat: 2.0, lon: 2.0, score: 2}]) + + ScoreCache.prune_older_than(~U[2026-04-12 12:00:00Z]) + + assert ScoreCache.fetch(10_000, ~U[2026-04-12 10:00:00Z]) == :miss + assert {:ok, [%{score: 2}]} = ScoreCache.fetch(10_000, ~U[2026-04-12 14:00:00Z]) + end + + test "keeps entries whose valid_time equals the cutoff" do + ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], [%{lat: 1.0, lon: 1.0, score: 1}]) + ScoreCache.prune_older_than(~U[2026-04-12 12:00:00Z]) + assert {:ok, [%{score: 1}]} = ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z]) + end + end + + describe "broadcast_put/3" do + test "populates the cache on the local node" do + scores = [%{lat: 32.0, lon: -97.0, score: 80}] + ScoreCache.broadcast_put(10_000, ~U[2026-04-12 12:00:00Z], scores) + ScoreCache.sync() + assert {:ok, ^scores} = ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z]) + end + end +end diff --git a/test/microwaveprop/propagation_test.exs b/test/microwaveprop/propagation_test.exs index 4c5903ae..752da3a8 100644 --- a/test/microwaveprop/propagation_test.exs +++ b/test/microwaveprop/propagation_test.exs @@ -1,8 +1,14 @@ defmodule Microwaveprop.PropagationTest do - use Microwaveprop.DataCase + use Microwaveprop.DataCase, async: false alias Microwaveprop.Propagation alias Microwaveprop.Propagation.GridScore + alias Microwaveprop.Propagation.ScoreCache + + setup do + ScoreCache.clear() + :ok + end describe "score_grid_point/4" do test "scores a single point for all bands" do @@ -81,6 +87,77 @@ defmodule Microwaveprop.PropagationTest do end end + describe "scores_at/3 with cache" do + test "populates the cache on a DB miss" do + valid_time = ~U[2026-07-15 13:00:00Z] + + Propagation.upsert_scores([ + %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}} + ]) + + assert ScoreCache.fetch(10_000, valid_time) == :miss + _ = Propagation.scores_at(10_000, valid_time) + assert {:ok, [%{lat: 35.0, lon: -97.0, score: 75}]} = ScoreCache.fetch(10_000, valid_time) + end + + test "returns cached bounds-filtered results without hitting the DB" do + valid_time = ~U[2026-07-15 13:00:00Z] + + # Seed the cache directly — DB is empty, so if the result matches the + # cached payload we know the DB was not consulted. + ScoreCache.put(10_000, valid_time, [ + %{lat: 32.0, lon: -97.0, score: 75}, + %{lat: 40.0, lon: -74.0, score: 50} + ]) + + assert Repo.aggregate(GridScore, :count) == 0 + + bounds = %{"south" => 30.0, "north" => 35.0, "west" => -100.0, "east" => -95.0} + result = Propagation.scores_at(10_000, valid_time, bounds) + + assert length(result) == 1 + assert [%{lat: 32.0, lon: -97.0, score: 75}] = result + end + + test "returns empty list when no data in cache or DB" do + assert Propagation.scores_at(10_000, ~U[2026-07-15 13:00:00Z]) == [] + end + end + + describe "warm_cache_and_broadcast/2" do + test "loads scores from DB into the cache" do + valid_time = ~U[2026-07-15 13:00:00Z] + + Propagation.upsert_scores([ + %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}}, + %{lat: 36.0, lon: -96.0, valid_time: valid_time, band_mhz: 10_000, score: 80, factors: %{}} + ]) + + assert ScoreCache.fetch(10_000, valid_time) == :miss + + Propagation.warm_cache_and_broadcast(10_000, valid_time) + ScoreCache.sync() + + assert {:ok, scores} = ScoreCache.fetch(10_000, valid_time) + assert length(scores) == 2 + end + + test "only warms the requested band" do + valid_time = ~U[2026-07-15 13:00:00Z] + + Propagation.upsert_scores([ + %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}}, + %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 30, factors: %{}} + ]) + + Propagation.warm_cache_and_broadcast(10_000, valid_time) + ScoreCache.sync() + + assert {:ok, [_]} = ScoreCache.fetch(10_000, valid_time) + assert ScoreCache.fetch(24_000, valid_time) == :miss + end + end + describe "latest_valid_time/0" do test "returns nil when empty" do assert Propagation.latest_valid_time() == nil