prop/lib/microwaveprop/weather/surface.ex

defmodule Microwaveprop.Weather.Surface do
  @moduledoc false

  import Ecto.Query

  alias Ecto.UUID
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.IemClient
  alias Microwaveprop.Weather.SolarIndex
  alias Microwaveprop.Weather.Station
  alias Microwaveprop.Weather.SurfaceObservation

  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

  @spec has_surface_observations?(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 """
  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?(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([{UUID.t(), Date.t()}]) ::
          MapSet.t({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([UUID.t()], DateTime.t(), DateTime.t()) :: MapSet.t(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 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 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} ->
        now = DateTime.utc_now()

        entries =
          Enum.map(stations, fn s ->
            %{
              id: UUID.generate(),
              station_code: s.station_code,
              station_type: type,
              name: s.name,
              lat: s.lat,
              lon: s.lon,
              inserted_at: now,
              updated_at: now
            }
          end)

        Repo.insert_all(Station, entries,
          on_conflict: :nothing,
          conflict_target: [:station_code, :station_type]
        )

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

      {:error, e} ->
        Logger.warning("Failed to sync #{network}: #{inspect(e)}")
    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

  # ── Solar Index ──

  @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

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

  # ── Private helpers ──

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