prop/lib/microwaveprop/weather.ex

defmodule Microwaveprop.Weather do
  @moduledoc false

  import Ecto.Query

  alias Microwaveprop.Propagation.ProfilesFile
  alias Microwaveprop.Radio
  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.GefsProfile
  alias Microwaveprop.Weather.GridCache
  alias Microwaveprop.Weather.HrrrClient
  alias Microwaveprop.Weather.HrrrNativeProfile
  alias Microwaveprop.Weather.HrrrProfile
  alias Microwaveprop.Weather.IemClient
  alias Microwaveprop.Weather.IemreObservation
  alias Microwaveprop.Weather.Metar5minObservation
  alias Microwaveprop.Weather.NarrProfile
  alias Microwaveprop.Weather.RtmaObservation
  alias Microwaveprop.Weather.SolarIndex
  alias Microwaveprop.Weather.Sounding
  alias Microwaveprop.Weather.SoundingParams
  alias Microwaveprop.Weather.Station
  alias Microwaveprop.Weather.SurfaceObservation
  alias Microwaveprop.Weather.WeatherLayers

  require Logger

  # Approximate km per degree latitude
  @km_per_deg_lat 111.0

  @spec find_or_create_station(map()) :: {:ok, Station.t()} | {:error, Ecto.Changeset.t()}
  def find_or_create_station(attrs) do
    code = attrs[:station_code] || attrs["station_code"]
    type = attrs[:station_type] || attrs["station_type"]

    if code && type do
      case Repo.get_by(Station, station_code: code, station_type: type) do
        nil ->
          %Station{}
          |> Station.changeset(attrs)
          |> Repo.insert()

        station ->
          {:ok, station}
      end
    else
      %Station{}
      |> Station.changeset(attrs)
      |> Repo.insert()
    end
  end

  @spec upsert_surface_observation(Station.t(), map()) :: {:ok, SurfaceObservation.t()} | {:error, Ecto.Changeset.t()}
  def upsert_surface_observation(%Station{} = station, attrs) do
    attrs = Map.put(attrs, :station_id, station.id)

    %SurfaceObservation{}
    |> SurfaceObservation.changeset(attrs)
    |> Repo.insert(
      on_conflict:
        from(s in SurfaceObservation,
          update: [
            set: [
              temp_f: fragment("EXCLUDED.temp_f"),
              dewpoint_f: fragment("EXCLUDED.dewpoint_f"),
              relative_humidity: fragment("EXCLUDED.relative_humidity"),
              wind_speed_kts: fragment("EXCLUDED.wind_speed_kts"),
              sea_level_pressure_mb: fragment("EXCLUDED.sea_level_pressure_mb"),
              sky_condition: fragment("EXCLUDED.sky_condition"),
              precip_1h_in: fragment("EXCLUDED.precip_1h_in"),
              wx_codes: fragment("EXCLUDED.wx_codes"),
              updated_at: fragment("EXCLUDED.updated_at")
            ]
          ],
          where:
            s.temp_f != fragment("EXCLUDED.temp_f") or
              s.dewpoint_f != fragment("EXCLUDED.dewpoint_f") or
              s.relative_humidity != fragment("EXCLUDED.relative_humidity") or
              s.wind_speed_kts != fragment("EXCLUDED.wind_speed_kts") or
              s.sea_level_pressure_mb != fragment("EXCLUDED.sea_level_pressure_mb")
        ),
      conflict_target: [:station_id, :observed_at],
      returning: true,
      stale_error_field: :id
    )
  end

  @doc """
  Bulk-upsert surface observations for a single station via one
  `Repo.insert_all` round-trip. ASOS fetches return 24–288 rows per
  station per call — collapsing them into a single statement avoids the
  per-row UPDATE-conflict round-trip of `upsert_surface_observation/2`,
  which is expensive against the Turing Pi 2 Postgres node.

  Rows missing `observed_at` are dropped (they can't satisfy the
  `(station_id, observed_at)` unique index). Returns the
  `{count, nil}` tuple from `Repo.insert_all/3`; `count` reflects
  affected rows (new + updated-when-changed). Returns `{0, nil}` for
  an empty input without touching the DB.
  """
  @spec upsert_surface_observations(Station.t(), [map()]) :: {non_neg_integer(), nil}
  def upsert_surface_observations(%Station{} = station, rows) when is_list(rows) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    entries =
      rows
      |> Enum.filter(&row_has_observed_at?/1)
      |> Enum.map(&surface_observation_entry(&1, station.id, now))
      |> dedupe_last_by_conflict_target()

    case entries do
      [] ->
        {0, nil}

      _ ->
        Repo.insert_all(SurfaceObservation, entries,
          on_conflict:
            from(s in SurfaceObservation,
              update: [
                set: [
                  temp_f: fragment("EXCLUDED.temp_f"),
                  dewpoint_f: fragment("EXCLUDED.dewpoint_f"),
                  relative_humidity: fragment("EXCLUDED.relative_humidity"),
                  wind_speed_kts: fragment("EXCLUDED.wind_speed_kts"),
                  wind_direction_deg: fragment("EXCLUDED.wind_direction_deg"),
                  sea_level_pressure_mb: fragment("EXCLUDED.sea_level_pressure_mb"),
                  altimeter_setting: fragment("EXCLUDED.altimeter_setting"),
                  sky_condition: fragment("EXCLUDED.sky_condition"),
                  precip_1h_in: fragment("EXCLUDED.precip_1h_in"),
                  wx_codes: fragment("EXCLUDED.wx_codes"),
                  updated_at: fragment("EXCLUDED.updated_at")
                ]
              ],
              where:
                s.temp_f != fragment("EXCLUDED.temp_f") or
                  s.dewpoint_f != fragment("EXCLUDED.dewpoint_f") or
                  s.relative_humidity != fragment("EXCLUDED.relative_humidity") or
                  s.wind_speed_kts != fragment("EXCLUDED.wind_speed_kts") or
                  s.sea_level_pressure_mb != fragment("EXCLUDED.sea_level_pressure_mb")
            ),
          conflict_target: [:station_id, :observed_at]
        )
    end
  end

  defp row_has_observed_at?(%{observed_at: %DateTime{}}), do: true
  defp row_has_observed_at?(%{"observed_at" => %DateTime{}}), do: true
  defp row_has_observed_at?(_), do: false

  # IEM ASOS occasionally returns two rows with the same timestamp
  # (e.g. routine + special METAR at the same minute). Passing both
  # to insert_all triggers Postgres 21000 cardinality_violation.
  # Keep the last occurrence — IEM's ordering is chronological, so
  # the later row is the final correction.
  defp dedupe_last_by_conflict_target(entries) do
    entries
    |> Enum.reverse()
    |> Enum.uniq_by(&{&1.station_id, &1.observed_at})
    |> Enum.reverse()
  end

  defp surface_observation_entry(row, station_id, now) do
    %{
      id: Ecto.UUID.generate(),
      station_id: station_id,
      observed_at: fetch_row(row, :observed_at),
      temp_f: fetch_row(row, :temp_f),
      dewpoint_f: fetch_row(row, :dewpoint_f),
      relative_humidity: fetch_row(row, :relative_humidity),
      wind_speed_kts: fetch_row(row, :wind_speed_kts),
      wind_direction_deg: fetch_row(row, :wind_direction_deg),
      sea_level_pressure_mb: fetch_row(row, :sea_level_pressure_mb),
      altimeter_setting: fetch_row(row, :altimeter_setting),
      sky_condition: fetch_row(row, :sky_condition),
      precip_1h_in: fetch_row(row, :precip_1h_in),
      wx_codes: fetch_row(row, :wx_codes),
      inserted_at: now,
      updated_at: now
    }
  end

  defp fetch_row(row, key) when is_atom(key) do
    case Map.fetch(row, key) do
      {:ok, value} -> value
      :error -> Map.get(row, Atom.to_string(key))
    end
  end

  @spec upsert_sounding(Station.t(), map()) :: {:ok, Sounding.t()} | {:error, Ecto.Changeset.t()}
  def upsert_sounding(%Station{} = station, attrs) do
    attrs = Map.put(attrs, :station_id, station.id)

    %Sounding{}
    |> Sounding.changeset(attrs)
    |> Repo.insert(
      on_conflict:
        from(s in Sounding,
          update: [
            set: [
              profile: fragment("EXCLUDED.profile"),
              level_count: fragment("EXCLUDED.level_count"),
              surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"),
              surface_temp_c: fragment("EXCLUDED.surface_temp_c"),
              surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"),
              surface_refractivity: fragment("EXCLUDED.surface_refractivity"),
              min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"),
              boundary_layer_depth_m: fragment("EXCLUDED.boundary_layer_depth_m"),
              precipitable_water_mm: fragment("EXCLUDED.precipitable_water_mm"),
              k_index: fragment("EXCLUDED.k_index"),
              lifted_index: fragment("EXCLUDED.lifted_index"),
              ducting_detected: fragment("EXCLUDED.ducting_detected"),
              duct_characteristics: fragment("EXCLUDED.duct_characteristics"),
              updated_at: fragment("EXCLUDED.updated_at")
            ]
          ],
          where:
            s.level_count != fragment("EXCLUDED.level_count") or
              s.surface_temp_c != fragment("EXCLUDED.surface_temp_c") or
              s.surface_refractivity != fragment("EXCLUDED.surface_refractivity")
        ),
      conflict_target: [:station_id, :observed_at],
      returning: true,
      stale_error_field: :id
    )
  end

  @spec upsert_solar_index(map()) :: {:ok, SolarIndex.t()} | {:error, Ecto.Changeset.t()}
  def upsert_solar_index(attrs) do
    %SolarIndex{}
    |> SolarIndex.changeset(attrs)
    |> Repo.insert(
      on_conflict:
        from(s in SolarIndex,
          update: [
            set: [
              sfi: fragment("EXCLUDED.sfi"),
              sfi_adjusted: fragment("EXCLUDED.sfi_adjusted"),
              sunspot_number: fragment("EXCLUDED.sunspot_number"),
              ap_index: fragment("EXCLUDED.ap_index"),
              kp_values: fragment("EXCLUDED.kp_values"),
              updated_at: fragment("EXCLUDED.updated_at")
            ]
          ],
          where:
            s.sfi != fragment("EXCLUDED.sfi") or
              s.ap_index != fragment("EXCLUDED.ap_index")
        ),
      conflict_target: [:date],
      returning: true,
      stale_error_field: :id
    )
  end

  @spec has_surface_observations?(Ecto.UUID.t(), DateTime.t(), DateTime.t()) :: boolean()
  def has_surface_observations?(station_id, start_dt, end_dt) do
    SurfaceObservation
    |> where([o], o.station_id == ^station_id)
    |> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
    |> Repo.exists?()
  end

  @doc """
  Flip `contacts.weather_status` from `:queued` to `:complete` for
  every contact whose ±2h / 150km window now contains at least one
  surface observation. Returns `{:ok, n}` where `n` is the number of
  contacts advanced.

  Background: `WeatherFetchWorker` upserts observations but has no
  back-pointer to the contacts that triggered the fetch — the same
  obs row satisfies many contacts. Previously the only place that
  flipped `:queued → :complete` was `MicrowavepropWeb.ContactLive.Show`
  on page view, which left thousands of contacts stuck in `:queued`
  even after their data had landed. This reconciler closes that loop
  as a single SQL UPDATE, invoked from the hourly enqueuer cron.

  Radius encoded as a conservative ±1.5° lat/lon bounding box — the
  same rectangle `weather_for_contact/2` uses at `radius_km: 150`
  around the equator/mid-latitudes.
  """
  @spec reconcile_weather_statuses() :: {:ok, non_neg_integer()}
  def reconcile_weather_statuses do
    sql = """
    UPDATE contacts c
    SET weather_status = 'complete'
    WHERE c.weather_status = 'queued'
      AND c.pos1 IS NOT NULL
      AND (
        EXISTS (
          SELECT 1
          FROM surface_observations o
          JOIN weather_stations s ON s.id = o.station_id
          WHERE o.observed_at >= c.qso_timestamp - interval '2 hours'
            AND o.observed_at <= c.qso_timestamp + interval '2 hours'
            AND s.lat BETWEEN ((c.pos1->>'lat')::float - 1.5)
                          AND ((c.pos1->>'lat')::float + 1.5)
            AND s.lon BETWEEN ((c.pos1->>'lon')::float - 1.5)
                          AND ((c.pos1->>'lon')::float + 1.5)
        )
        OR (
          c.pos2 IS NOT NULL
          AND EXISTS (
            SELECT 1
            FROM surface_observations o
            JOIN weather_stations s ON s.id = o.station_id
            WHERE o.observed_at >= c.qso_timestamp - interval '2 hours'
              AND o.observed_at <= c.qso_timestamp + interval '2 hours'
              AND s.lat BETWEEN ((c.pos2->>'lat')::float - 1.5)
                            AND ((c.pos2->>'lat')::float + 1.5)
              AND s.lon BETWEEN ((c.pos2->>'lon')::float - 1.5)
                            AND ((c.pos2->>'lon')::float + 1.5)
          )
        )
      )
    """

    %{num_rows: n} = Repo.query!(sql)
    {:ok, n}
  end

  @doc """
  True if the station already has at least one surface observation
  anywhere within the given UTC date. Used by the `asos_day` worker
  to short-circuit the IEM fetch when the day has already been
  ingested by a prior run.
  """
  @spec station_day_covered?(Ecto.UUID.t(), Date.t()) :: boolean()
  def station_day_covered?(station_id, date) do
    {:ok, start_dt} = DateTime.new(date, ~T[00:00:00], "Etc/UTC")
    {:ok, end_dt} = DateTime.new(date, ~T[23:59:59], "Etc/UTC")
    has_surface_observations?(station_id, start_dt, end_dt)
  end

  @doc """
  Returns the MapSet of `{station_id, date}` tuples already covered
  (at least one obs in that UTC day). Used by the enqueuer to avoid
  emitting jobs for combinations we've already fetched.
  """
  @spec station_day_pairs_covered([{Ecto.UUID.t(), Date.t()}]) ::
          MapSet.t({Ecto.UUID.t(), Date.t()})
  def station_day_pairs_covered([]), do: MapSet.new()

  def station_day_pairs_covered(pairs) do
    station_ids = pairs |> Enum.map(&elem(&1, 0)) |> Enum.uniq()
    dates = pairs |> Enum.map(&elem(&1, 1)) |> Enum.uniq()
    {:ok, start_dt} = DateTime.new(Enum.min(dates, Date), ~T[00:00:00], "Etc/UTC")
    {:ok, end_dt} = DateTime.new(Enum.max(dates, Date), ~T[23:59:59], "Etc/UTC")

    rows =
      SurfaceObservation
      |> where([o], o.station_id in ^station_ids)
      |> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
      |> select([o], {o.station_id, fragment("(?::date)", o.observed_at)})
      |> distinct(true)
      |> Repo.all()

    MapSet.new(rows)
  end

  @doc "Returns a MapSet of station_ids that have surface observations in the time window."
  @spec station_ids_with_surface_observations([Ecto.UUID.t()], DateTime.t(), DateTime.t()) :: MapSet.t(Ecto.UUID.t())
  def station_ids_with_surface_observations(station_ids, start_dt, end_dt) do
    SurfaceObservation
    |> where([o], o.station_id in ^station_ids)
    |> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
    |> select([o], o.station_id)
    |> distinct(true)
    |> Repo.all()
    |> MapSet.new()
  end

  @spec has_sounding?(Ecto.UUID.t(), DateTime.t()) :: boolean()
  def has_sounding?(station_id, observed_at) do
    Sounding
    |> where([s], s.station_id == ^station_id and s.observed_at == ^observed_at)
    |> Repo.exists?()
  end

  @doc "Returns a MapSet of {station_id, observed_at} tuples that have soundings."
  @spec station_ids_with_soundings([Ecto.UUID.t()], [DateTime.t()]) :: MapSet.t({Ecto.UUID.t(), DateTime.t()})
  def station_ids_with_soundings(station_ids, sounding_times) do
    Sounding
    |> where([s], s.station_id in ^station_ids and s.observed_at in ^sounding_times)
    |> select([s], {s.station_id, s.observed_at})
    |> distinct(true)
    |> Repo.all()
    |> MapSet.new()
  end

  @doc "Batch upsert solar indices using insert_all in chunks of 500."
  @spec upsert_solar_indices_batch([map()]) :: non_neg_integer()
  def upsert_solar_indices_batch(records) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    records
    |> Enum.chunk_every(500)
    |> Enum.reduce(0, fn chunk, acc ->
      entries =
        Enum.map(chunk, fn attrs ->
          %{
            id: Ecto.UUID.generate(),
            date: attrs[:date] || attrs.date,
            sfi: attrs[:sfi],
            sfi_adjusted: attrs[:sfi_adjusted],
            sunspot_number: attrs[:sunspot_number],
            ap_index: attrs[:ap_index],
            kp_values: attrs[:kp_values],
            inserted_at: now,
            updated_at: now
          }
        end)

      {count, _} =
        Repo.insert_all(SolarIndex, entries,
          on_conflict:
            from(s in SolarIndex,
              update: [
                set: [
                  sfi: fragment("EXCLUDED.sfi"),
                  sfi_adjusted: fragment("EXCLUDED.sfi_adjusted"),
                  sunspot_number: fragment("EXCLUDED.sunspot_number"),
                  ap_index: fragment("EXCLUDED.ap_index"),
                  kp_values: fragment("EXCLUDED.kp_values"),
                  updated_at: fragment("EXCLUDED.updated_at")
                ]
              ],
              where:
                s.sfi != fragment("EXCLUDED.sfi") or
                  s.ap_index != fragment("EXCLUDED.ap_index")
            ),
          conflict_target: [:date]
        )

      acc + count
    end)
  end

  @spec get_solar_index(Date.t()) :: SolarIndex.t() | nil
  def get_solar_index(date) do
    Repo.get_by(SolarIndex, date: date)
  end

  @spec existing_solar_dates() :: MapSet.t(Date.t())
  def existing_solar_dates do
    SolarIndex
    |> select([s], s.date)
    |> Repo.all()
    |> MapSet.new()
  end

  @spec sync_stations!() :: :ok
  def sync_stations! do
    asos =
      for s <-
            ~w(AK AL AR AZ CA CO CT DE FL GA HI IA ID IL IN KS KY LA MA MD ME MI MN MO MS MT NC ND NE NH NJ NM NV NY OH OK OR PA RI SC SD TN TX UT VA VT WA WI WV WY),
          do: "#{s}_ASOS"

    for network <- asos ++ ["RAOB"] do
      sync_network(network)
      Process.sleep(200)
    end

    count = Repo.aggregate(Station, :count)
    Logger.info("Weather stations sync complete: #{count} total")
    :ok
  end

  defp sync_network(network) do
    type = if String.contains?(network, "ASOS"), do: "asos", else: "sounding"

    case IemClient.fetch_network(network) do
      {:ok, stations} ->
        for s <- stations do
          %Station{}
          |> Station.changeset(Map.put(s, :station_type, type))
          |> Repo.insert(on_conflict: :nothing, conflict_target: [:station_code, :station_type])
        end

        Logger.info("Synced #{length(stations)} stations from #{network}")

      {:error, e} ->
        Logger.warning("Failed to sync #{network}: #{inspect(e)}")
    end
  end

  @spec nearby_stations(float(), float(), String.t(), number()) :: [Station.t()]
  def nearby_stations(lat, lon, station_type, radius_km) do
    dlat = radius_km / @km_per_deg_lat
    dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180))

    Station
    |> where([s], s.station_type == ^station_type)
    |> where(
      [s],
      s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
        s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
    )
    |> Repo.all()
  end

  @spec sounding_times_around(DateTime.t()) :: [DateTime.t()]
  def sounding_times_around(dt) do
    date = DateTime.to_date(dt)

    times =
      if dt.hour < 12 do
        [
          DateTime.new!(Date.add(date, -1), ~T[12:00:00], "Etc/UTC"),
          DateTime.new!(date, ~T[00:00:00], "Etc/UTC")
        ]
      else
        [
          DateTime.new!(date, ~T[00:00:00], "Etc/UTC"),
          DateTime.new!(date, ~T[12:00:00], "Etc/UTC")
        ]
      end

    Enum.uniq(times)
  end

  @spec weather_for_contact(map(), keyword()) :: %{
          surface_observations: [SurfaceObservation.t()],
          soundings: [Sounding.t()]
        }
  def weather_for_contact(contact_params, opts \\ []) do
    lat = contact_params[:lat] || contact_params.lat
    lon = contact_params[:lon] || contact_params.lon
    timestamp = contact_params[:timestamp] || contact_params.timestamp

    radius_km = Keyword.get(opts, :radius_km, 150)
    time_window_hours = Keyword.get(opts, :time_window_hours, 6)

    # Bounding box in degrees
    dlat = radius_km / @km_per_deg_lat
    dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180))

    time_start = DateTime.add(timestamp, -time_window_hours * 3600, :second)
    time_end = DateTime.add(timestamp, time_window_hours * 3600, :second)

    station_ids =
      Station
      |> where(
        [s],
        s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
          s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
      )
      |> select([s], s.id)

    surface_observations =
      SurfaceObservation
      |> where([o], o.station_id in subquery(station_ids))
      |> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
      |> preload(:station)
      |> Repo.all()

    soundings =
      Sounding
      |> where([s], s.station_id in subquery(station_ids))
      |> where([s], s.observed_at >= ^time_start and s.observed_at <= ^time_end)
      |> preload(:station)
      |> Repo.all()

    %{surface_observations: surface_observations, soundings: soundings}
  end

  @sounding_search_radii_km [150, 300, 600, 1000]

  @doc """
  Search for soundings in widening radii around the given location, stopping
  at the first radius that returns any. Returns `%{soundings, radius_km,
  exhausted}` where `exhausted: true` means the widest radius also came up
  empty — the caller should trigger a fetch for missing data at that point.
  """
  @spec soundings_with_widening_radius(map()) :: %{
          soundings: [Sounding.t()],
          radius_km: pos_integer(),
          exhausted: boolean()
        }
  def soundings_with_widening_radius(params) do
    Enum.reduce_while(@sounding_search_radii_km, nil, fn radius_km, _acc ->
      result = weather_for_contact(params, radius_km: radius_km)

      if result.soundings == [] do
        {:cont, %{soundings: [], radius_km: radius_km, exhausted: true}}
      else
        {:halt, %{soundings: result.soundings, radius_km: radius_km, exhausted: false}}
      end
    end)
  end

  @spec latest_grid_valid_time() :: DateTime.t() | nil
  def latest_grid_valid_time do
    cond do
      vt = GridCache.latest_valid_time() -> vt
      vt = ProfilesFile.latest_valid_time() -> vt
      true -> latest_grid_valid_time_from_db()
    end
  end

  # Last-resort fallback for historical data sitting in the legacy
  # hrrr_profiles table. PropagationGridWorker no longer writes
  # grid-point rows there, so in steady state this returns nil and
  # the ProfilesFile fallback above is the real source of truth.
  defp latest_grid_valid_time_from_db do
    Repo.one(
      from(h in HrrrProfile,
        where: h.is_grid_point == true,
        select: max(h.valid_time)
      )
    )
  end

  @doc """
  Cache-only read for the /weather LiveView mount hot path. Returns whatever
  is in `GridCache` for the latest valid_time and fires a deduped background
  fill task on a miss instead of blocking. The async task broadcasts
  `weather:updated` when done, triggering every connected LiveView to refresh.

  Callers that genuinely need synchronous data (tests, scripts) should use
  `load_weather_grid/1` instead.
  """
  @spec latest_weather_grid(%{optional(String.t()) => float()} | nil) :: [map()]
  def latest_weather_grid(bounds) do
    case latest_grid_valid_time() do
      nil ->
        []

      latest_vt ->
        case GridCache.fetch_bounds(latest_vt, bounds) do
          {:ok, rows} ->
            rows

          :miss ->
            kickoff_async_grid_fill(latest_vt)
            []
        end
    end
  end

  @doc """
  Synchronous cache-or-disk read. Blocks for ~1s on a cold cache while
  `ProfilesFile.read/1` loads the latest grid from `/data/profiles`.
  Used by tests and by `weather_point_detail/3` fallbacks. LiveView
  callers should prefer `latest_weather_grid/1`.
  """
  @spec load_weather_grid(%{optional(String.t()) => float()} | nil) :: [map()]
  def load_weather_grid(bounds) do
    case latest_grid_valid_time() do
      nil ->
        []

      latest_vt ->
        case GridCache.fetch_bounds(latest_vt, bounds) do
          {:ok, rows} ->
            rows

          :miss ->
            full = load_grid_rows_for(latest_vt)
            GridCache.put(latest_vt, full)
            filter_weather_bounds(full, bounds)
        end
    end
  end

  @doc """
  All persisted weather valid_times sorted ascending. The grid worker
  writes a ProfilesFile for every forecast hour (f00..f18), so this
  enumerates the 19-entry forecast timeline the /weather page renders
  at the bottom of the map.
  """
  @spec available_weather_valid_times() :: [DateTime.t()]
  def available_weather_valid_times do
    ProfilesFile.list_valid_times()
  end

  @doc """
  Read the weather grid for a specific `valid_time` and bounds. Like
  `load_weather_grid/1` but takes the valid_time explicitly so the
  timeline can scrub to any forecast hour, not just the analysis hour.
  Returns `[]` if no profile file exists for that valid_time.

  Deliberately does NOT write forecast-hour grids back into `GridCache`
  on a miss: caching 18 forecast hours × 92k points would add ~300 MB
  per pod. The ProfilesFile read is a single ~2 MB ETF decode per scrub,
  which is fast enough for a user click.
  """
  @spec weather_grid_at(DateTime.t(), %{optional(String.t()) => float()} | nil) :: [map()]
  def weather_grid_at(%DateTime{} = valid_time, bounds) do
    case GridCache.fetch_bounds(valid_time, bounds) do
      {:ok, rows} ->
        rows

      :miss ->
        case ProfilesFile.read(valid_time) do
          {:ok, grid_data} ->
            # Filter-before-derive: only derive the points inside the
            # viewport instead of all 92k CONUS grid points. On a DFW
            # viewport that's ~20× less `SoundingParams.derive` work
            # per timeline scrub.
            build_grid_cache_rows(grid_data, valid_time, bounds)

          {:error, _} ->
            []
        end
    end
  end

  # Build derived GridCache rows for a valid_time from whichever
  # source has data: the persisted ProfilesFile first (hot path in
  # steady state), then the legacy hrrr_profiles table (historical
  # data only).
  defp load_grid_rows_for(valid_time) do
    case ProfilesFile.read(valid_time) do
      {:ok, grid_data} -> build_grid_cache_rows(grid_data, valid_time)
      {:error, _} -> []
    end
  end

  defp filter_weather_bounds(rows, nil), do: rows

  defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
    Enum.filter(rows, fn %{lat: lat, lon: lon} ->
      lat >= s and lat <= n and lon >= w and lon <= e
    end)
  end

  defp kickoff_async_grid_fill(valid_time) do
    _ =
      if GridCache.claim_fill(valid_time) do
        {:ok, _pid} =
          Task.start(fn ->
            try do
              Logger.info("Weather.grid_cache async fill starting for #{valid_time}")
              warm_grid_cache_and_broadcast(valid_time)

              _ =
                Phoenix.PubSub.broadcast(
                  Microwaveprop.PubSub,
                  "weather:updated",
                  {:weather_updated, valid_time}
                )

              Logger.info("Weather.grid_cache async fill complete for #{valid_time}")
            rescue
              e ->
                Logger.error("Weather.grid_cache async fill failed: #{inspect(e)}")
            after
              GridCache.release_fill(valid_time)
            end
          end)
      end

    :ok
  end

  @doc """
  Eagerly populate the `GridCache` with the full CONUS weather grid for
  `valid_time` and broadcast it to every node in the cluster. Used by the cold
  cache fill path (`kickoff_async_grid_fill/1`) — prefer
  `build_grid_cache_rows/2` inside `PropagationGridWorker`, which already has
  the in-memory grid data and avoids the ~20s JSONB round trip.
  """
  @spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok
  def warm_grid_cache_and_broadcast(valid_time) do
    rows = load_grid_rows_for(valid_time)
    GridCache.broadcast_put(valid_time, rows)
    :ok
  end

  @doc """
  Warm the local `GridCache` from the latest persisted `ProfilesFile`
  on pod startup. Makes `/weather` usable immediately after a deploy
  instead of waiting for the next hourly PropagationGridWorker run.
  Local put only — every node reads the same NFS mount so no need to
  broadcast.
  """
  @spec warm_grid_cache_from_latest_profile() :: :ok
  def warm_grid_cache_from_latest_profile do
    case ProfilesFile.latest_valid_time() do
      nil ->
        :ok

      valid_time ->
        try do
          rows = load_grid_rows_for(valid_time)
          GridCache.put(valid_time, rows)
          Logger.info("Weather: warmed GridCache from ProfilesFile for #{valid_time} (#{length(rows)} rows)")
        rescue
          e ->
            Logger.warning("Weather: ProfilesFile warm failed: #{inspect(e)}")
        end

        :ok
    end
  end

  @doc """
  Build derived weather grid cache rows directly from an in-memory HRRR
  `grid_data` map. Used by the cold-cache fill path after `ProfilesFile.read/1`
  returns the grid written by the Rust worker.

  `grid_data` is `%{{lat, lon} => profile_map}` as produced by
  `ProfilesFile.read/1`. Each row is pushed through `derive_and_clean/1`
  to compute the derived fields consumed by the weather map LiveView.
  """
  @spec build_grid_cache_rows(
          %{{float(), float()} => map()},
          DateTime.t(),
          %{optional(String.t()) => float()} | nil
        ) :: [map()]
  def build_grid_cache_rows(grid_data, valid_time, bounds \\ nil) do
    grid_data
    |> filter_grid_data_bounds(bounds)
    |> Enum.flat_map(fn {{lat, lon}, profile} ->
      build_grid_cache_row(lat, lon, profile, valid_time)
    end)
  end

  defp filter_grid_data_bounds(grid_data, nil), do: grid_data

  defp filter_grid_data_bounds(grid_data, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
    :maps.filter(fn {lat, lon}, _ -> lat >= s and lat <= n and lon >= w and lon <= e end, grid_data)
  end

  defp build_grid_cache_row(lat, lon, profile, valid_time) do
    temp_c = profile[:surface_temp_c]

    if is_nil(temp_c) or temp_c < -80 or temp_c > 60 do
      []
    else
      sounding = derive_sounding(profile[:profile])

      # Prefer explicit values from the profile map (Rust-written ProfilesFile
      # persists derived sounding params alongside the raw profile) and fall
      # back to deriving from the profile list when absent. `prefer/3` uses
      # `Map.fetch/2` so a legitimate `false` / `0` persisted value is not
      # silently clobbered by a `||` fallback.
      row = %{
        lat: lat,
        lon: lon,
        valid_time: valid_time,
        temperature: temp_c,
        dewpoint_depression: depression(temp_c, profile[:surface_dewpoint_c]),
        bl_height: profile[:hpbl_m],
        pwat: profile[:pwat_mm],
        refractivity_gradient: prefer(profile, :min_refractivity_gradient, sounding[:min_refractivity_gradient]),
        ducting: prefer(profile, :ducting_detected, sounding[:ducting_detected]),
        surface_pressure_mb: profile[:surface_pressure_mb],
        surface_temp_c: temp_c,
        surface_dewpoint_c: profile[:surface_dewpoint_c],
        surface_refractivity: prefer(profile, :surface_refractivity, sounding[:surface_refractivity]),
        profile: profile[:profile] || [],
        duct_characteristics: prefer(profile, :duct_characteristics, sounding[:duct_characteristics])
      }

      [derive_and_clean(row)]
    end
  end

  # Fetch `key` from `profile` verbatim when present (including `false`,
  # `0`, or `[]`); only fall through to the derived value when the key is
  # absent. `profile[key] || default` would discard legitimate `false` /
  # `0` as if they weren't set.
  defp prefer(profile, key, default) do
    case Map.fetch(profile, key) do
      {:ok, value} -> value
      :error -> default
    end
  end

  defp derive_sounding(profile) when is_list(profile) and length(profile) >= 3 do
    SoundingParams.derive(profile) || %{}
  end

  defp derive_sounding(_), do: %{}

  defp depression(nil, _), do: nil
  defp depression(_, nil), do: nil
  defp depression(t, d), do: t - d

  @spec weather_point_detail(float(), float(), DateTime.t()) :: map() | nil
  def weather_point_detail(lat, lon, valid_time) do
    step = 0.125
    snapped_lat = Float.round(Float.round(lat / step) * step, 3)
    snapped_lon = Float.round(Float.round(lon / step) * step, 3)

    case GridCache.fetch_point(valid_time, snapped_lat, snapped_lon) do
      {:ok, row} ->
        row

      :miss ->
        case weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
          nil -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
          row -> row
        end
    end
  end

  # Derive a single GridCache-shaped row from a persisted ProfilesFile
  # entry for `(valid_time, lat, lon)`. Returns nil when the file
  # doesn't exist or the point has no profile.
  defp weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
    case ProfilesFile.read_point(valid_time, snapped_lat, snapped_lon) do
      nil ->
        nil

      profile ->
        case build_grid_cache_rows(%{{snapped_lat, snapped_lon} => profile}, valid_time) do
          [row] -> row
          _ -> nil
        end
    end
  end

  defp weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon) do
    from(h in HrrrProfile,
      where: h.lat == ^snapped_lat and h.lon == ^snapped_lon and h.valid_time == ^valid_time,
      select: %{
        lat: h.lat,
        lon: h.lon,
        valid_time: h.valid_time,
        temperature: h.surface_temp_c,
        dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c),
        bl_height: h.hpbl_m,
        pwat: h.pwat_mm,
        refractivity_gradient: h.min_refractivity_gradient,
        ducting: h.ducting_detected,
        surface_pressure_mb: h.surface_pressure_mb,
        surface_temp_c: h.surface_temp_c,
        surface_dewpoint_c: h.surface_dewpoint_c,
        surface_refractivity: h.surface_refractivity,
        profile: h.profile,
        duct_characteristics: h.duct_characteristics
      }
    )
    |> Repo.one()
    |> then(fn
      nil -> nil
      row -> derive_and_clean(row)
    end)
  end

  defp derive_and_clean(row) do
    derived = WeatherLayers.derive(row)

    row
    |> Map.merge(derived)
    |> Map.drop([:profile, :duct_characteristics, :surface_temp_c, :surface_dewpoint_c])
  end

  @spec upsert_gefs_profile(map()) :: {:ok, GefsProfile.t()} | {:error, Ecto.Changeset.t()}
  def upsert_gefs_profile(attrs) do
    changeset = GefsProfile.changeset(%GefsProfile{}, attrs)

    if changeset.valid? do
      Repo.insert(changeset,
        on_conflict: :nothing,
        conflict_target: [:lat, :lon, :valid_time]
      )
    else
      {:error, changeset}
    end
  end

  @spec upsert_gefs_profiles_batch([map()]) :: {non_neg_integer(), nil}
  def upsert_gefs_profiles_batch(profiles) do
    Microwaveprop.Instrument.span(
      [:db, :upsert_gefs_profiles],
      %{count: length(profiles)},
      fn -> do_upsert_gefs_profiles_batch(profiles) end
    )
  end

  defp do_upsert_gefs_profiles_batch(profiles) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    profiles
    |> Enum.chunk_every(500)
    |> Enum.reduce({0, nil}, fn chunk, {total_count, _} ->
      entries =
        Enum.map(chunk, fn attrs ->
          Map.merge(attrs, %{
            id: Ecto.UUID.generate(),
            inserted_at: now,
            updated_at: now
          })
        end)

      {count, rows} =
        Repo.insert_all(GefsProfile, entries,
          on_conflict: :nothing,
          conflict_target: [:lat, :lon, :valid_time]
        )

      {total_count + count, rows}
    end)
  end

  @spec upsert_hrrr_profile(map()) :: {:ok, HrrrProfile.t()} | {:error, Ecto.Changeset.t()}
  def upsert_hrrr_profile(attrs) do
    %HrrrProfile{}
    |> HrrrProfile.changeset(attrs)
    |> Repo.insert(
      on_conflict: :nothing,
      conflict_target: [:lat, :lon, :valid_time]
    )
  end

  @spec upsert_hrrr_profiles_batch([map()], keyword()) :: {non_neg_integer(), nil}
  def upsert_hrrr_profiles_batch(profiles, _opts \\ []) do
    Microwaveprop.Instrument.span(
      [:db, :upsert_hrrr_profiles],
      %{count: length(profiles)},
      fn -> do_upsert_hrrr_profiles_batch(profiles) end
    )
  end

  defp do_upsert_hrrr_profiles_batch(profiles) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    profiles
    |> Enum.chunk_every(500)
    |> Enum.reduce({0, nil}, fn chunk, {total_count, _} ->
      entries =
        Enum.map(chunk, fn attrs ->
          is_gp =
            rem(round(attrs.lat * 1000), 125) == 0 and
              rem(round(attrs.lon * 1000), 125) == 0

          Map.merge(attrs, %{
            id: Ecto.UUID.generate(),
            is_grid_point: is_gp,
            inserted_at: now,
            updated_at: now
          })
        end)

      {count, rows} =
        Repo.insert_all(HrrrProfile, entries,
          on_conflict: :nothing,
          conflict_target: [:lat, :lon, :valid_time]
        )

      {total_count + count, rows}
    end)
  end

  @spec has_hrrr_profile?(float(), float(), DateTime.t()) :: boolean()
  def has_hrrr_profile?(lat, lon, valid_time) do
    dlat = 0.07
    dlon = 0.07

    HrrrProfile
    |> where(
      [h],
      h.lat >= ^(lat - dlat) and h.lat <= ^(lat + dlat) and
        h.lon >= ^(lon - dlon) and h.lon <= ^(lon + dlon) and
        h.valid_time == ^valid_time
    )
    |> Repo.exists?()
  end

  @doc """
  Returns `true` when every path point for `contact` (pos1 → midpoint → pos2,
  or just pos1 for one-endpoint contacts) already has an HRRR profile at the
  nearest HRRR hour. Lets the enrichment enqueuer skip :queued → :queued
  churn when the backing data is already present.

  Returns `false` if `pos1` or `qso_timestamp` is nil — a contact with no
  position or no timestamp can't be looked up.
  """
  @spec hrrr_data_fully_present?(
          Contact.t()
          | %{
              required(:pos1) => term(),
              required(:qso_timestamp) => DateTime.t() | nil,
              optional(:pos2) => term()
            }
        ) :: boolean()
  def hrrr_data_fully_present?(%{pos1: nil}), do: false
  def hrrr_data_fully_present?(%{qso_timestamp: nil}), do: false

  def hrrr_data_fully_present?(contact) do
    rounded = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp)

    case Radio.contact_path_points(contact) do
      [] ->
        false

      points ->
        Enum.all?(points, fn {lat, lon} ->
          {rlat, rlon} = round_to_hrrr_grid(lat, lon)
          has_hrrr_profile?(rlat, rlon, rounded)
        end)
    end
  end

  @spec hrrr_for_contact(map()) :: HrrrProfile.t() | nil
  def hrrr_for_contact(%{pos1: nil}), do: nil

  def hrrr_for_contact(contact) do
    lat = contact.pos1["lat"]
    lon = contact.pos1["lon"]

    if lat && lon do
      find_nearest_hrrr(lat, lon, contact.qso_timestamp)
    end
  end

  @doc """
  Returns just `hrrr_native_profiles.best_duct_band_ghz` near the
  given cell. Convenience wrapper around `nearest_native_duct_info/3`
  for callers that don't need Richardson.
  """
  @spec nearest_native_duct_ghz(float(), float(), DateTime.t()) :: float() | nil
  def nearest_native_duct_ghz(lat, lon, %DateTime{} = timestamp) do
    case nearest_native_duct_info(lat, lon, timestamp) do
      {:ok, %{best_duct_band_ghz: ghz}} -> ghz
      _ -> nil
    end
  end

  @doc """
  Returns `{:ok, %{best_duct_band_ghz: ghz, bulk_richardson: r}}` for
  the nearest `hrrr_native_profiles` cell to (`lat`, `lon`) at
  `timestamp` within ±0.07° / ±1h, or `{:error, :not_found}`.

  Richardson gates the 1.15× refractivity boost in `Scorer` — a duct
  band reading under turbulent conditions (high Richardson) is likely
  to be mixed out before it supports the target path. The pair is
  cheap to fetch together, so callers that score a cell prefer this
  over the bare `nearest_native_duct_ghz/3`.
  """
  @spec nearest_native_duct_info(float(), float(), DateTime.t()) ::
          {:ok, %{best_duct_band_ghz: float() | nil, bulk_richardson: float() | nil}}
          | {:error, :not_found}
  def nearest_native_duct_info(lat, lon, %DateTime{} = timestamp) do
    dlat = 0.07
    dlon = 0.07
    time_start = DateTime.add(timestamp, -3600, :second)
    time_end = DateTime.add(timestamp, 3600, :second)

    from(h in HrrrNativeProfile,
      where:
        h.lat >= ^(lat - dlat) and h.lat <= ^(lat + dlat) and
          h.lon >= ^(lon - dlon) and h.lon <= ^(lon + dlon) and
          h.valid_time >= ^time_start and h.valid_time <= ^time_end,
      order_by:
        fragment(
          "ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
          h.lat,
          ^lat,
          h.lon,
          ^lon,
          h.valid_time,
          ^timestamp
        ),
      limit: 1,
      select: %{best_duct_band_ghz: h.best_duct_band_ghz, bulk_richardson: h.bulk_richardson}
    )
    |> Repo.one()
    |> case do
      nil -> {:error, :not_found}
      row -> {:ok, row}
    end
  end

  @doc """
  Nearest sounding to (`lat`, `lon`) within `radius_km` km and a ±3-hour
  window around `timestamp`. Joins `weather_stations` to `soundings` so
  the caller gets the raw sounding row (station_id set, derived duct
  fields populated) back.

  Returns `{:ok, sounding}` or `{:error, :not_found}`. Use this in the
  path calculator to surface the nearest RAOB's `ducting_detected` flag
  as an independent check on HRRR's pressure-level duct signal, which
  under-reads thin surface ducts.
  """
  @spec nearest_sounding_to(float(), float(), DateTime.t(), keyword()) ::
          {:ok, Sounding.t()} | {:error, :not_found}
  def nearest_sounding_to(lat, lon, timestamp, opts \\ []) do
    radius_km = Keyword.get(opts, :radius_km, 300)
    hours = Keyword.get(opts, :hours, 3)

    # 1 deg lat ≈ 111 km; lon scaled by cos(lat).
    dlat = radius_km / 111.0
    dlon = radius_km / (111.0 * max(0.1, :math.cos(lat * :math.pi() / 180.0)))
    time_start = DateTime.add(timestamp, -hours * 3600, :second)
    time_end = DateTime.add(timestamp, hours * 3600, :second)

    from(s in Sounding,
      join: station in assoc(s, :station),
      where:
        station.lat >= ^(lat - dlat) and station.lat <= ^(lat + dlat) and
          station.lon >= ^(lon - dlon) and station.lon <= ^(lon + dlon) and
          s.observed_at >= ^time_start and s.observed_at <= ^time_end,
      order_by:
        fragment(
          "SQRT(POW(? - ?, 2) + POW(? - ?, 2)) + ABS(EXTRACT(EPOCH FROM ? - ?)) / 86400.0",
          station.lat,
          ^lat,
          station.lon,
          ^lon,
          s.observed_at,
          ^timestamp
        ),
      limit: 1
    )
    |> Repo.one()
    |> case do
      nil -> {:error, :not_found}
      sounding -> {:ok, sounding}
    end
  end

  @spec find_nearest_hrrr(float(), float(), DateTime.t()) :: HrrrProfile.t() | nil
  def find_nearest_hrrr(lat, lon, timestamp) do
    dlat = 0.07
    dlon = 0.07
    time_start = DateTime.add(timestamp, -3600, :second)
    time_end = DateTime.add(timestamp, 3600, :second)

    HrrrProfile
    |> where(
      [h],
      h.lat >= ^(lat - dlat) and h.lat <= ^(lat + dlat) and
        h.lon >= ^(lon - dlon) and h.lon <= ^(lon + dlon) and
        h.valid_time >= ^time_start and h.valid_time <= ^time_end
    )
    |> order_by([h],
      asc:
        fragment(
          "ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
          h.lat,
          ^lat,
          h.lon,
          ^lon,
          h.valid_time,
          ^timestamp
        )
    )
    |> limit(1)
    |> Repo.one()
  end

  @spec hrrr_profiles_for_path(map()) :: [HrrrProfile.t()]
  def hrrr_profiles_for_path(%{pos1: nil}), do: []

  def hrrr_profiles_for_path(contact) do
    contact
    |> Radio.contact_path_points()
    |> Enum.map(fn {lat, lon} -> find_nearest_hrrr(lat, lon, contact.qso_timestamp) end)
    |> Enum.reject(&is_nil/1)
  end

  @spec find_nearest_native_profile(float(), float(), DateTime.t()) ::
          HrrrNativeProfile.t() | nil
  def find_nearest_native_profile(lat, lon, timestamp) do
    dlat = 0.07
    dlon = 0.07
    time_start = DateTime.add(timestamp, -3600, :second)
    time_end = DateTime.add(timestamp, 3600, :second)

    HrrrNativeProfile
    |> where(
      [n],
      n.lat >= ^(lat - dlat) and n.lat <= ^(lat + dlat) and
        n.lon >= ^(lon - dlon) and n.lon <= ^(lon + dlon) and
        n.valid_time >= ^time_start and n.valid_time <= ^time_end
    )
    |> order_by([n],
      asc:
        fragment(
          "ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
          n.lat,
          ^lat,
          n.lon,
          ^lon,
          n.valid_time,
          ^timestamp
        )
    )
    |> limit(1)
    |> Repo.one()
  end

  @doc """
  Find the best available atmospheric profile for a contact.
  Tries HRRR first (3 km, hourly), falls back to NARR (32 km, 3-hourly).
  Returns the profile struct or nil.
  """
  @spec best_profile_for_contact(map()) :: HrrrProfile.t() | NarrProfile.t() | nil
  def best_profile_for_contact(contact) do
    hrrr_for_contact(contact) || narr_for_contact(contact)
  end

  @doc "Find all atmospheric profiles along a contact's path, from any source."
  @spec profiles_along_path(map()) :: [HrrrProfile.t() | NarrProfile.t()]
  def profiles_along_path(contact) do
    hrrr_path = hrrr_profiles_for_path(contact)

    if hrrr_path == [] do
      narr_profiles_for_path(contact)
    else
      hrrr_path
    end
  end

  @spec narr_for_contact(map()) :: NarrProfile.t() | nil
  def narr_for_contact(%{pos1: nil}), do: nil

  def narr_for_contact(contact) do
    lat = contact.pos1["lat"]
    lon = contact.pos1["lon"]

    if lat && lon do
      find_nearest_narr(lat, lon, contact.qso_timestamp)
    end
  end

  @spec narr_profiles_for_path(map()) :: [NarrProfile.t()]
  def narr_profiles_for_path(%{pos1: nil}), do: []

  def narr_profiles_for_path(contact) do
    contact
    |> Radio.contact_path_points()
    |> Enum.map(fn {lat, lon} -> find_nearest_narr(lat, lon, contact.qso_timestamp) end)
    |> Enum.reject(&is_nil/1)
  end

  @spec find_nearest_narr(float(), float(), DateTime.t()) :: NarrProfile.t() | nil
  def find_nearest_narr(lat, lon, timestamp) do
    dlat = 0.15
    dlon = 0.15
    time_start = DateTime.add(timestamp, -1800, :second)
    time_end = DateTime.add(timestamp, 1800, :second)

    NarrProfile
    |> where(
      [p],
      p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and
        p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and
        p.valid_time >= ^time_start and p.valid_time <= ^time_end
    )
    |> order_by([p],
      asc:
        fragment(
          "ABS(? - ?) + ABS(? - ?)",
          p.lat,
          ^lat,
          p.lon,
          ^lon
        )
    )
    |> limit(1)
    |> Repo.one()
  end

  @spec find_nearest_rtma(float(), float(), DateTime.t()) :: RtmaObservation.t() | nil
  def find_nearest_rtma(lat, lon, timestamp) do
    dlat = 0.05
    dlon = 0.05
    time_start = DateTime.add(timestamp, -900, :second)
    time_end = DateTime.add(timestamp, 900, :second)

    RtmaObservation
    |> where(
      [o],
      o.lat >= ^(lat - dlat) and o.lat <= ^(lat + dlat) and
        o.lon >= ^(lon - dlon) and o.lon <= ^(lon + dlon) and
        o.valid_time >= ^time_start and o.valid_time <= ^time_end
    )
    |> order_by([o],
      asc:
        fragment(
          "ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
          o.lat,
          ^lat,
          o.lon,
          ^lon,
          o.valid_time,
          ^timestamp
        )
    )
    |> limit(1)
    |> Repo.one()
  end

  @spec round_to_hrrr_grid(float(), float()) :: {float(), float()}
  def round_to_hrrr_grid(lat, lon) do
    {Float.round(lat / 1.0, 2), Float.round(lon / 1.0, 2)}
  end

  @doc """
  Delete every `is_grid_point = true` row from `hrrr_profiles`, regardless of
  age. Grid-point profiles are historical — the propagation grid now lives in
  `/data/scores` binary files, nothing reads `is_grid_point = true` rows
  anymore, and forecast runs no longer write them. This purge walks each
  partition directly so a single DELETE can't scan the whole parent table.
  Preserves QSO-linked rows (`is_grid_point = false`), which remain the data
  path for contact enrichment and the `/path` calculator.
  """
  @spec purge_grid_point_profiles() :: non_neg_integer()
  def purge_grid_point_profiles do
    deleted =
      Enum.reduce(hrrr_profile_partitions(), 0, fn partition, acc ->
        %{num_rows: n} =
          Repo.query!(
            ~s(DELETE FROM "#{partition}" WHERE is_grid_point = true),
            [],
            timeout: 600_000
          )

        if n > 0 do
          require Logger

          Logger.info("Purged #{n} grid-point rows from #{partition}")
        end

        acc + n
      end)

    deleted
  end

  @spec hrrr_profile_partitions() :: [String.t()]
  defp hrrr_profile_partitions do
    {:ok, %{rows: rows}} =
      Repo.query(
        """
        SELECT child.relname
        FROM pg_inherits
        JOIN pg_class parent ON pg_inherits.inhparent = parent.oid
        JOIN pg_class child  ON pg_inherits.inhrelid  = child.oid
        WHERE parent.relname = 'hrrr_profiles'
        ORDER BY child.relname
        """,
        [],
        timeout: 60_000
      )

    Enum.map(rows, fn [name] -> name end)
  end

  @spec round_to_iemre_grid(float(), float()) :: {float(), float()}
  def round_to_iemre_grid(lat, lon) do
    {Float.round(lat * 8) / 8, Float.round(lon * 8) / 8}
  end

  @spec upsert_iemre_observation(map()) :: {:ok, IemreObservation.t()} | {:error, Ecto.Changeset.t()}
  def upsert_iemre_observation(attrs) do
    %IemreObservation{}
    |> IemreObservation.changeset(attrs)
    |> Repo.insert(
      on_conflict: :nothing,
      conflict_target: [:lat, :lon, :date]
    )
  end

  @spec has_iemre_observation?(float(), float(), Date.t()) :: boolean()
  def has_iemre_observation?(lat, lon, date) do
    IemreObservation
    |> where([i], i.lat == ^lat and i.lon == ^lon and i.date == ^date)
    |> Repo.exists?()
  end

  @spec iemre_for_contact(map()) :: IemreObservation.t() | nil
  def iemre_for_contact(%{pos1: nil}), do: nil

  def iemre_for_contact(contact) do
    lat = contact.pos1["lat"]
    lon = contact.pos1["lon"]

    if lat && lon do
      find_nearest_iemre(lat, lon, contact.qso_timestamp)
    end
  end

  @spec find_nearest_iemre(float(), float(), DateTime.t()) :: IemreObservation.t() | nil
  def find_nearest_iemre(lat, lon, timestamp) do
    {rlat, rlon} = round_to_iemre_grid(lat, lon)
    date = DateTime.to_date(timestamp)

    IemreObservation
    |> where([i], i.lat == ^rlat and i.lon == ^rlon and i.date == ^date)
    |> Repo.one()
  end

  @spec iemre_for_path(map()) :: [IemreObservation.t()]
  def iemre_for_path(%{pos1: nil}), do: []

  def iemre_for_path(contact) do
    contact
    |> Radio.contact_path_points()
    |> Enum.map(fn {lat, lon} -> find_nearest_iemre(lat, lon, contact.qso_timestamp) end)
    |> Enum.reject(&is_nil/1)
  end

  @doc """
  Find the nearest surface observation to a given (lat, lon, time),
  preferring 5-minute METAR data when available, falling back to the
  hourly `surface_observations` table.

  Returns a map with `:temp_f`, `:dewpoint_f`, `:wind_speed_kts`,
  `:observed_at`, etc. — the same shape regardless of which table the
  data came from. Returns `nil` if neither source has data.
  """
  @spec recent_surface_obs(float(), float(), DateTime.t()) :: Metar5minObservation.t() | SurfaceObservation.t() | nil
  def recent_surface_obs(lat, lon, timestamp) do
    dlat = 0.5
    dlon = 0.5
    time_start = DateTime.add(timestamp, -1800, :second)
    time_end = DateTime.add(timestamp, 1800, :second)

    station_ids =
      Station
      |> where(
        [s],
        s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
          s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
      )
      |> select([s], s.id)

    # Try 5-min first
    metar_5min =
      Metar5minObservation
      |> where([o], o.station_id in subquery(station_ids))
      |> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
      |> order_by([o], asc: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", o.observed_at, ^timestamp))
      |> limit(1)
      |> Repo.one()

    if metar_5min do
      metar_5min
    else
      # Fall back to hourly
      SurfaceObservation
      |> where([o], o.station_id in subquery(station_ids))
      |> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
      |> order_by([o], asc: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", o.observed_at, ^timestamp))
      |> limit(1)
      |> Repo.one()
    end
  end
end