prop/lib/microwaveprop/weather.ex

defmodule Microwaveprop.Weather do
  @moduledoc false

  import Ecto.Query

  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.Era5Profile
  alias Microwaveprop.Weather.GridCache
  alias Microwaveprop.Weather.HrrrProfile
  alias Microwaveprop.Weather.IemClient
  alias Microwaveprop.Weather.IemreObservation
  alias Microwaveprop.Weather.Metar5minObservation
  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

  @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 "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

  @spec latest_grid_valid_time() :: DateTime.t() | nil
  def latest_grid_valid_time do
    case GridCache.latest_valid_time() do
      %DateTime{} = vt ->
        vt

      nil ->
        Repo.one(
          from(h in HrrrProfile,
            where: h.is_grid_point == true,
            select: max(h.valid_time)
          )
        )
    end
  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(map()) :: [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-DB read. Blocks for several seconds on a cold cache
  because `load_weather_grid_from_db/1` parses 176k rows with JSONB columns.
  Used by tests and by `weather_point_detail/3` fallbacks. LiveView callers
  should prefer `latest_weather_grid/1`.
  """
  @spec load_weather_grid(map()) :: [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_weather_grid_from_db(latest_vt)
            GridCache.put(latest_vt, full)
            filter_weather_bounds(full, bounds)
        end
    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
      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

  defp load_weather_grid_from_db(latest_vt) do
    from(h in HrrrProfile,
      where: h.valid_time == ^latest_vt and h.is_grid_point == true,
      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.all(timeout: 60_000)
    |> Enum.map(&derive_and_clean/1)
  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_weather_grid_from_db(valid_time)
    GridCache.broadcast_put(valid_time, rows)
    :ok
  end

  @doc """
  Build derived weather grid cache rows directly from the in-memory HRRR
  `grid_data` that `PropagationGridWorker` already has after fetching a
  forecast hour. Avoids the 15s+ DB round trip that `load_weather_grid_from_db/1`
  pays to SELECT + JSONB-decode 92k hrrr_profiles rows — the source of the
  crash loop that stopped forecast hours from being written.

  `grid_data` is `%{{lat, lon} => profile_map}` as produced by
  `HrrrClient.fetch_grid/3` (optionally enriched via native duct merge). The
  output matches the shape of `load_weather_grid_from_db/1` after
  `derive_and_clean/1`.
  """
  @spec build_grid_cache_rows(%{{float(), float()} => map()}, DateTime.t()) :: [map()]
  def build_grid_cache_rows(grid_data, valid_time) do
    Enum.flat_map(grid_data, fn {{lat, lon}, profile} ->
      build_grid_cache_row(lat, lon, profile, valid_time)
    end)
  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])

      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: sounding[:min_refractivity_gradient],
        ducting: sounding[:ducting_detected],
        surface_pressure_mb: profile[:surface_pressure_mb],
        surface_temp_c: temp_c,
        surface_dewpoint_c: profile[:surface_dewpoint_c],
        surface_refractivity: sounding[:surface_refractivity],
        profile: profile[:profile] || [],
        duct_characteristics: sounding[:duct_characteristics]
      }

      [derive_and_clean(row)]
    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 -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
    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_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
    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

  @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"] || contact.pos1["lng"]

    if lat && lon do
      find_nearest_hrrr(lat, lon, contact.qso_timestamp)
    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
    |> Microwaveprop.Radio.contact_path_points()
    |> Enum.map(fn {lat, lon} -> find_nearest_hrrr(lat, lon, contact.qso_timestamp) end)
    |> Enum.reject(&is_nil/1)
  end

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

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

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

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

  def era5_for_contact(contact) do
    lat = contact.pos1["lat"]
    lon = contact.pos1["lon"] || contact.pos1["lng"]

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

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

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

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

    Era5Profile
    |> 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 HRRR grid profiles older than 24 hours that sit on the 0.125° grid.
  Preserves QSO-linked profiles which are at arbitrary lat/lon positions.
  With 19 forecast hours per run, this keeps ~2 runs worth of grid profiles.
  """
  @spec prune_old_grid_profiles() :: non_neg_integer()
  def prune_old_grid_profiles do
    cutoff = DateTime.add(DateTime.utc_now(), -24, :hour)
    # Bound the scan to recent partitions only (grid profiles are at most ~48h old).
    # Without a lower bound, Postgres scans every partition back to 2016.
    floor = DateTime.add(DateTime.utc_now(), -72, :hour)

    # Only delete profiles on the 0.125° propagation grid, preserving QSO-linked
    # profiles which sit at arbitrary lat/lon positions (HRRR's native ~3km grid).
    %{num_rows: deleted} =
      Repo.query!(
        """
        DELETE FROM hrrr_profiles
        WHERE valid_time >= $2 AND valid_time < $1
          AND is_grid_point = true
        """,
        [cutoff, floor],
        timeout: 300_000
      )

    if deleted > 0 do
      require Logger

      Logger.info("Pruned #{deleted} old HRRR grid profiles (valid_time < #{cutoff})")
    end

    deleted
  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"] || contact.pos1["lng"]

    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
    |> Microwaveprop.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