prop/lib/microwaveprop/weather.ex

defmodule Microwaveprop.Weather do
  @moduledoc false

  import Ecto.Query

  alias Microwaveprop.Propagation.Grid
  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.ScalarFile
  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.iemre_status` from `:queued` to `:complete` for every
  contact whose path points all have an `iemre_observations` row at
  the rounded grid cell for the QSO's UTC date.

  Per-contact check (via the existing helpers); kept in Elixir because
  `contact_path_points/1` produces 1–3 points depending on whether
  `pos2` is set, and rounding to the 0.125° IEMRE grid per point in
  pure SQL is awkward. Queue sizes are small (<20 stuck at steady
  state).
  """
  @spec reconcile_iemre_statuses() :: {:ok, non_neg_integer()}
  def reconcile_iemre_statuses do
    import Ecto.Query

    queued =
      Contact
      |> where([c], c.iemre_status == :queued and not is_nil(c.pos1))
      |> Repo.all()

    to_complete =
      Enum.filter(queued, fn c ->
        date = DateTime.to_date(c.qso_timestamp)

        c
        |> Radio.contact_path_points()
        |> Enum.all?(fn {lat, lon} ->
          {rlat, rlon} = round_to_iemre_grid(lat, lon)
          has_iemre_observation?(rlat, rlon, date)
        end)
      end)

    if to_complete == [] do
      {:ok, 0}
    else
      ids = Enum.map(to_complete, & &1.id)
      _ = Radio.set_enrichment_status!(ids, :iemre_status, :complete)
      {:ok, length(to_complete)}
    end
  end

  @doc """
  Reconcile `contacts.hrrr_status` for every `:queued` contact:

    * flips to `:complete` when every path point has an hrrr_profiles
      row at the rounded HRRR hour;
    * flips to `:unavailable` when any path point falls outside the
      HRRR CONUS grid (pos2 or the great-circle midpoint can drift
      into Canada, the mid-Atlantic, or the Pacific even when pos1 is
      stateside — the Rust hrrr-point-worker returns no profile for
      those points and the contact would otherwise sit `:queued`
      forever).

  Returns `{:ok, n}` where `n` is the total number of contacts advanced
  (completed + oconus).
  """
  @spec reconcile_hrrr_statuses() :: {:ok, non_neg_integer()}
  def reconcile_hrrr_statuses do
    import Ecto.Query

    queued =
      Contact
      |> where([c], c.hrrr_status == :queued and not is_nil(c.pos1))
      |> Repo.all()

    {oconus, rest} = Enum.split_with(queued, &contact_has_oconus_path_point?/1)
    to_complete = Enum.filter(rest, &hrrr_data_fully_present?/1)

    _ =
      if oconus != [] do
        ids = Enum.map(oconus, & &1.id)
        Radio.set_enrichment_status!(ids, :hrrr_status, :unavailable)
      end

    _ =
      if to_complete != [] do
        ids = Enum.map(to_complete, & &1.id)
        Radio.set_enrichment_status!(ids, :hrrr_status, :complete)
      end

    {:ok, length(oconus) + length(to_complete)}
  end

  defp contact_has_oconus_path_point?(contact) do
    contact
    |> Radio.contact_path_points()
    |> Enum.any?(fn {lat, lon} ->
      not Grid.contains?(%{"lat" => lat, "lon" => lon})
    end)
  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 ±1.5° latitude band; the longitude band is
  scaled by `1 / cos(lat)` so the box covers the same physical
  east-west distance (~150 km) at every latitude. A fixed 1.5° lon
  box collapses to ~110 km at lat 49° and would silently skip
  observations the per-contact `weather_for_contact/2` query would
  match.
  """
  @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 / GREATEST(cos(radians((c.pos1->>'lat')::float)), 0.01))
                          AND ((c.pos1->>'lon')::float + 1.5 / GREATEST(cos(radians((c.pos1->>'lat')::float)), 0.01))
        )
        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 / GREATEST(cos(radians((c.pos2->>'lat')::float)), 0.01))
                            AND ((c.pos2->>'lon')::float + 1.5 / GREATEST(cos(radians((c.pos2->>'lat')::float)), 0.01))
          )
        )
      )
    """

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

  @doc """
  Kick off `ANALYZE` on every public-schema table that hasn't been
  auto-analyzed recently. Returns immediately with the list of tables
  it will visit; the actual work runs in a supervised Task on the
  current node so a broken `kubectl exec` doesn't abort it mid-way.

  Callable from a release shell:

      bin/microwaveprop rpc 'Microwaveprop.Weather.analyze_all()'

  Surfaces the HRRR partition problem: most partitions have zero
  analyze stats, so the planner picks nested-loop joins that scan
  81 M rows. Running ANALYZE once gives it live_tup and column
  histograms to work with.

  `skip_recent` skips anything auto-analyzed in the last N seconds
  (default 6 h) so re-running is cheap. Progress is logged via
  `Logger.info` — tail `kubectl -n prop logs deploy/prop-backfill`
  to watch.
  """
  @spec analyze_all(keyword()) :: %{queued: non_neg_integer(), tables: [String.t()]}
  def analyze_all(opts \\ []) do
    skip_recent = Keyword.get(opts, :skip_recent_seconds, 6 * 3600)

    # `last_autoanalyze` is only set by the autovacuum daemon;
    # `last_analyze` is set by manual ANALYZE (including our Task
    # below). Take the newer of the two — otherwise a re-run within
    # `skip_recent` would re-analyze everything we just did.
    q = """
    SELECT relname
    FROM pg_stat_user_tables
    WHERE schemaname = 'public'
      AND COALESCE(
        GREATEST(last_analyze, last_autoanalyze),
        'epoch'::timestamptz
      ) < now() - ($1 || ' seconds')::interval
    ORDER BY n_live_tup ASC
    """

    %{rows: rows} = Repo.query!(q, [Integer.to_string(skip_recent)])
    tables = Enum.map(rows, fn [n] -> n end)

    # Fire-and-forget: run each ANALYZE on the pod's own node so the
    # caller returns before any single query finishes. Small tables
    # first keeps progress visible early.
    Task.Supervisor.start_child({:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}, fn ->
      Enum.each(tables, fn t ->
        Logger.info("Weather.analyze_all: ANALYZE #{t}")
        started = System.monotonic_time(:millisecond)

        try do
          Repo.query!("ANALYZE #{quote_ident(t)}", [], timeout: :infinity)
          elapsed = System.monotonic_time(:millisecond) - started
          Logger.info("Weather.analyze_all: ANALYZE #{t} done in #{elapsed}ms")
        rescue
          e -> Logger.warning("Weather.analyze_all: ANALYZE #{t} failed: #{inspect(e)}")
        end
      end)

      Logger.info("Weather.analyze_all: complete (#{length(tables)} tables)")
    end)

    %{queued: length(tables), tables: tables}
  end

  # Defensive: pg_stat_user_tables returns legitimate identifiers from
  # the catalog, but we still quote just in case a partition contains a
  # character the shell would interpret.
  defp quote_ident(name) when is_binary(name) do
    escaped = String.replace(name, ~s("), ~s(""))
    ~s("#{escaped}")
  end

  @doc """
  Derive `surface_refractivity` / `min_refractivity_gradient` /
  `ducting_detected` for every `hrrr_profiles` row where those columns
  are NULL but the pressure-level `profile` JSONB is populated.
  Required one-off after the Rust hrrr_points worker shipped without
  deriving these scalars — existing rows have the raw levels but the
  DB scalars are NULL. Idempotent; rows with scalars already set are
  skipped by the WHERE clause.

  Batches by `limit` rows per pass to keep transactions short on the
  Turing Pi 2 Postgres node. Returns the total number of rows advanced.
  """
  @spec backfill_hrrr_scalars(keyword()) :: non_neg_integer()
  def backfill_hrrr_scalars(opts \\ []) do
    batch = Keyword.get(opts, :batch_size, 500)
    max_batches = Keyword.get(opts, :max_batches, 1_000)

    Enum.reduce_while(1..max_batches, 0, fn _, total ->
      case backfill_hrrr_batch(batch) do
        0 -> {:halt, total}
        n -> {:cont, total + n}
      end
    end)
  end

  defp backfill_hrrr_batch(limit) do
    rows =
      HrrrProfile
      |> where([h], is_nil(h.surface_refractivity))
      |> limit(^limit)
      |> select([h], %{id: h.id, profile: h.profile})
      |> Repo.all()

    updates =
      Enum.flat_map(rows, fn %{id: id, profile: profile} ->
        case SoundingParams.derive(profile || []) do
          %{} = derived ->
            [
              %{
                id: id,
                surface_refractivity: Map.get(derived, :surface_refractivity),
                min_refractivity_gradient: Map.get(derived, :min_refractivity_gradient),
                ducting_detected: Map.get(derived, :ducting_detected, false),
                duct_characteristics: Map.get(derived, :duct_characteristics)
              }
            ]

          _ ->
            []
        end
      end)

    Enum.each(updates, fn u ->
      HrrrProfile
      |> where([h], h.id == ^u.id)
      |> Repo.update_all(
        set: [
          surface_refractivity: u.surface_refractivity,
          min_refractivity_gradient: u.min_refractivity_gradient,
          ducting_detected: u.ducting_detected,
          duct_characteristics: u.duct_characteristics,
          updated_at: DateTime.truncate(DateTime.utc_now(), :second)
        ]
      )
    end)

    length(updates)
  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), and the
  derived `ScalarFile` mirrors that for any hour that has been
  materialized. We union both so the timeline survives an aggressive
  retention sweep on either side as long as one artifact remains.
  """
  @spec available_weather_valid_times() :: [DateTime.t()]
  def available_weather_valid_times do
    scalar_times = ScalarFile.list_valid_times()
    profile_times = ProfilesFile.list_valid_times()

    (scalar_times ++ profile_times)
    |> Enum.uniq()
    |> Enum.sort(DateTime)
  end

  @doc """
  All persisted HRDPS valid_times (i.e. those with a `<iso>.hrdps`
  scalar dir on disk) sorted ascending. Backs the `/weather-ca`
  timeline.
  """
  @spec available_hrdps_valid_times() :: [DateTime.t()]
  def available_hrdps_valid_times do
    ScalarFile.list_valid_times_hrdps()
  end

  @doc """
  HRDPS-only counterpart to `weather_grid_at/2`. Reads from the
  `<vt>.hrdps` scalar dir and skips HRRR completely. Bypasses GridCache
  because the cache mixes HRRR + HRDPS rows by design — caching this
  separately would double the per-pod memory budget for marginal benefit
  (Canadian viewport reads are infrequent compared to CONUS).
  """
  @spec weather_grid_hrdps_at(DateTime.t(), %{optional(String.t()) => float()} | nil) :: [map()]
  def weather_grid_hrdps_at(%DateTime{} = valid_time, bounds) do
    ScalarFile.read_bounds_hrdps(valid_time, bounds)
  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 ->
        # Once a scalar file exists for `valid_time` it's the source of
        # truth — even an empty viewport read (e.g. over the ocean) is
        # authoritative, so don't fall back to a full ProfilesFile decode
        # in that case.
        if ScalarFile.exists?(valid_time) do
          read_via_scalar(valid_time, bounds)
        else
          read_and_derive_grid(valid_time, bounds)
        end
    end
  end

  defp read_via_scalar(valid_time, bounds) do
    fill_grid_cache_from_scalar(valid_time)

    case GridCache.fetch_bounds(valid_time, bounds) do
      {:ok, rows} -> rows
      :miss -> ScalarFile.read_bounds(valid_time, bounds)
    end
  end

  # Cap on cached valid_times. HRRR runs hourly with 18 forecast hours, so
  # 24 covers analysis + 18 forecasts plus a few late stragglers from the
  # previous run while a user is scrubbing.
  @grid_cache_valid_time_cap 24

  # Hydrate GridCache from the on-disk ScalarFile so concurrent viewport
  # reads for the same valid_time don't each gunzip+msgpack-decode the
  # same chunk files. Only the caller that wins `claim_fill` does the
  # work; losers wait briefly for the ETS write to land.
  defp fill_grid_cache_from_scalar(valid_time) do
    lock_key = {:scalar_to_grid_cache, valid_time}

    if GridCache.claim_fill(lock_key) do
      try do
        rows = ScalarFile.read_bounds(valid_time, nil)

        if rows != [] do
          GridCache.put(valid_time, rows)
          _ = GridCache.prune_keep_latest(@grid_cache_valid_time_cap)
        end
      after
        GridCache.release_fill(lock_key)
      end
    else
      wait_for_grid_cache(valid_time)
    end
  end

  defp wait_for_grid_cache(valid_time) do
    Enum.reduce_while(1..50, :miss, fn _, _ ->
      case GridCache.fetch(valid_time) do
        {:ok, _} ->
          {:halt, :ok}

        :miss ->
          Process.sleep(20)
          {:cont, :miss}
      end
    end)
  end

  # Cold path: ScalarFile didn't have anything for this valid_time, so
  # decode the raw ProfilesFile and derive the requested viewport. Kicks
  # off a background materialization of the full scalar file so the
  # next read hits the cheap path.
  defp read_and_derive_grid(valid_time, bounds) do
    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.
        rows = build_grid_cache_rows(grid_data, valid_time, bounds)
        kickoff_async_scalar_materialize(valid_time, grid_data)
        rows

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

  # Materialize the full ScalarFile for `valid_time` once per node,
  # asynchronously. Reuses GridCache.claim_fill so concurrent
  # `weather_grid_at` callers don't trigger N derivations of the same
  # 92k-cell grid. Lock key namespaced so it doesn't collide with the
  # GridCache cold-fill claim for the same valid_time.
  defp kickoff_async_scalar_materialize(valid_time, grid_data) do
    if ScalarFile.exists?(valid_time) do
      :ok
    else
      lock_key = {:scalar_materialize, valid_time}

      _ =
        if GridCache.claim_fill(lock_key) do
          {:ok, _pid} =
            Task.start(fn ->
              try do
                rows = build_grid_cache_rows(grid_data, valid_time)
                ScalarFile.write!(valid_time, rows)

                Logger.info("Weather.scalar_file materialized valid_time=#{valid_time} rows=#{length(rows)}")
              rescue
                e ->
                  Logger.error("Weather.scalar_file materialize failed valid_time=#{valid_time} #{inspect(e)}")
              after
                GridCache.release_fill(lock_key)
              end
            end)
        end

      :ok
    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)
    persist_scalar_file(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)
          persist_scalar_file(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

  # Persist a derived grid as a ScalarFile so subsequent `/weather` reads
  # don't have to re-decode the raw ProfilesFile or re-run
  # `SoundingParams.derive` + `WeatherLayers.derive`. Best-effort: an NFS
  # write failure is logged but never fatal — the cold-derive path keeps
  # working.
  defp persist_scalar_file(_valid_time, []), do: :ok

  defp persist_scalar_file(valid_time, rows) do
    ScalarFile.write!(valid_time, rows)
    :ok
  rescue
    e ->
      Logger.warning("Weather: ScalarFile.write failed valid_time=#{valid_time} #{inspect(e)}")
      :ok
  end

  @doc """
  Materialize the `ScalarFile` for `valid_time` from the on-disk
  `ProfilesFile`. Idempotent — if a scalar file already exists, returns
  `:ok` without re-deriving. Used by `NotifyListener` to pre-warm scalar
  artifacts the moment the Rust propagation pipeline fires
  `propagation_ready`, so the first `/weather` reader of a new forecast
  hour never falls back to the slow `ProfilesFile.read/1` + per-cell
  derive path.

  Synchronous; callers should run this inside a `Task` if they need
  not to block (the listener does).
  """
  @spec materialize_scalar_file(DateTime.t()) :: :ok
  def materialize_scalar_file(%DateTime{} = valid_time) do
    if ScalarFile.exists?(valid_time) do
      :ok
    else
      try do
        rows = load_grid_rows_for(valid_time)
        persist_scalar_file(valid_time, rows)
        Logger.info("Weather.materialize_scalar_file vt=#{valid_time} rows=#{length(rows)}")
        :ok
      rescue
        e ->
          Logger.warning("Weather.materialize_scalar_file failed vt=#{valid_time} #{inspect(e)}")
          :ok
      end
    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
      # `SoundingParams.derive` + `WeatherLayers.sort_profile` both call
      # `SoundingParams.normalize_profile_entry/1` internally, so we can
      # hand either shape (legacy string-keyed or Rust atom-keyed) through
      # unchanged. `surface_refractivity` and `min_refractivity_gradient`
      # aren't persisted by Rust — they flow through from the per-cell
      # sounding derivation; `refractivity_gradient` falls back to the
      # `native_min_gradient` scalar the Rust f01..f18 pipeline does write.
      sounding = derive_sounding(profile[:profile])

      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]) ||
            profile[:native_min_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 -> point_detail_off_cache(valid_time, snapped_lat, snapped_lon)
    end
  end

  defp point_detail_off_cache(valid_time, snapped_lat, snapped_lon) do
    case ScalarFile.read_point(valid_time, snapped_lat, snapped_lon) do
      {:ok, row} -> row
      :miss -> point_detail_from_disk(valid_time, snapped_lat, snapped_lon)
    end
  end

  defp point_detail_from_disk(valid_time, snapped_lat, snapped_lon) do
    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

  # 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