prop/lib/microwaveprop/weather.ex

defmodule Microwaveprop.Weather do
  @moduledoc false

  alias Ecto.UUID
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.GefsProfile
  alias Microwaveprop.Weather.Grid
  alias Microwaveprop.Weather.Hrrr
  alias Microwaveprop.Weather.Iemre
  alias Microwaveprop.Weather.Narr
  alias Microwaveprop.Weather.Soundings
  alias Microwaveprop.Weather.Surface

  require Logger

  # ── Delegates to sub-facades ──

  # HRRR
  defdelegate find_nearest_hrrr(lat, lon, timestamp), to: Hrrr
  defdelegate hrrr_profiles_for_path(contact), to: Hrrr
  defdelegate hrrr_profiles_for_contacts(contacts), to: Hrrr
  defdelegate find_nearest_native_profile(lat, lon, timestamp), to: Hrrr
  defdelegate best_profile_for_contact(contact), to: Hrrr
  defdelegate hrrr_for_contact(contact), to: Hrrr
  defdelegate hrrr_data_fully_present?(contact), to: Hrrr
  defdelegate has_hrrr_profile?(lat, lon, valid_time), to: Hrrr
  defdelegate hrrr_points_present_batch(points), to: Hrrr
  defdelegate round_to_hrrr_grid(lat, lon), to: Hrrr
  defdelegate purge_grid_point_profiles(), to: Hrrr
  defdelegate upsert_hrrr_profile(attrs), to: Hrrr
  defdelegate upsert_hrrr_profiles_batch(profiles, opts \\ []), to: Hrrr
  defdelegate nearest_native_duct_ghz(lat, lon, timestamp), to: Hrrr
  defdelegate nearest_native_duct_info(lat, lon, timestamp), to: Hrrr
  defdelegate backfill_hrrr_scalars(opts \\ []), to: Hrrr
  defdelegate reconcile_hrrr_statuses(), to: Hrrr
  defdelegate profiles_along_path(contact), to: Hrrr

  # NARR
  defdelegate narr_profiles_for_path(contact), to: Narr
  defdelegate find_nearest_narr(lat, lon, timestamp), to: Narr
  defdelegate narr_for_contact(contact), to: Narr

  # IEMRE
  defdelegate iemre_for_contact(contact), to: Iemre
  defdelegate iemre_for_path(contact), to: Iemre
  defdelegate find_nearest_iemre(lat, lon, timestamp), to: Iemre
  defdelegate round_to_iemre_grid(lat, lon), to: Iemre
  defdelegate upsert_iemre_observation(attrs), to: Iemre
  defdelegate has_iemre_observation?(lat, lon, date), to: Iemre
  defdelegate reconcile_iemre_statuses(), to: Iemre

  # Soundings
  defdelegate upsert_sounding(station, attrs), to: Soundings
  defdelegate has_sounding?(station_id, observed_at), to: Soundings
  defdelegate station_ids_with_soundings(station_ids, sounding_times), to: Soundings
  defdelegate sounding_times_around(dt), to: Soundings
  defdelegate weather_for_contact(contact_params, opts \\ []), to: Soundings
  defdelegate soundings_with_widening_radius(params), to: Soundings
  defdelegate nearest_sounding_to(lat, lon, timestamp, opts \\ []), to: Soundings

  # Surface
  defdelegate find_or_create_station(attrs), to: Surface
  defdelegate upsert_surface_observation(station, attrs), to: Surface
  defdelegate upsert_surface_observations(station, rows), to: Surface
  defdelegate has_surface_observations?(station_id, start_dt, end_dt), to: Surface
  defdelegate station_day_covered?(station_id, date), to: Surface
  defdelegate station_day_pairs_covered(pairs), to: Surface
  defdelegate station_ids_with_surface_observations(station_ids, start_dt, end_dt), to: Surface
  defdelegate nearby_stations(lat, lon, station_type, radius_km), to: Surface
  defdelegate sync_stations!(), to: Surface
  defdelegate reconcile_weather_statuses(), to: Surface
  defdelegate upsert_solar_index(attrs), to: Surface
  defdelegate upsert_solar_indices_batch(records), to: Surface
  defdelegate get_solar_index(date), to: Surface
  defdelegate existing_solar_dates(), to: Surface

  # Grid
  defdelegate latest_grid_valid_time(), to: Grid
  defdelegate latest_weather_grid(bounds), to: Grid
  defdelegate load_weather_grid(bounds), to: Grid
  defdelegate available_weather_valid_times(), to: Grid
  defdelegate available_hrdps_valid_times(), to: Grid
  defdelegate weather_grid_hrdps_at(valid_time, bounds), to: Grid
  defdelegate weather_grid_at(valid_time, bounds), to: Grid
  defdelegate warm_grid_cache_and_broadcast(valid_time), to: Grid
  defdelegate warm_grid_cache_from_latest_profile(), to: Grid
  defdelegate materialize_scalar_file(valid_time), to: Grid
  defdelegate build_grid_cache_rows(grid_data, valid_time, bounds \\ nil), to: Grid
  defdelegate weather_point_detail(lat, lon, valid_time), to: Grid

  # ── Remaining inline: GEFS ──

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

  # ── DB Maintenance ──

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