prop/lib/microwaveprop/pskr/client.ex

defmodule Microwaveprop.Pskr.Client do
  @moduledoc """
  MQTT client for the PSK Reporter live spot firehose.

  Connects to `mqtt.pskreporter.info:1883` (plain TCP — the broker
  doesn't expose TLS), subscribes to a curated set of topics, and
  forwards every payload to `Pskr.Aggregator.ingest/2`. Each
  topic anchors on USA (DXCC 291) on either the sender or receiver
  side so the broker pre-filters out the bulk of the firehose
  before it ever reaches us — the alternative would be subscribing
  to `pskr/filter/v2/#` and dropping ~95% client-side.

  ## Cluster singleton

  All replicas start this GenServer; **only one across the cluster
  actually connects to MQTT**. We elect a leader via
  `:global.register_name/2` — atomic across the BEAM cluster
  (libcluster gives us connectivity). The losing nodes stay in
  `:standby` mode: no MQTT connection, no DB writes, just a
  cheap idle process watching for the leader to disappear.

  When the leader's node goes down (deploy, OOM, network
  partition), `:global` evicts its registration cluster-wide and
  emits `{:nodedown, node}` to every subscriber. Standby clients
  race to re-register; one wins and starts a fresh connection
  inside `handle_continue/2`. The handover loses at most one flush
  cycle (~60 s) of in-flight aggregates from the old leader's
  Aggregator — acceptable, since the upsert-on-conflict logic
  merges any overlapping path-hours additively.

  ## Bands

  Subscribes to **VHF and up** by default (6m, 2m, 70cm, 23cm). HF
  spots are dominated by ionospheric propagation, which isn't what
  this project models — we deliberately skip HF rather than waste
  bandwidth. Microwave bands are listed too but spots there are
  vanishingly rare on PSK Reporter (FT8 activity is mostly
  144 MHz / 432 MHz / 1296 MHz).

  ## Off in dev/test

  `:microwaveprop, :pskr_mqtt_enabled` defaults to `false` in
  dev/test so `mix test` doesn't open an outbound connection. Prod
  flips it on via runtime.exs.
  """
  use GenServer

  alias Microwaveprop.Pskr.Aggregator

  require Logger

  @broker_host ~c"mqtt.pskreporter.info"
  @broker_port 1883

  # USA = 291 (continental). PSK Reporter splits Alaska (6) and
  # Hawaii (110) as separate DXCCs but neither is currently in the
  # HRRR domain; HRDPS will eventually add Canada (1).
  @us_dxcc 291

  @default_bands ~w(6m 2m 70cm 23cm 13cm 9cm 6cm 3cm 1.25cm)

  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__))
  end

  @impl true
  def init(opts) do
    Process.flag(:trap_exit, true)
    # `:net_kernel.monitor_nodes/1` lands `{:nodeup, node}` /
    # `{:nodedown, node}` messages in our mailbox — that's what
    # tells a standby client the leader vanished and election
    # should re-run.
    :ok = :net_kernel.monitor_nodes(true)

    bands = Keyword.get(opts, :bands, @default_bands)
    aggregator = Keyword.get(opts, :aggregator, Aggregator)
    client_id = Keyword.get(opts, :client_id, default_client_id())

    state = %{
      role: :standby,
      pid: nil,
      bands: bands,
      aggregator: aggregator,
      client_id: client_id,
      received: 0
    }

    {:ok, state, {:continue, :elect}}
  end

  @impl true
  def handle_continue(:elect, state) do
    case :global.register_name(__MODULE__, self()) do
      :yes ->
        Logger.info("Pskr.Client elected leader on #{node()} — connecting to MQTT")
        {:noreply, %{state | role: :active}, {:continue, :connect}}

      :no ->
        leader = :global.whereis_name(__MODULE__)
        Logger.info("Pskr.Client standby on #{node()} (leader: #{inspect(leader)})")
        {:noreply, state}
    end
  end

  def handle_continue(:connect, state) do
    case connect_and_subscribe(state) do
      {:ok, pid} ->
        Logger.info("Pskr.Client connected as #{state.client_id}, subscribed to #{length(state.bands)} bands")
        {:noreply, %{state | pid: pid}}

      {:error, reason} ->
        Logger.error("Pskr.Client connect failed: #{inspect(reason)} — retrying in 30s")
        Process.send_after(self(), :reconnect, 30_000)
        {:noreply, state}
    end
  end

  @impl true
  def handle_info(:reconnect, state), do: {:noreply, state, {:continue, :connect}}

  # When any node joins or leaves, standby clients re-attempt
  # leader election. The active leader is unaffected — its
  # registration is still valid until *its* node drops out.
  def handle_info({event, _node}, %{role: :standby} = state) when event in [:nodeup, :nodedown] do
    {:noreply, state, {:continue, :elect}}
  end

  def handle_info({event, _node}, state) when event in [:nodeup, :nodedown], do: {:noreply, state}

  # `:emqtt` delivers PUBLISH frames to the owner process as
  # `{:publish, %{topic: …, payload: …, qos: …}}`.
  def handle_info({:publish, %{payload: payload}}, state) do
    Aggregator.ingest(state.aggregator, payload)
    {:noreply, %{state | received: state.received + 1}}
  end

  def handle_info({:disconnected, reason_code, _props}, state) do
    Logger.warning("Pskr.Client disconnected: #{inspect(reason_code)} (received #{state.received} so far)")
    Process.send_after(self(), :reconnect, 5_000)
    {:noreply, %{state | pid: nil}}
  end

  def handle_info({:EXIT, pid, reason}, %{pid: pid} = state) do
    Logger.warning("Pskr.Client emqtt exited: #{inspect(reason)} — reconnecting in 5s")
    Process.send_after(self(), :reconnect, 5_000)
    {:noreply, %{state | pid: nil}}
  end

  def handle_info(_other, state), do: {:noreply, state}

  defp connect_and_subscribe(state) do
    opts = [
      host: @broker_host,
      port: @broker_port,
      clientid: state.client_id,
      clean_start: true,
      keepalive: 60,
      owner: self()
    ]

    with {:ok, pid} <- :emqtt.start_link(opts),
         {:ok, _props} <- :emqtt.connect(pid),
         :ok <- subscribe_all(pid, state.bands) do
      {:ok, pid}
    end
  end

  defp subscribe_all(pid, bands) do
    topics =
      Enum.flat_map(bands, fn band ->
        [
          {topic_for_band(band, :sender_us), 0},
          {topic_for_band(band, :receiver_us), 0}
        ]
      end)

    case :emqtt.subscribe(pid, %{}, topics) do
      {:ok, _props, _reasons} -> :ok
      other -> {:error, {:subscribe_failed, other}}
    end
  end

  # Topic order (from the broker's landing page):
  #   pskr/filter/v2/<band>/<mode>/<sc>/<rc>/<sl>/<rl>/<sa>/<ra>
  # Anchor either sender or receiver country on USA so the broker
  # delivers only the half of the firehose that touches our HRRR
  # domain.
  defp topic_for_band(band, :sender_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/#{@us_dxcc}/+"

  defp topic_for_band(band, :receiver_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/+/#{@us_dxcc}"

  defp default_client_id do
    suffix = 4 |> :crypto.strong_rand_bytes() |> Base.encode16(case: :lower)
    "microwaveprop-#{node_short()}-#{suffix}"
  end

  defp node_short do
    case System.get_env("POD_IP") do
      nil -> node() |> to_string() |> String.replace(["@", "."], "-")
      ip -> String.replace(ip, ".", "-")
    end
  end
end